Agent 并行 tool_calls：信号量、提供商 fan-out 上限与部分失败回灌

主流对话 API 允许模型在同一条 assistant 消息里发出多条工具调用（OpenAI 的 tool_calls[]、Anthropic 的多个 tool_use 块）。客户端若「全部 asyncio.create_task / 全开 goroutine」，会在一轮内把 shell、HTTP、SQL、浏览器会话同时打满 Worker 与下游依赖。本文只讨论 模型已返回一批 tool_calls 之后、结果写回下一轮 prompt 之前 的并行调度与失败策略；不重复各工具自身的 SSRF、路径边界、进程组超时与结果截断设计。

问题背景：一轮 fan-out 会叠加多种资源上限

典型数据流：

1. 用户消息进入编排器；模型流式或非流式返回，消息中含 N 条工具调用（N ≥ 1）。
2. 客户端按 tool_call_id / tool_use.id 分发到本地函数或 MCP tools/call。
3. 各调用完成后，把 tool / tool_result 消息批量附回，再请求下一轮模型。

在生产 Agent 中，下列因素会把「并行」从性能优化变成故障放大器：

• 提供商与模型语义：OpenAI Chat Completions 提供 parallel_tool_calls 布尔开关；Anthropic 支持并行 tool use，并可在 tool_choice 中通过 disable_parallel_tool_use 关闭。关闭并行只约束模型是否在一轮内发出多条，不约束客户端是否把历史中的多条串行执行。
• Worker 进程资源有界：每条 shell 工具可占一个进程组；每条 HTTP 占连接与出口带宽；每条 SQL 占连接池 checkout；浏览器类工具常独占一个 BrowserContext。N=16 的 fan-out 足以打满 4 核 Worker 的 FD、内存与 cgroup CPU。
• 下游限流正交：同一租户对 GitHub、Jira、内部 API 的 QPS 配额与「本机信号量」不是一回事；无租户级限流时，并行 tool_calls 会把单用户突发放大为对外部 API 的突发。
• 部分失败是常态：16 路中 1 路超时、1 路 403、其余成功时，若编排器「任一失败即整批 abort 且不回灌成功结果」，模型会丢失已完成工作；若「静默丢弃失败项」，模型会在缺失 tool_result 时行为未定义或再次重试。
• 隐式顺序依赖：模型可能并行发出 write_file 与依赖该文件的 run_tests，或并行两条写同一路径的调用。调度层默认不推断依赖图；错误并行会产生 TOCTOU 与不可复现结果。
• 取消与超时嵌套：用户点 Stop、外层 run deadline、单工具 timeout 三者同时存在时，必须把取消传播到仍在飞行的全部调用，并与进程组杀、HTTP 取消、SQL 会话终止对齐。

因此并行执行应拆成固定策略，而不是「有多少 call 就起多少 task」：

model tool_calls[N]
  → 策略裁决（允许并行？按工具名串行化？）
  → 准入（全局 / 租户 / 工具名 三级信号量 + 下游配额）
  → 有界并发执行（errgroup / TaskGroup + 每调用 timeout）
  → 汇聚（成功体 + 结构化错误，保留全部 tool_call_id）
  → 结果预算/脱敏（既有回灌流水线）
  → 写回下一轮消息（顺序建议与请求中 id 列表一致）

可落地实现：开关、信号量、配额与部分失败

1. 先对齐「模型侧 fan-out」与「执行侧并发」两个旋钮

执行侧示意（Python asyncio）：

from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable

@dataclass(frozen=True)
class ParallelPolicy:
    max_accepted_per_turn: int = 24
    max_parallel_per_turn: int = 6
    per_call_timeout_s: float = 60.0
    # 工具名 → 互斥组；同组内全局串行（跨 turn 也可持有同一把锁）
    mutex_groups: dict[str, str] = field(default_factory=dict)

ToolHandler = Callable[[dict[str, Any]], Awaitable[dict[str, Any]]]

async def run_tool_batch(
    calls: list[dict[str, Any]],
    handler: ToolHandler,
    policy: ParallelPolicy,
    *,
    global_sem: asyncio.Semaphore,
    tenant_sem: asyncio.Semaphore,
    tool_sems: dict[str, asyncio.Semaphore],
    group_locks: dict[str, asyncio.Lock],
) -> list[dict[str, Any]]:
    if len(calls) > policy.max_accepted_per_turn:
        return [
            {
                "tool_call_id": c["id"],
                "is_error": True,
                "content": (
                    f"rejected: {len(calls)} tool_calls exceed "
                    f"max_accepted_per_turn={policy.max_accepted_per_turn}"
                ),
            }
            for c in calls
        ]

    turn_sem = asyncio.Semaphore(policy.max_parallel_per_turn)
    results: list[dict[str, Any] | None] = [None] * len(calls)

    async def one(i: int, call: dict[str, Any]) -> None:
        name = call["name"]
        tool_sem = tool_sems.get(name, global_sem)
        group = policy.mutex_groups.get(name)
        lock = group_locks.setdefault(group, asyncio.Lock()) if group else None

        async def body() -> dict[str, Any]:
            try:
                raw = await asyncio.wait_for(
                    handler(call), timeout=policy.per_call_timeout_s
                )
                return {
                    "tool_call_id": call["id"],
                    "is_error": bool(raw.get("is_error")),
                    "content": raw.get("content", ""),
                }
            except asyncio.TimeoutError:
                return {
                    "tool_call_id": call["id"],
                    "is_error": True,
                    "content": f"timeout after {policy.per_call_timeout_s}s",
                }
            except asyncio.CancelledError:
                return {
                    "tool_call_id": call["id"],
                    "is_error": True,
                    "content": "cancelled",
                }
            except Exception as exc:  # 生产应分类并脱敏
                return {
                    "tool_call_id": call["id"],
                    "is_error": True,
                    "content": f"execution_error: {type(exc).__name__}",
                }

        # 准入顺序：租户 → 全局 → 每工具 → 本轮；互斥组在最内层持锁
        async with tenant_sem, global_sem, tool_sem, turn_sem:
            if lock is None:
                results[i] = await body()
            else:
                async with lock:
                    results[i] = await body()

    async with asyncio.TaskGroup() as tg:
        for i, c in enumerate(calls):
            tg.create_task(one(i, c))

    # 保序：与输入 calls 下标一致，便于对照 tool_call_id
    return [r if r is not None else {
        "tool_call_id": calls[i]["id"],
        "is_error": True,
        "content": "internal: missing result",
    } for i, r in enumerate(results)]

要点：

• 排队优于丢弃：max_parallel_per_turn 用信号量限制飞行中数量；max_accepted_per_turn 才是拒绝阈值。
• 三级信号量：全局保护 Worker，租户防吵闹邻居，按工具名限制「重」工具（如 browser_*、run_sql）的并发。
• 互斥组：对 apply_patch / write_file / git_commit 等共享工作区的写工具，用命名锁串行，即使模型并行发出。

2. 信号量与配额参数起点

Go 侧可用 golang.org/x/sync/errgroup 的 SetLimit，或自建 worker pool；语义与上表相同：Limit 管并发，不替代每调用 timeout 与租户配额。

3. 部分失败：必须为每条 call 回一条结果

提供商要求每个 tool_call_id / tool_use.id 都有对应结果消息，否则下一轮上下文不完整。推荐策略：

回灌给模型的错误应是短、稳定、可行动的文案，避免把内部主机名与栈轨迹塞进上下文（与结果脱敏流水线一致）。MCP 的 tool execution error 用结果内容 + isError: true 表达，而不是把业务失败提升为 JSON-RPC protocol error（后者通常不回给模型做自纠）。

结构化结果字段建议（写入会话存储的元数据，可按需暴露给模型）：

{
  "tool_call_id": "call_abc",
  "name": "http_fetch",
  "status": "timeout",
  "duration_ms": 60012,
  "is_error": true,
  "error_code": "TOOL_TIMEOUT",
  "content": "timeout after 60s"
}

4. 与取消、幂等、回灌层的衔接

• 取消传播：用户 Stop 应 cancel 整批 task，并触发各工具已有的终止路径（进程组 SIGTERM→SIGKILL、HTTP AbortSignal、SQL 连接 pg_cancel_backend 等）。批级 CancelledError 不应吞掉已完成结果的持久化。
• 幂等写工具：并行重试前必须有去重键（见 MCP ToolAnnotations.idempotentHint 与执行层 dedup）；并行 + 自动重试会放大「支付两遍 / 建两次资源」。
• 结果预算：max_round_tool_tokens 约束的是本轮全部 tool 结果之和；并行成功时更容易顶满，应在回灌层按 call 均分或按优先级截断，并标明截断元数据。
• 可观测性：为整批建 parent span，每条 call 建 child；属性至少含 tool.name、tool_call_id、tenant_id、queue_wait_ms、status。便于区分「模型 fan-out 过大」与「下游限流」。

5. 配置示例（网关 YAML）

tool_parallelism:
  max_accepted_per_turn: 24
  max_parallel_per_turn: 6
  per_call_timeout_s: 60
  turn_wall_clock_s: 120
  global_max_in_flight: 64
  tenant_max_in_flight: 8
  per_tool_max_in_flight:
    run_terminal: 2
    run_sql: 4
    http_fetch: 12
    browser_navigate: 1
  mutex_groups:
    write_file: workspace_write
    apply_patch: workspace_write
    git_commit: workspace_write
  partial_failure: all_settled   # or fail_fast_cancel_peers
model_request:
  # 按厂商映射到 parallel_tool_calls / disable_parallel_tool_use
  prefer_parallel_tool_calls: true

风险与边界

模型并行 ≠ 无依赖。 调度器不能可靠地从自然语言参数推断「call B 依赖 call A 的输出」。互斥组与「关闭模型侧并行」只能降低写冲突概率；对强依赖流水线，应改用单 call 串行多步，或由编排器暴露显式 plan/DAG 工具，而不是假设 fan-out 安全。

信号量不能替代下游正确性。 连接池耗尽、DB max_connections、浏览器崩溃、API 429 仍会发生。429 应映射为可重试错误并尊重 Retry-After；无抖动的立刻重试会在并行下形成重试风暴。

TaskGroup / errgroup 默认 fail-fast 语义。 Go errgroup 在第一个错误时取消同行；Python 3.11+ asyncio.TaskGroup 在子任务异常时也会取消兄弟任务。若产品选择 all_settled，不要直接把工具异常抛出 TaskGroup，应在 task 内部捕获并转为 is_error 结果（如上示意）。

保序与 id 对齐。 并行完成顺序是乱的，但回写消息列表应按原 call 列表顺序或严格一对 tool_call_id，避免客户端库按位置错配。流式 UI 可按完成顺序推送事件，与写回模型的顺序分离。

公平性与饥饿。 仅有全局信号量时，大 fan-out 租户会饿死小请求。租户级信号量是最低要求；更细可用加权公平队列（按租户 token bucket 再入全局队列）。

关闭模型侧并行的成本。 parallel_tool_calls: false 增加轮次与延迟，但简化写工具会话。对「仓库重构 + 测试」类 Coding Agent，常见做法是：读工具允许并行，写工具互斥组串行，而不是全局关闭并行。

安全边界不因并行而放宽。 每条 call 仍须独立做参数 schema 校验、路径 realpath、HTTP 出站 IP 裁决与结果脱敏。并行只共享调度配额，不共享「一次校验、多路复用放行」的捷径。

MCP 多 Server。 不同 MCP Server 上的 tools/call 可并行，但要注意每 Server 的会话亲和、速率限制与 OAuth token 刷新互斥；对同一 Server 的有状态会话，过度并行可能触发服务端 409 或会话损坏。

参考来源

• OpenAI — Function calling（并行 tool calls 与 parallel_tool_calls）
• OpenAI API Reference — Chat Completions（parallel_tool_calls 字段）
• Anthropic — Tool use with Claude（并行 tool use 与 disable_parallel_tool_use）
• Model Context Protocol — Tools（tools/call、isError、执行错误与协议错误区分）
• Model Context Protocol — ToolAnnotations（idempotentHint 等提示字段）
• Python 3 — asyncio.TaskGroup / Semaphore
• golang.org/x/sync/errgroup — SetLimit
• OWASP Gen AI — LLM06:2025 Excessive Agency
• OWASP Gen AI — LLM04:2025 Data and Model Poisoning / 相关资源耗尽讨论见 LLM Top 10

systems