当前位置：首页 > news >正文

多 Agent 协作的工程实现：用 Rust Actor 模型构建 agent 通信网络

news 2026/7/27 3:00:40

多 Agent 协作的工程实现：用 Rust Actor 模型构建 agent 通信网络

一、单 Agent 的天花板与复盘：我们是怎样遇到瓶颈的

去年年底，我所在的团队把一个 AI 代码审查工具从单 Agent 升级为多 Agent。起因很朴素：单个 Agent 在代码量超过 2000 行时会"失忆"——上下文窗口装不下全部文件，审查结果拼凑感极重，经常漏掉跨文件的依赖关系。

我们尝试了三种方案：

加大上下文窗口——GPT-4o 128K tokens，勉强能装 3000 行，但注意力在 30K tokens 后明显衰减；
分批审查 + 合并结果——让 Agent 分批审查，最后合并。问题在于"合并"环节变成了新的瓶颈；
多 Agent 协作——5 个 Agent 各司其职：语法检查 Agent、安全 Agent、风格 Agent、逻辑 Agent、汇总 Agent。这是我最终落地并跑在生产环境的方案。

这篇文章是我对多 Agent 协作的工程复盘——我们会从架构设计、消息通信、状态管理到生产部署完整拆解。

二、Actor 模型选型：为什么不用微服务

第一次讨论方案时，有人提出用微服务——每个 Agent 一个 HTTP 服务，通过 REST API 通信。我否决了这个方案，原因很简单：审查一个 2000 行的文件，如果 5 个 Agent 之间通过 HTTP 交换信息，光是网络延迟加起来就超过 500ms。

Actor 模型要解决的正是这个问题：同一进程内、异步、消息驱动的通信。不需要序列化/反序列化，不需要 TCP 握手，消息传递的开销接近于函数调用。

use tokio::sync::mpsc; use std::sync::Arc; /// Actor 消息枚举——所有 Agent 间通信的统一格式 #[derive(Debug, Clone)] pub enum AgentMessage { /// 任务分配——Orchestrator 发给工作 Agent TaskAssign { task_id: String, code_snippet: String, file_path: String, }, /// 审查结果——工作 Agent 发给汇总 Agent ReviewResult { task_id: String, agent_name: String, findings: Vec<Finding>, severity: Severity, }, /// 工作 Agent 请求额外上下文——跨 Agent 通信 ContextRequest { task_id: String, from_agent: String, to_agent: String, question: String, }, /// 被请求 Agent 的响应 ContextResponse { task_id: String, content: String, }, /// 系统消息 Shutdown, HealthCheck, } /// 审查发现——每个 Agent 输出统一的格式 #[derive(Debug, Clone)] pub struct Finding { pub line_number: usize, pub severity: Severity, pub category: String, pub description: String, pub suggestion: String, } #[derive(Debug, Clone, PartialEq)] pub enum Severity { Error, Warning, Info, Suggestion, } /// Agent trait——所有 Agent 的抽象接口 /// 使用 async_trait 宏支持异步 trait 方法 #[async_trait::async_trait] pub trait Agent: Send + Sync { /// Agent 名称——用于日志和路由 fn name(&self) -> &str; /// 处理单个任务——这是 Agent 的核心逻辑 async fn process(&self, task: AgentMessage) -> Vec<AgentMessage>; /// 启动 Agent 的消息循环 async fn run( &self, mut receiver: mpsc::Receiver<AgentMessage>, sender: mpsc::Sender<AgentMessage>, ) { while let Some(msg) = receiver.recv().await { // 检测到 Shutdown 消息，退出循环 if matches!(msg, AgentMessage::Shutdown) { break; } let agent_name = self.name().to_string(); let responses = self.process(msg).await; // 将返回消息发送到总线 for response in responses { if let Err(e) = sender.send(response).await { eprintln!("[{}] 发送消息失败: {}", agent_name, e); break; } } } } }

三、Orchestrator：多 Agent 的指挥官

Orchestrator 是整个系统的核心——它负责把用户请求拆解为子任务并分配给对应 Agent。

use std::collections::HashMap; use tokio::sync::oneshot; /// 编排器——多 Agent 系统的控制器 pub struct Orchestrator { /// 向工作 Agent 发送消息的通道 agent_senders: HashMap<String, mpsc::Sender<AgentMessage>>, /// 汇总 Agent 的发送通道 summarizer_tx: mpsc::Sender<AgentMessage>, /// 接收消息的总线 result_rx: tokio::sync::Mutex<mpsc::Receiver<AgentMessage>>, } impl Orchestrator { /// 执行代码审查——同步入口，隐藏异步细节 pub async fn review_code(&mut self, code: &str, file_path: &str) -> Vec<Finding> { let task_id = uuid::Uuid::new_v4().to_string(); let mut all_findings = vec![]; let mut received = 0; let expected = self.agent_senders.len(); // 第一步：向所有工作 Agent 分发任务 for (name, tx) in &self.agent_senders { let msg = AgentMessage::TaskAssign { task_id: task_id.clone(), code_snippet: code.to_string(), file_path: file_path.to_string(), }; if let Err(e) = tx.send(msg).await { eprintln!("[Orchestrator] 向 Agent {} 发送任务失败: {}", name, e); } } // 第二步：收集所有工作 Agent 的结果 let mut guard = self.result_rx.lock().await; while received < expected { if let Some(msg) = guard.recv().await { match msg { AgentMessage::ReviewResult { task_id: _, findings, .. } => { all_findings.extend(findings); received += 1; } AgentMessage::ContextRequest { task_id: _, to_agent, question, .. } => { // 转发上下文请求到目标 Agent if let Some(tx) = self.agent_senders.get(&to_agent) { let ctx_msg = AgentMessage::ContextResponse { task_id: task_id.clone(), content: format!("回答: {}", question), }; let _ = tx.send(ctx_msg).await; } } _ => {} } } } drop(guard); // 第三步：发给汇总 Agent 做聚合 let _ = self.summarizer_tx.send(AgentMessage::ReviewResult { task_id: task_id.clone(), agent_name: "orchestrator".to_string(), findings: all_findings.clone(), severity: Severity::Info, }).await; // 等待汇总结果（简化处理，直接返回） all_findings } }

实战踩坑记录

我在第一次实现 Orchestrator 时犯过一个低级错误——在review_code方法里用了await等待所有 Agent 返回，但没有设置超时。结果有一个 Agent 因为 LLM API 超时，整个审查流程卡了 2 分钟。后来加了tokio::select!宏做超时控制，问题才解决。

// 超时控制的正确写法——用 tokio::select! 同时等待消息和超时 use tokio::time::{timeout, Duration}; async fn recv_with_timeout( receiver: &mut mpsc::Receiver<AgentMessage>, ) -> Option<AgentMessage> { match timeout(Duration::from_secs(30), receiver.recv()).await { Ok(Some(msg)) => Some(msg), Ok(None) => None, // 通道关闭 Err(_) => { // 超时 eprintln!("[Orchestrator] 等待 Agent 响应超时"); None } } }

另一个坑是mpsc::channel的容量设置。我一开始设的是 1（默认），结果高并发时消息丢失——发送方在通道满时会直接丢弃消息。改成 100 后稳定了。这个经验后来写进了团队的 Rust 异步编程规范。

四、消息总线的设计与汇合 Agent

消息总线是多 Agent 系统的通信骨架。我们的实现用 mpsc channel 构建了一个简单的 hub。

/// 消息总线——所有 Agent 的消息中转站 pub struct MessageBus { /// 按 Agent 名称索引的发送通道 agent_senders: HashMap<String, mpsc::Sender<AgentMessage>>, } impl MessageBus { pub fn new() -> Self { Self { agent_senders: HashMap::new(), } } /// 注册一个 Agent——给它分配独立的 mpsc 通道 pub fn register(&mut self, name: &str, capacity: usize) -> mpsc::Receiver<AgentMessage> { let (tx, rx) = mpsc::channel(capacity); self.agent_senders.insert(name.to_string(), tx); rx } /// 向指定 Agent 发送消息 pub async fn send_to(&self, agent: &str, msg: AgentMessage) -> Result<(), BusError> { match self.agent_senders.get(agent) { Some(sender) => { sender.send(msg).await.map_err(|_| BusError::AgentDisconnected(agent.to_string())) } None => Err(BusError::AgentNotFound(agent.to_string())), } } /// 向所有 Agent 广播消息 pub async fn broadcast(&self, msg: AgentMessage) { for (name, sender) in &self.agent_senders { if let Err(e) = sender.send(msg.clone()).await { eprintln!("[Bus] 向 {} 广播失败: {:?}", name, e); } } } }

汇总 Agent 的实现逻辑：

/// 汇总 Agent——去重、排序、聚合多个工作 Agent 的发现 pub struct SummarizerAgent { name: String, } impl SummarizerAgent { fn merge_findings(&self, findings: Vec<Finding>) -> Vec<Finding> { let mut merged = Vec::new(); let mut seen = std::collections::HashSet::new(); for finding in findings { // 去重——同一行同一类的发现只保留一个 let key = (finding.line_number, finding.category.clone()); if !seen.contains(&key) { seen.insert(key); merged.push(finding); } } // 按行号排序——方便用户按文件顺序查看 merged.sort_by_key(|f| f.line_number); // 按严重程度排序——严重问题排前面 merged.sort_by_key(|f| match f.severity { Severity::Error => 0, Severity::Warning => 1, Severity::Info => 2, Severity::Suggestion => 3, }); merged } }

汇总 Agent 的去重逻辑在上线后暴露了一个边界问题：两个 Agent 报告同一行的不同类别问题时，merge_findings只保留了第一个。我们加了一个合并逻辑，把同类发现用;分隔描述。这个改动让报告的可读性显著提升。

性能方面，5 个 Agent 并行审查一个 3000 行的文件，P50 延迟是 8.2 秒，P99 是 15.7 秒。瓶颈在 LLM API 的响应时间，不是 Actor 通信本身。如果换成本地 ONNX 模型做初步筛选，预计能把 P50 压到 3 秒以内。

另外，Actor 的 channel buffer 必须设上限——我们没用unbounded_channel，防止某个慢 Agent 堆积消息撑爆内存。

五、总结

多 Agent 协作的本质是把"一个大问题"拆成"多个小问题"，让每个 Agent 专注自己的能力边界。Rust 的 actor 模型实现比微服务轻量太多——不需要 docker compose、不需要端口管理、不需要 protobuf 编译——写一个enum AgentMessage加几个mpsc::channel就能跑通。

这次多 Agent 改造成果：