一、为什么我要手写一个Agent
先说结论:现成的AutoGPT和LangChain AgentExecutor很好用,但一旦工具返回格式异常、模型开始胡编工具名、或者循环停不下来,你就抓瞎了。
我最近做一个内部运维助手,需求很简单:查服务器状态、执行重启、发通知。用LangChain的initialize_agent十分钟跑通demo,但上线后问题一堆——模型偶尔调用不存在的工具,工具报错后Agent直接卡死,连续对话超过5轮就开始丢失上下文。
痛定思痛,我把Agent拆开自己写了一遍。下面把工具定义、记忆管理、错误处理、循环控制四个模块的实现细节完整分享出来。环境是Python 3.11 + LangChain 0.2.16 + openai 1.40.0,模型用gpt-4o-mini(便宜,够用)。
二、方案设计:Agent就是个while循环
别被“智能体”这个词吓到。剥开外壳,一个Agent的核心逻辑就是:
while 未完成 and 轮次 str:
"""查询指定城市的当前天气。输入城市名,返回温度和天气状况。"""
# 模拟数据,实际替换为API调用
fake_db = {"北京": "晴, 28°C", "上海": "多云, 31°C", "深圳": "雷阵雨, 33°C"}
if city not in fake_db:
raise ValueError(f"暂不支持城市: {city}")
return fake_db[city]
@tool
def calculate(expression: str) -> str:
"""计算数学表达式,例如 '31 - 28'。仅支持四则运算。"""
allowed = set("0123456789+-*/.() ")
if not set(expression) str:
"""发送邮件。to为收件人,content为正文。"""
if "@" not in to:
raise ValueError("邮箱格式错误")
return f"邮件已发送至 {to},内容长度 {len(content)} 字符"
tools = [get_weather, calculate, send_email]
tool_map = {t.name: t for t in tools}
注意@tool装饰器的docstring必须写清楚——这就是给模型看的工具说明书。我踩过的坑:docstring写“查询天气”,模型会传{"city": "北京"};写“输入城市名,返回温度和天气状况”,模型调用准确率明显提升。
3.2 记忆管理
不用LangChain的ConversationBufferMemory,自己用list管更透明:
class AgentMemory:
def __init__(self, system_prompt: str, window_size: int = 6):
self.system = SystemMessage(content=system_prompt)
self.window_size = window_size
self.history = [] # 只存Human/AI/Tool消息
def add(self, msg):
self.history.append(msg)
# 滑动窗口:保留最近N条,但ToolMessage不能孤立存在
if len(self.history) > self.window_size:
self.history = self.history[-self.window_size:]
def get_messages(self):
return [self.system] + self.history
这里有个细节:裁剪时如果第一条是ToolMessage,模型会困惑(没有对应的AIMessage tool_call)。我的做法是裁剪后检查,若首条为ToolMessage则多丢一条。
3.3 循环控制与错误处理
这是最核心的部分,直接上完整Agent类:
class SimpleAgent:
def __init__(self, llm, tools, system_prompt, max_iter=8, max_retry=2):
self.llm = llm.bind_tools(tools)
self.tool_map = {t.name: t for t in tools}
self.memory = AgentMemory(system_prompt)
self.max_iter = max_iter
self.max_retry = max_retry
def _execute_tool(self, tool_call):
"""执行单个工具,带重试"""
name = tool_call["name"]
args = tool_call["args"]
if name not in self.tool_map:
return f"错误:工具 {name} 不存在,可用工具:{list(self.tool_map.keys())}"
for attempt in range(self.max_retry):
try:
result = self.tool_map[name].invoke(args)
return str(result)
except Exception as e:
if attempt == self.max_retry - 1:
return f"工具 {name} 执行失败:{e}。请换一种方式或告知用户。"
time.sleep(0.5)
def run(self, user_input: str) -> str:
self.memory.add(HumanMessage(content=user_input))
for i in range(self.max_iter):
try:
response = self.llm.invoke(self.memory.get_messages())
except Exception as e:
return f"模型调用失败:{e}"
self.memory.add(response)
# 无工具调用 → 最终答案
if not response.tool_calls:
return response.content
# 执行所有工具调用
for tc in response.tool_calls:
result = self._execute_tool(tc)
self.memory.add(ToolMessage(content=result, tool_call_id=tc["id"]))
return "达到最大轮次限制,任务未完成。请简化你的请求。"
几个关键点:
- bind_tools让模型知道有哪些工具,返回结构化tool_calls
- 工具不存在时不抛异常,而是返回错误信息给模型,让它自己纠正
- ToolMessage必须带tool_call_id,否则OpenAI API报400
- 达到max_iter强制退出,防止死循环烧token
3.4 组装运行
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0, timeout=15)
agent = SimpleAgent(
llm=llm,
tools=tools,
system_prompt="你是一个助手。需要数据时调用工具,不要编造。工具报错时尝试其他方案。"
)
print(agent.run("查一下北京和上海的天气,算一下温差,然后发邮件到 ops@test.com 告知结果"))
四、踩坑与优化
坑1:模型编造工具名。 gpt-4o-mini偶尔调用get_weather_info而不是get_weather。解决:在system prompt里明确列出工具名,且错误信息里返回可用工具列表,模型下一轮能自我纠正。
坑2:并行tool_calls导致记忆错乱。 模型一次返回多个tool_call时,必须按顺序为每个call追加ToolMessage。我最初用列表推导,顺序错乱导致tool_call_id对不上,API直接报错。
坑3:温度参数。 temperature=0时工具调用最稳定,但回答略显死板。我最终用0,因为Agent场景准确性优先。
坑4:超时设置。 默认无超时,一次网络抖动整个循环卡死。设timeout=15后,异常能被上层捕获。
五、效果数据
在50条测试用例上跑对比:
| 指标 | 初版(无错误处理) | 优化版 |
|---|---|---|
| 工具调用成功率 | 62% | 94% |
| 平均推理耗时 | 2.4s | 1.8s |
| 任务完成率(3轮内) | 71% | 92% |
| 死循环发生率 | 8% | 0% |
耗时下降主要因为错误重试减少了无效轮次,不是模型变快了。
六、总结
手写Agent没那么玄乎,核心就是:把工具描述写清楚、把错误信息喂回给模型、给循环设上限。LangChain的bind_tools省了schema转换的活,但循环控制和记忆管理自己写反而更稳。
下一步我打算把记忆换成向量检索,解决长对话丢失早期上下文的问题。如果你也在做Agent,建议先把这套最小实现跑通,再去套AutoGPT那套复杂框架——不然出了问题你都不知道从哪查。