AI Agent 的并发模型多个 agent 实例共享同一个工具注册表怎么办这其实是一个经典的并发资源管理问题。这篇文章我会记录我的完整思考过程——从问题出发尝试几种方案分析各自的优缺点。一、问题场景多个 Agent 抢同一个工具先把这个场景具象化。假设你有 3 个 Agent分别处理不同的用户请求但它们共享同一套工具sequenceDiagram participant U1 as 用户1 participant U2 as 用户2 participant U3 as 用户3 participant A1 as Agent实例1 participant A2 as Agent实例2 participant A3 as Agent实例3 participant T as 工具注册表(共享) participant API as 外部API U1-A1: 查一下天气 U2-A2: 帮我算个数 U3-A3: 查数据库 A1-T: 调用 search_weather 工具 A2-T: 调用 calculator 工具 A3-T: 调用 db_query 工具 Note over T: ⚠️ 三个Agent同时调用br/工具注册表怎么处理 T-API: search_weather(北京) T-API: calculator(2^10) T-API: db_query(SELECT...)工具注册表在这里成了共享热点。如果设计不好就会出现数据竞争两个 Agent 同时修改工具的内部状态死锁Agent A 等着工具释放Agent B 也等着性能瓶颈所有请求排队等一个锁二、方案分析三种工具注册表设计方案A全局 ArcMutex最直接use std::collections::HashMap; use std::sync::{Arc, Mutex}; /// 工具定义每个工具是一个可以被调用的函数 type ToolFn Boxdyn Fn(String) - String Send Sync; /// 工具注册表 —— 全局共享一份 /// ArcMutex... 保证线程安全 struct ToolRegistry { /// 工具名 → 工具函数 tools: HashMapString, ToolFn, /// 每个工具的调用次数统计内部状态 call_counts: HashMapString, u64, } impl ToolRegistry { fn new() - Self { ToolRegistry { tools: HashMap::new(), call_counts: HashMap::new(), } } /// 注册一个新工具 fn register(mut self, name: str, tool: ToolFn) { self.tools.insert(name.to_string(), tool); self.call_counts.insert(name.to_string(), 0); } /// 调用一个工具 fn call(mut self, name: str, input: str) - OptionString { if let Some(tool) self.tools.get(name) { // 统计调用次数 *self.call_counts.get_mut(name).unwrap() 1; // 执行工具 Some(tool(input.to_string())) } else { None } } } /// 使用 ArcMutexRegistry 的 Agent struct SharedToolAgent { /// 所有 Agent 共享同一个工具注册表 registry: ArcMutexToolRegistry, agent_id: u32, } impl SharedToolAgent { async fn run(self, tool_name: str, input: str) - String { // 获取互斥锁 —— 如果其他 Agent 正在用工具这里会阻塞 let mut registry self.registry.lock().unwrap(); // 调用工具在锁保护下 match registry.call(tool_name, input) { Some(result) result, None format!([Agent{}] 工具 {} 不存在, self.agent_id, tool_name), } // 锁在这里自动释放MutexGuard 离开作用域 } }这个方案的问题很明显所有调用串行化。Agent 1 在调用工具时Agent 2 和 Agent 3 只能干等即使它们调用的是不同的工具。方案B每个工具独立加锁粒度更细use std::sync::{Arc, RwLock}; use std::collections::HashMap; /// 单个工具的包装 —— 每个工具有自己的锁 struct ToolEntry { /// 工具函数不可变不需要锁 function: Arcdyn Fn(String) - String Send Sync, /// 调用次数可变需要保护 call_count: RwLocku64, } /// 细粒度锁的工具注册表 struct FineGrainedRegistry { /// 工具名 → 各自独立锁的工具条目 tools: HashMapString, ToolEntry, } impl FineGrainedRegistry { fn new() - Self { FineGrainedRegistry { tools: HashMap::new(), } } fn register(mut self, name: str, tool: impl Fn(String) - String Send Sync static) { self.tools.insert(name.to_string(), ToolEntry { function: Arc::new(tool), call_count: RwLock::new(0), }); } fn call(self, name: str, input: str) - OptionString { self.tools.get(name).map(|entry| { // 只对调用计数加写锁很小范围 { let mut count entry.call_count.write().unwrap(); *count 1; } // ← 写锁在这里立即释放 // 执行工具函数不需要锁因为函数是不可变的 (entry.function)(input.to_string()) }) } }优点Agent 1 调用工具 A 不会阻塞 Agent 2 调用工具 B因为它们用的是不同的锁。缺点如果 Agent 1 和 Agent 2 同时调用工具 A还是需要排队。方案CActor 模型消息驱动上面的方案都是用锁来保护共享状态换个思路——用Actor 模型。每个工具注册表是一个独立的 ActorAgent 通过发消息来调用工具use tokio::sync::{mpsc, oneshot}; /// Agent 发给工具注册表的消息 enum RegistryMessage { /// 调用工具请求 /// 携带一个 oneshot::Sender 用于回传结果 CallTool { tool_name: String, input: String, /// 回传通道注册表执行完工具后把结果发回来 reply_tx: oneshot::SenderString, }, /// 获取调用统计 GetStats { reply_tx: oneshot::SenderHashMapString, u64, }, } /// Actor 模式的工具注册表 struct ActorRegistry { /// 接收消息的通道 rx: mpsc::UnboundedReceiverRegistryMessage, /// 工具映射不需要锁因为只有一个任务在访问 tools: HashMapString, Arcdyn Fn(String) - String Send Sync, /// 调用统计同样不需要锁 call_counts: HashMapString, u64, } impl ActorRegistry { /// 启动 Actor 的消息循环 async fn run(mut self) { println!([工具注册表 Actor] 已启动); while let Some(msg) self.rx.recv().await { match msg { RegistryMessage::CallTool { tool_name, input, reply_tx } { // 处理工具调用 let result self.handle_call(tool_name, input); // 通过 oneshot 把结果发回给调用方 // 如果调用方已经取消等待比如超时send 会失败 let _ reply_tx.send(result); } RegistryMessage::GetStats { reply_tx } { let _ reply_tx.send(self.call_counts.clone()); } } } println!([工具注册表 Actor] 已关闭); } fn handle_call(mut self, name: str, input: str) - String { // 统计调用次数 *self.call_counts.entry(name.to_string()).or_insert(0) 1; // 执行工具函数 match self.tools.get(name) { Some(tool) tool(input.to_string()), None format!(工具 {} 未注册, name), } // 注意这里是同步执行如果工具执行时间很长 // 其他 Agent 的请求会被排队 // 解决办法在 Actor 内部用 tokio::spawn 执行工具 } } /// Agent 通过消息与工具注册表 Actor 交互 struct ActorAgent { /// 向 Actor 发送消息的通道 registry_tx: mpsc::UnboundedSenderRegistryMessage, agent_id: u32, } impl ActorAgent { async fn call_tool(self, tool_name: str, input: str) - String { // 创建一个 oneshot 通道用于接收结果 let (reply_tx, reply_rx) oneshot::channel(); // 发送调用请求到 Actor self.registry_tx.send(RegistryMessage::CallTool { tool_name: tool_name.to_string(), input: input.to_string(), reply_tx, }).expect(工具注册表 Actor 已关闭); // 等待 Actor 返回结果 reply_rx.await.unwrap_or_else(|_| { 工具注册表未响应.to_string() }) } } #[tokio::main] async fn main() { // 创建 Actor let (tx, rx) mpsc::unbounded_channel(); let registry ActorRegistry { rx, tools: HashMap::new(), call_counts: HashMap::new(), }; // 启动 Actor 的消息循环 tokio::spawn(registry.run()); // 创建多个 Agent共享 tx let agent1 ActorAgent { registry_tx: tx.clone(), agent_id: 1 }; let agent2 ActorAgent { registry_tx: tx.clone(), agent_id: 2 }; let agent3 ActorAgent { registry_tx: tx.clone(), agent_id: 3 }; // 三个 Agent 可以同时发送请求 let (r1, r2, r3) tokio::join!( agent1.call_tool(search, weather), agent2.call_tool(calc, 22), agent3.call_tool(db, SELECT * FROM users), ); println!(Agent1: {}, Agent2: {}, Agent3: {}, r1, r2, r3); }现在来看看三种方案的对比flowchart LR subgraph A[方案A: 全局大锁] A1[Agent1] -- A2[Agent2] -- A3[Agent3] A1 -- ALock[ Mutex] A2 -- ALock A3 -- ALock ALock -- ATools[工具注册表] end subgraph B[方案B: 细粒度锁] B1[Agent1] -- BLock1[ 工具A锁] B2[Agent2] -- BLock2[ 工具B锁] B3[Agent3] -- BLock3[ 工具C锁] BLock1 -- BTools[工具注册表] BLock2 -- BTools BLock3 -- BTools end subgraph C[方案C: Actor模型] C1[Agent1] -- CMsg[ 消息] C2[Agent2] -- CMsg C3[Agent3] -- CMsg CMsg -- CActor[Actorbr/单线程处理] CActor -- CTools[工具注册表] end三、我的实践选择说实话这三种方案我都试过最终选择取决于场景场景推荐方案理由小型项目、工具执行很快方案A全局锁代码最简单锁竞争不明显多个工具、调用互不干扰方案B细粒度锁不同工具可以并行工具执行时间长、需要超时取消方案CActor天然支持异步和超时在我当前的 Agent 项目中我选择了方案CActor 模型原因是工具调用可能很慢比如调用外部 APIActor 不会阻塞 AgentActor 天然支持背压back-pressure消息队列满了就拒绝新请求不需要操心锁的释放和死锁问题四、异步 Actor 的增强支持超时取消Actor 模型还可以很自然地加上超时取消/// 带超时的工具调用 impl ActorAgent { async fn call_tool_with_timeout( self, tool_name: str, input: str, timeout_duration: Duration, ) - ResultString, String { let (reply_tx, reply_rx) oneshot::channel(); self.registry_tx.send(RegistryMessage::CallTool { tool_name: tool_name.to_string(), input: input.to_string(), reply_tx, }).map_err(|_| 工具注册表不可用.to_string())?; // 用 tokio::time::timeout 包装等待 // 如果超时了reply_rx 会被 drop // Actor 那边的 send 会失败但不会 panic match tokio::time::timeout(timeout_duration, reply_rx).await { Ok(Ok(result)) Ok(result), Ok(Err(_)) Err(Actor 已关闭.to_string()), Err(_) Err(format!( 工具 {} 调用超时{}ms, tool_name, timeout_duration.as_millis() )), } // 超时后oneshot::Sender 在 Actor 端会被 drop // Actor 仍然会执行工具但结果无处可发被丢弃 // 如果要真正取消工具执行需要用 CancellationToken } }Actor 方案在实际生产中还有一个坑oneshot channel 的接收端被 drop 后发送端 send 会返回 Err。如果不处理这个 ErrActor 的任务会因为 unwrap 而 panic 退出。我在项目里就因为这个一个工具超时后整个 Actor 消息循环崩溃了。加上let _ reply_tx.send(result);忽略错误问题解决。Actor 的超时还可以更精细给每次工具调用设独立超时而不是整个 Actor 生命周期。否则一个工具卡住 30 秒所有等待这个 Actor 的请求都会排队超时然后雪崩。独立超时配合select!可以只取消卡住的那个调用。五、总结多 Agent 共享工具注册表本质上是一个并发资源管理的经典问题。三种方案各有适用场景全局锁简单直接适合工具执行很快的小项目细粒度锁性能最好但代码复杂度和工具数量成正比Actor 模型最灵活天然支持异步和超时适合工具执行时间不均的场景作为自学者在做这个设计的过程中学到的最重要的一课是并发问题的答案从来不是哪种方案最好而是在你的场景下trade-off 是什么。理解自己的需求选最合适的方案比盲目追求高性能锁更重要。如果你也在设计 Agent 的并发模型欢迎来交流你的方案。我们下篇见