一、为什么我要手写一个Agent
先说结论：LangChain的AgentExecutor和AutoGPT都很好用，但当你需要排查“为什么它调了三次同一个工具”“为什么记忆突然丢了”“为什么循环停不下来”时，黑盒会让你抓狂。
我上个项目用LangChain的create_openai_tools_agent搭了个查数据库+查天气的Agent，上线后遇到三个典型问题：
- 模型偶尔返回不存在的工具名，
AgentExecutor直接抛OutputParserException，整个请求500； - 多轮对话超过8轮后，历史消息把上下文撑爆，token费用飙升；
- 某个工具超时后，Agent不知道降级，反复重试同一个调用，直到
max_iterations耗尽。
于是我把Agent拆开自己写了一遍。本文就是这次拆解的完整记录。核心结论：Agent的本质就是一个while循环 + 工具注册表 + 消息历史 + 终止条件判断，没有魔法。
二、环境与版本
Python 3.11.8
langchain 0.1.16
langchain-openai 0.1.7
openai 1.30.1
pydantic 2.7.1
模型用gpt-4o-mini（便宜、function calling稳定），温度设为0（Agent场景下确定性比创造性重要）。如果你用本地模型，把ChatOpenAI换成ChatOllama即可，工具调用格式需要模型支持，qwen2和llama3.1都行。
三、方案设计：四个模块
我把Agent拆成四块，每块职责单一：
| 模块 | 职责 | 关键设计 |
|---|---|---|
| 工具层 | 定义可调用函数 + JSON Schema | Pydantic模型自动生成schema，带参数校验 |
| 记忆层 | 管理对话历史 | 滑动窗口保留最近6轮 + 超长时用LLM摘要压缩 |
| 执行层 | 解析模型输出、调用工具、捕获异常 | 重试2次 + 工具级降级 |
| 循环层 | 控制迭代 | max_steps=8 + 重复调用检测 |
主循环伪代码：
messages = [system_prompt] + memory
for step in range(max_steps):
response = llm.invoke(messages, tools=schemas)
if response has no tool_calls:
return response.content # 终止
for call in response.tool_calls:
result = safe_execute(call) # 带重试
messages.append(tool_message(result))
raise MaxStepsExceeded
下面逐块给代码。
四、核心实现
4.1 工具定义：Pydantic + Schema自动生成
不要手写JSON Schema，容易和函数签名不一致。用Pydantic的model_json_schema()生成，LangChain的convert_to_openai_tool可以直接转换。
from pydantic import BaseModel, Field
from langchain_core.utils.function_calling import convert_to_openai_tool
import json, time
class WeatherInput(BaseModel):
city: str = Field(description="城市名，如 北京")
unit: str = Field(default="celsius", description="温度单位，celsius 或 fahrenheit")
class WeatherTool:
name = "get_weather"
description = "查询指定城市的当前天气"
args_schema = WeatherInput
@staticmethod
def run(city: str, unit: str = "celsius") -> str:
# 模拟外部API，真实场景换成requests
if city not in {"北京", "上海", "深圳"}:
raise ValueError(f"不支持的城市: {city}")
time.sleep(0.1)
temp = {"北京": 22, "上海": 26, "深圳": 30}[city]
if unit == "fahrenheit":
temp = temp * 9 / 5 + 32
return json.dumps({"city": city, "temp": temp, "unit": unit}, ensure_ascii=False)
TOOLS = [WeatherTool]
TOOL_SCHEMAS = [convert_to_openai_tool({
"name": t.name, "description": t.description, "args_schema": t.args_schema
}) for t in TOOLS]
TOOL_MAP = {t.name: t for t in TOOLS}
这里有个细节：convert_to_openai_tool在langchain 0.1.x接受dict形式，直接传类会报错，我踩过这个坑（见第5节）。
4.2 记忆管理：滑动窗口 + 摘要压缩
简单用ConversationBufferMemory的问题是历史无限增长。我的策略：
- 保留最近
K=6轮完整消息（user + assistant + tool）； - 超过时，把更早的消息交给LLM生成一段不超过200字的摘要，插入到system消息之后；
- 摘要本身也做长度上限，避免二次膨胀。
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage, ToolMessage
class Memory:
def __init__(self, llm, window=6, summary_max_tokens=200):
self.llm = llm
self.window = window
self.summary_max_tokens = summary_max_tokens
self.summary = ""
self.buffer = [] # list of BaseMessage
def add(self, msg):
self.buffer.append(msg)
if len(self.buffer) > self.window * 2:
self._compress()
def _compress(self):
old, self.buffer = self.buffer[:-self.window * 2], self.buffer[-self.window * 2:]
text = "\n".join(f"{m.type}: {m.content}" for m in old if m.content)
prompt = f"把以下对话压缩成不超过{self.summary_max_tokens}字的中文摘要，保留关键事实和已完成的工具调用结果：\n{text}"
new_summary = self.llm.invoke([HumanMessage(content=prompt)]).content
self.summary = (self.summary + "\n" + new_summary)[-800:] # 硬上限
def build(self, system_prompt):
msgs = [SystemMessage(content=system_prompt)]
if self.summary:
msgs.append(SystemMessage(content=f"历史摘要：{self.summary}"))
return msgs + self.buffer
实测：10轮对话下，窗口方案比全量历史节省约34%的prompt token（从平均4200降到2780），摘要压缩再多省15%左右，但摘要会损失细节，只在长会话开启。
4.3 执行层：重试 + 降级 + 异常兜底
工具调用失败分三类，处理方式不同：
- 参数错误（Pydantic校验失败）：不重试，把错误信息回传给模型让它修正；
- 瞬时错误（超时、网络）：重试2次，指数退避0.5s/1s；
- 业务错误（城市不支持）：直接返回错误字符串，让模型决定是否换参数。
关键是：任何异常都不要让主循环崩掉，而是转成ToolMessage喂回模型。模型看到错误通常会自己调整。
import logging
logger = logging.getLogger(__name__)
def safe_execute(tool_name: str, args: dict, max_retry: int = 2) -> str:
tool = TOOL_MAP.get(tool_name)
if tool is None:
return f"错误：工具 {tool_name} 不存在，可用工具：{list(TOOL_MAP)}"
# 参数校验
try:
validated = tool.args_schema(**args).model_dump()
except Exception as e:
return f"参数校验失败：{e}。请检查参数格式后重试。"
last_err = None
for attempt in range(max_retry + 1):
try:
return tool.run(**validated)
except (TimeoutError, ConnectionError) as e:
last_err = e
if attempt str:
system_prompt = "你是一个助手。需要外部信息时调用工具。得到足够信息后直接回答，不要重复调用相同工具。"
memory.add(HumanMessage(content=user_input))
seen_calls = {}
for step in range(max_steps):
messages = memory.build(system_prompt)
resp = llm.invoke(messages, tools=TOOL_SCHEMAS)
if not resp.tool_calls:
memory.add(resp)
return resp.content
memory.add(resp)
for call in resp.tool_calls:
key = call["name"] + hashlib.md5(
json.dumps(call["args"], sort_keys=True).encode()).hexdigest()
seen_calls[key] = seen_calls.get(key, 0) + 1
if seen_calls[key] > 2:
logger.warning(f"检测到重复调用 {call['name']}，强制终止")
return "抱歉，我陷入了重复调用，请换个问法或补充信息。"
result = safe_execute(call["name"], call["args"])
memory.add(ToolMessage(content=result, tool_call_id=call["id"]))
return "已达到最大推理步数（8步），任务可能未完成。请拆分问题后重试。"
注意ToolMessage必须带tool_call_id，否则下一次invoke时OpenAI会报400，这个坑很隐蔽。
五、踩坑与优化
convert_to_openai_tool传类报错：langchain 0.1.16要求传dict且key为args_schema，直接传Pydantic类会KeyError。解法如上，包一层dict。tool_call_id丢失：自己拼消息时忘了带id，模型侧400。所有ToolMessage必须和AIMessage.tool_calls的id一一对应。- 温度不为0导致工具参数飘：温度0.7时，同一个问题两次调用
get_weather的unit参数一次celsius一次fahrenheit，测试用例直接挂。Agent场景建议温度0~0.2。 - 摘要压缩把工具结果丢了：早期摘要prompt没强调保留事实，导致多轮后Agent忘记已经查过的天气，重复调用。后来在prompt里明确“保留已完成的工具调用结果”。
- 重试把业务错误也重试了：一开始
except Exception全重试，城市不支持也重试3次，浪费2秒。后来按异常类型分流。
性能数据（GPT-4o-mini，20条测试用例，含单工具/多工具/需澄清三类）：
| 指标 | 裸写（无重试无校验） | 本文方案 |
|---|---|---|
| 任务完成率 | 78% | 96% |
| 平均步数 | 3.8 | 3.1 |
| 平均耗时 | 2.1s | 2.3s |
| 平均prompt token | 4200 | 2780 |
完成率提升主要来自参数校验回传和重试；耗时略增是因为重试和摘要，但token降了34%，综合成本反而更低。
六、总结
手写Agent最大的价值不是省依赖，而是让你对每一步都有控制权：工具schema自己生成所以不会漂，记忆策略自己定所以不会爆，异常自己分流所以不会500，循环自己掐所以不会跑飞。
如果你要上生产，建议在本文基础上补三件事：工具调用的结构化日志（便于回放）、按工具配置超时（concurrent.futures包一层）、以及对max_steps做业务分级（查询类3步、分析类8步）。LangChain的AgentExecutor可以继续用，但至少你现在知道它里面在干什么了。
