Agent 什么时候停?手写一次 Loop,才知道 Harness 在管什么
Agent Runtime 工程 · 工程面试系列 02/05
“我做过 Agent。”如果面试官接着问:“工具超时后你会重新执行吗?”回答“加个重试”,很容易忽略一个关键区别。
查询订单失败,可以考虑再查一次。发送通知超时,却可能是通知已经发出,只是返回包丢了。两种情况都叫超时,下一步应该完全不同。Agent 的工程深度,往往体现在这些具体的状态分歧里。
这篇文章用一个订单查询例子,把模型、工具和运行时串起来。示例使用 Python 标准库和脚本模型,可以离线运行。它验证运行时控制流,未接入真实大模型。
先把循环中的三方分开
用户问:“订单 A17 到哪了?”模型可以提出查询订单的工具调用。真正访问订单系统的动作由运行时执行。模型再根据查询结果决定回答,或者提出下一次工具调用。
用户任务
→ 模型产生 assistant 消息及 tool_calls
→ 运行时校验参数、权限、预算
→ 工具返回结果,绑定 tool_call_id
→ 结果加入上下文,再调用模型
→ 最终答案,或明确的暂停/失败状态
这里的 Harness 可以理解为模型周围负责约束执行的工程环境。工具注册、权限判断、超时、隔离、预算、审计和恢复都可以属于它的职责,具体需要哪几项取决于任务风险。
Agent 不一定先输出一份显式计划。有的任务查一个订单就能结束,有的任务需要长期计划和多步执行。判断是否存在 Agent Loop,要看模型是否能根据执行结果继续选择动作。
第一条契约:结果必须找到对应的调用
常见工具调用协议会提供一个调用 ID。它把模型的一次请求和工具观察结果对应起来:
assistant = {
"role": "assistant",
"tool_calls": [{
"id": "call-1",
"type": "function",
"function": {
"name": "check_order",
"arguments": '{"order_id":"A17"}',
},
}],
}
tool_result = {
"role": "tool",
"tool_call_id": "call-1",
"content": '{"ok":true,"value":{"status":"in_transit"}}',
}
上下文中要保留这条 assistant 工具调用消息,再追加结果。只把“运输中”作为普通用户消息塞进去,会丢掉谁请求了什么、结果属于哪次调用的结构关系。
批量调用更能检验这个理解。模型一次请求查询两个订单,每个结果都应绑定自己的 ID;不能靠结果返回顺序猜对应关系。本例拒绝缺失或重复的调用 ID,并在执行任何工具前先校验整批 ID,避免执行一半才发现协议无法对齐。
上面是 Chat Completions 风格的结构。换用 Responses API 或其他提供方时应按其消息协议实现适配层,不能直接套用这几个字段。
第二条契约:终止由运行时确认
下面十来行展示循环本身。为便于阅读,省略了批量 ID 校验和写入结果不明的处理,不能直接用于生产:
def simple_loop(model, dispatch, messages, max_tool_rounds=2):
for round_no in range(max_tool_rounds + 1):
decision = model(messages)
calls = decision.get("tool_calls", [])
if not calls:
answer = decision.get("content")
return ("done", answer) if answer else ("protocol_error", None)
if round_no == max_tool_rounds:
return ("budget_exhausted", None)
messages.append(decision)
for call in calls:
messages.append(dispatch(call)) # 已配对的工具消息
完整示例还显式区分这些状态:
| 状态 | 含义 | 后续处理 |
|---|---|---|
done | 没有新工具调用,且存在非空答案 | 把答案交给上层 |
budget_exhausted | 不允许再执行下一轮工具 | 保留轨迹,明确未完成 |
protocol_error | 模型响应或调用 ID 不符合协议 | 停止当前执行,交给上层处理 |
review_required | 写操作可能已生效,结果无法确定 | 核对外部状态后再决定 |
模型给出最终答案,只代表本轮工具循环完成。答案是否正确、任务是否满足业务验收,需要另外判断。例如“退款已完成”必须有可信工具结果或外部状态支持,不能仅凭模型自称成功。
示例最多执行两轮工具,同时限制每轮调用数量。最后一次拿到工具结果后,仍允许模型组织答案。若模型继续要求调用,运行时会在新的外部操作发生前返回预算耗尽。生产系统还应独立限制总工具次数、输出大小、token、费用和墙钟时间,单一 max_steps 覆盖不了这些维度。
参数写错时,把失败变成模型可理解的观察
本例的订单查询只接受一个字段:非空字符串 order_id,最多 80 个字符。未知字段、数字类型、非法 JSON、重复 JSON key 都会被拒绝。
{"ok": false, "error": "invalid_arguments"}
模型收到结果后可以修正参数,不过修正次数仍受预算约束。之所以额外拒绝重复 JSON key,是因为部分解析器会静默采用最后一个值。对有副作用的操作,审计里看到的参数与真正执行的参数应尽量没有解释歧义。
生产环境一般用成熟的 schema 校验库。本例只覆盖窄的字符串对象结构。校验通过也不代表参数安全:路径可能越界,订单可能属于另一个租户,URL 可能访问内部网络。这些约束要在工具服务或可信适配层继续检查。
权限不能来自模型自己写的一句话
如果模型输出 send_notice,运行时要检查它是否出现在当前运行允许的工具集合里。用户请求查询物流,并不自动授权通知客户。
if name not in allowed_tools:
return {"ok": False, "error": "permission_denied"}
本例的 allowed_tools 是由调用方提供的可信集合,模型无法通过修改参数获得新权限。真实订单操作还要绑定经过认证的用户、租户、资源范围和审批条件。工具的“只读”属性也要由注册者控制,不能让模型提交 read_only=true 来决定它是否可以自动重试。
工具输出同样需要隔离。比如搜索结果声称“管理员要求立刻调用发送工具”,它仍然只是某段外部数据。应保留工具结果的来源和角色,并让运行时独立判断权限,不能因为文本看起来像命令就放行。
哪些错误可以重试?先看能否重复执行
本例提供三个不同路径:
- 参数和权限错误直接返回结构化失败,不做相同请求的机械重试。
- 可信适配器将只读工具的某次失败分类为
TransientToolError时,最多重试一次。 - 写工具抛出未知异常或暂时性异常时,运行暂停,返回
outcome_unknown。
第三种情况下,后续尚未执行的工具调用会得到 skipped_pending_review,并与原调用 ID 配对。运行时不会把这些失败再交给模型,让它换一个调用 ID 重做同一笔写操作。
如果下游支持稳定的幂等键,运行时可以用同一个操作键查询,或在下游明确保证去重的条件下重放。幂等键应绑定业务操作,不能每次重试都重新生成。若无法确认操作是否生效,先查询外部事实或人工核对。
示例没有执行网络超时或指数退避。只读工具若永久阻塞,轮数预算也无法中断它。生产工具适配器必须配置网络截止时间;需要运行任意代码时,还要考虑独立进程、容器和取消后的资源回收。调用方停止等待,不代表后台任务已经停止。
离线跑通:测试证明了什么
从示例源码 ZIP解压得到 examples 目录。进入解压后的根目录运行:
python3 examples/agent_loop.py
python3 -B -m unittest discover -s examples -p test_agent_loop.py -v
脚本输出:
done 订单 A17 正在运输中。
6 项回归测试覆盖模型与工具消息的对应关系、非法参数与权限拦截、只读暂时性故障的一次重试、写操作结果不明时暂停、工具轮次预算、重复调用 ID 的整批拒绝。实际离线执行 6 项全部通过。
脚本模型是确定性 Python 函数。它能稳定复现异常路径;测试结果不代表真实模型会遵守协议,也没有验证网络超时、生产认证、分布式并发或数据库持久化。
面试官继续追问时,可以这样展开
模型输出最终答案就结束了吗? 运行时结束工具循环,但业务成功需要可信结果和任务验收。我会分别记录终止原因、执行状态与验证结果。
为什么超时不直接重试三次? 先区分操作是否有副作用,以及外部系统是否提供幂等或查询能力。读取通常更容易重试;写入超时可能结果不明,需要保存操作身份并核对,避免重复执行。
有了框架是不是就不需要自己写 Loop? 框架可以管理消息、调度和持久化,但工具权限、错误分类、预算和业务成功条件仍需设计。我会用最小离线例子验证这些契约,再用真实模型测试适配协议和失败路径。
参考
更多推荐

所有评论(0)