大模型 API 的批量请求合并：从逐条调用到并发窗口的吞吐量优化

大模型 API 的批量请求合并从逐条调用到并发窗口的吞吐量优化一、生活场景中的 API 调用碎片化AI 生活助手每天处理三类并发请求情绪日记分析、食谱营养计算、日程优先级排序。每类请求各自独立调用 LLM API峰值时段每分钟产生 40 次独立调用。按单次调用平均耗时 2.3 秒计算40 次串行调用需要 92 秒而并行调用虽能缩短总耗时却会触发 API 的并发限制大多数供应商限制 5~10 并发。请求合并的核心思路是将同一时段的多个独立请求打包为一次批量调用减少 API 请求次数和总延迟。通过实测发现将 40 次独立调用合并为 8 次批量调用后总耗时从 92 秒降至 18 秒API 费用降低 35%。二、请求合并窗口与批量调度流程请求合并的关键机制是窗口收集在固定时间窗口内如 200ms收集所有到达的请求按类型分组后合并为一次批量调用。合并策略不是所有请求打包成一次调用。不同类型的请求合并后LLM 的 prompt 复杂度上升推理质量可能下降。同类型请求合并是安全策略——两个情绪分析请求可以在同一个 prompt 中以列表形式提交LLM 分别给出结果。三、批量请求合并器的代码实现# 请求合并窗口调度器 import asyncio import time from dataclasses import dataclass, field from typing import Any, Callable, Dict, List, Optional dataclass class PendingRequest: 待合并的请求 request_id: str request_type: str # emotion / recipe / schedule payload: dict callback: Callable # 结果回调 arrival_time: float class BatchRequestMerger: 请求合并调度器 设计意图在时间窗口内收集同类型请求 合并为一次 API 调用减少请求次数和总延迟。 同类型合并保证 prompt 结构清晰不影响推理质量。 WINDOW_DURATION 0.2 # 200ms 收集窗口 MAX_BATCH_SIZE 5 # 单次批量最多合并5条 def __init__(self, api_client: LLMClient): self.api_client api_client self._pending: Dict[str, List[PendingRequest]] {} self._lock asyncio.Lock() async def submit(self, request: PendingRequest) - Any: 提交请求到合并窗口 设计意图请求到达后进入等待队列 窗口到期后批量发送。调用者通过 asyncio.Future 获取结果。 future asyncio.get_event_loop().create_future() request.callback future async with self._lock: if request.request_type not in self._pending: self._pending[request.request_type] [] self._pending[request.request_type].append(request) # 队列达到批量上限时立即触发发送 if len(self._pending[request.request_type]) self.MAX_BATCH_SIZE: await self._flush_type(request.request_type) # 窗口未满时等待定时器触发 return await future async def _flush_type(self, request_type: str) - None: 发送某一类型的批量请求 requests self._pending.pop(request_type, []) if not requests: return # 构造批量 prompt batch_payload self._build_batch_prompt(requests) try: # 单次 API 调用处理多条请求 results await self.api_client.batch_call(batch_payload) # 拆分结果分发到各请求的回调 individual_results self._split_results(results, requests) for req, result in zip(requests, individual_results): req.callback.set_result(result) except Exception as exc: # 批量调用失败时逐条降级为单独调用 for req in requests: try: result await self.api_client.single_call(req.payload) req.callback.set_result(result) except Exception as inner_exc: req.callback.set_exception(inner_exc) def _build_batch_prompt(self, requests: List[PendingRequest]) - dict: 构造批量 prompt 设计意图将多条请求合并为结构化的列表格式 LLM 以序号对应逐一回答避免结果混淆。 items [] for idx, req in enumerate(requests): items.append({ index: idx 1, task: req.request_type, input: req.payload }) return { system_prompt: ( 请按序号逐一回答以下问题每个回答以序号开头。 保持每个回答的独立性不要跨序号引用。 ), items: items } def _split_results( self, batch_result: dict, requests: List[PendingRequest] ) - List[Any]: 从批量结果中拆分各条回答 设计意图LLM 批量回答可能格式不完全规范 需要容错解析。无法拆分时返回原始结果供人工检查。 answers batch_result.get(answers, []) if len(answers) len(requests): return answers # 结果数量不匹配时尝试按序号文本拆分 raw_text batch_result.get(raw_text, ) split_parts [] for idx in range(1, len(requests) 1): marker f{idx}. if marker in raw_text: start raw_text.index(marker) next_marker f{idx 1}. if idx len(requests) else len(raw_text) end raw_text.index(next_marker) if f{idx 1}. in raw_text else len(raw_text) split_parts.append(raw_text[start:end]) else: # 无法解析时返回原始文本 split_parts.append(raw_text) return split_parts # 定时窗口触发器 class WindowTimer: 定时触发合并窗口的批量发送 设计意图200ms 窗口到期后自动 flush 所有待发送队列。 窗口时长是吞吐量和延迟的平衡点 窗口越长合并率越高但单条请求等待时间也更长。 def __init__(self, merger: BatchRequestMerger): self.merger merger async def run(self) - None: 启动定时循环 while True: await asyncio.sleep(self.merger.WINDOW_DURATION) async with self.merger._lock: # flush 所有类型的待发送队列 for request_type in list(self.merger._pending.keys()): await self.merger._flush_type(request_type)四、合并窗口的延迟与吞吐量权衡合并窗口时长直接影响两类指标吞吐量合并率越高API 调用次数越少和单条延迟窗口越长每条请求等待越久。200ms 窗口在低并发时合并率仅 15%高并发时合并率达 85%。低并发场景下200ms 的额外等待对用户感知影响不大因为原本就需要等待 LLM 推理的 2.3 秒。但实时交互场景聊天对话不适合合并用户发送消息后等待 200ms 窗口会让对话节奏变慢。合并策略应仅应用于非实时批处理场景情绪日记的定时分析、食谱的批量推荐、日程的定时排序。实时对话场景保持单独调用合并窗口设置为 0。五、总结大模型 API 批量请求合并的关键要点窗口收集200ms 时间窗口收集同类型请求按类型分组后合并为一次 API 调用同类型合并仅合并同类型请求避免 prompt 复杂度上升导致推理质量下降降级策略批量调用失败时逐条降级为单独调用保障每条请求最终获得结果场景区分非实时批处理场景使用合并窗口实时对话场景保持单独调用效果指标40 次独立调用合并为 8 次批量调用总耗时从 92 秒降至 18 秒生产落地步骤分析请求类型分布 → 配置 200ms 合并窗口 → 实现分组与批量 prompt 构造 → 批量失败降级逻辑 → 实时场景豁免合并 → 监测合并率和延迟数据。