1. 问题背景:为什么我要扔掉AgentExecutor

先说结论:不是AgentExecutor不好,而是它对“可控性”的牺牲让我无法接受。我用LangChain 0.2.1写了一个代码审查Agent,核心需求是让LLM(GPT-4o-mini)调用三个工具:read_filelint_codesuggest_fix。官方实现里,Agent循环由AgentExecutor管理,它内部有个max_iterations参数(默认15),但实际跑起来发现两个痛点:

  • 记忆无边界:每轮对话的IntermediateStep都会追加到memory.chat_memory.messages里。跑6轮后,prompt token数从2.1k涨到9.8k,响应时间从1.2秒涨到4.5秒,最后直接触发了上下文窗口溢出。
  • 错误处理是黑盒:当lint_code因为文件不存在抛异常时,AgentExecutor只会把错误字符串塞回给LLM,LLM经常误解为“继续重试”,导致死循环。

所以我的方案是:手写Agent循环,用显式的状态机控制每一步

2. 环境与版本

  • Python 3.10.12
  • langchain 0.2.1(注意:0.2.x的create_react_agent接口变了)
  • langchain-openai 0.1.3
  • pydantic 2.6.4
  • 模型:gpt-4o-mini-2024-07-18,temperature=0.2

这里提醒一下,LangChain 0.2.x中BaseTool必须用pydantic字段声明参数,否则会报TypeError: Can't instantiate abstract class

3. 方案设计:三个核心抽象

我的Agent循环由三个组件构成:

  1. ToolRegistry(工具注册表):一个字典,键是工具名,值是一个ToolSpec对象,包含funcdescriptionparameters(Pydantic模型)。
  2. MemoryManager(记忆管理):用collections.deque(maxlen=6)存储最近6轮(action, observation)对,每轮塞入prompt前做token数估算,超过1800 token时按比例丢弃最旧的20%内容。
  3. AgentLoop(主循环):一个while循环,包含step()方法执行一次“思考-行动-观察”周期,并用max_steps=8max_retries=2两个阈值控制退出。

整体流程如下(简化伪代码):

class SimpleAgent:
    def __init__(self, tools, llm, max_steps=8, max_retries=2):
        self.tools = {t.name: t for t in tools}
        self.llm = llm
        self.max_steps = max_steps
        self.max_retries = max_retries
        self.memory = deque(maxlen=6)
        self.step_count = 0
        self.retry_count = 0

    def run(self, task):
        messages = [{"role": "system", "content": SYSTEM_PROMPT}]
        messages.append({"role": "user", "content": task})
        while self.step_count  str:
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                lines = f.readlines()
            if end_line is None:
                end_line = len(lines)
            selected = lines[start_line:end_line]
            return "".join(selected)
        except FileNotFoundError:
            return f"[ERROR] 文件 {file_path} 不存在,请检查路径"
        except Exception as e:
            return f"[ERROR] {type(e).__name__}: {str(e)}"

关键点:_run方法必须返回字符串,且错误信息用[ERROR]前缀,这样LLM能明确知道这是工具执行失败,而不是正常输出。

4.2 循环控制与错误重试

我采用“JSON Action”格式,让LLM输出如下JSON:

{"action": "read_file", "action_input": {"file_path": "/tmp/test.py"}}

然后我用json.loads解析,如果解析失败,就重新生成一次prompt,并附加上一次的错误信息。重试逻辑如下:

def step(self, messages):
    for attempt in range(self.max_retries + 1):
        response = self.llm.invoke(messages)
        action_text = response.content
        try:
            action = self._parse_action(action_text)  # 自定义解析器
            break
        except Exception as e:
            if attempt == self.max_retries:
                return {"finish": True, "output": f"解析失败: {action_text}"}
            # 把错误回填给LLM
            messages.append({"role": "user", "content": f"你的输出无法解析为JSON: {e},请重新输出。"})

    # 执行工具
    tool = self.tools.get(action["name"])
    if tool is None:
        observation = f"[ERROR] 未知工具: {action['name']}"
    else:
        try:
            observation = tool.run(action["arguments"])
        except Exception as e:
            observation = f"[ERROR] 工具执行异常: {e}"

    # 记忆更新
    self.memory.append((action_text, observation))
    self.step_count += 1
    return {"action_text": action_text, "observation": observation}

这里有个坑:tool.run()内部会调用_run,但如果_run内部抛异常,LangChain会封装成ToolException。所以我在外层再包一层try-except,确保observation永远是字符串。

5. 记忆管理:滑动窗口+token裁剪

我观察到,在代码审查场景中,最近的3轮对话信息量最大,再往前的通常是“确认文件内容”之类的噪声。所以deque(maxlen=6)足够了。但为了控制token数,我加了一个_pack_memory方法:

def _pack_memory(self, current_messages):
    memory_blocks = []
    total_tokens = 0
    for action, obs in reversed(self.memory):
        block = f"Action: {action}\nObservation: {obs[:300]}"  # 截断超过300字符
        block_tokens = len(block) * 2  # 粗略估算,一个字符约2 token
        if total_tokens + block_tokens > 1800:
            break
        memory_blocks.append(block)
        total_tokens += block_tokens
    return "\n\n".join(reversed(memory_blocks))

注意第5行的obs[:300]——如果工具返回一个5000行的文件,我只截取前300字符作为记忆。这个截断策略让prompt大小稳定在2.5k token以内。

6. 踩坑与优化:三个真实问题

坑1:LLM把工具名拼错了
GPT-4o-mini偶尔会把read_file拼成readfile,导致工具找不到。我的解法是在SystemPrompt里显式列出工具名,并加一句“如果工具不存在,请使用list_tools查看可用工具。”当然,我没实现list_tools,但这句话让LLM的输出规范了很多。

坑2:AutoGPT风格的“子任务”陷阱
一开始我参考AutoGPT的思路,让Agent自己拆解任务、写plan。但实践发现,对于代码审查这种短任务,LLM会输出“第一步:读取文件,第二步:执行lint……”这种废话plan,浪费了两轮循环。后来我直接删掉了plan步骤,改为“直接给出第一个action”。

坑3:重复调用同一个工具
如果lint_code返回了10个错误,LLM可能会连续调用suggest_fix 10次,每次只处理一个错误。我加了一个“工具调用频率限制”:同一工具在连续3轮内最多调用2次,否则强制结束循环并返回当前进度。

7. 效果数据与总结

在12个真实代码文件(每个200-800行)的测试集上:

  • 手写Agent的任务完成率(指正确识别所有lint错误并给出修复建议)为91%,官方AgentExecutor为62%。
  • 平均循环轮数:手写Agent为4.3轮,官方为7.8轮(因为多了plan和重试)。
  • 峰值内存:手写Agent为320MB,官方为1.2GB,主要差异在于记忆管理策略。
  • 单轮响应时间:手写Agent为1.8秒(含LLM调用与工具执行),官方为2.7秒。

总结下来,手写Agent的核心收益是可控性:我能精确控制记忆窗口、错误重试次数、工具调用频率,而这些在LangChain的AgentExecutor里需要大量配置项甚至自定义回调才能实现。

如果你也在开发类似的中小型Agent(非复杂多智能体协作),我建议直接手写循环,成本不高(300行代码),但收益很大。另外,推荐使用langchain_core.tools.BaseTool而非@tool装饰器,因为前者能更好地集成Pydantic校验。

最后,不要迷信AutoGPT那种“自动拆解任务”的模式——对于目标明确的工具调用型Agent,显式状态机 + 严格JSON协议是最稳的路径。