AutoGPT Architecture

1"""
2Traditional vs Autonomous Agents
3
4Traditional (Orchestrated):
5┌─────────────────────────────────────────────┐
6│                 Orchestrator                 │
7│    (LangGraph, CrewAI, Human-defined)       │
8└──────────────────┬──────────────────────────┘
9                   │ Defines flow
10    ┌──────────────┼──────────────┐
11    ▼              ▼              ▼
12┌──────┐      ┌──────┐      ┌──────┐
13│Task 1│ ───▶ │Task 2│ ───▶ │Task 3│
14└──────┘      └──────┘      └──────┘
15
16Autonomous (Self-Directed):
17┌─────────────────────────────────────────────┐
18│                   Goal                       │
19│    "Build a market research report"         │
20└──────────────────┬──────────────────────────┘
21                   │
22                   ▼
23    ┌──────────────────────────────┐
24    │     Autonomous Agent Loop     │
25    │  ┌─────────────────────────┐ │
26    │  │   Think about goal      │ │
27    │  │         ↓               │ │
28    │  │   Decide next action    │ │
29    │  │         ↓               │ │
30    │  │   Execute action        │ │
31    │  │         ↓               │ │
32    │  │   Observe result        │ │
33    │  │         ↓               │ │
34    │  │   Update memory         │──┼──▶ Loop until done
35    │  └─────────────────────────┘ │
36    └──────────────────────────────┘
37"""
38
39# Key difference: The agent decides what to do next,
40# not a predefined workflow

1from typing import TypedDict, List, Optional
2from langchain_openai import ChatOpenAI
3from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
4import json
5
6
7class AgentState(TypedDict):
8    """State for autonomous agent."""
9    goals: List[str]
10    memory: List[dict]
11    current_task: Optional[str]
12    completed_tasks: List[str]
13    iteration: int
14    max_iterations: int
15    is_complete: bool
16
17
18class AutonomousAgent:
19    """Basic autonomous agent implementation."""
20
21    def __init__(
22        self,
23        goals: List[str],
24        max_iterations: int = 20
25    ):
26        self.llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
27        self.state: AgentState = {
28            "goals": goals,
29            "memory": [],
30            "current_task": None,
31            "completed_tasks": [],
32            "iteration": 0,
33            "max_iterations": max_iterations,
34            "is_complete": False
35        }
36        self.tools = self._setup_tools()
37
38    def _setup_tools(self) -> dict:
39        """Setup available tools."""
40        return {
41            "search": self._search,
42            "write_file": self._write_file,
43            "read_file": self._read_file,
44            "execute_code": self._execute_code,
45            "think": self._think,
46            "complete": self._mark_complete
47        }
48
49    def run(self) -> dict:
50        """Main autonomous loop."""
51        while not self.state["is_complete"]:
52            if self.state["iteration"] >= self.state["max_iterations"]:
53                print("Max iterations reached")
54                break
55
56            # Think about what to do next
57            action = self._decide_next_action()
58
59            # Execute the action
60            result = self._execute_action(action)
61
62            # Update memory with result
63            self._update_memory(action, result)
64
65            # Check if goals are achieved
66            self._evaluate_progress()
67
68            self.state["iteration"] += 1
69
70        return {
71            "completed_tasks": self.state["completed_tasks"],
72            "memory": self.state["memory"],
73            "iterations": self.state["iteration"]
74        }
75
76    def _decide_next_action(self) -> dict:
77        """Use LLM to decide next action."""
78        prompt = self._build_decision_prompt()
79
80        messages = [
81            SystemMessage(content=self._get_system_prompt()),
82            HumanMessage(content=prompt)
83        ]
84
85        response = self.llm.invoke(messages)
86        return self._parse_action(response.content)
87
88    def _get_system_prompt(self) -> str:
89        """System prompt for autonomous behavior."""
90        return """You are an autonomous AI agent working toward goals.
91
92For each step, you must output a JSON action:
93{
94    "thoughts": "Your reasoning about what to do",
95    "action": "tool_name",
96    "action_input": "input for the tool"
97}
98
99Available tools:
100- search: Search the web for information
101- write_file: Write content to a file
102- read_file: Read a file's contents
103- execute_code: Execute Python code
104- think: Internal reasoning/planning
105- complete: Mark the goal as achieved
106
107Always think step by step. Be thorough but efficient."""

1class AutonomousAgent:
2    # ... continued from above ...
3
4    def _build_decision_prompt(self) -> str:
5        """Build prompt for next action decision."""
6        recent_memory = self.state["memory"][-5:]  # Last 5 actions
7
8        return f"""
9Goals: {json.dumps(self.state['goals'])}
10
11Completed tasks: {json.dumps(self.state['completed_tasks'])}
12
13Recent actions and results:
14{json.dumps(recent_memory, indent=2)}
15
16Current iteration: {self.state['iteration']} / {self.state['max_iterations']}
17
18Based on your goals and progress, decide the next action.
19Output your response as JSON.
20"""
21
22    def _parse_action(self, response: str) -> dict:
23        """Parse LLM response into action."""
24        try:
25            # Extract JSON from response
26            start = response.find('{')
27            end = response.rfind('}') + 1
28            json_str = response[start:end]
29            return json.loads(json_str)
30        except Exception as e:
31            return {
32                "thoughts": f"Parse error: {e}",
33                "action": "think",
34                "action_input": response
35            }
36
37    def _execute_action(self, action: dict) -> str:
38        """Execute the decided action."""
39        tool_name = action.get("action", "think")
40        tool_input = action.get("action_input", "")
41
42        print(f"\n[Iteration {self.state['iteration']}]")
43        print(f"Thoughts: {action.get('thoughts', 'N/A')}")
44        print(f"Action: {tool_name}")
45        print(f"Input: {tool_input[:100]}...")
46
47        if tool_name in self.tools:
48            try:
49                result = self.tools[tool_name](tool_input)
50                print(f"Result: {result[:100]}...")
51                return result
52            except Exception as e:
53                return f"Error executing {tool_name}: {str(e)}"
54        else:
55            return f"Unknown tool: {tool_name}"
56
57    def _update_memory(self, action: dict, result: str):
58        """Add action and result to memory."""
59        memory_entry = {
60            "iteration": self.state["iteration"],
61            "thoughts": action.get("thoughts", ""),
62            "action": action.get("action", ""),
63            "action_input": action.get("action_input", ""),
64            "result": result[:500]  # Truncate long results
65        }
66        self.state["memory"].append(memory_entry)
67
68    def _evaluate_progress(self):
69        """Check if goals are achieved."""
70        if self.state["memory"]:
71            last_action = self.state["memory"][-1]
72            if last_action["action"] == "complete":
73                self.state["is_complete"] = True

1import chromadb
2from chromadb.config import Settings
3from typing import List
4import hashlib
5
6
7class AgentMemory:
8    """Memory system for autonomous agents."""
9
10    def __init__(self, agent_id: str):
11        self.agent_id = agent_id
12
13        # Short-term memory: Recent context
14        self.short_term: List[dict] = []
15        self.short_term_limit = 10
16
17        # Long-term memory: Vector store for retrieval
18        self.chroma_client = chromadb.Client(Settings(
19            anonymized_telemetry=False
20        ))
21        self.collection = self.chroma_client.get_or_create_collection(
22            name=f"agent_{agent_id}_memory"
23        )
24
25    def add_memory(self, entry: dict, long_term: bool = False):
26        """Add memory entry."""
27        # Always add to short-term
28        self.short_term.append(entry)
29        if len(self.short_term) > self.short_term_limit:
30            # Move oldest to long-term
31            old_entry = self.short_term.pop(0)
32            self._add_to_long_term(old_entry)
33
34        # Optionally add important items to long-term immediately
35        if long_term:
36            self._add_to_long_term(entry)
37
38    def _add_to_long_term(self, entry: dict):
39        """Add entry to vector store."""
40        text = f"""
41        Action: {entry.get('action', '')}
42        Input: {entry.get('action_input', '')}
43        Result: {entry.get('result', '')}
44        Thoughts: {entry.get('thoughts', '')}
45        """
46
47        entry_id = hashlib.md5(text.encode()).hexdigest()
48
49        self.collection.add(
50            documents=[text],
51            ids=[entry_id],
52            metadatas=[{"iteration": entry.get("iteration", 0)}]
53        )
54
55    def recall(self, query: str, n_results: int = 5) -> List[str]:
56        """Retrieve relevant memories."""
57        results = self.collection.query(
58            query_texts=[query],
59            n_results=n_results
60        )
61        return results["documents"][0] if results["documents"] else []
62
63    def get_recent(self, n: int = 5) -> List[dict]:
64        """Get recent short-term memories."""
65        return self.short_term[-n:]
66
67    def summarize(self) -> str:
68        """Summarize memory state."""
69        return f"""
70Memory Status:
71- Short-term entries: {len(self.short_term)}
72- Long-term entries: {self.collection.count()}
73- Recent actions: {[m.get('action') for m in self.short_term[-3:]]}
74"""

1from collections import deque
2from typing import List, Deque
3
4
5class TaskQueue:
6    """Task management for BabyAGI-style agents."""
7
8    def __init__(self):
9        self.tasks: Deque[dict] = deque()
10        self.completed: List[dict] = []
11        self.task_id = 0
12
13    def add_task(self, description: str, priority: int = 1):
14        """Add task to queue."""
15        self.task_id += 1
16        task = {
17            "id": self.task_id,
18            "description": description,
19            "priority": priority,
20            "status": "pending"
21        }
22        self.tasks.append(task)
23        self._sort_by_priority()
24
25    def _sort_by_priority(self):
26        """Sort tasks by priority."""
27        self.tasks = deque(sorted(
28            self.tasks,
29            key=lambda x: x["priority"],
30            reverse=True
31        ))
32
33    def get_next_task(self) -> dict | None:
34        """Get highest priority task."""
35        if self.tasks:
36            return self.tasks.popleft()
37        return None
38
39    def complete_task(self, task: dict, result: str):
40        """Mark task as complete."""
41        task["status"] = "completed"
42        task["result"] = result
43        self.completed.append(task)
44
45
46class BabyAGI:
47    """BabyAGI-style task-driven agent."""
48
49    def __init__(self, objective: str):
50        self.objective = objective
51        self.llm = ChatOpenAI(model="gpt-4o")
52        self.task_queue = TaskQueue()
53        self.memory = AgentMemory("babyagi")
54
55    def run(self, max_iterations: int = 10):
56        """Main BabyAGI loop."""
57
58        # Generate initial tasks
59        self._create_initial_tasks()
60
61        iteration = 0
62        while iteration < max_iterations:
63            # Get next task
64            task = self.task_queue.get_next_task()
65            if not task:
66                print("No more tasks!")
67                break
68
69            print(f"\nExecuting: {task['description']}")
70
71            # Execute task
72            result = self._execute_task(task)
73
74            # Complete task
75            self.task_queue.complete_task(task, result)
76
77            # Create new tasks based on result
78            self._create_new_tasks(task, result)
79
80            iteration += 1
81
82        return self.task_queue.completed
83
84    def _create_initial_tasks(self):
85        """Generate initial task list from objective."""
86        messages = [
87            SystemMessage(content="""Break down the objective into 3-5
88            initial tasks. Output as JSON list of task descriptions."""),
89            HumanMessage(content=f"Objective: {self.objective}")
90        ]
91
92        response = self.llm.invoke(messages)
93        tasks = json.loads(response.content)
94
95        for i, task in enumerate(tasks):
96            self.task_queue.add_task(task, priority=len(tasks) - i)
97
98    def _execute_task(self, task: dict) -> str:
99        """Execute a single task."""
100        context = self.memory.get_recent(3)
101
102        messages = [
103            SystemMessage(content=f"Objective: {self.objective}"),
104            HumanMessage(content=f"""
105Task: {task['description']}
106
107Recent context: {json.dumps(context)}
108
109Complete this task and provide the result.
110""")
111        ]
112
113        response = self.llm.invoke(messages)
114        return response.content
115
116    def _create_new_tasks(self, completed_task: dict, result: str):
117        """Generate new tasks based on completed task."""
118        messages = [
119            SystemMessage(content=f"""Objective: {self.objective}
120
121Based on the completed task and its result, determine if any new
122tasks should be added. Output as JSON list of new task descriptions,
123or empty list if no new tasks needed."""),
124            HumanMessage(content=f"""
125Completed: {completed_task['description']}
126Result: {result}
127
128Remaining tasks: {[t['description'] for t in self.task_queue.tasks]}
129""")
130        ]
131
132        response = self.llm.invoke(messages)
133        try:
134            new_tasks = json.loads(response.content)
135            for task in new_tasks:
136                self.task_queue.add_task(task)
137        except json.JSONDecodeError:
138            pass  # No new tasks