Appearance
第40天：AutoGPT与BabyAGI
学习目标
- 理解AutoGPT的架构和原理
- 掌握BabyAGI的设计思想
- 学习自主Agent系统的实现
- 掌握任务循环机制
- 了解实际应用案例
AutoGPT
AutoGPT简介
AutoGPT是一个实验性的开源应用程序，展示了GPT-4语言模型的功能。它能够：
- 自主决策：根据目标自主决定下一步行动
- 任务分解：将复杂目标分解为子任务
- 工具使用：使用各种工具完成任务
- 持续学习：从经验中学习和改进
AutoGPT架构
┌─────────────────────────────────────────┐
│ AutoGPT System │
├─────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Goal │───▶ User Input │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Planner │───▶ Task List │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Executor │───▶ Actions │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Memory │───▶ Experience │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Evaluator │───▶ Progress │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────┘AutoGPT核心组件
1. Goal Setting（目标设定）
python
class Goal:
def __init__(self, description: str, criteria: List[str]):
self.description = description
self.criteria = criteria
def is_achieved(self, result: Dict) -> bool:
for criterion in self.criteria:
if not self._check_criterion(criterion, result):
return False
return True
def _check_criterion(self, criterion: str, result: Dict) -> bool:
return criterion.lower() in str(result).lower()2. Task Planning（任务规划）
python
class TaskPlanner:
def __init__(self, llm):
self.llm = llm
def create_plan(self, goal: Goal) -> List[Task]:
prompt = f"""
Goal: {goal.description}
Break down this goal into specific, actionable tasks.
Each task should be:
1. Specific and clear
2. Achievable
3. Measurable
Return a list of tasks.
"""
response = self.llm.invoke(prompt)
return self._parse_tasks(response)
def _parse_tasks(self, response: str) -> List[Task]:
tasks = []
lines = response.split('\n')
for line in lines:
if line.strip():
task = Task(description=line.strip())
tasks.append(task)
return tasks3. Task Execution（任务执行）
python
class TaskExecutor:
def __init__(self, tools: List[Tool]):
self.tools = tools
def execute(self, task: Task) -> Dict:
for tool in self.tools:
if tool.can_handle(task):
return tool.execute(task)
return {
"status": "failed",
"error": "No suitable tool found"
}4. Memory Management（记忆管理）
python
class AutoGPTMemory:
def __init__(self):
self.short_term = []
self.long_term = []
self.reflections = []
def add_experience(self, experience: Dict):
self.short_term.append(experience)
if len(self.short_term) > 10:
self._compress_memory()
def _compress_memory(self):
summary = self._summarize_short_term()
self.long_term.append(summary)
self.short_term = []
def _summarize_short_term(self) -> Dict:
return {
"experiences": len(self.short_term),
"summary": " ".join([
exp.get("description", "")
for exp in self.short_term
])
}
def search(self, query: str) -> List[Dict]:
results = []
for memory in self.short_term + self.long_term:
if self._matches_query(memory, query):
results.append(memory)
return results
def _matches_query(self, memory: Dict, query: str) -> bool:
return query.lower() in str(memory).lower()AutoGPT实现示例
python
class AutoGPT:
def __init__(self, llm, tools: List[Tool]):
self.llm = llm
self.planner = TaskPlanner(llm)
self.executor = TaskExecutor(tools)
self.memory = AutoGPTMemory()
self.max_iterations = 10
def run(self, goal: Goal) -> Dict:
print(f"Starting AutoGPT with goal: {goal.description}")
plan = self.planner.create_plan(goal)
print(f"Created plan with {len(plan)} tasks")
results = []
iteration = 0
while plan and iteration < self.max_iterations:
iteration += 1
print(f"\nIteration {iteration}")
current_task = plan.pop(0)
print(f"Executing task: {current_task.description}")
result = self.executor.execute(current_task)
self.memory.add_experience({
"task": current_task.description,
"result": result
})
results.append(result)
if self._should_refine_plan(goal, results):
print("Refining plan...")
plan = self._refine_plan(goal, results)
if goal.is_achieved(result):
print("Goal achieved!")
break
return {
"goal": goal.description,
"iterations": iteration,
"results": results,
"status": "completed" if goal.is_achieved(result) else "incomplete"
}
def _should_refine_plan(self, goal: Goal, results: List[Dict]) -> bool:
return len(results) > 0 and len(results) % 3 == 0
def _refine_plan(self, goal: Goal, results: List[Dict]) -> List[Task]:
prompt = f"""
Goal: {goal.description}
Previous results:
{self._format_results(results)}
Based on the results, refine the remaining tasks.
"""
response = self.llm.invoke(prompt)
return self.planner._parse_tasks(response)
def _format_results(self, results: List[Dict]) -> str:
return "\n".join([
f"- {result.get('description', 'Unknown')}: {result.get('status', 'Unknown')}"
for result in results
])BabyAGI
BabyAGI简介
BabyAGI是一个更简单但功能强大的自主Agent系统，专注于任务管理和执行循环。
核心特点：
- 简洁的架构
- 高效的任务循环
- 灵活的工具集成
- 持续的任务优化
BabyAGI架构
┌─────────────────────────────────────────┐
│ BabyAGI System │
├─────────────────────────────────────────┤
│ │
│ ┌─────────────┐ │
│ │ Objective │───▶ Initial Goal │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Task List │───▶ Tasks │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Priority │───▶ Next Task │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Execution │───▶ Result │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Storage │───▶ Memory │
│ └─────────────┘ │
│ │ │
│ ▼ │
│ ┌─────────────┐ │
│ │ Creation │───▶ New Tasks │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────┘BabyAGI核心组件
1. Task Management（任务管理）
python
class TaskManager:
def __init__(self):
self.tasks = []
self.completed_tasks = []
def add_task(self, task: Task, priority: int = 0):
task.priority = priority
self.tasks.append(task)
self._prioritize_tasks()
def _prioritize_tasks(self):
self.tasks.sort(key=lambda t: t.priority, reverse=True)
def get_next_task(self) -> Optional[Task]:
if self.tasks:
return self.tasks.pop(0)
return None
def complete_task(self, task: Task, result: Dict):
task.result = result
task.completed_at = datetime.now()
self.completed_tasks.append(task)
def get_task_summary(self) -> str:
return "\n".join([
f"- {task.description} (Priority: {task.priority})"
for task in self.tasks
])2. Task Creation（任务创建）
python
class TaskCreator:
def __init__(self, llm):
self.llm = llm
def create_new_tasks(
self,
objective: str,
result: Dict,
task_description: str
) -> List[Task]:
prompt = f"""
Objective: {objective}
Current task: {task_description}
Task result: {result}
Based on the result, create new tasks to further the objective.
Each task should be:
1. Specific and actionable
2. Aligned with the objective
3. Not redundant with existing tasks
Return a list of tasks.
"""
response = self.llm.invoke(prompt)
return self._parse_tasks(response)
def _parse_tasks(self, response: str) -> List[Task]:
tasks = []
lines = response.split('\n')
for line in lines:
if line.strip():
task = Task(description=line.strip())
tasks.append(task)
return tasks3. Task Prioritization（任务优先级）
python
class TaskPrioritizer:
def __init__(self, llm):
self.llm = llm
def prioritize_tasks(
self,
tasks: List[Task],
objective: str
) -> List[Task]:
prompt = f"""
Objective: {objective}
Tasks:
{self._format_tasks(tasks)}
Prioritize these tasks based on:
1. Importance to the objective
2. Dependencies
3. Feasibility
Return the tasks in order of priority.
"""
response = self.llm.invoke(prompt)
prioritized_tasks = self._parse_tasks(response)
for i, task in enumerate(prioritized_tasks):
task.priority = len(tasks) - i
return prioritized_tasks
def _format_tasks(self, tasks: List[Task]) -> str:
return "\n".join([
f"{i+1}. {task.description}"
for i, task in enumerate(tasks)
])BabyAGI实现示例
python
class BabyAGI:
def __init__(self, llm, tools: List[Tool]):
self.llm = llm
self.task_manager = TaskManager()
self.task_creator = TaskCreator(llm)
self.task_prioritizer = TaskPrioritizer(llm)
self.executor = TaskExecutor(tools)
self.storage = TaskStorage()
self.max_iterations = 20
def run(self, objective: str) -> Dict:
print(f"Starting BabyAGI with objective: {objective}")
initial_task = Task(description=objective)
self.task_manager.add_task(initial_task, priority=10)
iteration = 0
while iteration < self.max_iterations:
iteration += 1
print(f"\nIteration {iteration}")
task = self.task_manager.get_next_task()
if not task:
print("No more tasks to execute")
break
print(f"Executing task: {task.description}")
result = self.executor.execute(task)
self.storage.store(task, result)
new_tasks = self.task_creator.create_new_tasks(
objective,
result,
task.description
)
for new_task in new_tasks:
self.task_manager.add_task(new_task)
self.task_manager.tasks = self.task_prioritizer.prioritize_tasks(
self.task_manager.tasks,
objective
)
self.task_manager.complete_task(task, result)
if self._is_objective_complete(objective, result):
print("Objective completed!")
break
return {
"objective": objective,
"iterations": iteration,
"completed_tasks": len(self.task_manager.completed_tasks),
"status": "completed"
}
def _is_objective_complete(self, objective: str, result: Dict) -> bool:
return objective.lower() in str(result).lower()任务循环机制
主循环
python
class TaskLoop:
def __init__(self, agent):
self.agent = agent
self.running = False
def start(self, objective: str):
self.running = True
iteration = 0
while self.running:
iteration += 1
try:
result = self.agent.execute_iteration(objective)
if self._should_stop(result):
self.running = False
break
if iteration >= self.agent.max_iterations:
print("Max iterations reached")
self.running = False
break
except Exception as e:
print(f"Error in iteration {iteration}: {e}")
self._handle_error(e)
return result
def _should_stop(self, result: Dict) -> bool:
return result.get("status") == "completed"
def _handle_error(self, error: Exception):
pass实际应用案例
案例1：研究助手
python
class ResearchAssistant(AutoGPT):
def __init__(self, llm):
tools = [
SearchTool(),
SummarizeTool(),
SaveTool()
]
super().__init__(llm, tools)
def research_topic(self, topic: str) -> Dict:
goal = Goal(
description=f"Research and summarize {topic}",
criteria=["summary", "key points", "references"]
)
return self.run(goal)案例2：内容创作
python
class ContentCreator(BabyAGI):
def __init__(self, llm):
tools = [
WriteTool(),
EditTool(),
PublishTool()
]
super().__init__(llm, tools)
def create_article(self, topic: str) -> Dict:
objective = f"Create a comprehensive article about {topic}"
return self.run(objective)实践练习
练习1：实现简单的AutoGPT
python
class SimpleAutoGPT:
def __init__(self, llm):
self.llm = llm
self.planner = TaskPlanner(llm)
self.executor = TaskExecutor([])
def run(self, goal: str):
goal_obj = Goal(description=goal, criteria=[goal])
plan = self.planner.create_plan(goal_obj)
for task in plan:
result = self.executor.execute(task)
print(f"Task: {task.description}")
print(f"Result: {result}")
return {"status": "completed"}练习2：实现简单的BabyAGI
python
class SimpleBabyAGI:
def __init__(self, llm):
self.llm = llm
self.task_manager = TaskManager()
self.task_creator = TaskCreator(llm)
def run(self, objective: str):
initial_task = Task(description=objective)
self.task_manager.add_task(initial_task)
for _ in range(5):
task = self.task_manager.get_next_task()
if not task:
break
result = {"status": "completed", "output": "Done"}
new_tasks = self.task_creator.create_new_tasks(
objective,
result,
task.description
)
for new_task in new_tasks:
self.task_manager.add_task(new_task)
return {"status": "completed"}总结
本节我们学习了AutoGPT和BabyAGI：
- AutoGPT的架构和核心组件
- AutoGPT的实现示例
- BabyAGI的设计思想和核心组件
- BabyAGI的实现示例
- 任务循环机制
- 实际应用案例
这些自主Agent系统展示了AI Agent的强大能力，为构建更复杂的Agent应用提供了参考。