更新于

Claude 多轮工具调用循环健壮性设计：崩溃恢复与语义死循环防护

ClaudeTool UseAgent技术深度

Claude Tool Use 最佳实践与陷阱 已经讲过工具调用循环的标准骨架，以及用 max_loops、token 预算监控防止死循环的基本做法。但当 Agent 真正跑起来处理复杂任务（比如连续调用十几次工具完成一个多步骤任务）时，“设置个循环次数上限”只是及格线，还有几类更隐蔽的健壮性问题需要专门设计：模型陷入语义死循环（次数没超但一直在做无效功）、并行工具调用里部分成功部分失败、进程崩溃后如何续跑、以及用户中途想取消一个正在跑的长循环。这篇文章聚焦这几个更深层的问题。

一、次数限制防不住的”语义死循环”

max_loops 能防住的是”循环次数爆炸”，但防不住模型陷入一种更隐蔽的状态：用相同或高度相似的参数反复调用同一个工具，没有实质进展。比如模型反复调用同一个搜索工具、传入几乎一样的关键词，或者对同一个文件反复读取又反复编辑却没有收敛。这种情况往往在循环次数上限之内就已经浪费了大量 token 和时间。

检测语义死循环的思路是对每次工具调用做指纹化，记录最近 N 次调用的”签名”（工具名 + 参数的规范化哈希），如果连续出现重复签名，提前中断并把这个情况反馈给模型：

import hashlib
import json

def call_signature(tool_name, tool_input):
    normalized = json.dumps(tool_input, sort_keys=True)
    return hashlib.sha256(f"{tool_name}:{normalized}".encode()).hexdigest()

recent_signatures = []
REPEAT_THRESHOLD = 3

def check_semantic_loop(tool_name, tool_input):
    sig = call_signature(tool_name, tool_input)
    recent_signatures.append(sig)
    if len(recent_signatures) > REPEAT_THRESHOLD:
        recent_signatures.pop(0)
    if len(recent_signatures) == REPEAT_THRESHOLD and len(set(recent_signatures)) == 1:
        return True  # 检测到连续重复调用
    return False

检测到重复后，不建议直接掐断对话报错，而是把情况作为一条工具结果反馈给模型，让它有机会调整策略：

if check_semantic_loop(tool_name, tool_input):
    tool_result = "检测到你连续多次用相同参数调用了这个工具，请换一种方式解决问题，或者说明遇到了什么阻碍。"

这种”提醒而非终止”的处理方式，给了模型一次自我纠正的机会，比直接报错退出的用户体验更好，也能减少不必要的循环中断。

二、并行工具调用的部分失败处理

Claude 支持在一轮里返回多个 tool_use 块（并行调用），这部分的基础用法可以参考 Claude 并行工具调用实践。但并行调用引入了一个单个调用场景不存在的问题：几个工具里有的成功有的失败，该怎么把结果组织后传回模型。

正确的处理方式是每个工具调用都必须有对应的 tool_result，无论成功还是失败，不能因为某个调用失败就跳过它的结果——Anthropic API 要求每个 tool_use 都有匹配的 tool_result，缺失会导致下一轮请求报错。

async def execute_parallel_tools(tool_calls):
    results = []
    tasks = [execute_single_tool(tc) for tc in tool_calls]
    outcomes = await asyncio.gather(*tasks, return_exceptions=True)

    for tc, outcome in zip(tool_calls, outcomes):
        if isinstance(outcome, Exception):
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": f"工具执行失败：{str(outcome)}",
                "is_error": True,
            })
        else:
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": outcome,
            })
    return results

关键点是用 asyncio.gather(..., return_exceptions=True) 而不是让某个任务的异常直接冒泡中断整个批次——一个工具失败不应该连累其他并行调用的正常结果被丢弃。同时给失败的结果标记 is_error: True，让模型明确知道这是一次失败而不是正常的空结果，避免模型误判继续往下走。

三、进程崩溃后的断点续跑

长时间运行的工具调用循环（比如涉及十几轮、跨越几分钟到几十分钟的任务）如果中途进程崩溃或者服务重启，从头重跑不仅浪费已经消耗的 token，用户体验也很差。断点续跑的核心是把每一轮的完整消息历史持久化，而不是只在内存里维护：

def persist_loop_state(session_id, messages, loop_count):
    db.execute(
        "INSERT INTO agent_loop_state (session_id, messages, loop_count, updated_at) "
        "VALUES (%s, %s, %s, NOW()) "
        "ON CONFLICT (session_id) DO UPDATE SET messages = %s, loop_count = %s, updated_at = NOW()",
        (session_id, json.dumps(messages), loop_count, json.dumps(messages), loop_count)
    )

def resume_loop(session_id):
    row = db.query_one(
        "SELECT messages, loop_count FROM agent_loop_state WHERE session_id = %s", (session_id,)
    )
    if row:
        return json.loads(row["messages"]), row["loop_count"]
    return [], 0

每完成一轮工具调用（模型响应 + 工具执行结果都拿到之后）就落一次盘，而不是等整个任务完成才保存。这样即使在第 8 轮崩溃，重启后也能从第 8 轮的状态继续，而不是回到第 1 轮。需要注意的是落盘操作本身不能拖慢循环的关键路径太多，如果消息历史很大，可以只存增量而不是每次全量覆盖，或者用异步写入不阻塞主循环。

四、用户中途取消：优雅中断而不是暴力杀进程

长循环任务如果允许用户主动取消（比如意识到 Agent 走错方向），直接杀掉后端进程是最简单但最粗暴的方式——正在执行的工具调用（比如一次数据库写入或者一次外部 API 调用）可能停在不确定的中间状态。更稳妥的做法是用取消信号，在循环的检查点响应它，而不是硬杀：

async def run_tool_loop(session_id, cancel_event):
    messages, loop_count = resume_loop(session_id)

    while loop_count < MAX_LOOPS:
        if cancel_event.is_set():
            persist_loop_state(session_id, messages, loop_count)
            return {"status": "cancelled", "loop_count": loop_count}

        response = await client.messages.create(model="claude-sonnet-5", messages=messages, tools=TOOLS)

        if response.stop_reason != "tool_use":
            return {"status": "completed", "final_response": response}

        # 工具执行也应该是可取消的，避免卡在一个长时间运行的工具里
        tool_results = await execute_parallel_tools_with_cancel(response.content, cancel_event)
        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": tool_results})
        loop_count += 1
        persist_loop_state(session_id, messages, loop_count)

    return {"status": "max_loops_reached", "loop_count": loop_count}

取消检查点放在每轮循环开始时，以及每个工具执行内部（如果工具本身耗时较长，比如调用外部 API），这样用户点击取消后能在合理的时间窗口内响应，而不是必须等当前这一整轮完全跑完。已经落盘的状态也保证了”取消”和”崩溃恢复”复用同一套持久化机制，不需要维护两套逻辑。

五、健壮性设计的取舍：不是所有场景都需要全套机制

语义死循环检测、断点续跑、优雅取消这几项机制都有额外的工程成本（持久化存储、状态管理复杂度），不是所有场景都值得全上：

场景建议
单次交互式对话，循环轮次少（<5 轮）只需要基础的 max_loops，无需断点续跑
长时间后台任务（如批量代码重构）断点续跑必要，语义死循环检测强烈建议
允许用户主动干预的交互式 Agent优雅取消机制必要
高频、低价值的简单工具调用（如查询天气）全套机制反而是过度设计，简单的错误重试就够

判断标准很直接：单次任务失败的成本（用户等待时间、已消耗的 token、重跑的代价）越高，越值得投入这些健壮性机制；对于轻量级、可以快速重试的场景，这些机制带来的复杂度可能得不偿失。

六、和上游限速、降级策略的配合

多轮循环里每一轮都是一次独立的 API 请求，如果循环跑到一半触发了上游的限速或者临时故障，循环层面的重试逻辑应该和 LLM 多厂商 Fallback 设计 里的降级策略打通——单轮请求失败不代表整个任务失败，应该在当前这一轮做重试或切换备用模型，而不是让整个已经跑了大半的循环从头作废。这也是断点续跑机制的另一个应用场景：把”上游临时故障”和”进程崩溃”用同一套恢复逻辑处理。

七、小结

工具调用循环的健壮性不止是设置一个 max_loops 那么简单。语义死循环需要用调用签名去重检测；并行调用要保证每个 tool_use 都有对应结果，失败的调用不能连累其他调用被丢弃；长时间任务要考虑崩溃后从检查点恢复而不是从头重跑；允许用户干预的场景需要优雅取消而不是暴力杀进程。这几项机制不是必须全部实现，根据任务的时长和失败成本决定投入的程度。

八、相关阅读

如果你的 Agent 需要在多轮工具调用循环里做上游故障的自动降级切换，用 YoTradeApi 中转可以统一管理多个模型的调用，配合断点续跑机制减少长任务因单次请求失败而全部作废的情况。

相关阅读