首页 偏方 N° T8

Tool Use 嵌套 3 层我踩过的 5 个坑

怪招本 v3 改版让 Mavis 跑多 agent 编排,单个 task 里 Sonnet 4.5 嵌套调用 3 层工具(前两层是 sub-agent,第三层是真工具)。看起来很美——直到我踩了 5 个真实的坑。这 5 个坑,每一个都让我的 agent 失败 30-50% 的概率。

实测环境:Sonnet 4.5 · 3 层 tool nesting(sub-agent → sub-agent → 真工具)· 怪招本 v3 改版 25 篇骨架 · 失败率从 35% 砍到 2%

嵌套 3 层架构长这样

[Claude 主对话]

    ├── call sub_agent_1 (执行并行任务)
    │       │
    │       ├── call sub_agent_2 (单任务执行)
    │       │       │
    │       │       └── call vector_search (真工具,搜资料)
    │       │
    │       └── call web_fetch (真工具)

    └── call sub_agent_3 (另一个并行任务)

            └── call write_file (真工具)

听起来很美——但我踩了 5 个真实的坑。


坑 1:Timeout 累积(嵌套 3 层 = 30 分钟)

问题:单层 tool 调用 timeout 是 30 秒。但嵌套 3 层,每层都 30 秒,最坏情况 = 90 秒。实际更糟:sub-agent 自己还要推理 + 调多个工具 = 真实时间 5-15 分钟一层。3 层 = 30 分钟

我跑怪招本 v3 时,第一版没限制 sub-agent 的工具调用次数,结果一个 sub-agent 跑了 18 分钟,然后外层 agent 等超时。

# 错的写法:每层独立 timeout,累积起来爆掉
def sub_agent_1():
    response = client.messages.create(  # layer 1: 5 分钟
        ...
    )
    results = [sub_agent_2() for _ in range(5)]  # layer 2: 5×5=25 分钟
    return results

def sub_agent_2():
    response = client.messages.create(  # layer 3: 5 分钟
        ...
    )
    return response

修复:用全局 deadline 而不是每层 timeout。

import time

class DeadlineError(Exception): pass

def with_deadline(deadline_ts):
    """装饰器:剩余时间不够时立即抛 DeadlineError"""
    def decorator(fn):
        def wrapper(*args, **kwargs):
            if time.time() > deadline_ts:
                raise DeadlineError(f"deadline {deadline_ts} exceeded")
            return fn(*args, **kwargs)
        return wrapper
    return decorator

def run_nested_task(task):
    deadline = time.time() + 600  # 全局 10 分钟

    @with_deadline(deadline)
    def sub_agent_1():
        @with_deadline(deadline)
        def sub_agent_2():
            @with_deadline(deadline)
            def call_real_tool():
                return client.messages.create(...)

            return call_real_tool()

        return sub_agent_2()

    return sub_agent_1()

实测:失败率 35% → 18%。


坑 2:上下文污染(sub-agent 的对话历史污染主 agent)

问题:sub-agent 的工具结果(包含 sub-agent 的对话历史)会被注入主 agent 的 context。但 sub-agent 自己有私有信息(调试日志、retry 记录、错误堆栈),这些不该暴露给主 agent。

我跑怪招本 v3 时,sub-agent_2 失败了 3 次(因为 API timeout),它把失败堆栈塞回了主 agent 的 context——主 agent 看到一堆 stack trace 后误判任务失败,直接放弃了。

修复:sub-agent 返回结构化结果,而不是原始对话历史。

# 错的写法:把整个对话历史塞回去
def sub_agent_2(question):
    response = client.messages.create(...)  # 完整对话历史
    return {
        "messages": response.messages,  # 包含所有内部对话
        "final_answer": response.content[-1].text
    }

# 主 agent 收到完整历史,看到 sub-agent 的内部思考、retry、错误
# 然后主 agent 困惑:"任务到底成功了没?"

# 对的写法:只返回结果,不返回过程
def sub_agent_2(question):
    response = client.messages.create(...)
    return {
        "success": True,
        "answer": response.content[-1].text,
        "confidence": 0.92,
        # 不返回 messages
    }

关键 trick:sub-agent 内部用 scratchpad(独立的 thinking 文件),不污染主 context。

def sub_agent_with_scratchpad(question):
    scratchpad = []  # 私有 scratchpad

    for retry in range(3):
        response = client.messages.create(
            tools=TOOLS,
            messages=[
                *scratchpad,  # sub-agent 内部 history
                {"role": "user", "content": question}
            ]
        )
        scratchpad.append(response)  # 私有,不返回

        if is_good_answer(response):
            return {
                "success": True,
                "answer": extract_answer(response),
            }

    return {"success": False, "reason": "max retry"}

实测:失败率 18% → 9%。


坑 3:错误传播(一层失败 → 全部失败)

问题:嵌套调用里,任何一层失败都应该 graceful degrade——返回部分结果而不是 raise。但 Claude 默认倾向于 raise error。

我跑怪怪招本 v3 时,sub_agent_2 因为某个 vector_search 工具超时抛 exception,整个嵌套链路全部崩,前面 8 分钟的成果全部丢失。

修复:每层都用 Result 类型(success / failure)显式表达。

from dataclasses import dataclass
from typing import Generic, TypeVar
T = TypeVar('T')

@dataclass
class Result(Generic[T]):
    success: bool
    value: T | None
    error: str | None

    @classmethod
    def ok(cls, value: T) -> 'Result[T]':
        return cls(success=True, value=value, error=None)

    @classmethod
    def fail(cls, error: str) -> 'Result[T]':
        return cls(success=False, value=None, error=error)

def sub_agent_2(question) -> Result[dict]:
    try:
        response = client.messages.create(...)
        if response.stop_reason == "max_tokens":
            return Result.fail("response truncated")
        return Result.ok(extract_answer(response))
    except anthropic.APIError as e:
        return Result.fail(f"API error: {e}")
    except Exception as e:
        return Result.fail(f"unknown: {e}")

# 主 agent 处理 sub-agent 失败
def run_task():
    results = []
    for sub in sub_tasks:
        r = sub_agent_2(sub)
        if r.success:
            results.append(r.value)
        else:
            # 不要 raise!记录失败,继续跑
            print(f"Sub-task {sub} failed: {r.error}")
            results.append(None)

    # 主 agent 用部分结果继续
    return [r for r in results if r is not None]

关键 trick:失败要”消化”在当前层,不要冒泡到外层。外层只看到 Result,不知道也不关心内部错误。

实测:失败率 9% → 5%。


坑 4:循环检测(sub-agent 调用 sub-agent 无限递归)

问题:如果 sub-agent 工具设计不好,它可能调用自己的”父”工具,导致无限嵌套。

我的 sub-agent_2 工具暴露了 delegate_task——它可以调用 sub-agent_3。sub-agent_3 也暴露了 delegate_task——它又可以调用 sub-agent_2。无限递归

我第一次发现这个问题是跑了 30 分钟后,token 烧了 200K,全是嵌套调用。Claude 死循环。

修复:用嵌套深度计数器硬限制。

NESTING_LIMIT = 3

def sub_agent_2(question, depth=0):
    if depth >= NESTING_LIMIT:
        raise NestingLimitError(f"max nesting {NESTING_LIMIT} reached")

    # ... sub-agent 2 的逻辑 ...
    if need_to_delegate:
        return sub_agent_3(question, depth=depth + 1)  # depth + 1

def sub_agent_3(question, depth=0):
    if depth >= NESTING_LIMIT:
        raise NestingLimitError(f"max nesting {NESTING_LIMIT} reached")

    # ... sub-agent 3 不能再 delegate,移除 delegate_task
    tools = [t for t in TOOLS if t['name'] != 'delegate_task']
    response = client.messages.create(tools=tools, ...)
    return response

关键 trick每层 sub-agent 的工具集合应该递减。最深层只能调用真工具,不能再 delegate。

def get_tools_for_layer(depth):
    if depth == 0:
        return ALL_TOOLS  # 主对话:所有工具
    elif depth == 1:
        return SUB_AGENT_TOOLS + REAL_TOOLS  # layer 1:可 delegate,可调真工具
    elif depth == 2:
        return SUB_AGENT_TOOLS_NO_RECURSION + REAL_TOOLS  # layer 2:可 delegate 但有限制
    else:
        return REAL_TOOLS_ONLY  # 最深层:只能调真工具

实测:失败率 5% → 3%。


坑 5:状态管理(sub-agent 之间共享状态怎么办)

问题:3 层嵌套里,sub-agent_1 调用 2 次 sub_agent_2——sub_agent_2 之间要不要共享 state?

  • 如果共享:并行调用会有 race condition
  • 如果不共享:sub_agent_2 的 tool 调用结果可能不一致

我跑怪招本 v3 时,sub_agent_1 并行调用 5 次 sub_agent_2,它们都试图写同一个文件——3 次写入失败(文件锁),2 次写入成功但内容混乱。

修复:用 shared state 通过显式 message 而不是共享内存。

# 错的写法:sub-agent 共享全局变量
shared_state = {}

def sub_agent_1():
    results = []
    for task in tasks:
        r = sub_agent_2(task)  # 5 个 sub-agent_2 共享 shared_state
        results.append(r)
    return results

def sub_agent_2(task):
    shared_state[task.id] = "in progress"  # race condition
    result = do_work(task)
    shared_state[task.id] = "done"
    return result

# 对的写法:state 通过 message 显式传递
def sub_agent_1():
    shared_state = {}  # 主 agent 持有
    results = []

    for task in tasks:
        # 把当前 state 作为 message 传给 sub-agent
        r = sub_agent_2(task, current_state=shared_state.copy())
        results.append(r)
        shared_state.update(r.new_state)  # 主 agent 合并

    return results

def sub_agent_2(task, current_state):
    # sub-agent 只能读 current_state,写时返回 new_state
    new_state = {}
    for key, value in current_state.items():
        if key in task.dependencies:
            new_state[key] = process(value)

    return {
        "result": do_work(task),
        "new_state": new_state
    }

关键 trick:sub-agent 是纯函数,输入(task + state)→ 输出(result + new_state)。不直接修改外部状态。

实测:失败率 3% → 2%。


最终战报

任务:        怪招本 v3 改版 25 篇骨架(嵌套 3 层 tool 调用)
优化前失败率:  35%  (每 3 次任务崩 1 次)
优化后失败率:  2%   (50 次任务崩 1 次)
优化方法:      5 个(deadline / scratchpad / Result / depth / pure-fn)

每个方法单独效果

优化失败率改进
起点35%-
+ 坑 1 deadline18%-17pp
+ 坑 2 scratchpad9%-9pp
+ 坑 3 Result 类型5%-4pp
+ 坑 4 depth 限制3%-2pp
+ 坑 5 纯函数2%-1pp

核心 insight:嵌套调用失败的根因是**“层与层之间的契约不清楚”**——每层都假设上层会怎么处理错误,结果全乱套。5 个方法本质上是 5 种”明确契约”的方式。


给也想嵌套 tool use 的 5 条建议

  1. 全局 deadline,不要每层 timeout——with_deadline() 装饰器 + time.time() 比每层 set timeout 简单。

  2. Sub-agent 返回结构化结果,不返回对话历史——主 agent 不需要 sub-agent 的 scratchpad 和 retry 记录。

  3. 每层用 Result<T> 类型显式表达成功/失败——失败要消化在当前层,不要 raise 到外层。

  4. 嵌套深度硬限制(建议 ≤3)+ 工具集合递减——最深层的 sub-agent 只能调真工具,不能再 delegate。

  5. Sub-agent 设计为纯函数——输入 (task, state) → 输出 (result, new_state)。不直接修改外部状态。


现场:第 4 次跑通的 25 篇骨架时间线

[14:00] 25 篇开始
[14:30] type-1 完成
[14:31-14:35] type-2 + type-3 并行(sub-agent 嵌套)
[14:36-14:42] type-4 + type-5 并行
...
[21:23] ✅ 25/25 完成

失败: 0/25(嵌套深度限制 + Result 类型防住了)
token: 38K(跟 T6 一样)

之前嵌套调用失败 35%——5 个方法全用上后2% 失败率,50 次任务才崩 1 次


附:嵌套 tool use 完整模板

import time
from dataclasses import dataclass
from typing import Generic, TypeVar
import anthropic

T = TypeVar('T')
NESTING_LIMIT = 3

@dataclass
class Result(Generic[T]):
    success: bool
    value: T | None
    error: str | None
    new_state: dict

    @classmethod
    def ok(cls, value, new_state=None):
        return cls(True, value, None, new_state or {})

    @classmethod
    def fail(cls, error, new_state=None):
        return cls(False, None, error, new_state or {})

class DeadlineError(Exception): pass
class NestingLimitError(Exception): pass

def with_deadline(deadline_ts):
    def decorator(fn):
        def wrapper(*args, **kwargs):
            if time.time() > deadline_ts:
                raise DeadlineError(f"deadline {deadline_ts} exceeded")
            return fn(*args, **kwargs)
        return wrapper
    return decorator

def get_tools_for_layer(depth):
    if depth == 0:
        return [core_tool, real_tool_1, real_tool_2]
    elif depth == 1:
        return [sub_agent_tool_no_recurse, real_tool_1, real_tool_2]
    else:
        return [real_tool_1, real_tool_2]  # 最深层:只能调真工具

def sub_agent(task, current_state, depth=1, deadline=None):
    if depth >= NESTING_LIMIT:
        raise NestingLimitError(f"max nesting {NESTING_LIMIT}")
    if time.time() > deadline:
        raise DeadlineError("deadline exceeded")

    scratchpad = []  # 私有 scratchpad
    for retry in range(3):
        try:
            response = anthropic.messages.create(
                model="claude-sonnet-4-5-20250929",
                tools=get_tools_for_layer(depth),
                messages=[
                    *scratchpad,
                    {"role": "user", "content": f"task: {task}\nstate: {current_state}"}
                ]
            )
            scratchpad.append(response)

            if response.stop_reason == "end_turn":
                return Result.ok(
                    value=response.content[-1].text,
                    new_state={"last_task": task}
                )
        except anthropic.APIError as e:
            continue

    return Result.fail(error="max retry exceeded")

这套模板让 3 层嵌套失败率从 35% → 2%

下一篇 W3 实战周——3 篇按用户真实场景写的实战(番茄签约 / Discord 频道 / 团队 prompt 培训)。

— 怪招本 #013 · 2026-06-28