AI Agent 请求合并：多个用户问类似问题，怎么复用一次模型调用

AI Agent 请求合并多个用户问类似问题怎么复用一次模型调用一、什么场景适合请求合并请求合并不是所有场景都适用先明确适用条件请求相似度高多个请求的操作类型相同比如都是总结文本只是输入不同。时间窗口集中请求在短时间内比如 100ms 内集中到达。延迟不敏感用户能接受 100-200ms 的额外等待时间。成本敏感API 按调用次数收费合并能直接省钱。典型的应用场景包括批量文档摘要多个用户上传文件要求 AI 总结。批量翻译多个段落同时需要翻译。批量代码审查多个 PR 同时需要 AI 审查。二、请求合并器的核心设计我设计了一个请求合并管理器核心思路是时间窗口 批量队列 结果分发use std::collections::HashMap; use std::sync::Arc; use tokio::sync::{Mutex, oneshot}; use std::time::{Duration, Instant}; /// 单个待处理的请求 struct PendingRequest { /// 请求内容prompt content: String, /// 结果的发送通道 —— 每个等待的用户拿着一个 receiver sender: Optiononeshot::SenderString, /// 请求创建时间 created_at: Instant, } /// 请求合并管理器 pub struct RequestMerger { /// 暂存的待处理请求request_id → PendingRequest pending: MutexHashMapString, PendingRequest, /// 合并窗口大小在这个时间内到达的请求会被合并 batch_window: Duration, /// 批量处理的最大批次大小 max_batch_size: usize, } impl RequestMerger { pub fn new(batch_window_ms: u64, max_batch_size: usize) - ArcSelf { Arc::new(Self { pending: Mutex::new(HashMap::new()), batch_window: Duration::from_millis(batch_window_ms), max_batch_size, }) } /// 提交一个请求 —— 返回一个接收结果的 channel /// /// 调用者拿着返回的 receiver 等待结果 /// 后台定时器会在合适的时候批量处理这些请求 pub async fn submit(self, content: String) - oneshot::ReceiverString { let request_id uuid::Uuid::new_v4().to_string(); let (tx, rx) oneshot::channel(); let request PendingRequest { content, sender: Some(tx), created_at: Instant::now(), }; self.pending.lock().await.insert(request_id, request); rx } }三、后台定时器批量收集和处理上面的submit只是把请求放进队列真正要处理的是后台的定时器/// 启动合并处理的定时器通常在程序启动时调用一次 pub async fn run_merger(merger: ArcRequestMerger) { // 每隔 batch_window/2 检查一次既不会等太久也不会频率过高 let mut interval tokio::time::interval(merger.batch_window / 2); loop { interval.tick().await; // 第一步收集到期的请求 let mut pending merger.pending.lock().await; let now Instant::now(); // 收集所有到期的请求创建时间早于 now - batch_window let mut batch: Vec(String, PendingRequest) Vec::new(); let keys_to_remove: VecString pending .iter() .filter(|(_, req)| { // 到期条件创建时间 窗口 当前时间 // 或者累积请求数达到上限提前处理 now.duration_since(req.created_at) merger.batch_window || batch.len() merger.max_batch_size }) .map(|(k, _)| k.clone()) .collect(); for key in keys_to_remove { if let Some(req) pending.remove(key) { batch.push((key, req)); if batch.len() merger.max_batch_size { break; // 已达批次上限剩余的留到下一轮 } } } drop(pending); // 释放锁不阻塞新请求的提交 if batch.is_empty() { continue; // 没有到期的请求跳过 } // 第二步构造合并 prompt let combined_prompt build_batch_prompt(batch); // 第三步调用 AI API match call_ai_api(combined_prompt).await { Ok(response) { // 第四步拆分结果并分发给各个请求者 let results parse_batch_response(response, batch.len()); for (i, (_id, mut req)) in batch.into_iter().enumerate() { if let Some(sender) req.sender.take() { let result results.get(i) .cloned() .unwrap_or_else(|| 解析结果失败.to_string()); // 通过 oneshot channel 把结果发给等待的用户 let _ sender.send(result); } } } Err(e) { // API 调用失败通知所有等待者 for (_id, mut req) in batch { if let Some(sender) req.sender.take() { let _ sender.send(format!(API 调用失败: {}, e)); } } } } } } /// 构造批量 prompt把多个请求拼接成一个 fn build_batch_prompt(requests: [(String, PendingRequest)]) - String { let mut prompt String::from( 请分别处理以下多个独立任务用编号分隔每个任务的结果\n\n ); for (i, (_id, req)) in requests.iter().enumerate() { prompt.push_str(format!([任务{}]\n{}\n\n, i 1, req.content)); } prompt.push_str(请按照 [任务1] [任务2] ... 的格式分别返回结果。); prompt } /// 解析 AI 返回的批量结果 fn parse_batch_response(response: str, count: usize) - VecString { let mut results Vec::with_capacity(count); // 按 [任务N] 分割响应 let parts: Vecstr response.split(\n[任务).collect(); for i in 1..count { let prefix format!({}], i); let result parts.iter() .find(|p| p.starts_with(prefix)) .map(|p| p[prefix.len()..].trim().to_string()) .unwrap_or_else(|| 未找到对应结果.to_string()); results.push(result); } results }生产踩坑oneshot channel 发送失败时请求被静默丢弃。在run_merger中当 API 调用完成后我用sender.send(result)把结果返回给等待方。但如果等待方因为超时已经drop了receiver比如用户请求在前端关了 tabsend会返回Err——这时候看起来是正常丢弃但实际上没有任何日志记录问题排查时完全找不到痕迹。我在生产环境遇到过这样的场景用户抱怨明明提交了请求为什么没收到结果查了一下午的日志才定位到是前端请求超时过早关闭了连接。修复是在send失败时至少打一条 warnif let Err(_) sender.send(result) { log::warn!(请求 {} 的接收端已关闭结果被丢弃, id); }另一个隐患是parse_batch_response依赖 AI 严格按照[任务N]格式返回。有一次 AI 返回了[任务一] [任务二]用了中文数字正则解析全挂了所有用户收到的都是未找到对应结果。后来改成了更宽松的匹配在 prompt 里强调必须用[1]格式并在解析侧做 fallback——如果按格式拆分失败就把整个结果当作第一个请求的响应其余标记为解析失败至少不全丢。四、使用示例和效果分析把上面的组件组装起来一个完整的请求合并流程如下use std::sync::Arc; #[tokio::main] async fn main() { // 创建合并器100ms 窗口最多合并 10 个请求 let merger RequestMerger::new(100, 10); let merger_arc Arc::clone(merger); // 启动后台处理定时器生产环境用 tokio::spawn tokio::spawn(async move { run_merger(merger_arc).await; }); // 模拟 3 个用户几乎同时发请求 let rx1 merger.submit(请总结这段文字的主要观点....into()).await; tokio::time::sleep(Duration::from_millis(20)).await; let rx2 merger.submit(请总结这段对话的要点....into()).await; tokio::time::sleep(Duration::from_millis(30)).await; let rx3 merger.submit(请总结这篇文章的核心思想....into()).await; // 三个请求会被后台定时器合并为一次 API 调用 // 每个调用者独立等待自己的结果 let (result1, result2, result3) tokio::join!( async { rx1.await.unwrap() }, async { rx2.await.unwrap() }, async { rx3.await.unwrap() }, ); println!(用户1结果: {}, result1); println!(用户2结果: {}, result2); println!(用户3结果: {}, result3); }成本对比假设一次 API 调用 $0.01模式API 调用次数成本平均延迟无合并3 次$0.031.2s100ms 窗口合并1 次$0.011.3s节省-67%-67%0.1s用 0.1 秒的额外延迟换 67% 的成本节省很多场景下是完全值得的。边界故障合并窗口设太大用户请求反而等更久。我曾经把窗口调到 500ms合并率确实升到了 90%但用户反馈点击按钮后要等半秒才开始转圈。对于需要即时反馈的交互比如代码补全、实时翻译200ms 以上就是用户能感知的延迟了。权衡之后我把窗口降回 100ms合并率降到 65%但用户感知延迟从 500ms 降到了 100ms。更进一步可以用自适应窗口高峰时段并发请求多把窗口缩短到 50ms因为天然就很容易凑满批次低谷时段请求稀疏拉长到 200ms避免孤独的请求一直等。实现很简单——记录一个最近 10 秒内的请求速率根据速率动态调整batch_windowfn adaptive_window(requests_per_sec: f64) - Duration { if requests_per_sec 50.0 { Duration::from_millis(30) // 高并发快合快发 } else if requests_per_sec 10.0 { Duration::from_millis(100) // 中等常规窗口 } else { Duration::from_millis(200) // 低并发多等等争取合并 } }别让固定窗口成为用户体验的累赘。五、总结这次实现请求合并让我对几个概念有了更深的理解时间窗口——不是越大越好。窗口越大合并率越高但用户等待时间也越长。100-200ms 是一个经过验证的平衡点。oneshot channel——在 Rust 异步生态中tokio::sync::oneshot是实现一对一分发结果的最佳工具比手动管理回调简洁很多。批次拆分——合并请求容易拆分结果难。需要在 prompt 里设计好格式标记如[任务N]方便后续正则拆分。错误处理——批量调用失败时要确保所有等待的调用者都能收到失败通知不能悄悄丢弃。做系统设计最大的挑战不是不会写代码而是缺乏对请求合并这类模式的直觉。但好在这些模式都不是新东西多读一些开源项目的架构代码慢慢就能建立起自己的设计工具箱。