一、从AutoGPT的失控到LangChain的僵化

上个月我接手了一个内部代码审查助手项目，需求是让Agent能自主完成“读取PR diff→定位相关测试→执行测试→修改失败用例→重新运行”的完整链路。最初尝试直接调用AutoGPT的cli，结果在第3轮循环时就出现了灾难性的上下文膨胀——AutoGPT的thoughts和reasoning字段会无条件追加到chat_history，导致单次请求Token消耗从初始的12K暴涨到89K，最终在GPT-4-turbo的128K窗口边缘反复触发截断。

随后切换到LangChain的AgentExecutor，虽然解决了循环控制问题，但新的痛点出现了：当工具调用链深度超过3层时（比如“读取文件→搜索函数定义→读取依赖→修改代码→执行测试”），其内部的ToolRuntimeError处理逻辑会简单粗暴地将异常信息全部塞给LLM，导致模型在后续步骤中频繁重复尝试同一个失败的调用。我在10个任务样本上统计，AgentExecutor的最终成功率只有78%，且平均每个任务多出4.2次无效工具调用。

于是决定自己写一个Agent循环——保留LangChain的工具抽象和模型封装，但把循环控制、记忆管理、错误恢复全部握在手里。

二、环境与版本锁定

这是我调试了三天才稳定的组合：

Python 3.11.7
langchain 0.1.10
langchain-openai 0.0.5
openai 1.13.3
tiktoken 0.6.0

注意：langchain 0.1.x的BaseTool接口与0.2.x完全不同，如果你用最新版，下面的run()方法签名需要改为_run()。另外，OpenAI的FunctionCalling模式必须使用gpt-4-turbo-preview或gpt-3.5-turbo-1106，旧版gpt-4不支持tools参数。

三、方案设计：四个核心模块的职责边界

整个Agent循环拆为四个独立组件，每个组件只做一件事：

  1. ToolManager：维护工具注册表，负责将Python函数转换为OpenAI tools Schema。每个工具必须声明name、description、parameters，其中description要精确到“什么场景下用，什么参数会报错”；
  2. MemoryController：维护一个滑动窗口的对话历史，当Token总数超过阈值时，用tiktoken计算每条消息的Token数，优先压缩工具结果（保留首尾各200字符，中间用...代替）；
  3. ErrorHandler：捕获工具执行异常，根据异常类型决定是重试（网络超时）、修正参数（JSON解析错误）还是终止（业务逻辑错误）；
  4. LoopController：负责判断何时停止循环——达到最大迭代次数（我设置为6）、LLM返回finish动作、或连续两次产生相同动作（防止死循环）。

四、核心实现：手写循环的骨架代码

下面是我裁剪后的核心循环逻辑，去掉了与业务相关的细节，保留最关键的框架：

from langchain_openai import ChatOpenAI
from langchain.tools import BaseTool
from typing import Dict, Any, List, Optional
import json, tiktoken

class SimpleAgent:
    def __init__(self, tools: List[BaseTool], model_name: str = "gpt-4-turbo-preview", max_iterations: int = 6, max_tokens: int = 30000):
        self.llm = ChatOpenAI(model=model_name, temperature=0.2, max_tokens=2048)
        self.tools = {tool.name: tool for tool in tools}
        self.tool_schemas = [self._convert_to_schema(tool) for tool in tools]
        self.encoder = tiktoken.encoding_for_model("gpt-4")
        self.max_iterations = max_iterations
        self.max_tokens = max_tokens
        self.memory: List[Dict[str, str]] = []

    def _convert_to_schema(self, tool: BaseTool) -> Dict[str, Any]:
        """将LangChain工具转为OpenAI function calling格式"""
        schema = {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description,
                "parameters": tool.args_schema.schema() if hasattr(tool, "args_schema") else {"type": "object", "properties": {}}
            }
        }
        return schema

    def _compress_memory(self):
        """当Token超限时，压缩工具结果消息"""
        total = sum(len(self.encoder.encode(m["content"])) for m in self.memory)
        if total  400:
                    self.memory[i]["content"] = original[:200] + "...[压缩]..." + original[-200:]
                    total = sum(len(self.encoder.encode(m["content"])) for m in self.memory)
                    if total  str:
        self.memory = [{"role": "system", "content": "你是一个代码审查助手，必须按顺序调用工具完成工作。"}]
        self.memory.append({"role": "user", "content": task})

        for iteration in range(self.max_iterations):
            self._compress_memory()
            response = self.llm.invoke(self.memory, tools=self.tool_schemas)

            if not response.tool_calls:
                # LLM决定不再调用工具，进入最终回答
                self.memory.append({"role": "assistant", "content": response.content})
                return response.content

            for call in response.tool_calls:
                tool_name = call["function"]["name"]
                tool_args = json.loads(call["function"]["arguments"])

                # 错误处理：捕获并注入错误信息
                try:
                    tool_result = self.tools[tool_name].run(tool_args)
                except Exception as e:
                    tool_result = f"ERROR: {type(e).__name__}: {str(e)}"
                    # 如果连续两次相同错误，直接放弃该工具
                    if self._check_repeated_error(tool_name, tool_result):
                        return f"无法完成，工具{tool_name}持续失败: {str(e)}"

                self.memory.append({"role": "assistant", "content": call["function"]["arguments"]})
                self.memory.append({"role": "tool", "content": str(tool_result), "tool_call_id": call["id"]})

        return "达到最大迭代次数，任务可能未完成"

这个骨架解决了三个问题：一是用tiktoken实现精确的Token预算控制；二是错误信息直接作为tool消息返回给LLM，让模型自己决定下一步；三是通过循环限制强制终止。

五、踩坑与优化：三个值得记录的细节

坑1：LangChain的.run()方法默认捕获异常。 在_convert_to_schema中，如果工具内部抛异常，LangChain会包装成ToolException并返回一个字符串，而不是抛出。这会导致我的try-except永远捕获不到错误。解决方法是给工具类设置handle_tool_error=False，或者直接调用工具的_run()（绕过包装层）。

坑2：OpenAI的tool_call_id必须与tool消息一一对应。 我发现如果一次返回多个tool_calls，而我只append了一个tool消息，API会报Invalid parameter: messages with role 'tool' must be a response to a preceding message with 'tool_calls'。正确做法是遍历response.tool_calls，每个call生成一个独立的tool消息，并携带对应的tool_call_id。

坑3：记忆压缩策略不能压缩system消息。 最初的压缩逻辑会对所有消息统一处理，结果把system提示词截断了，导致后续LLM行为混乱。现在的实现从index=1开始遍历，跳过system。

优化后的效果：在同样的10个代码审查任务上，成功率从78%提升到94%，平均工具调用次数从11.3次降到7.8次，单任务耗时从平均36秒降到24秒（主要省在减少了无效重试）。

六、效果数据与适用边界

用langchain.evaluation的AgentTrajectoryEvaluator做量化对比，结果如下：

指标 LangChain AgentExecutor 手写Agent
任务完成率 78% 94%
平均工具调用次数 11.3 7.8
平均耗时（秒） 36.2 24.1
Token消耗（平均） 48,200 41,300
错误循环次数 2.1次/任务 0.4次/任务

但必须承认，手写循环不适合所有场景。如果任务工具链非常固定（比如永远是A→B→C），LangChain的SequentialChain更合适；如果需要复杂的并行工具调用，建议直接上langgraph。我的方案适合那种“工具调用顺序不确定、依赖模型实时决策”的任务，比如代码审查、竞品分析、多源数据聚合。

最后提醒：这个循环里的_compress_memory是简单粗暴的截断，如果任务对中间结果敏感（比如需要精确引用某段代码），建议改成摘要式压缩——用LLM对工具结果生成50字摘要存入记忆，但这会额外消耗Token，需要权衡。