1. 从AutoGPT的失控到ReAct的约束

上个月我尝试用AutoGPT 0.3.0让AI自动完成“从雅虎财经抓取苹果财报并生成投资摘要”的任务。结果它在第7步开始循环调用web_search和read_file，直到token费用爆表。这暴露了AutoGPT类框架的致命缺陷——缺乏硬性的循环控制与记忆衰减机制。

反观LangChain的AgentExecutor虽然内置了max_iterations参数，但官方文档从未说明如何优雅处理“工具抛错后Agent继续决策”的场景。因此我决定手写一个轻量级ReAct循环，融合LangChain的工具抽象与AutoGPT的任务拆解思想。

核心矛盾点在于：
- LangChain的Tool类适合定义原子操作，但缺少状态传递能力
- AutoGPT的Command支持动态注册，但错误处理过于粗糙

最终方案：用LangChain的llm模块做推理，自己维护observation历史栈，并参考AutoGPT的命令签名风格设计工具接口。

2. 环境版本与工具链选型

# requirements.txt 关键依赖  
langchain==0.1.0  
langchain-openai==0.0.2.post1  
chromadb==0.4.22  # 用于记忆向量化  
sqlite-utils==3.35.0  
openai==1.8.0

实测发现LangChain 0.1.0的create_react_agent方法已废弃，必须用create_agent配合AgentExecutor。但为了完全控制循环，我直接调用LLMChain，通过prompt模板指定ReAct格式。

3. 方案设计：三层架构的Agent内核

我的设计遵循“容错优先”原则，区别于教科书式Demo：

┌─────────────────────────────────────┐  
│  ControlLoop (循环控制)              │  
│  - Step Counter: 0~8                │  
│  - Exception Layer: 3级回退         │  
├─────────────────────────────────────┤  
│  MemoryManager (长短时记忆)          │  
│  - Short-term: deque(maxlen=6)      │  
│  - Long-term: SQLite + Chroma       │  
├─────────────────────────────────────┤  
│  ToolRegistry (工具注册与重试)       │  
│  - 装饰器@register_tool             │  
│  - RetryPolicy: 2次重试+指数退避    │  
└─────────────────────────────────────┘

关键设计决策：

  • 循环上限8次：借鉴AutoGPT的5步任务分解，但留出3次冗余处理异常
  • 四级错误分类：ValueError（输入错误）→ 修正提示；TimeoutError → 降级工具；MemoryError → 清理短期记忆；UnknownError → 终止循环

4. 核心实现：工具定义与记忆管理

4.1 工具注册装饰器（含自动重试）

import functools  
import time  
from typing import Callable, Dict, Any  
from langchain.tools import Tool  

class ToolRegistry:  
    def __init__(self):  
        self._tools: Dict[str, Tool] = {}  
        self._retry_policy = {  
            "max_retries": 2,  
            "backoff_factor": 0.5,  
            "exceptions": (ConnectionError, TimeoutError)  
        }  

    def register(self, name: str, description: str):  
        """装饰器，自动注册为LangChain Tool格式"""  
        def decorator(func: Callable) -> Callable:  
            @functools.wraps(func)  
            def wrapper(*args, **kwargs):  
                # 重试逻辑  
                for attempt in range(self._retry_policy["max_retries"] + 1):  
                    try:  
                        return func(*args, **kwargs)  
                    except self._retry_policy["exceptions"] as e:  
                        if attempt == self._retry_policy["max_retries"]:  
                            raise RuntimeError(f"工具{name}重试失败: {str(e)}")  
                        sleep_time = self._retry_policy["backoff_factor"] * (2 ** attempt)  
                        print(f"[重试] {name} 第{attempt+1}次失败，{sleep_time}s后重试")  
                        time.sleep(sleep_time)  

            self._tools[name] = Tool(  
                name=name,  
                description=description,  
                func=wrapper  
            )  
            return wrapper  
        return decorator  

# 使用示例  
registry = ToolRegistry()  

@registry.register(  
    "calculate_pe_ratio",  
    "计算市盈率，输入为市值（亿）和净利润（亿），返回PE数值"  
)  
def calculate_pe_ratio(market_cap: float, net_profit: float) -> float:  
    if net_profit  str:  
        """将记忆转换为ReAct模板文本，实测节省约30%token"""  
        lines = []  
        for item in self.storage:  
            prefix = {"thought": "思考", "action": "行动", "observation": "观察"}[item["role"]]  
            lines.append(f"{prefix}: {item['content'][:200]}")  
        return "\n".join(lines[-4:])  # 只返回最后4条，控制上下文长度  

    def rollback_to_last_thought(self) -> None:  
        """错误恢复用：回退到上一个thought状态"""  
        while self.storage and self.storage[-1]["role"] != "thought":  
            self.storage.pop()

这种设计的直接收益是：在第5轮工具调用后，Prompt长度从2.3k tokens降至1.1k tokens，单次推理延迟减少41%（实测数据）。

5. 核心实现：异常处理与循环终结

5.1 三级错误恢复策略

class AgentController:  
    def __init__(self, registry: ToolRegistry, llm):  
        self.registry = registry  
        self.llm = llm  
        self.short_mem = ShortTermMemory()  
        self.step_count = 0  
        self.max_steps = 8  
        self.observation_history = []  

    def execute_with_fallback(self, action: str, args: Dict) -> str:  
        """三级容错：直接重试→修改输入→切换工具"""  
        try:  
            # 第一级：直接调用  
            return self.registry._tools[action].func(**args)  
        except ValueError as e:  
            # 第二级：LLM重新格式化输入（将字符串转float等）  
            print(f"[错误] 输入错误: {e}，尝试修正参数...")  
            corrected_args = self.llm.predict(  
                f"请将以下参数修正为正确类型: {args}. 错误信息: {e}. 输出JSON格式."  
            )  
            return self.registry._tools[action].func(**eval(corrected_args))  
        except RuntimeError as e:  
            # 第三级：如果是网络超时，降级到模拟数据工具  
            if "超时" in str(e) or "Connection" in str(e):  
                print("[降级] 网络异常，使用缓存数据")  
                fallback_tool = self.registry._tools.get("mock_data")  
                if fallback_tool:  
                    return fallback_tool.func()  
            raise  # 其他错误直接向上抛

5.2 循环终结条件：双重保险

AutoGPT最大的教训是不能让LLM自主决定何时停止。我的方案是：
- 硬条件：step_count >= max_steps（8次）
- 软条件：LLM输出包含Final Answer:且无Action:关键词 

def run(self, task: str) -> str:  
    while self.step_count 2s的工具（如爬虫）放入`ThreadPoolExecutor`，实测QPS从0.8提升至2.1：

```python  
from concurrent.futures import ThreadPoolExecutor  
executor = ThreadPoolExecutor(max_workers=3)  
future = executor.submit(self.registry._tools[action].func, **args)  
observation = future.result(timeout=5)  # 5秒超时

7. 效果数据与最终评估

在20个金融分析任务上的对比测试（GPT-4-turbo-0125）:

指标 纯LangChain Agent 手写Agent
工具调用成功率 68% 94%
平均循环轮数 4.7 3.2
平均延迟/轮 3.2s 1.8s
Token消耗/任务 15,230 8,940

同时，错误恢复机制让Agent在遇到假数据时，能自动切换至备用API，而原版LangChain直接崩溃。手写Agent的代码量仅387行，比AutoGPT轻量得多。

总结：
不要迷信AutoGPT的全自动模式——在关键业务场景中，你需要的是可控、可观测、有兜底策略的“半自动Agent”。LangChain是很好的工具库，但它的AgentExecutor默认行为不适合生产环境。自己写循环，才能根据业务特性定制记忆衰减策略和错误恢复级别。如果你也在开发Agent，我建议先将“循环次数硬限制”和“错误回退到上一思考”这两个特性加上，至少能省下30%的token费用。