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

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

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

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

典型的应用场景包括：

• 批量文档摘要：多个用户上传文件，要求 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) // 低并发：多等等，争取合并
}
}

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

五、总结

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