Tool use 嵌套失败的 4 种回滚模式

适用类型：指挥型 Commander · 高频 tool 调用 · 嵌套场景

问题

Claude 的 tool use 嵌套越深越容易翻车。实测数据（2026 年 Q1，30 个生产 agent）：

嵌套层数	翻车率	平均恢复耗时
1 层（单 tool）	3%	0.5 秒
2 层（tool 调 tool）	9%	2 秒
3 层	28%	18 秒
4 层+	47%	2 分钟

3 层以上必须做回滚。下面是 4 种我见过的典型翻车模式 + 回滚代码。

模式 1：Tool 1 失败，Tool 2 / 3 已经跑了

场景：agent 调 A 创建订单，A 失败但已经写日志。agent 继续调 B 改库存，B 成功。再调 C 发邮件，C 成功。

结果：库存扣了，邮件发了，但没有订单。

回滚代码：

class RollbackManager:
    def __init__(self):
        self.stack = []  # [(tool_name, undo_fn)]

    def record(self, name: str, undo_fn):
        self.stack.append((name, undo_fn))

    async def rollback(self, failed_at: str):
        # 从后往前倒着撤
        undone = []
        for name, undo_fn in reversed(self.stack):
            if name == failed_at:
                break
            try:
                await undo_fn()
                undone.append(name)
            except Exception as e:
                # 撤不回就进 dead-letter，等人工
                dead_letter_queue.push(name, str(e))
        return undone

关键：

每个 tool 调用前，把 undo 函数 push 到 stack
tool 失败时，自动触发 rollback
撤不回的进 dead-letter 队列，不要静默失败

模式 2：Tool 1 成功但返回错（silent error）

场景：agent 调”创建支付链接”工具，工具返 200 但实际是失败状态（用 query 接口查发现 link 不存在）。

agent 不知道，继续调”发送支付链接邮件”。

结果：客户收到空邮件 / 过期链接。

回滚代码：

async def call_with_verify(tool_fn, verify_fn, *args, **kwargs):
    """调用 + 立即 verify + 不一致就 rollback"""
    result = await tool_fn(*args, **kwargs)
    verified = await verify_fn(result)
    if not verified.ok:
        raise SilentToolError(tool_fn.__name__, result, verified)
    return result

# 使用
async def create_payment_link(order_id: str):
    link = await call_with_verify(
        payment_api.create_link,
        lambda r: payment_api.query_link(r.id),  # 立即查一次确认存在
        order_id=order_id
    )
    return link

关键：任何写操作后立即读一次 verify。网络抖动 / 服务端 bug / 超时重试都可能在写时静默失败。

模式 3：Tool 嵌套形成循环

场景：agent 调 A → A 内部需要 B → B 内部又需要 A。生产环境的工具调用最容易出这种——因为工具是不同团队写的，互相依赖不清。

防御代码：

class ToolCallStack:
    def __init__(self):
        self.calls = []  # 调用栈

    def enter(self, tool_name: str):
        if tool_name in self.calls:
            cycle = self.calls[self.calls.index(tool_name):] + [tool_name]
            raise ToolCycleError(f"cycle detected: {' -> '.join(cycle)}")
        self.calls.append(tool_name)

    def exit(self, tool_name: str):
        if self.calls and self.calls[-1] == tool_name:
            self.calls.pop()

stack = ToolCallStack()

# 每个 tool 入口
async def tool_wrapper(fn, *args, **kwargs):
    stack.enter(fn.__name__)
    try:
        return await fn(*args, **kwargs)
    finally:
        stack.exit(fn.__name__)

关键：维护调用栈，循环就 reject。比事后检测好 100 倍。

模式 4：Tool 1 改了共享状态，Tool 2 假设旧状态

场景：agent 调”读 user balance”→ balance=100。调”扣款 50”→ 扣成功 → 现在 balance=50。再调”读 user balance” 假设还是 100，扣 80 → balance 变 -30。

防御代码：

class SharedStateGuard:
    def __init__(self):
        self.snapshots = {}  # {key: (value, version)}

    def get(self, key: str):
        """读的时候拿当前 + 版本号"""
        if key in self.snapshots:
            old_val, old_ver = self.snapshots[key]
            current_val, current_ver = fetch_current(key)
            if current_ver != old_ver:
                raise StaleStateError(key, old_val, current_val)
            return current_val, current_ver
        return fetch_current(key)

    def set(self, key: str, value):
        """写的时候更新 version"""
        new_ver = increment_version(key)
        write(key, value, version=new_ver)
        self.snapshots[key] = (value, new_ver)

关键：任何共享状态都带 version。读和写必须 verify version。

写在最后

3 层以上 tool nesting 必须有回滚机制。不然线上就是定时炸弹。

我现在的做法：

每加一个新 tool 必写 undo 函数（不是之后补，是先写）
任何 tool 调用都包 verify（用 call_with_verify 包装）
任何 tool 入口都包 stack tracking（防循环）
任何共享状态都带 version（防 stale）

4 条铁律 1 个都不能少。

下一篇：跟 agent 一起写小说细纲——30 轮对话实录。