一、为什么不用现成的AgentExecutor?
先说结论:LangChain的AgentExecutor能覆盖80%的场景,但剩下20%的坑——比如自定义重试策略、控制token消耗、处理工具返回的异常——你迟早得读它的源码,甚至自己写一套。
我最近做一个内部知识库助手,需求很朴素:用户提问 → Agent决定查文档还是调API → 返回答案。用create_openai_tools_agent跑通只花了半小时,但上线后发现三个问题:
- 工具调用失败时,Agent会直接把异常信息当成"观察结果"喂回模型,导致模型胡言乱语
- 多轮对话时记忆无限增长,第15轮开始token直接爆到8k+
- 循环控制靠
max_iterations=15硬顶,但实际大部分任务3步就该结束
所以我决定手写一个Agent循环。下面把实现细节拆开讲。
二、环境与版本
Python 3.11.9
langchain==0.3.7
langchain-openai==0.2.8
openai==1.54.3
pydantic==2.9.2
模型用gpt-4o-mini,temperature=0,因为工具调用需要确定性输出。如果你用Claude或Qwen,接口逻辑类似,只是bind_tools的格式不同。
三、方案设计:Agent循环的四个核心模块
一个最小可用的Agent,本质是这样一个循环:
while not done and step str:
# 模拟向量检索
if "报错" in query:
return "错误码E1024:数据库连接超时,检查连接池配置"
return f"关于'{query}'的文档片段1...片段2...片段3"
search_tool = StructuredTool.from_function(
func=search_docs,
name="search_docs",
description="查询内部技术文档,适用于排查报错、查API用法",
args_schema=SearchInput,
)
关键点:description要写清楚"什么时候用",而不是"这是什么"。我测试过,把description从"搜索文档"改成"查询内部技术文档,适用于排查报错、查API用法"后,工具调用准确率从62%提升到94%——模型对使用场景的描述比对功能描述敏感得多。
4.2 记忆管理:分层的Message History
不要把所有消息塞进一个list。我分成两层:
conversation_history:用户和助手的对话,用于多轮上下文scratchpad:工具调用的中间轨迹,任务结束后清空
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
class AgentMemory:
def __init__(self, max_turns: int = 10):
self.conversation = []
self.scratchpad = []
self.max_turns = max_turns
def add_user(self, text):
self.conversation.append(HumanMessage(content=text))
self._trim()
def add_ai(self, msg):
self.conversation.append(msg)
self._trim()
def add_tool_result(self, tool_call_id, content):
self.scratchpad.append(
ToolMessage(content=content, tool_call_id=tool_call_id)
)
def _trim(self):
# 只保留最近N轮,避免token爆炸
if len(self.conversation) > self.max_turns * 2:
self.conversation = self.conversation[-self.max_turns * 2:]
def build_context(self):
return self.conversation + self.scratchpad
def clear_scratchpad(self):
self.scratchpad = []
实测:不trim的情况下,15轮对话后prompt token达到8200;trim到10轮后稳定在2400左右,单次调用成本降低约70%。
4.3 错误处理:把异常转成模型能理解的语言
这是最容易被忽略的部分。如果工具抛异常,直接str(e)塞回去,模型会懵。我的做法是包装成结构化结果:
``python
def safe_execute(tool, args, max_retry=2):
for attempt in range(max_retry + 1):
try:
result = tool.invoke(args)
return {"status": "success", "data": result}
except json.JSONDecodeError as e:
return {"status": "error", "error_type": "parse_error",
"message": f"参数格式错误:{e}", "retryable": False}
except TimeoutError:
if attempt 1时,强制串行执行,并在prompt里加一句"如果工具间有依赖,请一次只调用一个"。
坑2:ToolMessage的tool_call_id必须严格对应。 一开始我手动拼id,结果格式不对,OpenAI直接返回400。后来直接用call["id"]就没问题了。
坑3:max_steps设太小会截断合理任务。 我最初设3,结果"查报错→查文档→总结"刚好3步,再加一步就爆。最后设6,配合前面说的错误处理,实际平均步数是2.3步。
优化:缓存工具结果。 同一轮对话里重复查询相同参数的概率约12%,加个dict缓存后,平均响应时间从8.3秒降到4.1秒。
六、效果数据
在50条真实用户query上测试:
| 指标 | 优化前 | 优化后 |
|---|---|---|
| 工具调用准确率 | 62% | 94% |
| 平均步数 | 4.7 | 2.3 |
| 平均耗时 | 8.3s | 4.1s |
| 无效重试率 | 17% | 2% |
| 单次prompt token | 5200 | 2400 |
七、总结
手写Agent循环不难,难的是把错误处理、记忆trim、循环终止这些"脏活"做扎实。LangChain的AgentExecutor帮你省了80%的代码,但剩下20%的细节决定了你的Agent是demo还是能上线的产品。
如果你要上手,我的建议是:先用create_openai_tools_agent跑通流程,然后把它当黑盒用一段时间,等遇到瓶颈了,再按本文的思路自己实现一遍。代码不长,200行以内,但每一行你都知道在干什么。