一、为什么我要手写一个Agent
先说结论：如果你只会调 initialize_agent，出了问题你根本不知道怎么调。
我在做一个内部知识库问答机器人时，最开始用的是LangChain 0.1的 AgentExecutor + ConversationBufferMemory。跑起来确实快，但遇到三个问题：
- 工具抛异常时，整个chain直接崩，没有重试
- 模型偶尔陷入"调用工具→拿到结果→再调用同一个工具"的死循环，烧token
- 记忆无限增长，跑20轮后prompt超过8k token，延迟从1.2s涨到4.7s
后来我把LangChain的源码翻了一遍，发现Agent的本质没那么神秘：一个while循环 + 一份消息历史 + 一组工具描述 + 一个LLM。于是我决定手写一个，把每个环节的控制权拿回来。
这篇文章就是那次手写的完整记录。技术栈：Python 3.11、openai 1.51.0、pydantic 2.9.2。不依赖LangChain，但思路和LangChain的 AgentExecutor 是一致的，理解了这套，再去看LangChain/AutoGPT的源码会很轻松。
二、环境与版本
python==3.11.9
openai==1.51.0
pydantic==2.9.2
tiktoken==0.8.0
模型用 gpt-4o-mini，因为便宜、支持function calling、响应稳定。测试环境是本地macOS M2，网络走代理，单次LLM调用P50延迟约0.9s。
三、方案设计：Agent的四个核心模块
在动手前，先把架构定下来。我把它拆成四块：
┌─────────────────────────────────────┐
│ Agent Loop │
│ ┌──────────┐ ┌──────────────┐ │
│ │ Memory │──▶│ LLM Call │ │
│ └──────────┘ └──────┬───────┘ │
│ ▲ │ │
│ │ ▼ │
│ ┌──────┴──────┐ ┌──────────┐ │
│ │Tool Result │◀─│Tool Router│ │
│ └─────────────┘ └──────────┘ │
│ ▲ │ │
│ └── Error Handle & Retry ───┘
└─────────────────────────────────────┘
- 工具定义：用JSON Schema描述，交给OpenAI的
tools参数 - 记忆管理：消息列表 + 滑动窗口裁剪 + token预算
- 错误处理：工具异常捕获后作为observation回灌给模型，而不是抛出
- 循环控制：最大步数 + 重复工具调用检测
四、核心实现
4.1 工具定义
工具我用一个dataclass描述，关键是 to_openai_schema 要生成合法的JSON Schema。这里踩过一个坑：OpenAI要求 parameters 必须是 object 类型，且 required 字段必须存在。
from dataclasses import dataclass
from typing import Callable, Any
import json
@dataclass
class Tool:
name: str
description: str
parameters: dict # JSON Schema
func: Callable[..., Any]
def to_openai_schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
def run(self, **kwargs) -> str:
try:
result = self.func(**kwargs)
return str(result)
except Exception as e:
# 关键：异常不抛出，转成字符串回灌给模型
return f"[TOOL_ERROR] {type(e).__name__}: {e}"
# 示例工具
def get_weather(city: str) -> str:
fake_db = {"beijing": "晴 24℃", "shanghai": "多云 27℃"}
return fake_db.get(city.lower(), "未知城市")
def calc(expression: str) -> str:
# 生产环境别用eval，这里仅演示
return str(eval(expression, {"__builtins__": {}}, {}))
tools = [
Tool(
name="get_weather",
description="查询指定城市的当前天气",
parameters={
"type": "object",
"properties": {"city": {"type": "string", "description": "城市拼音"}},
"required": ["city"],
},
func=get_weather,
),
Tool(
name="calc",
description="计算数学表达式，例如 '2+3*4'",
parameters={
"type": "object",
"properties": {"expression": {"type": "string"}},
"required": ["expression"],
},
func=calc,
),
]
TOOL_MAP = {t.name: t for t in tools}
注意 run 方法里的异常处理——这是整个Agent健壮性的关键。如果工具抛异常直接往上冒，Agent循环就断了。转成 [TOOL_ERROR] 字符串回灌，模型下一轮会自己决定换参数重试或者换工具。
4.2 记忆管理
记忆我用一个 Memory 类封装，核心是两条规则：
- 始终保留system message和最近N轮
- token超过预算时从最老的非system消息开始丢
import tiktoken
class Memory:
def __init__(self, system_prompt: str, max_tokens: int = 3000, keep_recent: int = 8):
self.enc = tiktoken.encoding_for_model("gpt-4o-mini")
self.system = {"role": "system", "content": system_prompt}
self.messages: list[dict] = []
self.max_tokens = max_tokens
self.keep_recent = keep_recent
def add(self, msg: dict):
self.messages.append(msg)
def _count(self, msgs: list[dict]) -> int:
return sum(len(self.enc.encode(str(m.get("content") or ""))) for m in msgs)
def get(self) -> list[dict]:
# 从尾部往前保留keep_recent条，再算token
recent = self.messages[-self.keep_recent:]
total = self._count([self.system] + recent)
# 超预算就继续往前砍
idx = len(self.messages) - len(recent)
while total > self.max_tokens and idx > 0:
idx -= 1
candidate = self.messages[idx:]
total = self._count([self.system] + candidate)
return [self.system] + self.messages[idx:]
max_tokens=3000 是我实测出来的一个平衡点：低于2000时模型经常忘记前面的工具结果，高于4000延迟明显上升。
4.3 Agent主循环
现在把工具、记忆、循环控制拼起来。核心结构就是：
while step str:
return hashlib.md5(f"{name}:{args}".encode()).hexdigest()
def run(self, user_input: str) -> str:
self.memory.add({"role": "user", "content": user_input})
self._call_history.clear()
for step in range(self.max_steps):
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=self.memory.get(),
tools=self.tool_schemas,
temperature=0.2,
)
msg = resp.choices[0].message
# 无工具调用 → 最终答案
if not msg.tool_calls:
self.memory.add({"role": "assistant", "content": msg.content})
return msg.content
# 有工具调用 → 记录assistant消息（含tool_calls）
self.memory.add({
"role": "assistant",
"content": msg.content or "",
"tool_calls": [tc.model_dump() for tc in msg.tool_calls],
})
for tc in msg.tool_calls:
name = tc.function.name
args_str = tc.function.arguments
sig = self._sig(name, args_str)
# 循环检测：同一个工具+参数连续调用超过2次就强制终止
if self._call_history.count(sig) >= 2:
result = "[TOOL_ERROR] 检测到重复调用，请换一种方式"
else:
self._call_history.append(sig)
tool = self.tool_map.get(name)
if not tool:
result = f"[TOOL_ERROR] 未知工具 {name}"
else:
try:
kwargs = json.loads(args_str)
result = tool.run(**kwargs)
except json.JSONDecodeError as e:
result = f"[TOOL_ERROR] 参数解析失败: {e}"
self.memory.add({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
return "抱歉，任务超过最大步数限制，请简化你的问题。"
跑一个例子：
agent = Agent(tools)
print(agent.run("北京和上海今天哪个更热？温差多少度？"))
模型会先并行调两次 get_weather，拿到 24 和 27 后调 calc("27-24")，最后回答"上海更热，温差3度"。整个流程3步，token约 780。
五、踩过的坑
坑1：tool_calls 回灌格式错误。 OpenAI要求assistant消息里带 tool_calls，且每个tool结果消息必须有 tool_call_id。我一开始只塞了content，结果API直接报400。
坑2：并行工具调用的顺序。 gpt-4o-mini经常一次返回多个tool_calls。必须按顺序执行，且每个结果都要回灌，否则下一轮模型会"缺数据"。
坑3：重复调用死循环。 模型遇到 [TOOL_ERROR] 时，有时会原封不动重试。我用 _call_history 做签名去重，同一个 name+args 出现2次就注入错误提示，实测能把死循环率从约15%降到2%以下。
坑4：temperature不能太高。 我一开始设0.7，工具参数经常瞎编。改成0.2后，参数合法率从82%提升到97%。
六、效果数据
在30条测试用例（天气查询、数学计算、多步推理混合）上跑：
| 指标 | 数值 |
|---|---|
| 任务完成率 | 93.3% (28/30) |
| 平均步数 | 2.3 步 |
| 平均token | 约 760 |
| P50延迟 | 2.1s |
| 死循环触发 | 0 次 |
| 工具异常恢复率 | 11/12 |
相比直接用 gpt-4o-mini 裸答，token多消耗约40%，但准确率从 63% 提升到 93%。这个trade-off我认为是值的。
七、总结
手写Agent最大的收获不是省了依赖，而是每个环节都可控：
- 工具异常 → 转字符串回灌，不断链
- 记忆膨胀 → token预算裁剪，延迟稳定
- 死循环 → 签名去重，兜底max_steps
- 参数乱编 → 降temperature
理解了这300行，再回头看LangChain的 AgentExecutor 或AutoGPT的 plan-execute 循环，会发现它们无非是在这几个点上做了更多工程化封装。建议你也手写一遍，很多"玄学问题"会瞬间变得清晰。
下一步我打算把工具调用改成流式，再把记忆换成向量检索，有兴趣的可以关注后续。
