一、问题背景：为什么AutoGPT式的“万能Agent”在真实场景跑不通

上个月我在做一个内部数据查询助手，最初直接套用了AutoGPT那种“让LLM自己规划+执行+反思”的循环框架。结果非常惨烈：模型在“查询用户表”和“统计订单量”两个工具之间反复横跳，直到把上下文窗口塞满报错；更离谱的是，有一次它为了“确认结果正确性”，竟然连续调用了12次同一个查询工具，产生了近2万Token的冗余消耗。

事后复盘，AutoGPT那种完全开放的循环控制（while True + 模型自评）在沙盒Demo里很酷，但生产环境必须有硬性约束。LangChain提供了AgentExecutor这类封装，但它的默认循环控制（max_iterations=15）和记忆管理（ConversationBufferMemory无限增长）同样不理想。我决定拆开LangChain的轮子，手写一个满足以下需求的Agent核心：

  • 工具注册表：支持动态增删，每个工具带参数JSON Schema
  • 记忆管理：只保留最近K轮对话+工具结果摘要，超限自动压缩
  • 错误处理：单次工具调用失败自动重试1次，连续失败则降级为LLM直接回答
  • 循环控制：硬性最大步数（默认5步），每步结束强制校验“是否产生了新信息”

二、环境与版本：2024年2月相对稳定的组合

python==3.10.13
langchain==0.1.0
openai==1.10.0
tiktoken==0.5.1

注意：LangChain 0.1.x版本中，langchain.agents模块被重构过，initialize_agent和AgentExecutor的API与0.0.x完全不同。我踩坑后发现，直接使用langchain_core.messages和自定义BaseTool更稳妥。

三、方案设计：一个“有限自主”的Agent骨架

核心思路：不要信任模型的自我约束，要用代码强制边界。

整体架构分为四层：
1. 工具层：继承LangChain的BaseTool，声明工具名、描述、参数格式
2. 记忆层：自定义TokenMemory类，基于tiktoken计算Token数，超过阈值时丢弃最旧消息并插入摘要
3. 控制层：主循环run()，负责调用LLM获取下一步动作（Action）或最终答案（Final Answer）
4. 容错层：捕获所有异常，区分“工具不存在”“参数校验失败”“网络超时”等类型

关键设计决策：
- 循环控制不用while True，而是for step in range(max_steps)，配合条件检查提前终止
- 每步结束后，将当前(observation, thought)追加到记忆，同时检查重复动作检测——如果连续两步调用同一工具且参数相同，直接终止并报错

四、核心实现：手写Agent循环与工具定义

4.1 工具定义：继承BaseTool而非用装饰器

我推荐用类继承方式而非@tool装饰器，因为这样能更清晰控制_run方法的异常抛出。

from langchain_core.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Optional, Type
import requests

class WeatherInput(BaseModel):
    city: str = Field(description="城市名称，中文，如'北京'")
    days: int = Field(default=1, description="未来天数，1-3")

class WeatherTool(BaseTool):
    name = "weather_query"
    description = "查询指定城市未来天气，返回温度与降水概率"
    args_schema: Type[BaseModel] = WeatherInput

    def _run(self, city: str, days: int = 1) -> str:
        # 模拟真实API调用，生产环境替换为requests.get
        if not city or len(city)  {observation}"))
                # 关键：把observation作为新的输入给模型继续推理
                self.memory.add_message(HumanMessage(content=f"工具返回结果: {observation}，请继续决策"))
            else:
                return f"模型输出缺少action或answer字段：{parsed}"

        return "达到最大步数限制，未得到最终答案，请简化问题。"

这段代码的核心思想：每一步的observation都被当作新的用户输入塞回对话，这比LangChain默认的AgentExecutor更透明——你可以在每一步print出记忆内容来调试。

五、记忆管理：TokenMemory类的实现细节

AutoGPT失败的一个关键原因是记忆无限膨胀。我实现的TokenMemory基于tiktoken精确计算Token，当超过阈值时，不是简单丢弃最旧消息，而是保留系统消息和最近一轮对话，将中间内容压缩为一行摘要。

```python
import tiktoken
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

class TokenMemory:
def init(self, token_limit=2000):
self.enc = tiktoken.encoding_for_model("gpt-4")
self.token_limit = token_limit
self.messages = []

def add_message(self, message):
    self.messages.append(message)
    self._trim_if_needed()

def get_messages(self):
    return self.messages

def _trim_if_needed(self):
    # 计算总Token
    total_tokens = sum(len(self.enc.encode(m.content)) for m in self.messages)
    if total_tokens 0.8，直接终止。

七、效果数据与总结

在内部数据集（20个查询任务，包含多跳工具调用）上对比：
- 原生LangChain AgentExecutor（max_iterations=15）：工具选择准确率72%，平均Token消耗6200/任务
- 本文手写Agent（max_steps=5 + TokenMemory 2000）：工具选择准确率91%，平均Token消耗3850/任务，且未出现一次死循环

核心结论：Agent不是越“自主”越好，而是需要可量化的边界。手写循环控制虽然代码量多200行左右，但你能精确掌握每一步的输入输出，这对于调试和成本控制是决定性优势。

最后提醒：以上代码基于LangChain 0.1.0，如果你用0.2+版本，langchain_core的API有细微变化（如BaseTool的_run签名变了）。建议锁版本生产，不要追新。如果读者用AutoGPT那种完全开放式框架，建议至少加上max_steps=5和duplicate_action_check=True两个参数。