Agent 到底是什么?一个 while 循环
Agent = 模型 + 工具 + 循环。模型说"我要调这个工具",你的代码去调,把结果塞回对话,再问一次模型。直到它说"我做完了",或者你让它停。
循环本身二三十行。难的是什么时候停、停得对不对。每一个停止条件都是一个可靠性问题,也是测开最能发挥的地方。
1三个部件:模型、工具、循环
先说清楚一件事:模型从来不执行工具。它只会在回复里写一段结构化的"调用请求"(tool_use 块):调哪个工具、参数是什么。真正去读文件、跑测试、查数据库的,是你写的代码。
| 部件 | 谁负责 | 做什么 |
|---|---|---|
| 模型 | Claude API | 看完整个对话历史,决定:直接回答,还是调工具、调哪个 |
| 工具 | 你的代码 | 普通函数。输入是模型给的 JSON 参数,输出是字符串(或内容块) |
| 循环(harness) | 你的代码 | 调模型 → 看 stop_reason → 执行工具 → 把结果追加进 messages → 再调模型;并决定什么时候停 |
框架(LangChain、Agent SDK、各种 Tool Runner)做的也是这件事,只是替你写好了循环。面试问"不用框架怎么写一个 agent",考的就是你知不知道这个循环里每一步的细节。
2一次请求、一次响应长什么样
场景:让 Agent "跑一下 login 模块的测试,有失败就查清楚原因"。请求里带上工具定义:
POST /v1/messages
{
"model": "claude-opus-5-5",
"max_tokens": 16000,
"tools": [{
"name": "run_tests",
"description": "运行某个模块的单元测试,返回 pytest 输出。需要确认测试是否通过时调用。",
"input_schema": {
"type": "object",
"properties": {"module": {"type": "string"}},
"required": ["module"]
}
}],
"messages": [{"role": "user", "content": "跑一下 login 模块的测试,有失败就查清楚原因。"}]
}
模型想调工具时,响应的 stop_reason 是 "tool_use",content 里有一个或多个 tool_use 块,每个块有唯一的 id:
{
"role": "assistant",
"stop_reason": "tool_use",
"content": [
{"type": "text", "text": "先跑一下 login 模块的测试。"},
{"type": "tool_use", "id": "toolu_01", "name": "run_tests", "input": {"module": "login"}}
]
}
你执行完工具,用一条 role 为 user 的消息把结果送回去。tool_use_id 必须等于上面那个 id:
{"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "toolu_01", "content": "1 failed, 2 passed ..."}
]}
Claude API 没有单独的 tool 角色:工具调用放在 assistant 消息里,工具结果放在 user 消息里。示例里的 id 是简写,真实的 id 形如 toolu_01A09q90qw90lq917835lq9。
3stop_reason:循环的分岔口
每次响应都带一个 stop_reason,告诉你模型为什么停下。循环根据它决定下一步。官方文档列出的取值:
| stop_reason | 含义 | 循环该怎么做 |
|---|---|---|
end_turn | 模型自然说完了 | 正常结束,取最后的文本 |
tool_use | 模型要调工具 | 执行全部 tool_use,结果追加回去,继续 |
max_tokens | 输出撞到了你设的 max_tokens | 不是完成。输出被截断;如果最后一块是没写完的 tool_use,不能执行,要调大 max_tokens 重试 |
stop_sequence | 命中了你设的停止序列 | 按业务处理,一般 agent loop 不设 |
pause_turn | 服务端工具(如 web search)的内部循环到了迭代上限 | 把 assistant 内容原样发回去让它接着跑 |
refusal | 安全分类器拒绝了(HTTP 200,不是报错) | 读 stop_details,不要执行这一轮的工具 |
model_context_window_exceeded | 撞到了模型上下文窗口上限 | 当成截断;需要压缩或拆分历史 |
while resp.stop_reason == "tool_use",循环一退出就当"任务完成"。这样 max_tokens、refusal 都会被当成成功,测试报告里是绿的,用户拿到的是半句话。只有 end_turn 才算正常完成,其他一律按异常分支处理。4消息历史就是状态
API 是无状态的。每调一次模型,都要把整个 messages 列表重新发一遍。Agent 的"记忆"、"进度"、"做过什么",全在这个列表里。一次正常跑完的 loop,messages 长这样:
[
{"role": "user", "content": "跑一下 login 模块的测试,有失败就查清楚原因。"},
{"role": "assistant", "content": [text, tool_use(id=toolu_01, run_tests)]}, # stop_reason: tool_use
{"role": "user", "content": [tool_result(tool_use_id=toolu_01)]},
{"role": "assistant", "content": [text, tool_use(id=toolu_02, read_file log),
tool_use(id=toolu_03, read_file src)]}, # 并行调用
{"role": "user", "content": [tool_result(toolu_02), tool_result(toolu_03)]}, # 一条消息装两个结果
{"role": "assistant", "content": [text("test_login_lockout 失败:...")]} # stop_reason: end_turn
]
这个列表必须满足的规则(来自官方文档)
- 配对:每个
tool_use都要有一个tool_use_id相同的tool_result,放在紧接着的下一条 user 消息里。中间不能插别的消息,否则 400:tool_use ids were found without tool_result blocks immediately after。 - 并行调用一次性返回:一条 assistant 消息里有几个
tool_use,下一条 user 消息里就放几个tool_result。只带一部分结果就发请求会 400;拆成两条相邻的 user 消息,API 会合并成一个 turn,不一定报错,但官方文档把它列为错误格式,会让模型以后少做并行调用。 - tool_result 放最前:同一条 user 消息里如果还有文本,文本必须在所有 tool_result 之后;最好干脆只放 tool_result,否则模型容易直接结束这一轮(返回空的
end_turn)。 - 原样追加 assistant 内容:把
response.content整个追加回去,不要只抽文本。里面可能有 thinking 块、tool_use 块,丢了下一轮就对不上。 - 工具报错也要回结果:工具抛异常时,返回
tool_result并设"is_error": true,内容写清楚错在哪。不能不回,也不能让循环崩掉。
5停止条件:每一条都是一个可靠性问题
"模型说 end_turn 就停"只是理想情况。模型可能一直调工具停不下来,工具可能卡住,费用可能失控。所以一个能上线的 loop 至少要有五个停止条件:
| 停止条件 | 防的是什么 | 怎么测 |
|---|---|---|
模型 end_turn | —(正常出口) | 断言最终状态是"完成",最后一条是 assistant 文本 |
| 最大轮数 max_turns | 模型反复调工具、死循环 | 假模型每次都返回同一个 tool_use,断言恰好在第 max_turns 轮停,状态是"超出轮数" |
| token / 成本预算 | 历史越滚越长,一个任务烧掉几十美元 | 假模型返回固定 usage,断言累计超预算后不再发请求 |
| 超时 | 工具卡住、模型响应慢 | 假工具 sleep,断言整体耗时不超过上限,并且返回的是"超时"而不是挂起 |
| 重复调用检测 | 同一个工具、同一组参数反复调,没有进展 | 假模型重复同一调用,断言在第 N 次时停下;换参数时不误杀 |
token 为什么会失控:每一轮都要重发全部历史。设 system 和工具定义占 \(S\) 个 token,每轮新增 \(\Delta\) 个(模型输出 + 工具结果),第 \(n\) 次调用的输入是 \(S + (n-1)\Delta\),跑 \(N\) 轮的累计输入是
$$\sum_{n=1}^{N}\bigl(S + (n-1)\Delta\bigr) = NS + \frac{N(N-1)}{2}\Delta$$和轮数是平方关系。\(S = 400,\ \Delta = 300\) 时,10 轮累计 17,500 个输入 token,20 轮就是 65,000,轮数翻倍,token 约 3.7 倍。所以轮数上限和 token 预算是两个不同的闸门,都要有。
6四个常见坑
| 坑 | 现象 | 正确做法 |
|---|---|---|
tool_use_id 没对上 | 400 invalid_request_error;或者结果张冠李戴,模型拿 A 的结果回答 B | 结果按 block.id 回填,不要按下标或工具名猜 |
| 并行调用的结果分几次发 | 只回了一部分结果就发请求:400;拆成两条相邻的 user 消息:API 会合并,不一定报错,但模型以后会少做并行调用 | 一条 assistant 里的全部 tool_use,结果放进同一条 user 消息 |
把 max_tokens 当成正常结束 | 报告显示成功,输出是半句话;截断的 tool_use 参数不完整 | 只有 end_turn 算完成;截断的 tool_use 不执行,调大 max_tokens 重试 |
| 无限循环 | 模型反复调同一个工具,费用一直涨 | max_turns + token 预算 + 超时 + 重复检测,缺一个就有一条路径没兜住 |
▶互动演示:单步执行一个 agent loop
不调真实 API:模型的回复是脚本写好的,工具结果也是模拟的,但循环逻辑、停止条件、messages 校验都是真的在跑。token 按 JSON 字符数 ÷ 3 粗估(每次调用都把工具定义和全部历史算一遍),耗时是模拟的秒数。
✎练习
stop_reason: "tool_use",content 里有两个 tool_use(toolu_07 和 toolu_08)。下一条消息应该怎么发?stop_reason 是 "max_tokens",content 最后一块是 tool_use,input 里的 path 看起来是完整的。harness 应该怎么做?7面试要点与代码
一句话讲清楚
Agent 就是模型加工具加一个循环:把 messages 发给模型,stop_reason 是 tool_use 就执行工具,把 tool_result 按 tool_use_id 配对、放进同一条 user 消息追加回去,再调模型;只有 end_turn 算正常完成。上线的 loop 还要有最大轮数、token 预算、超时和重复调用检测四道闸门,每一道都用脚本化的假模型去测。
追问准备
- 工具报错了循环怎么办?捕获异常,返回
is_error: true的 tool_result,把错误信息写清楚(最好写"下一步该怎么做"),让模型自己调整。循环不能崩。 - 为什么 API 是无状态的还能做多轮?状态全在 messages 里,每次重发全部历史。代价是 token 平方增长,所以要有预算,长任务要压缩历史(第 3 周)。
- 超时怎么做才对?两层:单次请求的 HTTP 超时(SDK 的
timeout参数)和单个工具的执行超时,加上整个任务的总时长上限。下面的代码只在每轮开始时检查总时长,一个卡住的工具要靠工具自己的超时兜住。 - 停下来之后 messages 还能接着用吗?如果停在 tool_use 之后(比如重复检测触发),最后一条 assistant 消息的 tool_use 还没有结果。要续跑,得先补上 tool_result(可以是 is_error 说明"被 harness 中止"),否则下一次请求就是 400。
常见错误说法
❌ "
stop_reason 不是 tool_use 就说明做完了":max_tokens、refusal 都不是完成。❌ "并行调用的结果按顺序一条一条发回去":必须放进同一条 user 消息。
❌ "设了 max_turns 就不会失控":10 轮以内 token 也可能超预算,单个工具也可能卡死,要多道闸门。
❌ "真实模型跑了很多次都没死循环,所以不用测":rule of three,30 次零发生只能说上界约 10%。
最小实现(Python,anthropic SDK)
import json
import time
import anthropic
client = anthropic.Anthropic() # 从 ANTHROPIC_API_KEY 读取凭证
MODEL = "claude-opus-5-5"
TOOLS = [{
"name": "read_file",
"description": "读取仓库里的一个文本文件并返回内容。需要看源码或日志时调用。",
"input_schema": {
"type": "object",
"properties": {"path": {"type": "string", "description": "相对仓库根目录的路径"}},
"required": ["path"],
},
}]
def read_file(path: str) -> str:
with open(path, encoding="utf-8") as f:
return f.read()[:20_000]
TOOL_IMPLS = {"read_file": read_file}
def run_tool(block) -> dict:
"""执行一个 tool_use 块。无论成败,都返回 tool_use_id 相同的 tool_result。"""
try:
output = TOOL_IMPLS[block.name](**block.input)
return {"type": "tool_result", "tool_use_id": block.id, "content": output}
except Exception as e: # 工具报错回给模型,而不是让循环崩掉
return {"type": "tool_result", "tool_use_id": block.id,
"content": f"{type(e).__name__}: {e}", "is_error": True}
def run_agent(task, max_turns=10, token_budget=200_000, timeout_s=300, max_repeats=3):
messages = [{"role": "user", "content": task}]
used, start, seen = 0, time.monotonic(), {}
for _ in range(max_turns):
if used >= token_budget:
return "stopped: token_budget", messages
if time.monotonic() - start >= timeout_s:
return "stopped: timeout", messages
response = client.messages.create(
model=MODEL, max_tokens=16000, tools=TOOLS, messages=messages,
)
# 开了 prompt caching 时,input_tokens 不含缓存部分,还要加上
# usage.cache_read_input_tokens 和 usage.cache_creation_input_tokens
used += response.usage.input_tokens + response.usage.output_tokens
messages.append({"role": "assistant", "content": response.content}) # 整个 content 原样追加
if response.stop_reason == "end_turn":
return "done", messages
# 本例没有服务端工具,不会出现 pause_turn;用了 web search 等服务端工具时,
# 要把 assistant 内容追加回去再请求一次
if response.stop_reason != "tool_use": # max_tokens / refusal / ... 都不算完成
return f"stopped: {response.stop_reason}", messages
calls = [b for b in response.content if b.type == "tool_use"]
for b in calls:
key = (b.name, json.dumps(b.input, sort_keys=True))
seen[key] = seen.get(key, 0) + 1
if seen[key] >= max_repeats:
return f"stopped: repeated {b.name}", messages
# 并行调用:全部结果放进同一条 user 消息,而且只放 tool_result
messages.append({"role": "user", "content": [run_tool(b) for b in calls]})
return "stopped: max_turns", messages
if __name__ == "__main__":
status, history = run_agent("读一下 README.md,用三句话总结这个项目。")
print(status, len(history))
写法和官方文档的手动循环一致:client.messages.create(...)、response.stop_reason、response.content 里 type == "tool_use" 的块带 id / name / input,response.usage 里有 input_tokens / output_tokens。SDK 自带的 Tool Runner(client.beta.messages.tool_runner)能替你写这个循环,但先手写一遍,才知道它替你做了什么、没做什么。
下一章:工具调用与结构化输出。工具的 schema 怎么设计、strict: true 能保证什么,以及怎么让模型稳定地输出 JSON。
公式显示依赖 KaTeX(CDN),断网时公式会显示成原始 LaTeX 源码,交互部分不受影响。