3.2 停止条件与循环上限
上一节的 run() 有一个隐含假设：模型总会自己停下来。真实世界不成立。模型会为了一个查不到的结果反复搜索、会固执地重复同一个动作、会因为输出被截断而给不出完整答案。如果循环只认「没有工具调用就结束」这一条出口，遇到这些情况就会一直转下去，直到 max_steps 用尽——而每一轮都在真金白银地烧 token。
本节把 minagent 推进到「会自己喊停」：在
minagent/loop.py里补齐五条终止路径，用AgentLoopError把异常终止连同部分结果一起抛出，让调用方能诊断「为什么停的、停在哪一步」。
3.2.1 五种终止路径
一个健壮的循环不是只有一个出口，而是一组出口。把它们列全，是写对循环的第一步：
| 终止路径 | 触发条件 | 性质 | 期望行为 |
|---|---|---|---|
| 自然收敛 | 某轮 msg.tool_calls 为空 | 成功 | 返回 RunResult |
| 达到步数上限 | 循环跑满 max_steps | 异常 | 抛错，附已走轨迹 |
| 输出被截断 | resp.stop_reason == "max_tokens" | 异常 | 抛错，保留半截文本 |
| 工具连续失败 | 连续 N 次工具返回 error | 异常 | 抛错，附最后一次错误 |
| 重复动作 | 相邻两次工具调用签名相同 | 异常 | 抛错，附已走轨迹 |
只有第一条是「好」的结束。后四条都是「坏」的结束——但「坏」不代表程序该崩，而是意味着我们要把失败变成一种可读的数据，交给上层去决定是重试、降级还是报错。这是整个章节的设计基调。
3.2.2 无工具调用：唯一的「成功」出口
这是上一节已有的逻辑，原样保留：
if not msg.tool_calls:
steps.append(Step(index=i, kind="final", detail=msg.content))
return RunResult(messages=messages, steps=steps, final=msg.content)
判据是「这一轮没有要求任何工具」。注意它和「content 非空」不是一回事：模型可能既说了话又调了工具，那种情况仍要执行工具、继续循环。只有工具列表为空，才轮到文本答案登场。
3.2.3 max_steps：兜底保险丝
max_steps 是最粗暴也最可靠的一道防线。它不判断模型「该不该停」，只规定「最多跑几步」。上一节它在超限时返回一个带 error 步的结果，本节改成立即抛出异常：
raise AgentLoopError("max_steps", RunResult(messages, steps, ""))
为什么默认是 8？这是一个经验值：绝大多数单文件任务（查资料、算数、读一篇文章）在 3～6 步内收敛，8 步留了余量又能把最坏成本压在可控范围。复杂任务可以调大，但要清楚每加一步就多一次模型调用的开销。第 11 章讲规划时，会用一个更大的上限配合计划检查点，而不是简单地把它调大。
3.2.4 输出被截断：先看 stop_reason
ModelResponse.stop_reason 是个容易被忽略但极重要的信号。它告诉我们模型为什么停下了：
stop_reason | 含义 | 循环该怎么反应 |
|---|---|---|
end_turn | 正常说完 | 正常处理 |
tool_use | 要求调工具 | 执行工具 |
max_tokens | 输出达到长度上限被截断 | 这一轮结果不完整，应中断 |
被截断的 content 是一句话说到一半，它看起来像「最终答案」，实则不是。如果不管 stop_reason，循环会把这个半截答案当成功返回，用户拿到一个残缺的结果却毫不知情。所以在追加消息之后、判断工具调用之前，先检查这一条：
if resp.stop_reason == "max_tokens":
raise AgentLoopError("max_tokens", RunResult(messages, steps, msg.content))
这里把半截文本放进 final 一起抛出，是为了让调用方能看到「模型说到哪了」，而不是只拿到一个干巴巴的错误码。
3.2.5 工具连续失败
工具失败本身不是问题——上一节说过，registry.call() 出错时返回 {"error": ...} 的 JSON，让模型有机会读到错误并换个思路。真正的问题是连续失败：模型反复重试同一个坏参数，每次都拿到同样的错误，却始终不改。
做法是维护一个连续失败计数，任一成功就清零：
failures = failures + 1 if '"error"' in result else 0
if failures >= max_consecutive_failures:
raise AgentLoopError("tool_failures", RunResult(messages, steps, ""))
判据是「工具返回的字符串里含 "error" 键」。这依赖第 4 章会强化的约定：工具的错误必须统一以 {"error": ...} 形式返回。默认阈值 3 次——给模型两次纠错机会，第三次还失败就认定它卡住了。
3.2.6 重复动作检测
最难缠的一种死循环是「原地打转」：模型每次都要调同一个工具、传同样的参数，比如反复搜同一个查不到的词。它每一步都在「正常」调用工具，max_steps 拦得住，但拦得太晚。
检测方法是给每次工具调用算一个签名，和上一次比对：
def _signature(tc: ToolCall) -> tuple[str, str]:
return tc.name, json.dumps(tc.arguments, sort_keys=True)
sort_keys=True 保证 {"a": 1, "b": 2} 和 {"b": 2, "a": 1} 被视为同一个签名。只比较相邻两次——因为「查 A、查 B、再查 A」是合理的工作流，而「查 A、查 A」才是打转：
sig = _signature(tc)
if sig == last_sig:
raise AgentLoopError("repeated_action", RunResult(messages, steps, ""))
last_sig = sig
这是一个极简但有效的启发式。它的漏网之鱼是「周期更长的循环」（A、B、A、B……），那种要用窗口内的集合去重，留到第 14 章讲重试与幂等时再展开。
3.2.7 把失败变成可诊断的数据
五条路径里有四条是异常终止。与其让它们各自抛不同的内置异常，不如统一成一个自定义异常，并把已经产生的部分结果挂上去：
class AgentLoopError(RuntimeError):
"""循环异常终止。携带 reason 与已产生的部分结果，便于诊断。"""
def __init__(self, reason: str, result: RunResult) -> None:
super().__init__(reason)
self.reason = reason
self.result = result
reason 是机器可读的字符串（max_steps / max_tokens / tool_failures / repeated_action），调用方据此决定后续动作；result 是出问题时已经走过的轨迹，日志、监控、调试全靠它。这就是「失败即数据」：错误不再是程序崩溃的终点，而是带着上下文的一个值。
完整的 run() 把五条路径串起来：
# minagent/loop.py
def run(model: Model, registry: ToolRegistry, user_input: str,
system: str = "", max_steps: int = 8,
max_consecutive_failures: int = 3) -> RunResult:
messages: list[Message] = []
if system:
messages.append(Message(role="system", content=system))
messages.append(Message(role="user", content=user_input))
steps: list[Step] = []
last_sig: tuple[str, str] | None = None
failures = 0
for i in range(max_steps):
resp = model.complete(messages, registry.schemas())
msg = resp.message
messages.append(msg)
if resp.stop_reason == "max_tokens":
raise AgentLoopError("max_tokens", RunResult(messages, steps, msg.content))
if not msg.tool_calls:
steps.append(Step(index=i, kind="final", detail=msg.content))
return RunResult(messages=messages, steps=steps, final=msg.content)
for tc in msg.tool_calls:
sig = _signature(tc)
if sig == last_sig:
raise AgentLoopError("repeated_action", RunResult(messages, steps, ""))
last_sig = sig
result = registry.call(tc)
messages.append(Message(role="tool", content=result, tool_call_id=tc.id))
steps.append(Step(index=i, kind="tool_call",
detail=f"{tc.name}({tc.arguments}) -> {result}"))
failures = failures + 1 if '"error"' in result else 0
if failures >= max_consecutive_failures:
raise AgentLoopError("tool_failures", RunResult(messages, steps, ""))
raise AgentLoopError("max_steps", RunResult(messages, steps, ""))
3.2.8 跑一遍：死循环与重复动作
验证止损线，最好的办法是故意制造它们。先写一个永远返回同一个工具调用的模型：
class LoopingModel:
"""永远返回同一个 tool_call。"""
def __init__(self, tc: ToolCall) -> None:
self.tc = tc
def complete(self, messages, tools) -> ModelResponse:
return ModelResponse(message=Message(role="assistant", tool_calls=[self.tc]),
stop_reason="tool_use")
再写一个「参数每次都变」的模型，用来绕过重复检测、单独验证 max_steps：
class CounterModel:
"""参数每次都变，躲过重复检测，但永不收敛。"""
def __init__(self) -> None:
self.n = 0
def complete(self, messages, tools) -> ModelResponse:
self.n += 1
tc = ToolCall(id=f"c{self.n}", name="echo", arguments={"text": f"turn {self.n}"})
return ModelResponse(message=Message(role="assistant", tool_calls=[tc]),
stop_reason="tool_use")
跑三种场景，看它们各自撞上哪条止损线：
try:
run(CounterModel(), registry(), "开始", max_steps=3)
except AgentLoopError as exc:
print("A) max_steps:", exc.reason, "| 已完成", len(exc.result.steps), "步")
same = ToolCall(id="c1", name="echo", arguments={"text": "ping"})
try:
run(LoopingModel(same), registry(), "开始", max_steps=8)
except AgentLoopError as exc:
print("B) repeated_action:", exc.reason, "| 已完成", len(exc.result.steps), "步")
truncated = ModelResponse(
message=Message(role="assistant", content="我先算加法得"),
stop_reason="max_tokens",
)
try:
run(FakeModel([truncated]), registry(), "请计算 (2+3)*4")
except AgentLoopError as exc:
print("C) max_tokens:", exc.reason, "| 保留文本", repr(exc.result.final))
输出：
A) max_steps: max_steps | 已完成 3 步
B) repeated_action: repeated_action | 已完成 1 步
C) max_tokens: max_tokens | 保留文本 '我先算加法得'
三个结果值得逐条品味：
- A 里
CounterModel每步参数都不同，重复检测抓不到它，于是老老实实跑满 3 步后被max_steps拦下，exc.result.steps恰好是 3 条。 - B 里
LoopingModel每次签名都一样，第二次就触发repeated_action，只走了 1 步就停——比等到max_steps=8早了 7 步，省下的都是真金白银。 - C 里模型没说完整就被截断，循环靠
stop_reason提前收手，并把半截文本保留在exc.result.final里供排查。
三个场景都证明了一件事：止损不是靠运气，而是靠显式写死的条件。每一条路径都有对应的假模型可以复现，也就意味着每一条路径都可以写进自动化测试。
3.2.9 真实模型下会怎样
假模型是「为演示而生」的：CounterModel 无限循环、LoopingModel 无限重复，都是极端构造。真实模型的行为处在两个极端之间——它通常会收敛，但会偶尔陷入打转；它通常输出完整，但偶尔会被 max_tokens 截断。所以这些止损线平时看起来「从没触发过」，恰恰是因为它们在暗中兜底。
真实场景里还有两个假模型模拟不出来的失败模式。其一是语义重复：模型换了措辞、改了参数名（q 变成 query），签名比对认不出来，但实质上在问同一个问题——这需要更聪明的相似度判断。其二是工具失败但模型不自知：它拿到 {"error": ...} 后不纠正，反而把错误当成正常结果继续推进。这两种都只能靠更精细的策略去覆盖，第 14 章会系统展开。
但方向是对的：先把能显式判断的都判断掉，把异常变成带 reason 的数据。剩下的模糊地带，才是真正需要「智能」去处理的部分。
run() 现在既会跑也会停，但它是函数式的——每次调用都要重传 model、registry、system，状态也散在局部变量里。下一节 3.3 抽成 Agent 类：状态、步骤与回调
会把它收进一个对象，让配置、历史与回调各归其位。想回看上一节的循环骨架，见 3.1 朴素循环：跑到没有工具调用为止
。
阅读导航：上一节：3.1 朴素循环：跑到没有工具调用为止 · 下一节：3.3 抽成 Agent 类：状态、步骤与回调 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。