并发跑工具，Go 的天然优势

本文是《从 0 写 Agent》系列第 5 章。这个系列基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版），用 Go 从零实现一个生产级 Agent 运行时；本章讲「并发工具执行：Go 的天然优势」。

当模型同时要调 5 个工具，查天气、查股票、查地图、查汇率、查新闻，你是串行调（一个一个等），还是并行调（同时发出）？

答案显然是并行。但"并行"在 Python 和 Go 里，体验天差地别。这一章拆 agents-go 的 runFunctionTools，看它怎么用 errgroup 优雅地并发调工具，以及为什么这是 Go 做 Agent 的天然优势。

引子：为什么工具要并发

模型在一轮 turn 里，可能同时返回多个工具调用。比如你问"帮我查北京天气、腾讯股价、美元汇率"，模型可能一次性返回三个 function_call：get_weather、get_stock、get_exchange_rate。

串行执行：调天气（等 500ms）→ 调股价（等 800ms）→ 调汇率（等 300ms）= 总共 1.6 秒 并行执行：三个同时发，总耗时 ≈ 最慢的那个 = 800ms

并行快了一倍。 而且工具越多，差距越大，10 个工具串行 5 秒，并行可能只要 800ms。

对 Agent 这种"频繁调多个工具"的场景，并发是刚需。

原理：errgroup，Go 并发的瑞士军刀

Go 标准库的 errgroup 包，是并发执行多个任务并收集错误的标准工具。它的核心思想：

g, ctx := errgroup.WithContext(parentCtx)
for _, task := range tasks {
    g.Go(func() error {  // 每个任务一个 goroutine
        return doWork(ctx)
    })
}
if err := g.Wait(); err != nil {  // 等所有完成，返回第一个错误
    // 处理错误
}

三个关键特性：

1. 自动并发。 每个 g.Go 起一个 goroutine，所有任务并行。

2. 共享 context。 errgroup.WithContext 返回一个派生 context，任何一个任务返回错误，这个 context 立即 cancel。其他还在跑的任务能通过 ctx.Done() 感知，主动退出。

3. 错误聚合。 g.Wait() 等所有任务结束，返回第一个非 nil 错误。

对比 Python 的 asyncio.gather：

# Python
results = await asyncio.gather(
    get_weather(city),
    get_stock(ticker),
    get_exchange_rate(pair),
    return_exceptions=True  # 错误处理要单独写
)

Python 要 async/await 满天飞，每个工具函数都要是 async def，错误处理要 return_exceptions=True 再单独处理。Go 的版本没有 async 关键字，goroutine 是隐式的，函数还是普通函数。

源码对照：agents-go 的 runFunctionTools

agents-go 的并发工具执行在 agents/run_step.go:277，是 errgroup 的教科书级应用：

// agents/run_step.go:277（骨架）
func (r *runner) runFunctionTools(ctx context.Context, agent *Agent, runs []toolRunFunction) ([]functionToolResult, error) {
    results := make([]functionToolResult, len(runs))  // 预分配结果切片
    g, gctx := errgroup.WithContext(ctx)

    // 限流：防止一次性起太多 goroutine 把下游打挂
    if r.opts.MaxToolConcurrency > 0 {
        g.SetLimit(r.opts.MaxToolConcurrency)
    }

    for i, run := range runs {
        i, run := i, run  // 经典闭包陷阱防御（Go 1.22 前必需）
        g.Go(func() error {
            // 完整的工具调用流水线（第 4 章讲的九步）
            tc := &ToolContext{...}
            callToolStart(gctx, ...)               // hook
            r.runToolInputGuardrails(gctx, ...)    // 输入护栏
            span := r.trace.StartFunctionSpan(...) // trace span
            out, err := invokeTool(gctx, run.Tool, tc, run.Call.Arguments)
            // 错误处理...
            r.runToolOutputGuardrails(gctx, ...)   // 输出护栏
            callToolEnd(gctx, ...)                 // hook
            results[i] = functionToolResult{...}   // 写入预分配位置
            return nil
        })
    }

    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

注意四个设计细节：

1. 预分配结果切片，按索引写入

results := make([]functionToolResult, len(runs))
// ...
results[i] = functionToolResult{...}  // 每个 goroutine 写自己的位置

不是用 channel 收集结果，而是预分配固定大小的切片，每个 goroutine 写自己的索引位置。 这是并发安全的，不同 goroutine 写不同索引，不需要锁。比 channel 简单，比 mutex 高效。

2. SetLimit 限流

if r.opts.MaxToolConcurrency > 0 {
    g.SetLimit(r.opts.MaxToolConcurrency)
}

为什么不无限并发？ 因为工具背后通常是外部 API（数据库、第三方服务）。10 个工具同时打过去，可能触发限流、打挂下游、或者被 ban。MaxToolConcurrency 是个"礼貌的"上限，告诉下游"我最多同时来 N 个"。

默认 0 = 不限。 你可以根据下游能力设置。这是工程纪律，并发不是越多越好，要匹配下游容量。

3. 共享 context 的取消传播

g, gctx := errgroup.WithContext(ctx)

gctx 传给每个工具调用。任何一个工具返回错误，gctx 立即 cancel，其他工具通过 gctx.Done() 感知，主动退出。

这意味着，"一个工具失败，其他工具不再继续跑"。这是合理的默认（避免无意义的计算），但你也可以设计成"失败不取消其他"（如果你要"尽力而为，收集所有结果"）。agents-go 选了"fast-fail"语义。

4. 闭包变量捕获的防御

for i, run := range runs {
    i, run := i, run  // 防御 Go 1.22 前的闭包陷阱
    g.Go(func() error {
        // 用 i 和 run
    })
}

Go 1.22 之前，循环变量会被所有闭包共享，如果不做 i, run := i, run 复制，所有 goroutine 看到的都是最后一个值。这是 Go 的经典坑。agents-go 显式复制防御（虽然 Go 1.22+ 已修复这个语义，但保持防御性编程是好习惯）。

设计权衡：为什么用 errgroup 不用裸 goroutine

有人会问：直接 go func() 起一堆 goroutine，用 sync.WaitGroup 等，不行吗？

能跑，但 errgroup 解决了三个裸 goroutine 解决不了的问题：

1. 错误传播。 裸 goroutine 里，goroutine 内的 error 没法直接返回给调用方（goroutine 没有"返回值"）。errgroup 通过 g.Wait() 返回第一个错误。

2. 取消传播。 裸 goroutine 失败了，没法通知其他 goroutine 停。errgroup 的共享 context 自动做这事。

3. panic 恢复。 errgroup 的 g.Go 会 recover panic 并转成 error（通过 PanicError）。裸 goroutine 里 panic 会直接崩进程。

所以 errgroup 是 Go 并发任务的"标配"，除非你有特殊需求（比如"所有任务都要跑完，不管错没错"），否则都用 errgroup。

Python 的并发对比

来对比一下 Python 怎么做同样的事。假设工具是 async 的：

import asyncio

async def run_function_tools(runs):
    # 用 asyncio.gather 并发
    tasks = [run_single_tool(run) for run in runs]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return results

看着也简单，但有几个隐藏代价：

1. 整个调用链必须 async。 run_function_tools 是 async，调用它的 executeToolsAndSideEffects 也得 async，再往上 loop、Run 全得 async。async 是传染性的，一个函数 async，整条调用链都要 async。这就是为什么 Python 的 Agent 框架到处是 async def。

2. 阻塞操作会卡死事件循环。 如果某个工具内部用了同步 IO（比如 requests.get 而不是 httpx.AsyncClient），它会卡住整个事件循环，所有并发任务都被阻塞。必须严格保证所有 IO 都是 async 的，这是 Python 异步编程最大的坑。

3. CPU 密集型要单独处理。 Python 的 GIL 让 async 不能利用多核。要做 CPU 密集的并发，要用 ProcessPoolExecutor，复杂度飙升。

Go 完全没有这些问题，

• goroutine 不是"传染性的"，普通函数就能 go
• 阻塞 IO 不卡其他 goroutine（Go 的调度器处理）
• 多核并行天然支持（GOMAXPROCS）

这就是 agents-go 作者说的"Go 的并发模型天然适合多工具并行调用"。 不是 Go 更强大，是 Go 的并发模型更省心，你不用思考"这个函数要不要 async"，不用怕"阻塞调用卡死事件循环"。

设计权衡：MaxToolConcurrency 该设多少

这是个真实的运维问题。设太大，打挂下游；设太小，浪费并发优势。

经验值：

关键原则，并发上限要匹配"最弱的下游"。 你的 Agent 可能调 5 种不同的工具，每种工具的下游容量不同。MaxToolConcurrency 是全局的，所以它要照顾最弱的那个。

如果你有不同类型的工具，更好的做法是，给每类工具单独的并发池。但 agents-go 目前是全局 MaxToolConcurrency，简化处理。生产环境可以考虑扩展。

小结

读完这一章，你应该能回答：

1. 为什么工具要并发？，多个工具调用并行，总耗时≈最慢的那个，而非累加
2. errgroup 的三个特性？，自动并发、共享 context 取消传播、错误聚合
3. 为什么用预分配切片而非 channel？，不同索引并发安全，无需锁，比 channel 简单
4. MaxToolConcurrency 该设多少？，匹配最弱的下游容量，默认 0=不限
5. Go 并发比 Python 优雅在哪？，goroutine 不传染（不用 async 满天飞）、不卡事件循环、支持多核

并发工具执行是 Go 做 Agent 的杀手锏。 在 Python 里，要做到同样的事，整个代码库都要 async 化，还要小心翼翼避免阻塞调用。在 Go 里，一个 errgroup 搞定，普通函数照样跑。

下一篇

第 6 章，我们看 Agent 的流式输出，从阻塞 Run 到流式 RunStreamed 的本质区别。agents-go 用 channel + goroutine 的生产者消费者模型，把"模型边想边吐"变成事件流推给前端。这是你看到 ChatGPT 打字效果背后的机制。

本文是《从 0 写 Agent》系列第 5 章，案例代码基于开源项目 agents-go（OpenAI Agents SDK 的 Go 移植版）。全系列目录见总览。