多 Agent 协作系统架构设计：从编排模式到生产落地

多 Agent 协作系统架构设计：从编排模式到生产落地

一、单个 Agent 能做很多事，但做不了复杂事：多 Agent 协作的工程驱动因素

单 Agent 架构在处理复杂任务时有两个根本限制：一是 LLM 的上下文窗口有限，一个 Agent 无法同时掌握全局规划和局部执行所需的所有信息；二是单 Agent 的工具调用缺乏分工，当一块业务逻辑出错时，很难定位责任边界。

多 Agent 协作（Multi-Agent Collaboration）的出现不是追求架构上的花哨，而是来自实际工程约束：当业务系统需要同时对接三个数据源、执行五步校验、调用两个第三方 API，并且每一步都有独立的错误处理和降级逻辑时，用一个 Agent 的思维链（Chain-of-Thought）去编排所有步骤，既不精确也不可观测。将任务拆解为若干子任务，每个子任务由一个专注的 Agent 处理，再由协调者（Orchestrator）或路由器（Router）串联结果，才是可工程化的方案。

多 Agent 系统在交付时需要考虑：Agent 之间的通信协议、任务分解与分配策略、结果聚合与冲突解决、全局状态管理、错误传播与隔离。本文从这些维度展开，给出可落地的设计方案。

二、三种主流的 Agent 编排模式

flowchart TD subgraph Mode1[模式一：Orchestrator 集中编排] O[Orchestrator Agent\n全局调度器] --> A1[Analyzer Agent\n分析任务] O --> A2[Extractor Agent\n数据提取] O --> A3[Validator Agent\n校验任务] A1 --> O A2 --> O A3 --> O O --> Result1[聚合结果] end subgraph Mode2[模式二：Sequential Pipeline] Input --> AA[Agent A\n意图识别] AA --> BA[Agent B\n数据检索] BA --> CA[Agent C\n结果生成] CA --> Output end subgraph Mode3[模式三：Debate/Review 协作] G[Generator Agent\n生成方案] --> R1[Reviewer 1\n审查安全性] G --> R2[Reviewer 2\n审查性能] R1 --> D[Decision Agent\n裁决] R2 --> D D --> G D --> Final[最终方案] end

2.1 Orchestrator 模式

Orchestrator（编排器）是所有 Agent 的中心调度节点。它接收用户请求，拆解为子任务列表，按依赖关系分配给对应的 Worker Agent，收集结果后再聚合输出。这是最接近传统微服务架构的模式。

优势：全局可见性，Orchestrator 拥有所有子任务的状态和结果，便于监控和排障。

劣势：Orchestrator 是单点，其上下文窗口和决策能力决定了整个系统的上限。如果 Orchestrator 的任务拆解能力不够，Worker Agent 拿到的子任务质量就差。

适用场景：任务步骤可枚举且有明确顺序依赖的场景，如文档审核、订单履约。

2.2 Sequential Pipeline 模式

每个 Agent 只处理一个步骤，输出作为下一个 Agent 的输入。没有中心调度节点，Agent 之间通过共享的 Context 对象传递中间结果。

优势：每个 Agent 职责单一，prompt 设计简单，端到端延迟最低（无调度开销）。

劣势：Pipeline 中任何一步出错都会导致整个链路失败，且错误难以回滚。Agent A 输出的格式如果不符合 Agent B 的预期，需要额外的格式适配逻辑。

适用场景：数据转换类任务，如 ETL Pipeline、文本翻译流水线。

2.3 Debate/Review 模式

多个 Agent 从不同维度审查或生成同一份内容，由裁决者合并为最终结果。这种模式借鉴了软件开发中的 Code Review 流程。

优势：结果质量高，单一 Agent 的"幻觉"或偏差可以被另一个维度的 Agent 纠正。

劣势：Token 消耗高，延迟大，每条路径都可能产生大量输入输出。

适用场景：代码审查、安全合规审核、金融风控等需要多维度验证的场景。

三、生产级多 Agent 系统的 Go 实现

下面实现一个轻量级的多 Agent 编排引擎，支持以上三种模式的灵活组合。

3.1 Agent 抽象接口

package agent import ( "context" ) // Message 是 Agent 之间传递的数据单元。 type Message struct { Role string // "user" | "assistant" | "system" Content string // 消息内容 Meta map[string]string // 元数据，如来源 Agent 名称、步骤序号 } // Result 是 Agent 执行的输出。 type Result struct { AgentName string // 执行 Agent 的名称 Output string // 输出内容 Messages []Message // 完整对话历史，可用于 Debug Err error // 执行过程中的错误 } // Agent 定义了所有 Agent 组件必须实现的接口。 type Agent interface { Name() string Run(ctx context.Context, input Message) Result }

3.2 Orchestrator 实现

package orchestrator import ( "context" "fmt" "log" "strings" "your/agent" ) // Task 定义 Orchestrator 分配给 Worker 的一个子任务。 type Task struct { ID string Agent agent.Agent Input agent.Message Depends []string // 依赖的任务 ID 列表 } // Orchestrator 是集中式编排器。 // 它使用 LLM 将用户请求拆解为子任务列表，再按依赖关系调度 Worker Agent。 type Orchestrator struct { Name string Workers map[string]agent.Agent // 名称到 Agent 的映射 } // Run 执行 Orchestrator 的主流程。 // 1. Plan：调用 LLM 将请求拆解为子任务列表 // 2. Execute：按依赖顺序执行子任务 // 3. Summarize：聚合所有子任务结果 func (o *Orchestrator) Run(ctx context.Context, input agent.Message) agent.Result { // Step 1: Plan - 将用户请求拆解为子任务 tasks, err := o.plan(ctx, input.Content) if err != nil { return agent.Result{AgentName: o.Name, Err: fmt.Errorf("plan failed: %w", err)} } // Step 2: Execute - 按依赖顺序调度 results := make(map[string]agent.Result) for _, task := range tasks { // 检查依赖是否全部完成 for _, depID := range task.Depends { if _, ok := results[depID]; !ok { return agent.Result{ AgentName: o.Name, Err: fmt.Errorf("dependency %s not satisfied for task %s", depID, task.ID), } } } agentResult := task.Agent.Run(ctx, task.Input) results[task.ID] = agentResult if agentResult.Err != nil { log.Printf("[Orchestrator] task %s failed: %v", task.ID, agentResult.Err) // 错误隔离：一个 Worker 失败不影响其他无关任务 } } // Step 3: Summarize - 聚合结果 var summaries []string for id, result := range results { if result.Err == nil { summaries = append(summaries, fmt.Sprintf("[%s]: %s", id, result.Output)) } else { summaries = append(summaries, fmt.Sprintf("[%s]: ERROR - %s", id, result.Err)) } } return agent.Result{ AgentName: o.Name, Output: strings.Join(summaries, "\n"), } } // plan 调用 LLM 将用户请求拆解为子任务。 // 实际实现时应使用结构化输出的 LLM 调用，这里简化示意。 func (o *Orchestrator) plan(ctx context.Context, userRequest string) ([]Task, error) { // 伪代码： // prompt := fmt.Sprintf(` // 你是一个任务编排器。用户请求是： // %s // 请将上述请求拆解为子任务，每个子任务指定对应的 Agent 名称。 // 可用的 Agent 有：%s // 输出格式为 JSON 数组。`, userRequest, availableAgentNames(o.Workers)) // // response := callLLM(ctx, prompt) // tasks := parseJSON(response) return []Task{}, nil }

3.3 Sequential Pipeline 实现

package pipeline import ( "context" "fmt" "your/agent" ) // Pipeline 按顺序执行一组 Agent。 type Pipeline struct { Steps []agent.Agent } func (p *Pipeline) Run(ctx context.Context, input agent.Message) agent.Result { currentInput := input for i, step := range p.Steps { result := step.Run(ctx, currentInput) if result.Err != nil { return agent.Result{ AgentName: fmt.Sprintf("pipeline.step.%d.%s", i, step.Name()), Err: result.Err, } } // 当前 Agent 的输出作为下一个 Agent 的输入 currentInput = agent.Message{ Role: "assistant", Content: result.Output, Meta: map[string]string{"source": step.Name()}, } } return agent.Result{ AgentName: "pipeline", Output: currentInput.Content, } }

3.4 带超时和重试的 Agent 执行器

单个 Agent 的 LLM 调用可能因限流、网络抖动而失败，生产环境中必须引入超时和重试：

package executor import ( "context" "fmt" "time" "your/agent" ) // RetryConfig 重试配置。 type RetryConfig struct { MaxRetries int BaseBackoff time.Duration MaxBackoff time.Duration } // DefaultRetryConfig 提供默认的重试配置。 var DefaultRetryConfig = RetryConfig{ MaxRetries: 3, BaseBackoff: 500 * time.Millisecond, MaxBackoff: 10 * time.Second, } // Executor 包装 Agent，添加超时和重试能力。 type Executor struct { Agent agent.Agent Timeout time.Duration Retry RetryConfig } func (e *Executor) Run(ctx context.Context, input agent.Message) agent.Result { var lastErr error for attempt := 0; attempt <= e.Retry.MaxRetries; attempt++ { if attempt > 0 { // 退避：指数级递增，上限为 MaxBackoff backoff := e.Retry.BaseBackoff * (1 << (attempt - 1)) if backoff > e.Retry.MaxBackoff { backoff = e.Retry.MaxBackoff } select { case <-ctx.Done(): return agent.Result{AgentName: e.Agent.Name(), Err: ctx.Err()} case <-time.After(backoff): } } // 带超时的 Agent 执行 runCtx, cancel := context.WithTimeout(ctx, e.Timeout) result := e.Agent.Run(runCtx, input) cancel() if result.Err == nil { return result } lastErr = result.Err log.Printf("[Executor] agent %s attempt %d failed: %v", e.Agent.Name(), attempt+1, lastErr) } return agent.Result{ AgentName: e.Agent.Name(), Err: fmt.Errorf("agent %s failed after %d retries: %w", e.Agent.Name(), e.Retry.MaxRetries, lastErr), } }

四、编排模式的选型对比与工程取舍

各模式核心对比

落地建议

• 优先用 Pipeline：如果任务步骤是确定的，不要为了"用多 Agent"而引入 Orchestrator。Pipeline 的延迟最低，工程实现最简单。
• 需要容错时加 Orchestrator：当单个步骤可能失败，且失败后需要跳过或使用默认值时，Orchestrator 可以捕获子任务的结果并做出后续决策。
• 涉及安全合规时加 Review 环节：生成式结果的场景（如代码生成、合同起草），至少需要一个 Reviewer Agent 从不同角度复查结果。
• 不要过度拆解：多 Agent 不是 Agent 越多越好。每个额外的 Agent 都意味着一次 LLM 调用的延迟和成本。如果一个 Agent 能做的事，就不要拆成两个。
• 禁用场景：Agent 之间的通信频率高于 LLM 调用的收益时。如果 80% 的请求可以由单 Agent 处理，只有 20% 的复杂请求需要协作，应该优先走单 Agent 路径，只有识别到复杂请求时才路由到多 Agent 系统。

五、总结

多 Agent 协作是解决复杂任务的有效架构手段。三种编排模式各有适用场景：Pipeline 适合确定步骤的流水线任务、Orchestrator 适合需要动态拆解和全局状态管理的复杂流程、Debate/Review 适合要求多维度验证的高质量场景。工程落地的核心实践是：每个 Agent 的职责必须单一、Agent 间通信协议必须明确、错误处理必须隔离且可降级。多 Agent 和单 Agent 之间应该有清晰的流量路由策略，避免为简单请求付费不必要的 Agent 开销。