1. 问题背景：为什么不用现成的AgentExecutor
最近在做一个内部知识库问答机器人，需要让LLM自主调用搜索引擎、数据库查询、代码执行器三个工具。一开始直接用LangChain的AgentExecutor，但遇到三个实际问题：
- 工具调用失败无感知：天气API偶发超时，Agent直接报错终止，不会重试。
- 上下文无限膨胀：AutoGPT式的ReAct循环中，每次观察结果都塞进Prompt，到第5轮就超出GPT-4o的128k窗口（实际按token算更早爆）。
- 循环控制死板：无法自定义“最大轮次”和“关键结果命中即停”。
我决定基于LangChain的BaseTool和BaseChatMessageHistory手写一个循环控制器，只保留必要组件，去掉黑盒抽象。
2. 环境与版本
Python 3.11.5
langchain==0.3.7
langchain-openai==0.2.8
openai==1.54.3
tiktoken==0.8.0
注意：LangChain 0.3.x中AgentExecutor已经标记为legacy，但BaseTool、Tool、ChatPromptTemplate这些底层组件依然稳定。我用的模型是gpt-4o-mini（便宜，单轮调用成本约0.003美元）。
3. 方案设计：三个核心模块
我的Agent结构分为三层：
┌─────────────────────────────────────┐
│ LoopController (循环控制/熔断) │
├─────────────────────────────────────┤
│ MemoryManager (双缓冲记忆池) │
├─────────────────────────────────────┤
│ ToolRegistry (工具注册/调用/错误) │
└─────────────────────────────────────┘
关键设计决策：
- 工具定义：用
@tool装饰器+args_schema强制参数校验，比裸函数多一层Pydantic检查。 - 记忆管理：维护两个列表——
short_term（当前任务轮次）和long_term（关键事实摘要）。每次循环前，只把short_term最后2轮和long_term全部塞入Prompt。 - 错误处理：捕获工具异常后，把错误信息作为“观察”返回给LLM，而不是终止循环。连续3次相同工具失败则熔断。
- 循环控制：最大轮次5次，当LLM输出的Action为
Final Answer时立即停止。
4. 核心实现：工具定义与注册
先看工具定义。我用@tool装饰器，并显式声明args_schema避免LLM传入错误类型：
# tools.py
from langchain_core.tools import BaseTool, tool
from pydantic import BaseModel, Field
import requests, json, time
class SearchInput(BaseModel):
query: str = Field(description="搜索关键词，必须为英文或中文短句")
@tool(args_schema=SearchInput, return_direct=False)
def web_search(query: str) -> str:
"""模拟搜索接口，实际项目替换为SerpAPI或Bing Search。"""
# 模拟50%概率超时，用于演示错误处理
if len(query) > 20:
time.sleep(2.5) # 模拟超时
raise TimeoutError("search service timeout after 2s")
return json.dumps({"result": f"找到3条关于{query}的结果"}, ensure_ascii=False)
# 注册表：用dict存工具名和实例
TOOL_REGISTRY = {
"web_search": web_search,
}
def execute_tool(name: str, args: dict) -> tuple[bool, str]:
"""统一执行入口，返回(是否成功, 结果/错误信息)"""
if name not in TOOL_REGISTRY:
return False, f"未知工具: {name}，可用工具: {list(TOOL_REGISTRY.keys())}"
try:
result = TOOL_REGISTRY[name].invoke(args)
return True, str(result)
except Exception as e:
return False, f"[工具错误] {type(e).__name__}: {str(e)}"
这里有个坑：@tool装饰器默认会包装函数，但invoke时如果Pydantic校验失败会抛ValidationError，必须在execute_tool里捕获。否则异常会直接穿透到循环控制器。
5. 核心实现：记忆管理与循环控制
记忆管理我采用双缓冲策略。short_term只保留最近2轮的思考/行动/观察，防止Prompt膨胀；long_term存关键事实，比如“用户偏好简洁回答”。
# agent_core.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
import json, tiktoken
class SimpleAgent:
def __init__(self, model_name="gpt-4o-mini", max_rounds=5, max_fail_count=3):
self.llm = ChatOpenAI(model=model_name, temperature=0.2)
self.max_rounds = max_rounds
self.max_fail_count = max_fail_count
self.short_term = [] # 最近2轮
self.long_term = [] # 关键事实
self.encoder = tiktoken.encoding_for_model("gpt-4o-mini")
def _build_prompt(self, user_query: str) -> str:
# 动态计算token，超过4000则裁剪short_term
system = SystemMessage(content=(
"你是Agent，必须按格式输出："
'{"action": "工具名", "args": {...}} 或 {"action": "Final Answer", "content": "最终回答"}'
))
memory_msgs = []
for item in self.long_term[-3:]: # 最多3条长期记忆
memory_msgs.append(HumanMessage(content=f"[记忆] {item}"))
for item in self.short_term[-2:]: # 最近2轮
memory_msgs.append(AIMessage(content=item["thought"]))
memory_msgs.append(HumanMessage(content=item["observation"]))
prompt = [system] + memory_msgs + [HumanMessage(content=f"问题: {user_query}")]
# token截断：如果超过4000，只保留最后3条消息
total_tokens = sum(len(self.encoder.encode(m.content)) for m in prompt)
if total_tokens > 4000:
prompt = prompt[-3:]
return prompt
def run(self, user_query: str) -> str:
fail_count = 0
for round_idx in range(1, self.max_rounds + 1):
prompt = self._build_prompt(user_query)
response = self.llm.invoke(prompt).content
# 解析JSON动作
try:
action_obj = json.loads(response)
except json.JSONDecodeError:
# 容错：如果LLM输出非JSON，直接要求重新生成
self.short_term.append({"thought": response, "observation": "输出格式错误，必须输出合法JSON"})
continue
if action_obj.get("action") == "Final Answer":
return action_obj["content"]
tool_name = action_obj.get("action")
tool_args = action_obj.get("args", {})
success, result = execute_tool(tool_name, tool_args)
if not success:
fail_count += 1
if fail_count >= self.max_fail_count:
return f"连续{self.max_fail_count}次工具失败，熔断终止。最后错误: {result}"
self.short_term.append({
"thought": f"尝试调用{tool_name}但失败，我需要换一种方式",
"observation": result
})
else:
fail_count = 0 # 成功则重置计数
self.short_term.append({
"thought": f"调用{tool_name}成功，结果: {result[:100]}...",
"observation": result
})
# 如果结果包含“找到3条”，存入长期记忆
if "找到3条" in result:
self.long_term.append(f"用户问题[{user_query}]的搜索结果数量为3")
# 控制short_term长度，只保留最后4条（2轮）
if len(self.short_term) > 4:
self.short_term = self.short_term[-4:]
return "达到最大轮次5，未得到最终答案"
核心循环逻辑：
- 每轮先构建Prompt，检查token数。
- 让LLM输出严格JSON格式的动作。
- 解析JSON，如果格式错误直接计入
short_term并继续。 - 工具调用失败时，不终止，而是把错误信息作为观察给LLM，让它“自我纠错”。
- 连续3次失败触发熔断（类似断路器模式）。
- 每次成功后重置失败计数。
6. 踩坑与优化：三个真实教训
坑1：LangChain 0.3的Tool.invoke参数类型
web_search.invoke({"query": "hello"})正常，但直接web_search.invoke("hello")会抛TypeError。因为args_schema强制要求dict输入。我最初用字符串测试浪费了半小时。
坑2：LLM输出JSON时偶尔带Markdown代码块
实测GPT-4o-mini有约12%的概率输出```json ... ```包裹的JSON。必须在解析前做清洗：
import re
def clean_json_response(raw: str) -> str:
raw = raw.strip()
if raw.startswith("```"):
raw = re.sub(r"^```(?:json)?|```$", "", raw, flags=re.MULTILINE).strip()
return raw
坑3：记忆池的token计算不准
用tiktoken按字符估算误差大，gpt-4o-mini的tokenizer是o200k_base，但encoding_for_model自动处理。实测len(encoder.encode("中文"))约等于1.5个token，中文场景下4000token限制大约能存2800个汉字，够用。
7. 效果数据与总结
在本地20组测试问题（含5个超时场景、3个非法参数场景）中：
| 指标 | 手写Agent | LangChain AgentExecutor |
|---|---|---|
| 工具调用成功率 | 94% | 72% |
| 平均完成轮次 | 2.3 | 3.8 |
| 平均耗时(秒) | 1.8 | 2.9 |
| 死循环率 | 0% | 5% |
手写Agent的主要优势在于失败回传机制——LLM看到工具错误信息后能自我修正（比如把query缩短），而AgentExecutor会直接抛出AgentActionException。
如果你也遇到类似问题，建议不要盲目上LangGraph，先手写一个200行的循环体，理解工具、记忆、错误之间的关系，再考虑更重的框架。代码已上传至GitHub（文中简化版），有问题评论区交流。
