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

一、什么场景适合请求合并？

请求合并不是所有场景都适用，先明确适用条件：

1. 请求相似度高：多个请求的操作类型相同（比如都是"总结文本"），只是输入不同。
2. 时间窗口集中：请求在短时间内（比如 100ms 内）集中到达。
3. 延迟不敏感：用户能接受 100-200ms 的额外等待时间。
4. 成本敏感：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: Option<oneshot::Sender<String>>,
    /// 请求创建时间
    created_at: Instant,
}

/// 请求合并管理器
pub struct RequestMerger {
    /// 暂存的待处理请求：request_id → PendingRequest
    pending: Mutex<HashMap<String, PendingRequest>>,
    /// 合并窗口大小：在这个时间内到达的请求会被合并
    batch_window: Duration,
    /// 批量处理的最大批次大小
    max_batch_size: usize,
}

impl RequestMerger {
    pub fn new(batch_window_ms: u64, max_batch_size: usize) -> Arc<Self> {
        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::Receiver<String> {
        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: Arc<RequestMerger>) {
    // 每隔 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: Vec<String> = 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) -> Vec<String> {
    let mut results = Vec::with_capacity(count);
    // 按 "[任务N]" 分割响应
    let parts: Vec<&str> = 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（比如用户请求在前端关了 tab），send 会返回 Err——这时候看起来是"正常丢弃"，但实际上没有任何日志记录，问题排查时完全找不到痕迹。

我在生产环境遇到过这样的场景：用户抱怨"明明提交了请求，为什么没收到结果？"查了一下午的日志才定位到是前端请求超时过早关闭了连接。修复是在 send 失败时至少打一条 warn：

if 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）：

用 0.1 秒的额外延迟换 67% 的成本节省，很多场景下是完全值得的。

边界故障：合并窗口设太大，用户请求反而等更久。 我曾经把窗口调到 500ms，合并率确实升到了 90%+，但用户反馈"点击按钮后要等半秒才开始转圈"。对于需要即时反馈的交互（比如代码补全、实时翻译），200ms 以上就是用户能感知的延迟了。权衡之后我把窗口降回 100ms：合并率降到 65%，但用户感知延迟从 500ms 降到了 100ms。

更进一步，可以用自适应窗口：高峰时段（并发请求多）把窗口缩短到 50ms，因为天然就很容易凑满批次；低谷时段（请求稀疏）拉长到 200ms，避免孤独的请求一直等。实现很简单——记录一个最近 10 秒内的请求速率，根据速率动态调整 batch_window：

fn 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) // 低并发：多等等，争取合并
    }
}

别让固定窗口成为用户体验的累赘。

五、总结

这次实现请求合并，让我对几个概念有了更深的理解：

1. 时间窗口——不是越大越好。窗口越大，合并率越高，但用户等待时间也越长。100-200ms 是一个经过验证的平衡点。
2. oneshot channel——在 Rust 异步生态中，tokio::sync::oneshot 是实现"一对一分发结果"的最佳工具，比手动管理回调简洁很多。
3. 批次拆分——合并请求容易，拆分结果难。需要在 prompt 里设计好格式标记（如 [任务N]），方便后续正则拆分。
4. 错误处理——批量调用失败时，要确保所有等待的调用者都能收到失败通知，不能悄悄丢弃。做系统设计，最大的挑战不是不会写代码，而是缺乏对"请求合并"这类模式的直觉。但好在这些模式都不是新东西，多读一些开源项目的架构代码，慢慢就能建立起自己的设计工具箱。