Claude 并行工具调用：一次问完 4 个香港数据源，少回一条 tool_result 直接报错

patrickstar231

673人浏览 · 2026-09-13 21:15:00

patrickstar231 · 2026-09-13 21:15:00 发布

在这里插入图片描述

文章目录

1. 先说结论：一批调用，三条硬契约

Claude 默认可能在一次回应里同时调用多个工具。响应里 stop_reason 是 tool_use，content 数组里可以躺着好几个 tool_use 区块——这一点很多人知道。真正容易翻车的是回填那一步。

官方规范把这个动作写得很死，归纳成三条：

#	契约	原文
1	每个 `tool_use` 都要拿到一个 `tool_result`，全部放在下一条 user 消息里	return one `tool_result` for each `tool_use` block, all together in the next user message
2	每个 `tool_result` 必须排在该消息中任何 text 之前	put every `tool_result` block before any text content in that message
3	没执行的那个调用也要回填，带 `is_error: true`	still return a `tool_result` for it with `is_error: true`

第 3 条最反直觉：一个调用因为上游失败而根本没跑，很多人会顺手跳过它——跳过就等着报错。官方给的错误信息长这样：

tool_use ids were found without tool_result blocks immediately after

还有一条不在契约里、但同样吃时间的事实：并发不是万能药。它只在"整批调用都没有长尾"时成立。我实测了一批香港官方端点，4 个独立只读接口串行 0.62 秒、并发 0.23 秒（×2.75）；但同一批里只要混进一个慢接口，加速比立刻掉到 ×1.20。

下面把数据源、执行器代码、三条契约的本地校验、以及一条真实的依赖链完整拆开。

2. 环境与数据源

Python 3.13.12 (macOS)
matplotlib 3.11.1
只用标准库：json / subprocess / time / concurrent.futures

工具层挂了 4 个零鉴权香港公开端点，都是日常真在用的：

工具名	数据	端点
`hk_weather_now`	实时天气	`data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType=rhrread`
`hk_public_holidays`	公众假期	`www.1823.gov.hk/common/ical/tc.json`
`hk_aqhi_now`	空气质量健康指数	`dashboard.data.gov.hk/api/aqhi-individual`
`hk_forecast_9day`	九天预报	`data.weather.gov.hk/weatherAPI/opendata/weather.php?dataType=fnd`

选它们的理由很直接：互相独立、只读、无顺序要求——正是官方说的"通常可以安全并行"那一类。反例见第 9 节。

3. 并发到底省多少时间

在这里插入图片描述

左半组是 4 个独立只读端点：串行 3 轮中位数 0.624s，并发 0.227s，加速 ×2.75。差距来自网络往返被叠在了一起。

右半组是同一批再加一个慢接口（小巴路线清单）：串行中位 24.883s，并发中位 20.781s，加速只有 ×1.20——整批耗时被最慢的那个工具吃掉了，并发只是把其余 4 个 0.1–1.2 秒的等待藏进了它的影子里。

这里还有个更值得记的细节：混合组串行的 3 轮原始值是 1.43 / 24.88 / 25.50 秒。同一个端点，前后两分钟差 17 倍。所以在这种批次里平均值没有意义，必须看中位数和分布——这也解释了为什么并发节省是"有时 3 倍、有时 1.2 倍"，取决于那一轮有没有踩到慢态。

4. 批次执行器：分组、并发、回填

先写工具层。http_json 统一返回 (payload, error) 二元组，让异常在边界收口；is_silent_empty 专门盯一种"假成功"，第 10 节会讲它为什么必要。

import json, subprocess, time
from concurrent.futures import ThreadPoolExecutor

TOOLS = {
    "hk_weather_now":     {"url": "https://data.weather.gov.hk/weatherAPI/opendata/weather.php",
                           "query": {"dataType": "rhrread", "lang": "tc"},
                           "parallel_safe": True, "depends_on": []},
    "hk_public_holidays": {"url": "https://www.1823.gov.hk/common/ical/tc.json",
                           "query": {},
                           "parallel_safe": True, "depends_on": []},
    "hk_aqhi_now":        {"url": "https://dashboard.data.gov.hk/api/aqhi-individual",
                           "query": {"format": "json"},
                           "parallel_safe": True, "depends_on": []},
    "hk_forecast_9day":   {"url": "https://data.weather.gov.hk/weatherAPI/opendata/weather.php",
                           "query": {"dataType": "fnd", "lang": "tc"},
                           "parallel_safe": True, "depends_on": []},
}

def http_json(tool, args, timeout=40):
    """返回 (payload, error)；error 非 None 时 payload 恒为 None。"""
    spec = TOOLS[tool]
    url = spec["url"].format(**args)
    if spec["query"]:
        url += "?" + "&".join(f"{k}={v}" for k, v in spec["query"].items())
    out = subprocess.run(["curl", "-sL", "-m", str(timeout), "-w", "\n%{http_code}", url],
                         capture_output=True, text=True).stdout
    body, _, code = out.rpartition("\n")
    if code.strip() != "200":
        return None, f"HTTP {code.strip() or '000'}"
    try:
        return json.loads(body.lstrip("\ufeff")), None   # 这批接口有带 BOM 的
    except ValueError as exc:
        return None, f"not-json: {exc}"

def is_silent_empty(payload):
    """HTTP 200，但业务数据是空的。工具层不报错，只能在这一层判定。"""
    if not isinstance(payload, dict):
        return False
    if payload.get("data") in ({}, []):
        return True
    inner = payload.get("data")
    if isinstance(inner, dict):
        for key in ("routes", "route_stops"):
            if key in inner and not inner[key]:
                return True
    return False

执行器本身不长，关键在最后两行断言——契约是可以被代码强制的，不用靠人记：

def _wrap(tu, payload, err):
    if err:
        return {"type": "tool_result", "tool_use_id": tu["id"], "is_error": True,
                "content": f"{tu['name']} failed: {err}"}
    if is_silent_empty(payload):
        return {"type": "tool_result", "tool_use_id": tu["id"], "is_error": True,
                "content": f"{tu['name']} returned HTTP 200 but no business data."}
    return {"type": "tool_result", "tool_use_id": tu["id"], "is_error": False,
            "content": json.dumps(payload, ensure_ascii=False)[:800]}

def run_batch(tool_uses):
    """执行一批 tool_use，返回带契约保证的 tool_results。"""
    order = [tu["id"] for tu in tool_uses]
    group = [tu for tu in tool_uses
             if TOOLS.get(tu["name"], {}).get("parallel_safe")
             and not TOOLS.get(tu["name"], {}).get("depends_on")]
    chained = [tu for tu in tool_uses if tu not in group]

    results, done = {}, {}                   # done 按工具名记成功过的调用
    if group:
        with ThreadPoolExecutor(max_workers=len(group)) as ex:
            for tu, payload, err in ex.map(
                    lambda t: (t, *http_json(t["name"], t.get("input") or {})), group):
                results[tu["id"]] = _wrap(tu, payload, err)
                if not results[tu["id"]]["is_error"]:
                    done[tu["name"]] = True

    for tu in chained:                       # 有依赖的按序走，依赖没满足也照样回填
        deps = TOOLS.get(tu["name"], {}).get("depends_on", [])
        if [d for d in deps if d not in done]:
            results[tu["id"]] = {"type": "tool_result", "tool_use_id": tu["id"],
                                 "is_error": True,
                                 "content": "Not executed: the preceding call did not "
                                            "return usable data."}
            continue
        payload, err = http_json(tu["name"], tu.get("input") or {})
        results[tu["id"]] = _wrap(tu, payload, err)

    ordered = [results[i] for i in order]    # 按输入顺序重建，一个都不能少
    assert len(ordered) == len(tool_uses)
    assert [r["tool_use_id"] for r in ordered] == order
    return {"tool_results": ordered,
            "n_error": sum(1 for r in ordered if r["is_error"])}

这套执行器可以直接拿走用，建议收藏备用——换成任何一批工具，只要改 TOOLS 里那张表；parallel_safe 与 depends_on 两个字段就是分组依据，其余代码不用动。

5. 契约一：一个都不能少

ordered = [results[i] for i in order] 这一行看起来多余——results 里本来就有全部结果。但如果某条分支忘了写入 results，字典取值会直接 KeyError，在本地就炸，而不是等 API 返回 400。这就是把契约写进代码的价值：错误暴露在成本最低的那一层。

原因很实在：tool_result 必须全部塞进同一条 user 消息。拆成两条、每条回一个，模型就没法把这些结果和上一轮的调用对上。Troubleshooting 页里「并行调用不生效」这一栏写的正是这条：

Send multiple tool_result blocks in ONE user message, not one per turn.

6. 契约二：result 必须在 text 之前

这条最容易被忽略，因为写完 tool_result 顺手补一句"以上是结果"是很自然的动作——但这句话必须放在所有 tool_result 之后：

def build_user_message(tool_results, text=None):
    """构造回填消息：所有 tool_result 排在任何 text 之前。"""
    content = list(tool_results)
    if text:
        content.append({"type": "text", "text": text})
    return {"role": "user", "content": content}

顺序错了不会静默降级，而是直接失败——解析器只认"result 在前"这一种形态。

7. 契约三：没跑的也要回填

这条是三条里最反直觉、也最容易漏的。

场景很常见：一批调用里有依赖关系，第 1 步失败了，第 2 步压根没跑。直觉是"没跑就没有结果，跳过"。但官方的要求是照样回填一条，标记 is_error: true，并说明为什么没跑：

{
  "type": "tool_result",
  "tool_use_id": "toolu_02",
  "is_error": true,
  "content": "Not executed: the preceding write_file call failed."
}

道理也顺：模型只能通过 tool_result 了解每次调用的下场。跳过等于留一个永远不会兑现的悬空 id。"没执行"本身就是结果，而且是模型下一步决策必须知道的结果——它要据此判断是重试、换参数，还是放弃整条链。

8. 把三条契约变成一次本地校验

四处散着记三条规则容易漏，写成一个校验器更省事：

OFFICIAL_ERR = "tool_use ids were found without tool_result blocks immediately after"

def validate_history(history):
    """检查每个 assistant 回合的 tool_use 是否都拿到了回填。"""
    problems = []
    for i, msg in enumerate(history):
        if msg.get("role") != "assistant" or not isinstance(msg.get("content"), list):
            continue
        ids = [b["id"] for b in msg["content"]
               if isinstance(b, dict) and b.get("type") == "tool_use"]
        if not ids:
            continue
        nxt = history[i + 1] if i + 1 < len(history) else None
        if not nxt or nxt.get("role") != "user":
            problems.append({"at": i, "kind": "no_next_user_message",
                             "official_error": OFFICIAL_ERR})
            continue
        blocks = nxt.get("content") if isinstance(nxt.get("content"), list) else []
        returned = [b.get("tool_use_id") for b in blocks
                    if isinstance(b, dict) and b.get("type") == "tool_result"]
        missing = [x for x in ids if x not in returned]
        if missing:
            problems.append({"at": i, "kind": "missing_tool_result",
                             "detail": missing, "official_error": OFFICIAL_ERR})
        for j, b in enumerate(blocks):           # text 之后不得再出现 tool_result
            if b.get("type") == "text" and any(
                    x.get("type") == "tool_result" for x in blocks[j + 1:]):
                problems.append({"at": i, "kind": "text_before_tool_result",
                                 "official_error": OFFICIAL_ERR})
                break
    return problems

这段校验器建议收藏，任何多工具编排都能直接复用。拿四种历史形态喂进去，结果是这样：

在这里插入图片描述

四种写法里只有第一种通过。注意第 2 种和第 3 种报的是同一个类别——“拆成两条消息各回一个"在结构上等价于"只回了一个”，因为它们都让某个 tool_use_id 在自己的下一条消息里没拿到结果。

9. 什么时候不能并发：一条真实的依赖链

parallel_safe 这个标记不是随便打的。反例用香港小巴的真实接口来演示——它的数据是三段链式的：

# 第 1 步：拿到路线清单（108 条）
#   GET /route/HKI                 -> {"data": {"routes": ["1", "10", "10P", ...]}}
# 第 2 步：用清单里的路线代号取详情，才拿得到 route_id
#   GET /route/HKI/1               -> {"data": [{"route_id": 2006408, ...}]}
# 第 3 步：route_id 只能来自第 2 步
#   GET /route-stop/{route_id}/{route_seq}  -> 沿途站点

在这里插入图片描述

实测这条链走完是 64.5 秒（20.8 + 21.0 + 22.7），每一步的参数都来自上一步的返回：

第 2 步的 route_code 取自第 1 步返回的 108 条清单；
第 3 步的 route_id=2006408 由第 2 步给出，模型不可能凭空猜对。

这类调用没有办法并发——不是"并发会慢一点"，而是并发时第 2 步根本没有参数可用。把它和前面那批只读接口放进同一个 tool_uses 列表时，执行器靠 depends_on 字段把它们分到串行组，并发组照常并行，两者互不拖累。

判据一句话：参数是否来自同批次其他调用的返回。是，就必须串行；否，就可以并行。官方对这条的表述是"有副作用、共享状态或顺序要求的工具，更适合按顺序执行"，并且指出 computer use / browser use 这类工具更严格——必须按出现顺序串行，且在第一次失败处停止。

10. 更隐蔽的坑：HTTP 200，但 data 是空的

最后一个坑和第 3 条契约是配套的。

我用一个不存在的路线去请求城巴接口，得到的是：

HTTP 200   99 字节
{"type": "Route", "version": "2.0", "generated_timestamp": "...", "data": {}}

状态码是 200，响应是合法 JSON，只是 data 里什么都没有。 如果工具层只判断 status == 200，这次调用会被记成"成功"，交给模型的是一句"查到了"——而实际什么都没查到。

这正是 is_silent_empty 存在的理由："接口正常返回"和"业务上有数据"是两件事，前者查状态码，后者得看结构。这类"假成功"在并行批次里更危险，因为一个静默失败会被另外几条成功结果盖住，你不去逐条看根本发现不了。

11. 踩坑清单与几条结论

坑	现象	处理
只回了部分结果	报 `tool_use ids were found without tool_result blocks immediately after`	每个 `tool_use` 都要一条，全部放进同一条 user 消息
把回填拆成多个回合	并行失效，模型对不上结果	一条 user 消息装完全部 `tool_result`
text 排在 result 之前	消息被判定为格式错误	`tool_result` 全部前置，正文放最后
跳过错过的调用	历史里留下悬空 id	也要回填 `is_error: true` 并写明原因
以为并发一定更快	混入慢接口后加速比从 ×2.75 掉到 ×1.20	先看该批有没有长尾，再决定并发还是串行
`depends_on` 漏标	并发时后续调用拿不到参数	参数来自同批其他返回的，一律进串行组
只看 HTTP 状态码	200 + `{"data": {}}` 被当成成功	加一层空结构判定，判成 `is_error`

值得记住的三条：

契约要写进代码，不要写在文档里。 三条规则里两条都可以用两行断言兜住——assert len(ordered) == len(tool_uses) 和一条"必须写进 results"的取值，漏了就在本地炸。
并发是长尾的函数，不是调用数的函数。 这一批里 hk_gmb_routes 单次要 21 秒，它一个人决定了整批的墙钟；把另外 4 个并发起来，省下的 0.6 秒在大数面前没有意义。
"没执行"和"没数据"都是结果。 前者要回填 is_error 说明原因，后者要在工具层判空——两者都不能让模型收到一句含糊的"查到了"。

可复用的三块：批次执行器（parallel_safe + depends_on 分组）、历史契约校验器 validate_history、空结构判定 is_silent_empty。如果这篇对你有用，收藏 + 点赞——下次接一批数据源时，可以直接把这三块搬过去。

这类多工具编排的实测会继续更新，关注不迷路。