AutoGPT 与 BabyAGI：自主智能体的先驱与遗产

用户给出目标 → AI 自主分解任务 → 执行 → 评估 → 继续

# AutoGPT 的核心循环
class AutoGPT:
    def run(self, goal):
        while not self.is_complete(goal):
            # 1. 思考：基于当前状态，下一步做什么
            plan = self.think(goal, self.history)
            
            # 2. 行动：执行计划
            result = self.act(plan)
            
            # 3. 观察：记录结果
            self.history.append({"plan": plan, "result": result})
            
            # 4. 评估：是否完成
            if self.evaluate(goal, result):
                break

class AutoGPT:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4")
        self.memory = PineconeMemory()        # 长期记忆
        self.tools = [
            WebSearch(),                        # 网页搜索
            FileWriter(),                       # 文件读写
            CodeExecutor(),                     # 代码执行
            ShellExecutor(),                    # 命令行执行
        ]
        self.workspace = "./workspace/"         # 工作目录
        self.history = []                       # 对话历史
    
    async def think(self, goal, history):
        prompt = f"""
        目标：{goal}
        历史：{history[-5:]}
        可用工具：{[t.name for t in self.tools]}
        
        请决定下一步行动。输出 JSON：
        {{"thoughts": "...", "command": {{"name": "...", "args": {...}}}}}
        """
        return await self.llm.invoke(prompt)
    
    async def act(self, plan):
        tool = self.get_tool(plan.command.name)
        return await tool.execute(**plan.command.args)

auto_gpt_problems = {
    "无限循环": "Agent 卡在"搜索→思考→搜索"循环中无法跳出",
    "目标漂移": "执行几步后偏离原始目标",
    "费用爆炸": "GPT-4 每步消耗大量 Token，任务成本 $10+",
    "可靠性低": "文件操作经常失败，代码执行经常报错",
    "上下文丢失": "对话历史过长导致 LLM 遗忘早期信息",
    "缺乏验证": "Agent 自我评估不可靠，经常错误地认为任务完成",
}

auto_gpt_legacy = {
    "ReAct 模式": "思考→行动→观察循环成为 Agent 标准模式",
    "工具使用": "LLM + 工具的范式被广泛采纳",
    "长期记忆": "向量数据库作为 Agent 记忆成为标配",
    "自主性理念": "从'回答问题'到'完成任务'的范式转变",
    "Prompt工程": "系统提示词设计影响深远",
}

任务队列 → 优先级排序 → 执行 → 生成子任务 → 加入队列

class BabyAGI:
    def __init__(self):
        self.task_list = TaskList()           # 任务队列
        self.memory = VectorMemory()          # 向量记忆
        self.llm = ChatOpenAI(model="gpt-4")
    
    async def run(self, objective):
        # 初始任务
        self.task_list.add({"task": f"完成目标：{objective}"})
        
        while self.task_list.has_tasks():
            # 1. 取出最高优先级任务
            task = self.task_list.pop()
            
            # 2. 执行任务
            result = await self.execute(task)
            
            # 3. 存入记忆
            self.memory.add(task, result)
            
            # 4. 基于结果创建子任务
            new_tasks = await self.create_subtasks(objective, result)
            self.task_list.add_many(new_tasks)
            
            # 5. 重新排序任务
            self.task_list.prioritize(objective)
    
    async def create_subtasks(self, objective, result):
        prompt = f"""
        目标：{objective}
        刚完成的任务结果：{result}
        
        基于结果，列出下一步需要完成的 3 个子任务。
        如果目标已完成，返回空列表。
        """
        response = await self.llm.invoke(prompt)
        return parse_tasks(response)

class BabyAGI:
    async def execution_agent(self, task):
        """执行 Agent：完成单个任务"""
        context = self.memory.retrieve(task, top_k=5)
        return await self.llm.invoke(task, context=context)
    
    async def creation_agent(self, objective, result):
        """创建 Agent：基于结果生成子任务"""
        return await self.llm.invoke(
            f"目标：{objective}\n结果：{result}\n生成子任务"
        )
    
    async def prioritization_agent(self, task_list, objective):
        """优先级 Agent：重新排序任务"""
        return await self.llm.invoke(
            f"目标：{objective}\n任务列表：{task_list}\n重新排序"
        )

baby_agi_legacy = {
    "任务队列": "任务分解+优先级排序成为 Agent 标准组件",
    "向量记忆": "Pinecone/Milvus 作为 Agent 记忆的范式",
    "三 Agent 架构": "执行+创建+优先级的分工模式",
    "简洁设计": "不到 200 行代码实现自主 Agent",
}

# 2023 AutoGPT/BabyAGI 的设计 → 2026 年的标准实践
legacy_to_modern = {
    "ReAct 循环": "→ LangGraph 的 think-act-observe 图",
    "任务队列": "→ LangGraph 的状态图 + 条件边",
    "向量记忆": "→ RAG + 长期记忆系统",
    "工具使用": "→ MCP 协议 + Function Calling",
    "自我评估": "→ LLM-as-Judge + 输出检测",
}

lessons_learned = {
    "不要完全自主": "纯自主 Agent 可靠性低，人机协作更实际",
    "限制循环次数": "最大 10 步，防止无限循环",
    "成本控制": "每步设置 Token 上限，总成本设上限",
    "验证要外部": "Agent 自我评估不可靠，需要外部验证器",
    "上下文管理": "长任务需要摘要+记忆，不能无限累积历史",
    "目标要具体": "目标越具体，Agent 表现越好",
}

class ModernAgent:
    """2026 年的自主 Agent：吸收了 AutoGPT/BabyAGI 的教训"""
    
    def __init__(self):
        self.llm = LLM(model="qwen3-72b")
        self.memory = HybridMemory(
            working=ConversationBuffer(max_tokens=4000),
            long_term=VectorDB(top_k=5),
            episodic=SummaryBuffer(max_summaries=10),
        )
        self.tools = MCPClient()  # MCP 标准工具
        self.max_steps = 15        # 限制步数
        self.max_cost = 1.0        # 限制成本 $1
        self.human_checkpoint = True  # 人机协作
    
    async def run(self, goal):
        plan = await self.plan(goal)
        
        for step in plan.steps[:self.max_steps]:
            # 执行
            result = await self.execute(step)
            
            # 验证
            if not await self.verify(step, result):
                # 自动修正
                result = await self.retry(step, result)
            
            # 人机检查点
            if self.human_checkpoint and self.needs_approval(step):
                approved = await self.ask_human(step, result)
                if not approved:
                    break
            
            # 成本检查
            if self.total_cost > self.max_cost:
                return "成本超限，停止执行"
        
        return await self.summarize(goal)