第5讲：多 Agent 协作——Orchestrator-Worker 模式从零实现-恒创新域建站

★ FEATURED ARTICLE

一、为什么需要多 Agent单一 Agent 的瓶颈问题表现后果上下文污染一个 Agent 既要规划又要执行思考步骤挤占工具调用空间容易丢失上下文忘记前面的结论工具集膨胀所有工具塞给一个 AgentPrompt 越来越长LLM 选择困难调用准确率下降职责不清同一个 Agent 既写代码又审代码自己写的自己审发现不了问题无法并行串行执行一步卡住全部卡住响应时间长用户体验差单点故障Agent 出错整个任务失败缺乏容错和兜底机制多 Agent 的核心思路分而治之——每个 Agent 只做一件事做深做专。┌─────────────────────────────────────────────────────────────┐ │ 多 Agent 协作架构 │ │ │ │ 用户任务 │ │ │ │ │ ▼ │ │ ┌──────────────────────────────────────────────────────┐ │ │ │ Orchestrator (调度器) │ │ │ │ • 任务理解与分解 │ │ │ │ • 子任务分配 │ │ │ │ • 进度跟踪 │ │ │ │ • 结果聚合 │ │ │ └────────┬──────────┬──────────┬───────────────────────┘ │ │ │ │ │ │ │ ┌─────▼────┐ ┌──▼─────┐ ┌──▼──────┐ │ │ │Researcher│ │ Coder │ │Reviewer │ ... │ │ │ Agent │ │ Agent │ │ Agent │ │ │ └────┬─────┘ └──┬─────┘ └──┬──────┘ │ │ │ │ │ │ │ ▼ ▼ ▼ │ │ ┌─────────────────────────────────────┐ │ │ │ 结果聚合器 │ │ │ │ (合并、去重、冲突仲裁) │ │ │ └──────────────────┬──────────────────┘ │ │ │ │ │ ▼ │ │ 最终输出 │ └─────────────────────────────────────────────────────────────┘二、核心数据结构// multiagent/types.go package multiagent import ( context encoding/json time ) // AgentID Agent 唯一标识 type AgentID string // TaskStatus 任务状态 type TaskStatus string const ( StatusPending TaskStatus pending StatusRunning TaskStatus running StatusCompleted TaskStatus completed StatusFailed TaskStatus failed StatusCancelled TaskStatus cancelled ) // Task 任务定义 type Task struct { ID string json:id ParentID string json:parent_id,omitempty // 父任务 ID Description string json:description // 任务描述 AgentID AgentID json:agent_id // 分配给哪个 Agent Input json.RawMessage json:input Output json.RawMessage json:output,omitempty Status TaskStatus json:status Dependencies []string json:dependencies // 依赖的任务 ID Priority int json:priority // 优先级 0-10 MaxRetries int json:max_retries RetryCount int json:retry_count CreatedAt time.Time json:created_at StartedAt time.Time json:started_at,omitempty CompletedAt time.Time json:completed_at,omitempty Error string json:error,omitempty Metadata map[string]interface{} json:metadata,omitempty } // SubTaskResult 子任务结果 type SubTaskResult struct { TaskID string json:task_id AgentID AgentID json:agent_id Output json.RawMessage json:output Success bool json:success Error string json:error,omitempty Duration time.Duration json:duration } // AgentCapability Agent 能力声明 type AgentCapability struct { AgentID AgentID json:agent_id Name string json:name Description string json:description Tools []string json:tools // 拥有的工具列表 MaxConcurrent int json:max_concurrent // 最大并发数 AvgLatency time.Duration json:avg_latency } // AgentWorker Agent 工作单元接口 type AgentWorker interface { ID() AgentID Capability() AgentCapability Execute(ctx context.Context, task Task) (json.RawMessage, error) CanHandle(task Task) bool } // Message Agent 间消息 type Message struct { From AgentID json:from To AgentID json:to Type string json:type // request / response / broadcast Content json.RawMessage json:content TaskID string json:task_id,omitempty Timestamp time.Time json:timestamp Metadata map[string]interface{} json:metadata,omitempty }三、Orchestrator 实现3.1 任务分解器// multiagent/orchestrator.go package multiagent import ( context encoding/json fmt log sync time ) // Orchestrator 调度器 type Orchestrator struct { mu sync.RWMutex workers map[AgentID]AgentWorker taskQueue chan Task tasks map[string]*Task results map[string]SubTaskResult decomposer TaskDecomposer aggregator ResultAggregator maxWorkers int } type TaskDecomposer interface { Decompose(ctx context.Context, taskDescription string) ([]Task, error) } type ResultAggregator interface { Aggregate(ctx context.Context, results []SubTaskResult) (json.RawMessage, error) } func NewOrchestrator(decomposer TaskDecomposer, aggregator ResultAggregator, maxWorkers int) *Orchestrator { return Orchestrator{ workers: make(map[AgentID]AgentWorker), taskQueue: make(chan Task, 1000), tasks: make(map[string]*Task), results: make(map[string]SubTaskResult), decomposer: decomposer, aggregator: aggregator, maxWorkers: maxWorkers, } } // RegisterWorker 注册 Worker func (o *Orchestrator) RegisterWorker(worker AgentWorker) { o.mu.Lock() defer o.mu.Unlock() o.workers[worker.ID()] worker log.Printf(Worker registered: %s (%s), worker.ID(), worker.Capability().Name) } // SubmitTask 提交主任务 func (o *Orchestrator) SubmitTask(ctx context.Context, description string) (string, error) { // 1. 任务分解 subTasks, err : o.decomposer.Decompose(ctx, description) if err ! nil { return , fmt.Errorf(task decomposition failed: %w, err) } // 2. 创建主任务 mainTask : Task{ ID: generateID(), Description: description, Status: StatusRunning, CreatedAt: time.Now(), MaxRetries: 3, } o.mu.Lock() o.tasks[mainTask.ID] mainTask o.mu.Unlock() // 3. 分配子任务 for i : range subTasks { subTasks[i].ParentID mainTask.ID o.assignTask(subTasks[i]) } return mainTask.ID, nil } // assignTask 分配任务到合适的 Worker func (o *Orchestrator) assignTask(task *Task) { o.mu.Lock() defer o.mu.Unlock() // 找到最适合的 Worker bestWorker : o.selectWorker(task) if bestWorker nil { task.Status StatusFailed task.Error no suitable worker found return } task.AgentID bestWorker.ID() task.Status StatusPending o.tasks[task.ID] task // 异步执行 go o.executeTask(task, bestWorker) } // selectWorker 选择合适的 Worker func (o *Orchestrator) selectWorker(task *Task) AgentWorker { var bestWorker AgentWorker var bestScore float64 for _, worker : range o.workers { if !worker.CanHandle(*task) { continue } // 评分匹配度 × 负载因子 cap : worker.Capability() matchScore : o.calculateMatchScore(task, cap) loadFactor : 1.0 - float64(o.countActiveTasks(worker.ID()))/float64(cap.MaxConcurrent) totalScore : matchScore * loadFactor if totalScore bestScore { bestScore totalScore bestWorker worker } } return bestWorker } // calculateMatchScore 计算任务与 Worker 的匹配度 func (o *Orchestrator) calculateMatchScore(task *Task, cap AgentCapability) float64 { // 基于工具重叠度计算 taskTools : extractToolNames(task.Description) if len(taskTools) 0 { return 0.5 // 默认中等匹配 } matched : 0 for _, tt : range taskTools { for _, wt : range cap.Tools { if tt wt { matched break } } } return float64(matched) / float64(len(taskTools)) } // executeTask 执行任务 func (o *Orchestrator) executeTask(task *Task, worker AgentWorker) { task.Status StatusRunning task.StartedAt time.Now() ctx, cancel : context.WithTimeout(context.Background(), 5*time.Minute) defer cancel() output, err : worker.Execute(ctx, *task) task.CompletedAt time.Now() if err ! nil { task.Status StatusFailed task.Error err.Error() // 重试逻辑 if task.RetryCount task.MaxRetries { task.RetryCount log.Printf(Retrying task %s (attempt %d/%d), task.ID, task.RetryCount, task.MaxRetries) o.assignTask(task) return } } else { task.Status StatusCompleted task.Output output } // 保存结果 result : SubTaskResult{ TaskID: task.ID, AgentID: worker.ID(), Output: output, Success: err nil, Error: task.Error, Duration: task.CompletedAt.Sub(task.StartedAt), } o.mu.Lock() o.results[task.ID] result o.mu.Unlock() // 检查主任务是否完成 o.checkMainTaskCompletion(task.ParentID) } // checkMainTaskCompletion 检查主任务是否完成 func (o *Orchestrator) checkMainTaskCompletion(mainTaskID string) { o.mu.RLock() mainTask, exists : o.tasks[mainTaskID] o.mu.RUnlock() if !exists { return } // 收集所有子任务结果 var allResults []SubTaskResult allCompleted : true o.mu.RLock() for _, task : range o.tasks { if task.ParentID mainTaskID { if task.Status ! StatusCompleted { allCompleted false break } if result, ok : o.results[task.ID]; ok { allResults append(allResults, result) } } } o.mu.RUnlock() if !allCompleted { return } // 聚合结果 ctx : context.Background() aggregated, err : o.aggregator.Aggregate(ctx, allResults) if err ! nil { mainTask.Status StatusFailed mainTask.Error fmt.Sprintf(aggregation failed: %v, err) return } mainTask.Status StatusCompleted mainTask.Output aggregated mainTask.CompletedAt time.Now() log.Printf(Main task %s completed with %d subtasks, mainTaskID, len(allResults)) } // GetResult 获取任务结果 func (o *Orchestrator) GetResult(taskID string) (*Task, error) { o.mu.RLock() defer o.mu.RUnlock() task, exists : o.tasks[taskID] if !exists { return nil, fmt.Errorf(task %s not found, taskID) } return task, nil } // countActiveTasks 统计 Worker 正在执行的任务数 func (o *Orchestrator) countActiveTasks(agentID AgentID) int { count : 0 for _, task : range o.tasks { if task.AgentID agentID task.Status StatusRunning { count } } return count }四、Worker 实现示例4.1 Researcher Worker// multiagent/workers/researcher.go package workers import ( context encoding/json fmt time your-project/multiagent ) // ResearcherWorker 研究员 Worker type ResearcherWorker struct { id multiagent.AgentID tools map[string]interface{} // 搜索、读取工具 } func NewResearcherWorker() *ResearcherWorker { return ResearcherWorker{ id: researcher-001, tools: map[string]interface{}{ web_search: nil, read_doc: nil, summarize: nil, }, } } func (rw *ResearcherWorker) ID() multiagent.AgentID { return rw.id } func (rw *ResearcherWorker) Capability() multiagent.AgentCapability { return multiagent.AgentCapability{ AgentID: rw.id, Name: Researcher, Description: 擅长搜索和整理信息能够从互联网和文档中提取关键信息, Tools: []string{web_search, read_doc, summarize}, MaxConcurrent: 3, AvgLatency: 5 * time.Second, } } func (rw *ResearcherWorker) CanHandle(task multiagent.Task) bool { // 判断是否包含研究类关键词 keywords : []string{搜索, 查找, 调研, 研究, 收集, 查询, research, search, find} for _, kw : range keywords { if contains(task.Description, kw) { return true } } return false } func (rw *ResearcherWorker) Execute(ctx context.Context, task multiagent.Task) (json.RawMessage, error) { // 模拟研究过程 result : map[string]interface{}{ findings: []map[string]interface{}{ { topic: task.Description, summary: 这是关于 task.Description 的研究结果..., sources: []string{source1, source2}, }, }, confidence: 0.85, } data, _ : json.Marshal(result) return data, nil }4.2 Coder Worker// multiagent/workers/coder.go package workers import ( context encoding/json fmt time your-project/multiagent ) // CoderWorker 程序员 Worker type CoderWorker struct { id multiagent.AgentID } func NewCoderWorker() *CoderWorker { return CoderWorker{ id: coder-001, } } func (cw *CoderWorker) ID() multiagent.AgentID { return cw.id } func (cw *CoderWorker) Capability() multiagent.AgentCapability { return multiagent.AgentCapability{ AgentID: cw.id, Name: Coder, Description: 擅长编写代码、调试和代码优化, Tools: []string{write_code, run_test, debug, refactor}, MaxConcurrent: 2, AvgLatency: 10 * time.Second, } } func (cw *CoderWorker) CanHandle(task multiagent.Task) bool { keywords : []string{编码, 编程, 写代码, 实现, 开发, code, implement, develop} for _, kw : range keywords { if contains(task.Description, kw) { return true } } return false } func (cw *CoderWorker) Execute(ctx context.Context, task multiagent.Task) (json.RawMessage, error) { // 模拟编码过程 result : map[string]interface{}{ code: // Generated code for: task.Description, language: Go, tests: []string{Test1 passed, Test2 passed}, coverage: 92.5, } data, _ : json.Marshal(result) return data, nil }4.3 Reviewer Worker// multiagent/workers/reviewer.go package workers import ( context encoding/json fmt time your-project/multiagent ) // ReviewerWorker 审查员 Worker type ReviewerWorker struct { id multiagent.AgentID } func NewReviewerWorker() *ReviewerWorker { return ReviewerWorker{ id: reviewer-001, } } func (rw *ReviewerWorker) ID() multiagent.AgentID { return rw.id } func (rw *ReviewerWorker) Capability() multiagent.AgentCapability { return multiagent.AgentCapability{ AgentID: rw.id, Name: Reviewer, Description: 擅长代码审查、质量检查和改进建议, Tools: []string{code_review, quality_check, suggest_improvements}, MaxConcurrent: 4, AvgLatency: 3 * time.Second, } } func (rw *ReviewerWorker) CanHandle(task multiagent.Task) bool { keywords : []string{审查, 检查, 审核, 评审, review, check, audit, inspect} for _, kw : range keywords { if contains(task.Description, kw) { return true } } return false } func (rw *ReviewerWorker) Execute(ctx context.Context, task multiagent.Task) (json.RawMessage, error) { result : map[string]interface{}{ issues: []map[string]interface{}{ { severity: medium, line: 42, message: 变量命名不规范建议改为驼峰命名, }, }, score: 85, suggestions: []string{添加错误处理, 优化性能}, approved: false, } data, _ : json.Marshal(result) return data, nil }五、任务分解器与结果聚合器5.1 LLM 驱动的任务分解// multiagent/decomposer.go package multiagent import ( context encoding/json fmt strings ) // LLMDecomposer 基于 LLM 的任务分解器 type LLMDecomposer struct { llm LLMClient } type LLMClient interface { Chat(ctx context.Context, messages []Message) (string, error) } func NewLLMDecomposer(llm LLMClient) *LLMDecomposer { return LLMDecomposer{ llm: llm, } } func (ld *LLMDecomposer) Decompose(ctx context.Context, taskDescription string) ([]Task, error) { prompt : fmt.Sprintf(你是一个任务分解专家。请将以下任务分解为多个子任务。任务%s 要求 1. 每个子任务应该是独立的、可执行的 2. 标记子任务之间的依赖关系 3. 为每个子任务指定需要的工具类型search / code / review / analyze / write 4. 按执行顺序排列请以 JSON 数组格式返回每个元素包含 - description: 子任务描述 - dependencies: 依赖的子任务序号从0开始 - required_tools: 需要的工具列表返回格式示例 [ {description: 搜索相关资料, dependencies: [], required_tools: [search]}, {description: 分析搜索结果, dependencies: [0], required_tools: [analyze]} ], taskDescription) resp, err : ld.llm.Chat(ctx, []Message{ {Role: system, Content: 你是一个任务分解专家输出严格的 JSON 格式。}, {Role: user, Content: prompt}, }) if err ! nil { return nil, fmt.Errorf(LLM call failed: %w, err) } // 解析 JSON var rawTasks []struct { Description string json:description Dependencies []int json:dependencies RequiredTools []string json:required_tools } // 清理可能的 Markdown 代码块标记 resp strings.TrimSpace(resp) resp strings.TrimPrefix(resp, json) resp strings.TrimPrefix(resp, ) resp strings.TrimSuffix(resp, ) resp strings.TrimSpace(resp) if err : json.Unmarshal([]byte(resp), rawTasks); err ! nil { return nil, fmt.Errorf(parse decomposition failed: %w, err) } // 转换为 Task tasks : make([]Task, len(rawTasks)) depMap : make(map[int]string) // 序号 - task ID for i, rt : range rawTasks { taskID : generateID() depMap[i] taskID var deps []string for _, depIdx : range rt.Dependencies { if depID, ok : depMap[depIdx]; ok { deps append(deps, depID) } } tasks[i] Task{ ID: taskID, Description: rt.Description, Dependencies: deps, Status: StatusPending, CreatedAt: time.Now(), MaxRetries: 2, Metadata: map[string]interface{}{ required_tools: rt.RequiredTools, }, } } return tasks, nil }5.2 结果聚合器// multiagent/aggregator.go package multiagent import ( context encoding/json fmt strings ) // LLMAggregator 基于 LLM 的结果聚合器 type LLMAggregator struct { llm LLMClient } func NewLLMAggregator(llm LLMClient) *LLMAggregator { return LLMAggregator{ llm: llm, } } func (la *LLMAggregator) Aggregate(ctx context.Context, results []SubTaskResult) (json.RawMessage, error) { // 构建聚合 Prompt var parts []string for i, r : range results { parts append(parts, fmt.Sprintf( 【子任务 %d】由 %s 完成:\n%s\n, i1, r.AgentID, string(r.Output), )) } prompt : fmt.Sprintf(请将以下多个子任务的结果整合成一个完整的最终输出。子任务结果 %s 要求 1. 去除重复信息 2. 保持逻辑连贯 3. 突出关键结论 4. 标注信息来源 5. 如有冲突说明分歧点请给出整合后的最终答案。, strings.Join(parts, \n)) resp, err : la.llm.Chat(ctx, []Message{ {Role: system, Content: 你是一个结果整合专家善于将分散的信息组织成完整的答案。}, {Role: user, Content: prompt}, }) if err ! nil { return nil, fmt.Errorf(aggregation LLM call failed: %w, err) } result : map[string]interface{}{ final_answer: resp, subtask_count: len(results), aggregated_at: time.Now(), } data, _ : json.Marshal(result) return data, nil }六、完整使用示例// examples/multiagent/main.go package main import ( context encoding/json fmt log time your-project/multiagent your-project/multiagent/workers ) func main() { // 1. 创建 LLM 客户端实际项目中替换为真实 LLM llm : mockLLM{} // 2. 创建分解器和聚合器 decomposer : multiagent.NewLLMDecomposer(llm) aggregator : multiagent.NewLLMAggregator(llm) // 3. 创建 Orchestrator orchestrator : multiagent.NewOrchestrator(decomposer, aggregator, 10) // 4. 注册 Worker orchestrator.RegisterWorker(workers.NewResearcherWorker()) orchestrator.RegisterWorker(workers.NewCoderWorker()) orchestrator.RegisterWorker(workers.NewReviewerWorker()) // 5. 提交任务 ctx : context.Background() taskID, err : orchestrator.SubmitTask(ctx, 开发一个用户登录功能包括前端页面和后端API) if err ! nil { log.Fatal(err) } fmt.Printf(Task submitted: %s\n, taskID) // 6. 等待结果 time.Sleep(5 * time.Second) result, err : orchestrator.GetResult(taskID) if err ! nil { log.Fatal(err) } fmt.Printf(\n 最终结果 \n) fmt.Printf(Status: %s\n, result.Status) if result.Output ! nil { var prettyJSON map[string]interface{} json.Unmarshal(result.Output, prettyJSON) data, _ : json.MarshalIndent(prettyJSON, , ) fmt.Printf(Output:\n%s\n, string(data)) } } // mockLLM 模拟 LLM type mockLLM struct{} func (m *mockLLM) Chat(ctx context.Context, messages []multiagent.Message) (string, error) { // 模拟任务分解 return [ {description: 设计登录页面前端UI, dependencies: [], required_tools: [code]}, {description: 实现后端登录API, dependencies: [], required_tools: [code]}, {description: 前后端联调测试, dependencies: [0, 1], required_tools: [code, review]} ], nil }七、关键设计决策7.1 同步 vs 异步模式优点缺点适用场景同步实现简单结果即时整体耗时各任务之和步骤强依赖、任务少异步可并行吞吐高需要轮询/回调任务多、有独立子任务本讲采用异步模式Orchestrator 提交任务后立即返回 taskIDWorker 异步执行完成后回调。7.2 通信方式Agent 间通信有三种模式 1. 共享内存本实现采用 - Orchestrator 维护全局任务状态 - Worker 读写共享的任务表 - 优点延迟低实现简单 - 缺点耦合度高不适合分布式 2. 消息队列生产环境推荐 - RabbitMQ / Kafka / NATS - Worker 订阅自己的任务队列 - 优点解耦、可靠、可扩展 - 缺点引入额外组件 3. HTTP/gRPC分布式场景 - Worker 作为独立服务部署 - Orchestrator 通过 RPC 调用 - 优点物理隔离独立扩缩容 - 缺点网络开销序列化成本7.3 错误处理策略错误类型处理方式Worker 超时重试最多 N 次更换 WorkerWorker 崩溃检测心跳重新分配任务结果不一致多个 Worker 执行相同任务投票仲裁依赖任务失败级联失败或启用兜底方案LLM 调用失败指数退避重试八、本讲总结┌─────────────────────────────────────────────────────────────┐ │ 第5讲核心收获 │ │ │ │ ✅ 理解了为什么需要多 Agent │ │ - 单一 Agent 的瓶颈上下文污染、工具膨胀、职责不清 │ │ - 分而治之每个 Agent 只做一件事做深做专 │ │ │ │ ✅ 实现了 Orchestrator-Worker 架构 │ │ - Orchestrator任务分解、分配、进度跟踪、结果聚合 │ │ - Worker独立执行单元声明自己的能力 │ │ - 智能调度基于匹配度和负载因子的 Worker 选择 │ │ │ │ ✅ 实现了三种典型 Worker │ │ - Researcher搜索和信息整理 │ │ - Coder编码和调试 │ │ - Reviewer审查和质量检查 │ │ │ │ ✅ 实现了 LLM 驱动的任务分解和结果聚合 │ │ │ └─────────────────────────────────────────────────────────────┘九、下一讲预告第6讲《Agent 安全体系——Prompt 注入防御、工具调用审计与输出合规》你将学到Prompt 注入的常见手法与检测工具调用的权限控制与审计日志输出的敏感信息过滤与合规检查完整的 Agent 安全框架实现开发之余的小工具推荐调试多 Agent 系统时经常需要查看 JSON 格式的任务分解结果和 Worker 间的消息。我用 zz365.top 的 JSON 格式化器来处理这些所有数据都在浏览器本地处理不上传服务器。免费无广告适合开发时随手使用。

阅读完成 · 觉得有帮助？

咨询建站