一、为什么我要手写一个Agent
上个月团队要做一个内部运维助手，需求很朴素：根据用户问题调用几个内部API（查监控、查日志、重启服务），然后汇总结果。第一反应是用LangChain的AgentExecutor，结果调了两天，被它的回调机制和中间步骤日志搞得头大——出问题时根本不知道是prompt的问题、工具描述的问题，还是它内部解析输出格式失败。
后来索性花一个下午自己写了一个。核心逻辑其实就一个while循环：让模型决定调用哪个工具，执行工具，把结果塞回对话历史，再问模型，直到它给出最终答案。就这么简单。
这篇文章把实现过程完整拆开讲。代码基于Python 3.11 + openai 1.35.0，不依赖LangChain，全部代码可以直接跑。
二、整体设计：一个Agent需要哪几块
先明确一个最小可用Agent的组成：
- 工具定义：用JSON Schema描述每个工具的名称、参数、用途，让模型知道有什么可调用
- 消息历史（记忆）：维护一个messages列表，包含system prompt、用户输入、模型回复、工具执行结果
- 循环控制：最大迭代次数限制 + 终止条件判断（模型不再请求工具即为结束）
- 错误处理：工具执行异常时不能直接崩溃，要把错误信息作为工具结果返回给模型，让它自己决定重试还是换方案
整个执行流程是：
用户输入 → 组装messages → 调用LLM
↓
LLM返回tool_calls?
├─ 是 → 执行工具 → 结果追加到messages → 回到调用LLM（循环）
└─ 否 → 返回最终文本答案，结束
这个循环就是所有Agent框架的核心，AutoGPT也好、LangChain的AgentExecutor也好，本质都是这个循环加上各种增强（反思、规划、多Agent协作）。
三、核心实现
3.1 工具定义与注册
工具定义的关键是描述要写清楚用途和边界。模型选错工具，90%是因为description写得含糊。
import json
import inspect
from typing import Callable
class Tool:
def __init__(self, name: str, description: str, func: Callable, parameters: dict):
self.name = name
self.description = description
self.func = func
self.parameters = parameters
def to_openai_schema(self) -> dict:
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": self.parameters,
},
}
class ToolRegistry:
def __init__(self):
self._tools: dict[str, Tool] = {}
def register(self, tool: Tool):
self._tools[tool.name] = tool
def get(self, name: str) -> Tool | None:
return self._tools.get(name)
def all_schemas(self) -> list[dict]:
return [t.to_openai_schema() for t in self._tools.values()]
注册两个示例工具。注意参数schema用标准JSON Schema，required字段一定要写，否则模型可能传空参数：
def get_weather(city: str) -> str:
# 模拟实现，真实场景换成HTTP请求
fake_db = {"北京": 18, "上海": 24, "广州": 29}
if city not in fake_db:
raise ValueError(f"不支持的city参数: {city}，仅支持北京/上海/广州")
return json.dumps({"city": city, "temp_c": fake_db[city]}, ensure_ascii=False)
def calculate(expression: str) -> str:
# 只允许数字和四则运算，防止注入
allowed = set("0123456789+-*/(). ")
if not set(expression) 200:
msg["content"] = "[历史工具结果已省略]"
self.messages = keep_head + keep_tail
3.3 循环控制与错误处理
这是整个Agent的"心脏"。我踩过的坑集中在这里：
坑1：模型可能无限调用工具。 必须有max_iterations，我设的是8。超过后强制让模型输出最终答案（追加一条user消息："请基于已有信息直接给出结论"）。
坑2：工具抛异常直接崩。 必须try/except，把异常信息转成字符串作为tool消息返回。模型看到错误后通常会自己调整参数重试，实测GPT-4o-mini对"参数非法"这类错误的自愈率大概70%。
坑3：模型返回的arguments不是合法JSON。 虽然function calling模式下概率很低，但出现过一次（模型把中文引号写进去了）。用json.loads包一层try，失败就返回错误消息让它重试。
坑4：并行tool_calls。 模型一次可能返回多个tool_call，必须全部执行、全部追加结果，缺一个都会报错。
from openai import OpenAI
client = OpenAI(api_key="sk-xxx", base_url="https://api.openai.com/v1")
SYSTEM_PROMPT = """你是一个运维助手。你可以调用工具获取信息。
规则：
1. 需要数据时必须调用工具，不要编造。
2. 工具返回错误时，分析错误原因并调整参数重试，最多重试2次。
3. 信息足够后直接给出简洁结论。"""
def run_agent(user_input: str, max_iterations: int = 8) -> str:
memory = Memory(SYSTEM_PROMPT)
memory.add({"role": "user", "content": user_input})
tool_retry_count: dict[str, int] = {}
for i in range(max_iterations):
resp = client.chat.completions.create(
model="gpt-4o-mini",
messages=memory.messages,
tools=registry.all_schemas(),
tool_choice="auto",
temperature=0,
)
msg = resp.choices[0].message
# 终止条件：模型不再请求工具
if not msg.tool_calls:
memory.add({"role": "assistant", "content": msg.content})
return msg.content
# 把assistant的tool_calls消息原样存回历史（必须！）
memory.add({
"role": "assistant",
"content": msg.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {"name": tc.function.name,
"arguments": tc.function.arguments},
} for tc in msg.tool_calls
],
})
# 逐个执行（支持并行tool_calls）
for tc in msg.tool_calls:
result = _execute_tool(tc, tool_retry_count)
memory.add({
"role": "tool",
"tool_call_id": tc.id,
"content": result,
})
# 迭代超限，强制收敛
memory.add({"role": "user", "content": "已达到最大工具调用次数，请基于现有信息直接给出最终结论。"})
final = client.chat.completions.create(
model="gpt-4o-mini", messages=memory.messages, temperature=0
)
return final.choices[0].message.content
def _execute_tool(tc, retry_count: dict) -> str:
tool = registry.get(tc.function.name)
if tool is None:
return f"错误：工具 {tc.function.name} 不存在"
try:
args = json.loads(tc.function.arguments)
except json.JSONDecodeError as e:
return f"错误：参数不是合法JSON（{e}），请重新生成"
try:
return str(tool.func(**args))
except TypeError as e:
return f"错误：参数不匹配（{e}）"
except Exception as e:
cnt = retry_count.get(tc.function.name, 0) + 1
retry_count[tc.function.name] = cnt
if cnt >= 3:
return f"错误：{e}。该工具已失败{cnt}次，请放弃调用并基于已有信息作答。"
return f"错误：{e}。请调整参数后重试。"
跑一下：
print(run_agent("北京和上海现在温差多少度？用华氏度表示温差"))
实际执行路径是：get_weather(北京) → get_weather(上海) → calculate((24-18)*1.8) → 最终答案。三次工具调用，一次循环内完成两个并行调用。
四、踩坑与优化记录
1. temperature必须设0。 一开始用默认值0.7，同一句话两次运行走了不同工具路径，有一次甚至跳过工具直接编了个温度。改0之后行为稳定。
2. description里写"不要用于X"比写"用于Y"更有效。 我一开始把calculate描述成"数学计算工具"，结果模型查完天气后直接心算温差，压根不调calculate。改成"所有数值计算必须调用此工具，不要自行心算"后解决。
3. tool消息的content不能是None。 有次工具函数返回None，直接传进去API报错。统一str()包一层。
4. token成本。 每个tool schema大约占80-120 tokens，4个工具的情况下每轮请求的固定开销约400 tokens。加上历史累积，一个三步骤任务实测平均1200 tokens、耗时3.2秒（gpt-4o-mini，含3次LLM调用）。如果工具超过10个，建议做工具动态筛选，不然光schema就吃掉上千token。
5. 错误信息要写给模型看，不是给人看。 早期我返回KeyError: 'city'，模型完全不知道怎么改。改成"缺少必填参数city，请补充"之后，自愈率明显上升。
五、效果数据
在20条测试用例上（覆盖单工具、多工具串联、参数错误、不存在的城市四类场景）：
| 指标 | 结果 |
|---|---|
| 任务完成率 | 19/20（95%） |
| 平均LLM调用次数 | 3.1 次 |
| 平均耗时 | 3.2 s |
| 平均token消耗 | 1210 tokens |
| 参数错误自愈率 | 7/10（70%） |
| 唯一失败案例 | 用户问"火星天气"，模型重试3次后放弃，但没给出明确"无法回答"的结论 |
失败案例后来通过在第3次重试的错误信息里加"请直接告知用户该信息无法获取"解决。
六、总结
手写Agent之后我的感受是：LangChain和AutoGPT的价值在于生态和开箱即用，但核心循环本身不到100行。自己写一遍的好处是每个环节都可控——工具为什么没被调用、错误为什么没被自愈、循环为什么停不下来，都能直接定位到具体的那几行代码。
如果要做生产级Agent，在这个骨架上还需要补的东西：流式输出、工具调用的并发执行（现在是串行for循环，可以改成asyncio.gather）、结构化日志（记录每次tool_call的入参出参耗时）、以及工具级别的权限控制。但这些都是在同一个循环上做加法，核心逻辑不会变。
代码全部在上面，复制粘贴就能跑，建议自己改改工具描述和max_iterations，观察模型行为的变化，比看十篇框架文档有用。
