第11章 Go的并发，Agent的加速器

前七章我们用Eino搭起了Agent的完整骨架。但上线之后你会发现：一个请求跑得挺好，一百个请求同时来呢？一千个呢？这一章，我们充分发挥Go的并发优势——goroutine编排工作流、channel传递结果、errgroup管理错误——让你的Agent快起来。

11.1 一千个请求同时来，Python在排队，Go在飞

11.1.1 真实的生产场景

假设你部署了一个Agent服务，单次请求需要调用3个外部工具，每个工具耗时300ms。串行处理：

请求1：工具A(300ms) → 工具B(300ms) → 工具C(300ms) = 900ms
请求2：等着请求1跑完……900ms
请求3：等着请求2跑完……900ms

1000个请求串行跑完 = 900秒 = 15分钟。用户早跑了。

Go的解决方式：每个请求一个goroutine，请求内的3个工具也并发调：

// 生产级 Agent 服务——每个请求一个 goroutine
func agentHandler(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 5*time.Second)
    defer cancel()

    // 并发调3个工具
    result := queryWithConcurrency(ctx, r.FormValue("q"))
    
    json.NewEncoder(w).Encode(result)
}

func main() {
    http.HandleFunc("/ask", agentHandler)
    http.ListenAndServe(":8080", nil)
    // Go 的 net/http 对每个请求自动创建 goroutine
    // 1000个并发请求 = 1000个 goroutine，无压力
}

11.1.2 字节的实践参考

Eino 的设计目标之一就是"高性能 Agent"。字节内部用 Go + Eino 跑的 Agent 服务，单机 QPS 轻松上万——这不是 Go 的极限，是 LLM API 的速率限制。

11.2 goroutine编排Agent工作流

11.2.1 Agent的典型工作流

一个 Agent 请求进来，通常要做这些事：

1. 意图识别（调 LLM）
2. 如果需要工具 → 并发调用多个工具
3. 工具结果汇总，再调 LLM 生成回答
4. （可选）反思检查

步骤2是并发的天然场景——各个工具之间没有依赖，等串行跑就是浪费。

func (a *Agent) Process(ctx context.Context, question string) (string, error) {
    // 步骤1：意图识别（这一步不能并发，必须先做）
    intent, err := a.classifyIntent(ctx, question)
    if err != nil {
        return "", err
    }

    // 步骤2：根据意图，并发调用相关工具
    toolResults := a.runToolsConcurrently(ctx, intent, question)

    // 步骤3：汇总，让 LLM 生成最终回答
    return a.generateResponse(ctx, question, intent, toolResults)
}

11.2.2 并发调用工具

func (a *Agent) runToolsConcurrently(ctx context.Context, 
    intent Intent, question string) []ToolResult {
    
    var tools []Tool
    switch intent.Type {
    case "weather":
        tools = []Tool{a.weatherTool}
    case "news":
        tools = []Tool{a.newsTool}
    case "comprehensive":
        tools = []Tool{a.weatherTool, a.newsTool, a.stockTool}
    }

    results := make([]ToolResult, len(tools))
    var wg sync.WaitGroup

    for i, tool := range tools {
        wg.Add(1)
        go func(idx int, t Tool) {
            defer wg.Done()
            
            result, err := t.Run(ctx, question)
            if err != nil {
                results[idx] = ToolResult{Error: err.Error()}
                return
            }
            results[idx] = result
        }(i, tool)
    }

    wg.Wait()
    return results
}

注意：goroutine 闭包里一定要传参数副本（idx, t），不能直接用循环变量——这是 Go 并发最常见的坑。

11.3 channel传递工具调用结果

11.3.1 缓冲channel：谁先回来先处理

sync.WaitGroup 适合"等全部完成"。但如果工具A只要200ms，工具B要2000ms——你不想等B。

func (a *Agent) runToolsWithStreaming(ctx context.Context, 
    question string, tools []Tool) []ToolResult {
    
    results := make(chan ToolResult, len(tools))
    
    for _, tool := range tools {
        go func(t Tool) {
            // 检查 ctx——用户可能关了页面
            select {
            case <-ctx.Done():
                return
            default:
            }
            
            result, err := t.Run(ctx, question)
            if err != nil {
                results <- ToolResult{Name: t.Name(), Error: err.Error()}
                return
            }
            results <- result
        }(tool)
    }

    // 收集结果——谁先回来先处理
    var collected []ToolResult
    for i := 0; i < len(tools); i++ {
        select {
        case r := <-results:
            collected = append(collected, r)
            fmt.Printf("[%s] 已完成\n", r.Name)
        case <-ctx.Done():
            return collected // 超时了，不等了
        }
    }
    return collected
}

11.3.2 带优先级的channel

更进一步——不是所有工具结果同等重要。天气数据对"今天出门要带什么"更关键，股价数据可有可无：

type PrioritizedResult struct {
    Result   ToolResult
    Priority int // 1=关键，2=重要，3=可选
}

func (a *Agent) runWithPriority(ctx context.Context, 
    question string) []ToolResult {
    
    highPriority := make(chan PrioritizedResult, 2)  // 天气、日历
    lowPriority := make(chan PrioritizedResult, 3)   // 新闻、股价、其他
    
    // 高优先级工具
    go func() {
        r := a.weatherTool.Run(ctx, question)
        highPriority <- PrioritizedResult{r, 1}
    }()
    go func() {
        r := a.calendarTool.Run(ctx, question)
        highPriority <- PrioritizedResult{r, 1}
    }()
    
    // 低优先级工具
    go func() {
        r := a.newsTool.Run(ctx, question)
        lowPriority <- PrioritizedResult{r, 3}
    }()
    
    // 先收高优先级的，有富余时间再收低优先级的
    var results []ToolResult
    timeout := time.After(2 * time.Second)
    
    for i := 0; i < 3; i++ {
        select {
        case r := <-highPriority:
            results = append(results, r.Result)
        case r := <-lowPriority:
            results = append(results, r.Result)
        case <-timeout:
            return results // 超时了，有多少返回多少
        }
    }
    return results
}

11.4 sync.WaitGroup和errgroup：管好并发任务

11.4.1 WaitGroup：最基础的并发控制

var wg sync.WaitGroup

for _, tool := range tools {
    wg.Add(1)
    go func(t Tool) {
        defer wg.Done()
        t.Run(ctx, question)
    }(tool)
}

wg.Wait() // 等所有工具返回
fmt.Println("全部完成")

WaitGroup 够用，但有两个问题：

goroutine 里的错误不好收集
一个goroutine panic 了，其他的还在跑

11.4.2 errgroup：带错误管理的并发

golang.org/x/sync/errgroup 是 WaitGroup 的增强版——任何一个 goroutine 出错，自动取消其他所有 goroutine：

import "golang.org/x/sync/errgroup"

func (a *Agent) runToolsWithErrgroup(ctx context.Context, 
    question string, tools []Tool) ([]ToolResult, error) {
    
    g, ctx := errgroup.WithContext(ctx)
    results := make([]ToolResult, len(tools))

    for i, tool := range tools {
        i, tool := i, tool // 闭包陷阱的修复方式
        g.Go(func() error {
            result, err := tool.Run(ctx, question)
            if err != nil {
                return fmt.Errorf("工具 %s 失败: %w", tool.Name(), err)
            }
            results[i] = result
            return nil
        })
    }

    // 任何一个出错，Wait 返回第一个错误，其他 goroutine 的 ctx 会被取消
    if err := g.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

errgroup 的妙处：ctx 会自动取消其他 goroutine。比如天气工具失败了，新闻和股价工具收到 ctx.Done()，立刻停止——不浪费 API 调用。

11.4.3 限制并发数：别把LLM API打爆

LLM API 有速率限制（Rate Limit）。你需要像信号量一样控制并发：

func (a *Agent) runWithLimit(ctx context.Context, 
    tasks []Task, maxConcurrency int) []Result {
    
    sem := make(chan struct{}, maxConcurrency) // 信号量
    results := make([]Result, len(tasks))
    var wg sync.WaitGroup

    for i, task := range tasks {
        wg.Add(1)
        go func(idx int, t Task) {
            defer wg.Done()
            
            sem <- struct{}{}        // 获取信号量
            defer func() { <-sem }() // 释放信号量
            
            results[idx] = t.Run(ctx)
        }(i, task)
    }

    wg.Wait()
    return results
}

// 使用：最多同时3个 goroutine 调 LLM
results := agent.runWithLimit(ctx, tasks, 3)

11.5 实战：用goroutine同时调10个工具

11.5.1 完整示例

把前面学的串起来——一个真实的"并发工具调用"场景：

func (a *Agent) HandleComplexQuery(ctx context.Context, 
    question string) (*AgentResponse, error) {
    
    // 第一步：意图识别（必须串行）
    intent, err := a.classifyIntent(ctx, question)
    if err != nil {
        return nil, err
    }

    // 第二步：并发调用所有相关工具
    tools := a.selectTools(intent)
    g, ctx := errgroup.WithContext(ctx)
    
    toolResults := &sync.Map{} // 并发安全的 map

    for _, tool := range tools {
        tool := tool
        g.Go(func() error {
            // 检查 ctx 是否已取消
            if ctx.Err() != nil {
                return ctx.Err()
            }
            
            result, err := tool.Run(ctx, question)
            if err != nil {
                return fmt.Errorf("%s: %w", tool.Name(), err)
            }
            toolResults.Store(tool.Name(), result)
            return nil
        })
    }

    // 等所有工具返回（或第一个错误）
    if err := g.Wait(); err != nil {
        // 部分工具成功了，部分失败了
        // Agent 可以用已有的结果继续回答
        log.Printf("部分工具调用失败: %v", err)
    }

    // 第三步：汇总结果，生成回答
    return a.generateResponse(ctx, question, toolResults)
}

11.5.2 性能对比

方案	10个工具（每个300ms）	说明
串行	3000ms	一个等一个
goroutine 并发	~300ms	取最慢的那个
errgroup + 限流	~600ms	限流5个并发，分两批

不是所有场景都适合全部并发。LLM API 有速率限制，盲目地开100个 goroutine 同时调，会被限流。errgroup + 信号量才是生产环境的正确姿势。

11.6 本章小结

这一章把Go的并发能力用到了Agent开发中：

goroutine编排工作流：每个请求一个goroutine，请求内工具并发调用——Go的 net/http 原生支持。
缓冲channel：工具之间没有依赖，谁先回来先处理。还能加优先级——关键数据优先收。
errgroup替代WaitGroup：任何一个工具出错，自动取消其他goroutine的context——不浪费API调用。
信号量限流：chan struct{} 做并发控制，避免打爆LLM API的速率限制。
实战组合：意图识别串行 + 工具调用并发 + errgroup 错误管理 + sync.Map 收集结果。

✅ 知识点检查

学完这一章，试试回答这几个问题：

[ ] goroutine 闭包为什么要传参数副本？
[ ] errgroup 比 WaitGroup 好在哪里？
[ ] 信号量限制并发数用什么数据结构？为什么？
[ ] Agent 工作流中哪些步骤可以并发，哪些必须串行？

📚 延伸阅读

Go 并发模式：https://go.dev/blog/pipelines
errgroup 文档：https://pkg.go.dev/golang.org/x/sync/errgroup
本书配套源码：关注公众号「图解AI系列」免费领取

🎯 下一章预告

第12章，Agent上线之后——

"上了线，就放心了？安全、可观测性、Eino Callbacks——让Agent每个动作都看得见。"

第11章 Go的并发，Agent的加速器 ​

11.1 一千个请求同时来，Python在排队，Go在飞 ​

11.1.1 真实的生产场景 ​

11.1.2 字节的实践参考 ​

11.2 goroutine编排Agent工作流 ​

11.2.1 Agent的典型工作流 ​

11.2.2 并发调用工具 ​

11.3 channel传递工具调用结果 ​

11.3.1 缓冲channel：谁先回来先处理 ​

11.3.2 带优先级的channel ​

11.4 sync.WaitGroup和errgroup：管好并发任务 ​

11.4.1 WaitGroup：最基础的并发控制 ​

11.4.2 errgroup：带错误管理的并发 ​

11.4.3 限制并发数：别把LLM API打爆 ​

11.5 实战：用goroutine同时调10个工具 ​

11.5.1 完整示例 ​

11.5.2 性能对比 ​

11.6 本章小结 ​

✅ 知识点检查 ​

📚 延伸阅读 ​

🎯 下一章预告 ​

喜欢这篇内容？关注我，别错过后续更新