Agent 流式 tool_calls 增量拼装：index 对齐、arguments 缓冲与完成门闩

生产 Agent 几乎都开流式输出：前端要首 token、网关要 idle 心跳、可观测性要 TTFT。文本 delta 可以直接推 UI，但 tool 调用的参数往往是「按字符串碎片到达的 JSON」—— 在流结束前既不保证可解析，更不保证 schema 合法。若在半包上 json.loads 失败就重试、或拼到「看起来像完整对象」就立刻执行 shell/SQL，会把一次正常的流式补全变成参数截断执行、重复副作用或协议非法的下一轮消息。本文只讨论 模型流式响应中 tool_calls /tool_use 的客户端拼装与完成门闩，不重复并行调度、结果回灌预算、run 级 deadline 与各工具沙箱设计。

问题背景：工具参数不是「一个完整 JSON 事件」

典型数据流：

1. 编排器以 stream=true（或 SSE / Anthropic streaming）请求模型，附带 tools 定义。
2. 流上交错出现：assistant 文本 token、一条或多条 tool 调用的 元数据（id、name）与 参数碎片。
3. 流正常结束后，客户端应得到 0..N 条完整 tool call，再进入既有：schema 校验 → 并行 / 串行执行 → 结果脱敏回灌。

各厂商事件形状不同，但共同约束是：

在缺少统一拼装层时，常见故障包括：

• 半包执行：arguments 已拼出 {"path":"/tmp/a 就调用写文件；或 json.loads 偶发成功于截断对象（尾部恰好可被容错解析器吞掉）导致静默缺字段。
• index /block 错位：并行两条 tool 时，把 call B 的参数碎片 append 到 call A，产生合法 JSON、错误语义。
• 过早 tools/call：流仍在进行，客户端已对「当前缓冲」执行 MCP；随后又来一段 delta，无法撤销已发生副作用。
• 取消后脏状态：用户 Stop 时缓冲是非法 JSON，却仍向模型回灌「假成功」或丢弃 tool_call_id，破坏「每个 call 必须有 result」的协议约定。
• 无界缓冲：模型把整份大文件内容塞进 tool 参数字符串，Worker 在拼装阶段 OOM，尚未进入结果预算层。
• UI 与执行语义混淆：前端可以展示「正在生成参数…」；执行门闩必须仍是「流结束 + 可解析 + schema 通过」。

因此应把流式 tool 路径建模为 只追加的缓冲状态机，执行发生在 committed 之后：

on_stream_event(ev):
  route by provider → update buffers[key]
  emit ui_partial (optional, never execute)

on_stream_end_or_block_stop:
  for each buffer:
    if cancelled → mark cancelled, no execute
    else finalize:
      parse JSON → schema validate → enqueue for tool runner

tool runner (既有并行/超时/沙箱) 只消费 committed calls

可落地实现：统一缓冲、完成门闩与取消

1. 统一内部模型（与厂商事件解耦）

不要在业务代码里直接 if openai: ... elif anthropic 执行工具。先归一为：

from __future__ import annotations

from dataclasses import dataclass, field
from enum import Enum
from typing import Any


class CallAssembleState(str, Enum):
    OPEN = "open"                 # 已见 id/name，参数可能仍在追加
    ARGS_COMPLETE = "args_complete"  # 厂商声明该 call/块结束或整段流结束
    COMMITTED = "committed"       # JSON 解析 + schema 通过，可进入执行队列
    FAILED = "failed"             # 解析/校验失败，将回灌 is_error
    CANCELLED = "cancelled"       # 用户/deadline 取消，不执行


@dataclass
class ToolCallBuffer:
    key: str                      # openai: str(index) 或 id；anthropic: content block index / id
    call_id: str | None = None
    name: str | None = None
    arguments_buf: list[str] = field(default_factory=list)
    arguments_bytes: int = 0
    state: CallAssembleState = CallAssembleState.OPEN
    parse_error: str | None = None
    parsed_args: dict[str, Any] | None = None

    def append_args_fragment(self, fragment: str, *, max_args_bytes: int) -> None:
        if self.state is not CallAssembleState.OPEN:
            raise RuntimeError(f"append on non-open buffer: {self.state}")
        b = fragment.encode("utf-8")
        if self.arguments_bytes + len(b) > max_args_bytes:
            self.state = CallAssembleState.FAILED
            self.parse_error = "arguments_exceed_max_bytes"
            return
        self.arguments_buf.append(fragment)
        self.arguments_bytes += len(b)

    def raw_arguments(self) -> str:
        return "".join(self.arguments_buf)

建议默认参数：

2. OpenAI Chat Completions：按 index 对齐增量

流式 chunk 中 delta.tool_calls 是数组，元素常带：

• index：本 choice 内 tool call 的稳定下标（拼装主键）
• 首包常见：id、type、function.name
• 后续包：function.arguments 为 JSON 文本碎片（不是已解析对象）

import json
from typing import Iterator


def ingest_openai_chat_chunk(
    buffers: dict[str, ToolCallBuffer],
    chunk: dict,
    *,
    max_args_bytes: int,
    max_calls: int,
) -> None:
    choice0 = (chunk.get("choices") or [None])[0]
    if not choice0:
        return
    delta = choice0.get("delta") or {}
    for tc in delta.get("tool_calls") or []:
        idx = tc.get("index")
        if idx is None:
            continue
        key = f"idx:{idx}"
        buf = buffers.get(key)
        if buf is None:
            if len(buffers) >= max_calls:
                # 超额：记录失败占位，避免静默丢事件
                buffers[key] = ToolCallBuffer(
                    key=key,
                    state=CallAssembleState.FAILED,
                    parse_error="too_many_tool_calls",
                )
                continue
            buf = ToolCallBuffer(key=key)
            buffers[key] = buf
        if tc.get("id"):
            buf.call_id = tc["id"]
        fn = tc.get("function") or {}
        if fn.get("name"):
            buf.name = fn["name"]
        if "arguments" in fn and fn["arguments"] is not None:
            if buf.state is CallAssembleState.OPEN:
                buf.append_args_fragment(fn["arguments"], max_args_bytes=max_args_bytes)

    # finish_reason 出现在某一 chunk 的 choice 上；仍应读到 stream 终止
    # 真正 finalize 放在 for chunk in stream 循环退出之后

整段流结束后：

def finalize_buffers(buffers: dict[str, ToolCallBuffer]) -> list[ToolCallBuffer]:
    ordered = [buffers[k] for k in sorted(buffers, key=lambda x: int(x.split(":")[1]))]
    for buf in ordered:
        if buf.state in (
            CallAssembleState.FAILED,
            CallAssembleState.CANCELLED,
            CallAssembleState.COMMITTED,
        ):
            continue
        if not buf.call_id or not buf.name:
            buf.state = CallAssembleState.FAILED
            buf.parse_error = "missing_id_or_name"
            continue
        raw = buf.raw_arguments() or "{}"
        try:
            val = json.loads(raw)
        except json.JSONDecodeError as e:
            buf.state = CallAssembleState.FAILED
            buf.parse_error = f"invalid_json: {e}"
            continue
        if not isinstance(val, dict):
            # 多数 function tools 约定 parameters 为 object；非 object 直接失败更安全
            buf.state = CallAssembleState.FAILED
            buf.parse_error = "arguments_not_object"
            continue
        buf.parsed_args = val
        buf.state = CallAssembleState.ARGS_COMPLETE
        # 下一步：既有 JSON Schema / strict 子集校验 → COMMITTED 或 FAILED
    return ordered

要点：

• 主键用 index，不要用「当前数组位置」或「最后一次看到的 id」；稀疏 / 乱序 delta 时 index 才稳定。
• arguments 必须当 opaque string concat；不要对碎片做 json.loads，也不要用「括号计数启发式」提前 commit。
• 空参数工具：部分模型可能不发 arguments 或发 ""；finalize 时按 "{}" 处理，再交给 schema（required 会拦截真缺字段）。

3. Anthropic Messages：partial_json 追加与 content_block_stop

Anthropic 流式下，client tool 常见顺序：

1. content_block_start：content_block.type == "tool_use"，带 id、name，input 初始多为空对象。
2. 多个 content_block_delta：delta.type == "input_json_delta"，字段 partial_json 为字符串碎片。
3. content_block_stop：该块参数文本结束。
4. message_delta / message_stop：消息级 stop_reason（如 tool_use）。

def ingest_anthropic_event(
    buffers: dict[str, ToolCallBuffer],
    event: dict,
    *,
    max_args_bytes: int,
    max_calls: int,
) -> None:
    et = event.get("type")
    if et == "content_block_start":
        block = event.get("content_block") or {}
        if block.get("type") != "tool_use":
            return
        idx = event.get("index")
        key = f"block:{idx}"
        if key not in buffers and len(buffers) >= max_calls:
            buffers[key] = ToolCallBuffer(
                key=key, state=CallAssembleState.FAILED, parse_error="too_many_tool_calls"
            )
            return
        buf = buffers.setdefault(key, ToolCallBuffer(key=key))
        buf.call_id = block.get("id")
        buf.name = block.get("name")
        # 忽略 start 里的 input 对象，以 partial_json 拼接结果为准
    elif et == "content_block_delta":
        delta = event.get("delta") or {}
        if delta.get("type") != "input_json_delta":
            return
        key = f"block:{event.get('index')}"
        buf = buffers.get(key)
        if not buf or buf.state is not CallAssembleState.OPEN:
            return
        frag = delta.get("partial_json") or ""
        buf.append_args_fragment(frag, max_args_bytes=max_args_bytes)
    elif et == "content_block_stop":
        key = f"block:{event.get('index')}"
        buf = buffers.get(key)
        if buf and buf.state is CallAssembleState.OPEN:
            # 标记块级参数文本结束；JSON finalize 仍可放到 message_stop 后批量做
            pass

与 OpenAI 路径共用 finalize_buffers（排序键改为 block index）。不要在每个 content_block_stop 上立刻执行工具，除非产品明确要求「块级流水线」且已处理：同轮后续块、run 预算、并行互斥组。默认更稳的是 message 流结束后批处理，与 Chat Completions 对齐。

4. 完成门闩：何时允许进入 tool runner

推荐门闩（全部满足才 COMMITTED）：

1. 传输完成：提供商流正常结束，或（若采用块级策略）对应 block stop 且不再接受该 key 的 append。
2. 身份完整：call_id 与 name 非空；name 落在本轮注册工具表（防模型编造工具名）。
3. JSON 可解析：json.loads 成功且类型符合约定（一般为 object）。
4. Schema 通过：走既有「执行前 JSON Schema /strict 对齐」校验；失败则 FAILED，不调用下游。
5. 预算允许：max_tool_calls_per_run / 并行信号量等在 enqueue 前 再检一次（流式拼装期间模型可能已超额发射）。

对 FAILED / CANCELLED 的 call，仍应生成 结构化 tool 错误结果（is_error / 等价），保证下一轮每个 tool_call_id 有配对 result—— 与并行部分失败回灌同一纪律。

def to_error_result(buf: ToolCallBuffer) -> dict:
    return {
        "tool_call_id": buf.call_id or buf.key,
        "name": buf.name or "unknown",
        "is_error": True,
        "content": buf.parse_error or buf.state.value,
    }

5. 取消与断连：半成品不可「当成功」

与 run 级取消传播衔接时，拼装层规则：

不要把 CANCELLED 的半包 arguments 写入审计明文以外的「可再执行队列」；若需断点续跑，应保存 已 COMMITTED 且已完成 的 tool 结果，而不是半截 JSON。

6. 配置示例（网关 YAML）

agent_stream_assemble:
  max_tool_argument_bytes: 262144      # 256 KiB
  max_tool_calls_in_flight_assemble: 32
  assemble_idle_timeout_s: 60
  execute_before_stream_end: false
  ui_forward_partial_args: true
  arguments_must_be_object: true
  on_invalid_json: error_result        # error_result | drop_with_protocol_repair
  on_args_overflow: error_result
  finalize_order: provider_index       # 与回灌顺序策略一致

可观测性建议指标（低基数标签：provider、tenant、tool_name 可选）：

• tool_assemble_args_bytes（histogram）
• tool_assemble_invalid_json_total
• tool_assemble_overflow_total
• tool_assemble_to_commit_seconds（流结束 → committed）
• tool_execute_rejected_not_committed_total（防回归：执行路径误吃 OPEN 缓冲）

风险与边界

「能 parse 成功」≠「参数完整」。 截断后的 {"a":1} 若模型本意是更多字段，JSON 仍合法。门闩只能保证结构可解析；语义完整性依赖模型 finish 与 schema required。禁止用宽松解析器（如注释容忍、尾逗号自动修）隐藏协议错误。

不要在 delta 上做安全裁决的唯一依据。 路径 realpath、SSRF IP 裁决、SQL 角色等必须在 committed 参数 上执行；UI 上展示的 partial 字符串可能尚未包含最终路径。

厂商事件版本差异。 Chat Completions、Responses、Anthropic、各类兼容网关的字段名与「id 首次出现时机」不一致。适配层应单测：单 tool、并行双 tool、无参数 tool、超大 arguments 截断、中途 cancel 五类夹具。

流式 UI 泄露。 partial arguments 可能含用户粘贴的密钥或 PII；若把拼装中的字符串推到多租户共享日志或前端错误上报，需与回灌脱敏策略一致，或仅推送「工具名 + 字节数」直到 commit。

块级提前执行的诱惑。 在 Anthropic content_block_stop 后立刻跑工具可降 TTFT-to-side-effect，但会与「同轮并行互斥」「run 预算一次性检查」「用户看到完整 assistant 文本再确认」冲突。默认批处理；若启用流水线，必须按工具名配置 allow_pipelined_execute 白名单，且写工具默认关闭。

与 prompt cache / 历史持久化。 写入会话存储的应是 finalize 后的规范 assistant 消息（完整 tool_calls / tool_use），而不是 delta 数组。用半包历史续聊会导致下一轮 400 或模型行为漂移。

兼容代理可能重切分。 经 LLM 网关再分片时，arguments 碎片边界会变，但 concat 结果应不变。单测应用「同一逻辑字符串的随机切分」属性测试，而不是写死 chunk 边界。

执行层幂等不能替代拼装正确性。 即使写工具有幂等键，错误拼装仍可能对错误参数做幂等成功，后果更难察觉。拼装错误应在执行前失败并回灌，而不是依赖下游去重。

参考来源

• OpenAI — Function calling
• OpenAI API Reference — Chat Completions（流式 delta、tool_calls、finish_reason）
• OpenAI — Responses API（流式事件与 function call arguments delta）
• Anthropic — Streaming Messages
• Anthropic — Tool use with Claude
• Anthropic — Handle tool calls（tool_use / tool_result 配对与格式约束）
• Model Context Protocol — Tools（tools/call、执行错误与协议错误）
• JSON — ECMA-404 / IETF RFC 8259（完整文本解析语义，非流式半包）
• OWASP Gen AI — LLM06:2025 Excessive Agency
• MDN — AbortController（流读取与执行侧取消协同）

systems