并行工具调用与流式

并行工具调用与流式

本节摘要:三个独立的天气查询串行跑就是三次往返;并行跑,总时间塌缩到最慢的那一次。如今每家前沿厂商都能在单轮里发出多个工具调用。收益是真的,但管线很微妙。本节走通两半:并行扇出(parallel fan-out)与流式参数重组(streamed-argument reassembly),重点放在 id 关联陷阱上——当三个调用乱序完成,结果必须携带匹配的 tool_call_id;当结果流式抵达,你必须把参数碎片拼成完整 JSON 才能执行。

学习目标

阅读完本节,你应当能够:

1. 解释 parallel_tool_calls: true 为什么存在,以及何时该禁用它。
2. 在并行扇出时,把流式抵达的参数分块按 id 关联到正确的调用。
3. 把部分 arguments 字符串重组成完整 JSON,且绝不提前解析。
4. 跑一个三城市天气基准,演示串行 vs 并行的延迟差异。

一、问题与直觉

没有并行调用时,一个回答「班加罗尔、东京、苏黎世天气如何」的 Agent 会这么做:

user -> LLM
LLM -> 调用 get_weather(Bengaluru)
host -> 运行执行器,回复结果
LLM -> 调用 get_weather(Tokyo)
host -> 运行执行器,回复结果
LLM -> 调用 get_weather(Zurich)
host -> 运行执行器,回复结果
LLM -> 最终文本答案
​

三次 LLM 往返,每次还要付执行器延迟,墙钟时间约为理想值的 4 倍。

有了并行调用:

user -> LLM
LLM -> 调用 get_weather(Bengaluru); 调用 get_weather(Tokyo); 调用 get_weather(Zurich)
host -> 并发跑三个执行器,回复三个结果
LLM -> 最终文本答案
​

一次 LLM 往返,执行器时间取三者最大而非之和。三家厂商的生产基准显示,扇出负载下墙钟时间能减少 60%~70%。

代价是关联复杂度:三个调用乱序完成时,结果必须携带匹配的 tool_call_id,模型才能对上号;结果流式抵达时,你必须在执行前把参数碎片拼成完整 JSON。Gemini 3 加唯一 id,部分就是为了解决一个真实问题——同一工具的两个并行调用无法区分。

二、从零实现

启用并行

• OpenAI:parallel_tool_calls: true 默认开;设 false 强制串行。
• Anthropic:通过 disable_parallel_tool_use: false(Claude 3.5+ 默认)启用;设 true 走串行。
• Gemini:始终具备并行能力;tool_config.function_calling_config.mode = "AUTO" 让模型自决。

该禁用并行的场景:工具有顺序依赖(create_file 然后 write_file)、一个调用的输出影响另一个的输入、或限流器扛不住扇出。

id 关联

模型发出的每个调用都有一个 id;host 返回的每个结果必须带上同一个 id。没有它,结果就是歧义的:

• OpenAI:每条 tool 角色消息上的 tool_call_id。
• Anthropic:每个 tool_result 块上的 tool_use_id。
• Gemini:每个 functionResponse 上的 id(Gemini 3+;Gemini 2 按名字匹配,对同名并行调用会失效)。

并发执行调用

host 把每个调用的执行器跑在自己的线程、协程或远程 worker 上。最简单的脚手架用线程池;生产用 asyncio 的 asyncio.gather 或结构化并发。完成顺序不可预测——id 才是标识符。

一个常见 bug:按调用列表顺序而非完成顺序回复结果。这通常能跑,因为模型只认 tool_call_id,但若结果被丢弃或重复,乱序提交会让调试更难。优先按完成顺序、带显式 id 回复。

流式工具调用

模型流式输出时,arguments 是分块抵达的。三个并行调用的三股分块流在线上交错。你需要每个 id 一个累加器:

• OpenAI:每个 chunk 是 choices[0].delta.tool_calls[i].function.arguments(部分字符串),chunk 带 index(在调用列表中的位置)。你按 index 累加,等 id 首次出现时读出,在 finish_reason = "tool_calls" 时才解析 JSON。
• Anthropic:流事件是 message_start,然后每个块一个 content_block_start(类型 tool_use,含 id、name、空 input);content_block_delta 事件携带 input_json_delta 分块;content_block_stop 关闭每个块。
• Gemini:streamFunctionCallArguments(Gemini 3+)发射带 functionCallId 的分块,使调用干净交错;Gemini 3 之前,流式一次返回一个完整调用。

部分 JSON 与「提前解析」陷阱

arguments 在完整之前不能解析。像 {"city": "Beng 这样的部分 JSON 不合法,会抛异常。正确的闸门是厂商的调用结束信号:OpenAI 的 finish_reason = "tool_calls"、Anthropic 的 content_block_stop、或 Gemini 的流结束事件。只有到那一刻才尝试 json.loads。

更稳健的做法用增量 JSON 解析器,在结构完成时产出事件——OpenAI 的流式指南推荐这种,用于显示实时「思考中」指示器的 UX。数括号作为完整性测试不可靠(引号字符串内的括号或转义内容会造成假阳性),只能当非正式的调试启发式用。

乱序完成

# 三个调用完成顺序不可预测,id 是唯一可靠标识
results = []
for fut in concurrent.futures.as_completed([fa, fb, fc]):
    call_id, output = fut.result()
    results.append({"role": "tool", "tool_call_id": call_id, "content": output})
# 回复顺序对 OpenAI/Anthropic 正确性无关,Gemini 只要 id 匹配就接受任意顺序
​

基准:串行 vs 并行

code/main.py 的脚手架用 400、600、800 毫秒延迟模拟三个执行器。串行共 1800 毫秒;并行是 max(400,600,800) = 800 毫秒。差异是常数而非比例,所以工具越多节省越大。

⚠️ 现实告诫:并行调用会压垮下游 API。对一个限流服务做 10 路扇出会失败。第 17 节覆盖网关级背压;重试语义留待后续章节。

流式扇出的墙钟优化

如果模型本身在流式输出,你可以在一个调用的参数刚完整时就开跑,而不必等所有调用收尾。这是 OpenAI 文档记录但并非所有 SDK 都暴露的优化。本节脚手架就这么做:模拟流一吐出某个 id 的完整参数对象,host 立刻触发该调用。

三、框架对比

💡 心法:并行是默认优化,但只在调用彼此独立时安全;一旦有顺序依赖或下游限流,就得回到串行或加依赖图调度。

四、可复用产物

本节产出 outputs/skill-parallel-call-safety-check.md——给定一份工具注册表,它审计哪些工具可安全并行、哪些有顺序依赖、哪些会压垮下游限流,返回一份带每工具 parallel_safe 标志的修订注册表。

五、练习

1. 变延迟跑基准:运行 code/main.py,改变模拟延迟。确认并行/串行比约为 max/sum(实际因线程调度、序列化、脚手架开销略偏离理想)。什么延迟分布下并行就不再重要了?

2. 处理流中取消:扩展累加器,处理「调用在流中被取消」的情况——丢弃其缓冲并发射 cancelled 事件。哪家的文档明确写了这种情况?查 Anthropic 的 content_block_stop 语义与 OpenAI 的 finish_reason:"length" 行为。

3. 换 asyncio:用 asyncio.gather 替换线程池并基准对比。执行器做真实 I/O 时应能看到 async 因上下文切换成本更低而略胜。

4. 依赖图:挑两个不该并行的工具(如 create_file 然后 write_file),给注册表加一个 ordering_dependency 图,让并行扇出受该图门控。这是依赖感知调度的最小机制,后续 Agent 工程章会形式化。

5. 读厂商文档:读 OpenAI 的并行函数调用章节与 Anthropic 的 disable_parallel_tool_use 文档,找出 Anthropic 建议禁用并行的那一类真实工具。(提示:对同一资源的有副作用变更。)

本节要点回顾

1. 并行把 N 次往返压成 1 次:执行器时间从 sum 变 max,扇出负载下墙钟减少 60%~70%。
2. 启用方式各异:OpenAI parallel_tool_calls:true 默认开;Anthropic disable_parallel_tool_use:false 默认开;Gemini 始终能并行。
3. id 关联是命门:结果必须携带匹配 tool_call_id/tool_use_id/functionResponse.id,否则乱序结果歧义。
4. 流式需每 id 一累加器:OpenAI 按 index、Anthropic 按 block、Gemini 按 functionCallId。
5. 提前解析陷阱:部分 JSON 不可解析,以厂商调用结束信号为闸门;增量 JSON 解析器更稳健。
6. 乱序完成:按完成顺序、带显式 id 回复;OpenAI/Anthropic 顺序无关,Gemini 只要 id 匹配。
7. 流式扇出优化:一个调用参数完整就开跑,不必等全部收尾。
8. 何时禁用并行:顺序依赖、输出喂输入、下游限流扛不住扇出。

下一节,我们进入结构化输出——JSON Schema、Pydantic、Zod 与受限解码,把「模型保证吐合法 JSON」从概率变成保证。