流式工具调用实操

共同的增量形状（以及它在哪里不再共同）。

三家厂商在同一份大致宣传下交付工具调用流：模型先发一个命名工具的初始事件（带 id / call_id / block_id、一个 index 或内容块 index，以及函数名），随后是一串每个携带 JSON arguments 字符串片段的增量事件，最后是一次"调用完成"事件。按顺序拼片段，JSON 解析结果，派发。K7 厂商矩阵覆盖非流式的调用形状；流式复用同一份底层事件内容，却包在各厂商的 SSE 或流封装里，起名不同。

三条轴发散，制造了多数 bug。第一，片段粒度：OpenAI 的 arguments 增量可能一次一个 token，也可能是整行，视模型而定；Gemini 的块通常更大，有时是整个值；Anthropic 的通常是小段 JSON 增量文本。第二，调用身份：OpenAI 用 tool-call 数组上的 index 加一个首块出现且随后重复的 tool_call_id；Anthropic 用一个跨块整段生命周期都稳定的内容块 index；Gemini 用 candidate/part index 对。第三，并行调用交错：三家都可能在一条流里发出多个并行调用的 token，但顺序保证与"块完成哨兵"存在与否并不一致。拼接器必须按身份键值，而不是按位置假设，否则会静默污染并行调用。

OpenAI 拼接与 GPT-4.1-nano 重复调用 bug。

OpenAI 的 Chat Completions 流形状稳定、文档清晰。每个流块携带一个 choices[0].delta.tool_calls 数组；数组每个元素有一个 index，要么携带首块字段（id、type、function.name），要么携带进行中的片段字段（function.arguments——一段跨块拼接的部分 JSON 字符串）。拼接器按 index 键值、首见时初始化条目、每块有 arguments 时追加、外层 finish_reason 变为 "tool_calls" 时关闭。

# OpenAI Chat Completions — accumulator keyed on tool-call index
calls = {}
for chunk in stream:
    for tc in chunk.choices[0].delta.tool_calls or []:
        e = calls.setdefault(tc.index, {"id": None, "name": None, "args": ""})
        if tc.id: e["id"] = tc.id
        if tc.function and tc.function.name: e["name"] = tc.function.name
        if tc.function and tc.function.arguments:
            e["args"] += tc.function.arguments
final = [{"id": v["id"], "name": v["name"], "args": json.loads(v["args"])}
         for _, v in sorted(calls.items())]

GPT-4.1-nano 的重复调用 bug 值得点名。在某些提示上，nano 模型会发出与首次同 index、同 id 的第二个 tool-call 块，等于把整次调用重复一遍。朴素拼接器看到重复，把它的 arguments 追加到已完成的 args 字符串上，JSON 解析失败，因为拼出来的字符串是两个 JSON 对象粘一起。防御修法是幂等：一旦一个调用的 args 干净解析，再看到同 id 的块就作为 no-op 处理。OpenAI SDK 的 issue tracker 有一份具体的报告；绕开只需几行代码，能挡住一个否则会让你花一小时追的静默污染。

Responses API 稍微改了形状，拼接器模式相同——你按 function_call 项的 item_id 键值、追加 arguments 增量。类型化流事件名不同（response.function_call_arguments.delta），但状态机一一对应。

Gemini：块更大、聚合更粗。

Gemini 的流式发出 GenerateContentResponse 块，每块携带 candidates[0].content.parts。函数调用 part 是 functionCall: {name, args}——与 OpenAI 的增量字符串不同，args 字段常常在一块中就以基本完整或整份 JSON 对象到达。拼接器仍需容忍同一次调用收到多块（Gemini 较新模型偶尔会把大参数对象拆成两块），但常见情形是每条流一份函数调用 part，来得晚，完整。

# Gemini — accumulate function-call parts, tolerant of coarser chunks
calls = []
for chunk in stream:
    for part in (chunk.candidates[0].content.parts or []):
        fc = getattr(part, "function_call", None)
        if fc is None: continue
        # If Gemini split args across chunks, merge into the last-open call
        if calls and calls[-1]["name"] == fc.name and not calls[-1]["closed"]:
            calls[-1]["args"].update(dict(fc.args))
        else:
            calls.append({"name": fc.name, "args": dict(fc.args), "closed": False})
# Close on stream end
for c in calls: c["closed"] = True

两个 Gemini 具体点要留意。第一，args 以结构化对象到达，不是 JSON 字符串——你不解析，你消费。这比 OpenAI 的形状更好，但会打破任何假设"流式意味着拼字符串"的代码。第二，Gemini 的并行调用以同一块 parts 数组里的多个 functionCall parts、或跨块出现；拼接器要逐块遍历 parts 列表，别假设每块一次调用。

Anthropic：内容块、并行调用交错。

Anthropic 的流式是事件类型化的：message_start、content_block_start、content_block_delta、content_block_stop、message_stop。一次工具调用以 content_block_start（携带一个 tool_use 块，带 id 与 name）开始，随后是一串 content_block_delta，携带 input_json_delta 部分 JSON 字符串，最后是 content_block_stop。多个并行工具调用作为多个不同 index 的内容块出现，Anthropic 会在同一条流里交错来自不同块的增量——拼接器按块 index 键值，等到 content_block_stop 才认调用完成。

# Anthropic — event-typed stream, index-keyed accumulator
blocks = {}
for event in stream:
    if event.type == "content_block_start" and event.content_block.type == "tool_use":
        blocks[event.index] = {"id": event.content_block.id,
                               "name": event.content_block.name,
                               "args": "", "done": False}
    elif event.type == "content_block_delta" and event.index in blocks:
        if getattr(event.delta, "type", None) == "input_json_delta":
            blocks[event.index]["args"] += event.delta.partial_json
    elif event.type == "content_block_stop" and event.index in blocks:
        blocks[event.index]["done"] = True
        blocks[event.index]["args"] = json.loads(blocks[event.index]["args"] or "{}")

并行调用陷阱是真的。两个并行运行的 tool_use 块可能交错：流可以载 block 1 的增量、再载 block 2、再载 block 1，而按"最后一个块"（而非事件里的块 index）键值的拼接器会把片段错路。这在 Anthropic SDK 里有文档，但下游框架胶水未必总同步；工具错误恢复一文处理拼接器错路后的清理模式，更便宜的修法则是上面这份按 index 键值的拼接器。

再来一条 Anthropic 特有的：tool_use 的初始 content_block_start 带一个通常为空的 input: {} 字段——真正的 args 在增量里。某些拼接器把空的 input 误当作最终 args、错过增量，最终拿到一次 args 为空对象的工具调用。修法是永远把增量拼进一段新字符串，忽略空的初始 input。

一份可移植的拼接器形状。

三种拼接器，一个共同状态机。每一种看起来都是：(1) 首次出现"命名工具调用"事件时，为该调用按其身份（index、块 index 或 item id）分配一份状态；(2) 每次收到已知调用的参数片段事件时，追加它；(3) 流结束（或每调用完成事件）时，解析并校验 args，把完成的调用列表交给外壳。一份把各厂商流映射到同一组"start / append / close"事件三元组的可移植适配器不大——每家厂商大约一百来行——能让你外壳的其余部分与厂商无关。

从一开始就该接进两件事。第一，逐调用幂等：跟踪已派发过的调用 id，丢重复。这挡住 OpenAI GPT-4.1-nano 的重复调用 bug，也挡住其他厂商模型演化时可能冒出的同类问题。第二，畸形 JSON 容忍：如果 args 字符串在关闭时解析不通，别静默交付一次坏工具调用——把畸形 args 上抛给外壳，让它发出一段模型可据以重试的纠错性 tool result。把错误消息当作提示一文处理"那段纠错消息该长什么样"。

把五步合起来读，流式工具调用就不再是逐厂商的谜题，而是适配层的一件差事。形状哪里都一样：按身份键值的拼接器、片段追加、完成时解析关闭。凸起全在各厂商特有的字段名、块粒度，以及特定模型上出现的具体 bug。适配器一次做对，把各厂商 bug 备注留在分支旁的注释里，流式循环就不再是你外壳在生产里翻船的地方——它成了所有有趣推理下面那份无聊的底座。