一、为什么我要手写一个Agent
先说结论：LangChain的AgentExecutor和AutoGPT的规划循环，在生产环境里我踩过太多坑。LangChain 0.1.x时代，max_iterations默认15，但工具报错后模型会反复重试同一个错误调用，直到打满迭代；AutoGPT更激进，它的任务队列会无限拆分，曾经有个查询任务跑了40多分钟、烧掉11万Token还没停。
问题不在框架，而在于Agent的四个核心机制被封装成了黑盒：工具怎么描述、记忆怎么裁剪、错误怎么回灌、循环什么时候终止。你不改源码就很难控制。
所以我用openai==1.30.1 + pydantic==2.7.1，不引入LangChain，手写一个最小可用Agent。目标很明确：单文件、可调试、每个决策点都能打日志。
二、环境与整体设计
环境很简单：
python==3.11.8
openai==1.30.1
pydantic==2.7.1
tiktoken==0.6.0
模型用gpt-4o-mini，temperature=0，max_tokens=1024。选mini是因为Agent循环里90%的调用是工具选择和参数填充，不需要强推理，成本能压到1/10。
整体循环用经典的ReAct变体：
用户输入 → 组装Prompt(系统+工具+记忆) → LLM输出
→ 解析是工具调用还是最终答案
→ 工具调用：执行 → 结果写入记忆 → 回到LLM
→ 最终答案：返回用户
四个模块对应四个类：Tool、Memory、AgentExecutor、ToolRegistry。下面逐个拆。
三、工具定义：别让模型猜参数
工具定义最容易犯的错是只给一个函数名。模型不知道参数类型，就会传"2024-01-01"还是datetime全靠猜。
我的做法是用Pydantic定义参数Schema，再生成JSON Schema喂给模型的tools字段：
from pydantic import BaseModel, Field
from typing import Callable, Any
import json
class ToolParameter(BaseModel):
pass
class WeatherParams(ToolParameter):
city: str = Field(..., description="城市名，例如 '北京'")
unit: str = Field("celsius", description="温度单位，celsius 或 fahrenheit")
class Tool:
def __init__(self, name: str, description: str,
params_model: type[BaseModel], func: Callable):
self.name = name
self.description = description
self.params_model = params_model
self.func = func
def to_openai_schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.params_model.model_json_schema(),
}
}
def run(self, raw_args: str) -> str:
# 关键：先校验再执行
args = self.params_model.model_validate_json(raw_args)
result = self.func(**args.model_dump())
return str(result)
注意model_validate_json这一步。模型返回的arguments是字符串，经常出现多一个逗号、少一个引号的情况。Pydantic校验失败时抛的ValidationError信息很具体，我会把它原样回灌给模型，让它自己修——这一步是后面错误自愈的基础。
四、记忆管理：滑动窗口 + 工具结果压缩
记忆是Agent最容易爆的地方。一个查天气+查股价的任务，工具返回的JSON动辄上千Token，5轮下来上下文就过万。
我的策略是两层：
- 对话消息保留滑动窗口：只保留最近
max_turns=6轮完整消息。 - 工具结果超长截断：超过
800字符的结果，保留前400 + 后200，中间用...[truncated N chars]...占位。
import tiktoken
class Memory:
def __init__(self, max_turns: int = 6, tool_result_limit: int = 800):
self.messages: list[dict] = []
self.max_turns = max_turns
self.tool_result_limit = tool_result_limit
self.enc = tiktoken.encoding_for_model("gpt-4o-mini")
def add_user(self, content: str):
self.messages.append({"role": "user", "content": content})
def add_assistant(self, message: dict):
self.messages.append(message)
def add_tool_result(self, tool_call_id: str, content: str):
content = self._truncate(content)
self.messages.append({
"role": "tool",
"tool_call_id": tool_call_id,
"content": content,
})
def _truncate(self, text: str) -> str:
if len(text) list[dict]:
# 保留 system + 最近 max_turns 轮
if len(self.messages) int:
return sum(len(self.enc.encode(m.get("content") or "")) for m in self.messages)
token_count是我加的可观测性钩子，每次循环打印一次，超过6000就强制触发裁剪告警。实测一个3工具任务，不裁剪平均消耗4200 Token，裁剪后降到2900左右。
五、循环控制与错误处理：三个终止条件
这是全文最关键的部分。循环必须同时满足三个终止条件才退出：
- 模型返回最终答案（没有
tool_calls） - 达到最大迭代次数
max_iterations=8 - 连续工具错误次数
max_consecutive_errors=3
第3条是我踩坑后加的。之前只靠迭代次数兜底，模型会在同一个错误工具上重试7次，浪费Token还污染记忆。
import json
from openai import OpenAI
client = OpenAI()
class AgentExecutor:
def __init__(self, tools: list[Tool], system_prompt: str,
max_iterations: int = 8, max_consecutive_errors: int = 3):
self.tools = {t.name: t for t in tools}
self.system_prompt = system_prompt
self.max_iterations = max_iterations
self.max_consecutive_errors = max_consecutive_errors
self.memory = Memory()
def _build_tool_schemas(self):
return [t.to_openai_schema() for t in self.tools.values()]
def run(self, user_input: str) -> str:
self.memory.add_user(user_input)
consecutive_errors = 0
for step in range(self.max_iterations):
messages = [{"role": "system", "content": self.system_prompt}] \
+ self.memory.get_context()
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=messages,
tools=self._build_tool_schemas(),
tool_choice="auto",
temperature=0,
max_tokens=1024,
)
msg = resp.choices[0].message
self.memory.add_assistant(msg.model_dump(exclude_none=True))
# 终止条件1：没有工具调用，直接返回
if not msg.tool_calls:
return msg.content or ""
# 执行所有工具调用
for call in msg.tool_calls:
name = call.function.name
raw_args = call.function.arguments
if name not in self.tools:
err = f"Error: tool '{name}' not found. Available: {list(self.tools)}"
self.memory.add_tool_result(call.id, err)
consecutive_errors += 1
continue
try:
result = self.tools[name].run(raw_args)
self.memory.add_tool_result(call.id, result)
consecutive_errors = 0
except Exception as e:
# 关键：把错误信息回灌给模型，让它自我修正
err_msg = f"Error calling {name}: {type(e).__name__}: {e}"
self.memory.add_tool_result(call.id, err_msg)
consecutive_errors += 1
print(f"[step {step}] tool error: {err_msg}")
# 终止条件3：连续错误过多，强制中断
if consecutive_errors >= self.max_consecutive_errors:
return f"[Agent aborted] {consecutive_errors} consecutive tool errors."
# 终止条件2：迭代耗尽
return "[Agent aborted] max_iterations reached."
三个细节值得说：
msg.model_dump(exclude_none=True)：OpenAI SDK 1.x的message对象直接塞进messages会带一堆None字段，某些版本会报schema错。- 错误信息里带
type(e).__name__：模型看到ValidationError和KeyError的反应不一样，前者会重试参数，后者会换工具。 consecutive_errors = 0只在成功时重置：这是"连续"的定义，别写成累计。
六、效果数据与踩坑记录
跑了一组对比测试，任务集是20条混合指令（查天气、算数、多步组合），结果：
| 指标 | 裸ReAct(无错误回灌) | 本文实现 |
|---|---|---|
| 平均迭代轮数 | 4.1 | 2.7 |
| 工具调用成功率 | 68% | 93% |
| 平均Token消耗 | 4100 | 2870 |
| 任务完成率 | 75% | 95% |
三个印象最深的坑：
坑1：tool_choice="auto"不是万能的。 模型有时会在应该调工具时直接编答案。解决办法是在system prompt里硬写一句"如果需要外部信息，必须先调用工具，禁止凭记忆回答"。
坑2：并行工具调用会打乱顺序。 GPT-4o-mini支持一次返回多个tool_calls，但执行顺序不确定。我的做法是串行执行，虽然慢一点，但记忆顺序稳定。
坑3：截断工具结果会丢关键信息。 有一次查股票返回的JSON，关键价格字段正好在中间被截掉了。后来改成"优先保留JSON的顶层key"，但那是另一个话题了，简单场景下头尾截断够用。
七、总结
手写Agent的价值不在于替代LangChain，而在于你知道每一行在干什么。工具定义用Pydantic锁住参数、记忆用滑动窗口+截断控成本、错误回灌让模型自愈、三重终止条件防死循环——这四件事做好，一个300行的Agent就能覆盖80%的日常场景。
下一步我打算把工具执行改成异步+超时控制，现在同步调用遇到慢接口会把整个循环卡住。如果你也在做Agent，建议先把max_consecutive_errors这个参数加上，能省不少钱。
