Agent 不是「更聪明的聊天机器人」，而是能调用工具、观察结果、决定下一步的循环。这篇讲怎么把这个循环写稳——难点不在调 API，在错误处理、循环终止、结果校验。
一、Agent 的最小骨架：一个循环，四个动作
循环 {
1. 把「历史消息 + 可用工具清单」发给模型
2. 模型返回：要么是文本答案（结束），要么是工具调用请求
3. 执行工具，把结果作为新消息追加进历史
4. 回到 1
}
最多循环 N 次（必须有上限）就这么简单。所有复杂 Agent 都是这个骨架加东西——加记忆、加多工具、加规划、加多 Agent 协作。所以骨架必须先写对。
二、工具定义：描述写不好，模型就不会用
TOOLS = [
{
"name": "search_docs",
"description": "在技术文档库中检索。当用户问具体的 API 用法、配置项、错误码含义时使用。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "检索关键词。用名词短语，不要用完整句子。例如「Nginx 502 排查」而非「我想知道 Nginx 报 502 怎么办」"
},
"top_k": {"type": "integer", "default": 5,
"description": "返回条数，1-10"}
},
"required": ["query"]
}
},
{
"name": "run_sql",
"description": "执行只读 SQL 查询（仅 SELECT）。用于回答需要统计数据的问题，如「上周有多少订单」。",
"parameters": {
"type": "object",
"properties": {
"sql": {"type": "string", "description": "只允许 SELECT 语句"}
},
"required": ["sql"]
}
}
]| 描述要素 | 为什么重要 | 反例 |
|---|---|---|
| 何时使用（触发条件） | 模型据此判断要不要调 | 只写「检索文档」，模型不知道何时用 |
| 参数怎么写（格式约束） | 决定调用质量 | 不写「用名词短语」，模型会传整句 |
| 边界与限制 | 防止越权 | 不写「仅 SELECT」，模型会试 DELETE |
工具描述是提示词工程的一部分，不是文档注释。写得含糊，模型就会调错或不调。经验：description 里写清「什么时候该用」，比写清「它能做什么」更重要。
三、循环实现：三个必须写的防护
async def run_agent(user_msg: str, max_steps: int = 8):
messages = [{"role": "user", "content": user_msg}]
used_tools = [] # 记录已用过的工具，防重复
for step in range(max_steps):
resp = await llm.chat(messages=messages, tools=TOOLS)
msg = resp.choices[0].message
messages.append(msg)
1. 1. 没有工具调用 = 结束
if not msg.tool_calls:
return msg.content
1. 2. 逐个执行工具
for call in msg.tool_calls:
name, args = call.function.name, json.loads(call.function.arguments)
1. 防重复调用同一工具同一参数（模型容易陷入死循环）
sig = f"{name}:{json.dumps(args, sort_keys=True)}"
if sig in used_tools:
result = "该工具已用相同参数调用过，请基于已有结果作答或换方法。"
else:
used_tools.append(sig)
result = await execute_tool(name, args)
messages.append({
"role": "tool",
"tool_call_id": call.id,
"content": str(result)[:2000], # 截断，防爆上下文
})
1. 3. 超过步数上限，强制收尾
return "已达到最大工具调用次数，以下是基于现有信息的回答：\n" + summarize(messages)max_steps是硬防护：没有它，模型可能无限循环烧光 token- 重复调用检测：模型遇到「结果不满意」时会反复用同一参数再试，这是最常见的浪费
- 工具结果要截断：一个 SQL 返回 1000 行会把上下文撑爆，
[:2000]是必要的 - 超步数要有兜底：不能直接报错，要把已有信息整理给用户
四、工具执行：错误必须温柔地告诉模型
async def execute_tool(name: str, args: dict) -> str:
try:
if name == "search_docs":
rows = await search(args["query"], args.get("top_k", 5))
if not rows:
return "未检索到相关文档。可以尝试更通用的关键词。"
return "\n\n".join(f"[{i+1}] {r['content']}" for i, r in enumerate(rows))
if name == "run_sql":
sql = args["sql"].strip()
1. 只读校验：必须 SELECT 开头，且不含危险关键字
if not sql.lower().startswith("select") or \
any(w in sql.lower() for w in ("insert", "update", "delete", "drop", "alter")):
return "只允许 SELECT 查询。请改写。"
rows = await db.fetch(sql, timeout=10)
if len(rows) > 50:
return f"共 {len(rows)} 行，仅展示前 20 行：\n" + str(rows[:20])
return str(rows)
if name == "open_url":
text = await fetch_text(args["url"])
return text[:3000]
return f"未知工具：{name}"
except asyncio.TimeoutError:
return "工具执行超时（10 秒）。可以缩小查询范围重试。"
except Exception as e:
1. 关键：不要抛异常终止 Agent，把错误当「观察结果」告诉模型
return f"工具执行失败：{type(e).__name__}: {e}。请换一种方式或告知用户。"错误不要抛异常，要作为工具结果返回。Agent 的设计理念是「观察 → 调整」，模型看到「超时了，可以缩小范围」会自己改策略；直接抛异常终止，等于放弃了 Agent 的核心能力。
五、状态管理：为什么必须持久化
# 存在数据库里的会话（不是一个内存 dict）
CREATE TABLE agent_sessions (
id UUID PRIMARY KEY,
user_id BIGINT,
messages JSONB, -- 完整消息历史
created_at TIMESTAMPTZ DEFAULT now(),
updated_at TIMESTAMPTZ DEFAULT now()
);
1. 追加消息（不覆盖，便于审计与回溯）
async def append_message(sid: str, msg: dict):
await db.execute("""
UPDATE agent_sessions
SET messages = messages || $2::jsonb, updated_at = now()
WHERE id = $1
""", sid, json.dumps([msg]))- 必须持久化：Agent 一次任务可能跨几分钟，进程重启就丢
- 追加而非覆盖：便于审计「它为什么这么做」，也便于回放
- 要设上限：历史太长会撑爆上下文，超出时做摘要压缩
六、上下文膨胀：Agent 最容易忽略的成本
# 上下文增长的实际速度
1. 第 1 步：用户问题 50 token
1. 第 2 步：工具调用 + 结果 2000 token
1. 第 3 步：又一轮 2000 token
1. ... 8 步之后：16000+ token
1. 每步都要把「全部历史」重发一遍 → 成本是 O(n²)
1. 解法 1：工具结果截断（前面已做）
1. 解法 2：老消息摘要压缩
def compress_history(messages, keep_recent=4):
if len(messages) <= keep_recent + 2:
return messages
old = messages[:-keep_recent]
summary = summarize(old) # 让模型压缩成要点
return [{"role": "system", "content": f"此前的执行摘要：{summary}"}] + messages[-keep_recent:]| 措施 | 效果 | 代价 |
|---|---|---|
| 工具结果截断 | 减少最大单条占用 | 可能丢关键信息 |
| 历史摘要压缩 | 控制总长度 | 摘要本身要一次模型调用 |
| 只保留最近的工具结果 | 大幅减少 | 早期细节丢失 |
| 换更便宜的模型做中间步 | 成本降 5-10 倍 | 规划质量略降 |
七、多工具编排：模型自己会选，别硬编码流程
# 错误：代码里写死「先查文档再查数据库」
async def bad_flow(q):
docs = await search_docs(q)
data = await run_sql(...) # 但用户可能只是问文档
return llm.summary(docs, data)
1. 正确：只给工具，让模型自己决定要不要全用、什么顺序
1. 模型看到「Nginx 502 怎么解决」只会调 search_docs
1. 看到「上周订单数 + 相关文档」会先 run_sql 再 search_docs
result = await run_agent(user_msg)只有在流程有强制约束时才硬编码（比如「必须先查用户权限才能查数据」）。其余情况交给模型判断，这是 Agent 的意义所在。
八、可观测性：出问题要能回答「它为什么这么做」
@dataclass
class AgentTrace:
session_id: str
steps: list # 每一步：输入、模型输出、工具调用、结果、耗时
total_tokens: int
total_cost: float
status: str # success / max_steps / error
1. 每一步都记，方便回放：
1. 第 3 步 | 模型决定调用 search_docs(query="Nginx 502")
1. | 结果：3 条文档，最长 480 字
1. | 耗时 120ms，token +1800
1. 第 4 步 | 模型给出最终答案- 记录每步的 token 与耗时：才知道成本花在哪一步
- 记录工具调用参数与结果摘要：模型选错工具时能定位是描述不清还是理解错
- 记录循环次数：频繁触顶说明工具描述或任务拆解有问题
九、五个真实的坑
| 坑 | 现象 | 正解 |
|---|---|---|
| 工具描述含糊 | 模型该调时不调，或调错工具 | description 写清「何时使用」 |
| 工具结果不截断 | 几千行 SQL 结果撑爆上下文 | 限制行数 + 字符截断 |
| 错误抛异常终止 | 一次超时整个任务失败 | 错误作为工具结果返回 |
| 无步数上限 | 模型死循环烧光配额 | max_steps + 重复调用检测 |
| 历史不压缩 | 越跑越慢、越跑越贵 | 摘要压缩 + 只留最近结果 |
十、总结：Agent 稳不稳看这四点
- 骨架先写对：循环 + 上限 + 兜底，这三样缺一个就不稳
- 工具描述当提示词写：写「何时用」比写「能做什么」重要
- 错误是观察结果，不是异常——让模型看到错误并自己调整
- 上下文与成本要主动管理：截断、摘要、换模型，否则跑几步就失控
- 没有 trace 就没法优化：每步都要能回放