模型推理的请求去重相同输入在并发窗口内的合并处理一、100 个用户同时问今天天气怎么样你的推理服务做了 100 次一模一样的计算模型推理和传统的 API 调用有一个本质区别推理服务的边际成本不是近似为零的。每次推理调用都需要 GPU 计算、显存分配、KV Cache 占用。如果你的应用场景中用户问题有较高的重复率——比如热门话题问答、标准化客服对话、模板化内容生成——请求去重Request Deduplication可以显著降低推理成本。请求去重的核心逻辑是在短时间内对完全相同的输入做 hash 比对。如果检测到正在处理同一个 question 的推理新请求不应该重复执行而应该搭车——等第一个请求完成后共享结果。如果输入相似但不完全相同则可以用向量相似度做近似去重——将高度相似的请求合并为一次推理对差异部分做增量处理。这里的挑战在并发窗口的管理——窗口太短去重率低窗口太长第一个请求还没完成后续请求等待太久。另外即使输入文本完全相同推理温度temperature也可能不同——这种情况下不应该去重因为用户期望的是不同的随机输出。二、底层机制与原理剖析具体而言请求处理流程始于指纹计算随后进行精确去重检查。若命中相同指纹且请求正在处理中新请求将搭车等待原请求完成并复用结果若未命中则进入近似去重检查。当相似度超过阈值时请求被合并并标记为可共享结果必要时对差异部分做增量推理若仍未命中则作为新请求加入处理队列执行推理。最终所有结果均会被缓存并在 TTL 过期后清理。关键设计点请求指纹的构成hash(prompt model max_tokens top_p stop_sequences)。注意排除 temperature、seed、request_id 这些每次可能不同的参数。指纹的目的是判断推理结果是否可共享而非请求是否完全相同。并发去重窗口这是一个关键参数。如果窗口设得太小如 100ms同一个热问题在 100ms 内可能只有 2-3 个请求去重率不够。如果窗口设得太大如 5 秒用户等待时间过长。通常设置 500ms-2s 的窗口或者自适应——根据当前并发请求数和重复率动态调整。结果缓存策略去重和缓存是互补的。去重是当前正在计算的相同请求合并为一次。缓存是之前计算过的相同请求直接返回结果。两者结合使用去重处理并发窗口内的重复缓存处理跨窗口的重复。三、生产级代码实现 模型推理请求去重器 精确去重 近似去重 结果缓存 import asyncio import hashlib import time from dataclasses import dataclass, field from typing import Dict, List, Optional, Any, Callable from collections import OrderedDict import threading dataclass class InferenceRequest: 推理请求 prompt: str model: str max_tokens: int 256 temperature: float 0.7 # 不参与指纹计算 top_p: float 0.9 request_id: str # 不参与指纹计算 def fingerprint(self) - str: 计算请求指纹 设计决策 - 包含 prompt model 确定性参数 - 排除 temperature / request_id 等每请求不同的参数 - 这是判断结果是否可共享的关键 content f{self.prompt}|{self.model}|{self.max_tokens}|{self.top_p} return hashlib.sha256(content.encode()).hexdigest() dataclass class CachedResult: 缓存的结果 fingerprint: str result: Any created_at: float ttl: float # 存活时间秒 def is_expired(self) - bool: return time.time() - self.created_at self.ttl class RequestDeduplicator: 推理请求去重器 三个层次 1. 并发去重相同指纹的请求共享一次推理 2. 结果缓存历史结果跨窗口复用 3. 近似去重高度相似的请求标记不强制合并 def __init__( self, cache_ttl: float 60.0, # 缓存 60 秒 dedup_window: float 1.0, # 去重窗口 1 秒 max_cache_size: int 1000, ): self.cache_ttl cache_ttl self.dedup_window dedup_window self.max_cache_size max_cache_size # 并发去重fingerprint - asyncio.Event (等待原请求完成) self._pending: Dict[str, asyncio.Event] {} self._pending_results: Dict[str, Any] {} self._lock threading.Lock() # 结果缓存fingerprint - CachedResult self._cache: OrderedDict OrderedDict() # 统计 self.stats { total_requests: 0, dedup_hits: 0, cache_hits: 0, new_inferences: 0, } async def execute( self, request: InferenceRequest, inference_fn: Callable, ) - Any: 执行推理带去重 流程 1. 检查结果缓存 - 命中直接返回 2. 检查并发去重 - 命中则等待原请求完成 3. 都不命中 - 执行推理 缓存结果 self.stats[total_requests] 1 fingerprint request.fingerprint() # 第 1 层结果缓存检查 cached self._check_cache(fingerprint) if cached is not None: self.stats[cache_hits] 1 return cached.result # 第 2 层并发去重检查 with self._lock: if fingerprint in self._pending: # 搭车等待原请求完成 self.stats[dedup_hits] 1 event self._pending[fingerprint] if event in locals(): # 等待原请求完成 try: await asyncio.wait_for(event.wait(), timeout30.0) with self._lock: result self._pending_results.get(fingerprint) if result is not None: return result except asyncio.TimeoutError: pass # 超时后降级为独立请求 # 第 3 层执行推理 with self._lock: if fingerprint not in self._pending: self._pending[fingerprint] asyncio.Event() try: self.stats[new_inferences] 1 # 执行实际的推理 result await inference_fn(request) # 缓存结果 self._add_cache(fingerprint, result) # 通知等待方 with self._lock: self._pending_results[fingerprint] result self._pending[fingerprint].set() return result except Exception as e: # 推理失败通知等待方并传播错误 with self._lock: if fingerprint in self._pending: self._pending[fingerprint].set() # 不阻塞等待方 raise finally: # 延迟清理等所有等待方读取结果后再清理 asyncio.get_event_loop().call_later( 5.0, self._cleanup_pending, fingerprint ) def _check_cache(self, fingerprint: str) - Optional[CachedResult]: 检查结果缓存 with self._lock: if fingerprint in self._cache: cached self._cache[fingerprint] if not cached.is_expired(): # LRU移到末尾 self._cache.move_to_end(fingerprint) return cached else: # 过期清理 del self._cache[fingerprint] return None def _add_cache(self, fingerprint: str, result: Any): 添加结果到缓存 with self._lock: # LRU 淘汰 while len(self._cache) self.max_cache_size: self._cache.popitem(lastFalse) self._cache[fingerprint] CachedResult( fingerprintfingerprint, resultresult, created_attime.time(), ttlself.cache_ttl, ) def _cleanup_pending(self, fingerprint: str): 清理并发去重状态 with self._lock: self._pending.pop(fingerprint, None) self._pending_results.pop(fingerprint, None) def get_stats(self) - Dict: 获取去重统计 with self._lock: dedup_rate 0.0 if self.stats[total_requests] 0: dedup_rate ( self.stats[dedup_hits] self.stats[cache_hits] ) / self.stats[total_requests] return { **self.stats, dedup_rate: round(dedup_rate, 4), cache_size: len(self._cache), pending_count: len(self._pending), } def evict_cache(self, pattern: str None): 手动驱逐缓存 with self._lock: if pattern is None: self._cache.clear() else: keys_to_remove [ k for k in self._cache if pattern in k ] for k in keys_to_remove: del self._cache[k] # 近似去重 class ApproximateDeduplicator: 基于文本相似度的近似去重 注意仅标记相似请求不强制合并。 因为近似合并的风险两个稍有差异的 prompt 结果不通用较高。 def __init__(self, embed_fn, similarity_threshold: float 0.95): self.embed_fn embed_fn # 嵌入函数 self.similarity_threshold similarity_threshold self._recent_prompts: List[tuple] [] # (embedding, prompt, timestamp) self._max_recent 100 def find_similar(self, prompt: str) - Optional[str]: 查找最近的相似 prompt try: prompt_embed self.embed_fn(prompt) except Exception: return None now time.time() best_sim 0 best_prompt None for embed, cached_prompt, ts in self._recent_prompts: if now - ts 10: # 只考虑 10 秒内的 continue # 余弦相似度 sim self._cosine_sim(prompt_embed, embed) if sim best_sim and sim self.similarity_threshold: best_sim sim best_prompt cached_prompt # 添加当前 prompt 到最近列表 self._recent_prompts.append((prompt_embed, prompt, now)) if len(self._recent_prompts) self._max_recent: self._recent_prompts.pop(0) return best_prompt if best_sim self.similarity_threshold else None staticmethod def _cosine_sim(a, b): import numpy as np return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b) 1e-8)四、边界分析与架构权衡去重适用的前提推理去重的收益取决于请求的重复率。如果你的用户问题高度去中心化如开放式聊天去重率可能不到 5%管理开销大于收益。如果你的场景是模板化知识库问答或标准化客服去重率可以达到 20-40%。去重窗口与用户体验去重引入的额外等待是一个重要的用户体验问题。如果可以接受毫秒级延迟的增量搭车的用户多等 500ms 而不是重新推理的 2 秒去重是净收益。但如果第一个请求推理太慢如 10 秒搭车用户等待时间不可接受——此时需要放弃去重、允许并行处理。temperature 的处理如果 temperature0.7 和 temperature0.0 的两个相同请求去重了前者失去了随机性。正确处理是请求指纹包含 temperature不同 temperature 的请求不去重。除非明确知道某个请求类型如分类任务的温度不影响结果。适用边界最适合标准化问题回答、模板内容生成等请求模式重复度高的场景。多租户 AI SaaS 平台多个客户运营相似场景是去重发挥最大价值的地方。禁用场景不适合创意性生成temperature 0.8的去重——用户期望每次得到不同结果去重违背了这个期望。也不适合对延迟要求极低 100ms的场景。五、总结推理请求去重的本质是在并发窗口内以空间缓存换时间GPU 计算。指纹计算决定了结果是否可共享的边界——排除 temperature、保留 prompt deterministic params。三层设计——缓存做跨窗口复用、并发去重做窗口内合并、近似去重做跨请求标记。在客服问答、模板生成等场景下20-40% 的去重率意味着同等 GPU 成本下 20-40% 的吞吐提升。