模型推理的请求去重：相同输入在并发窗口内的合并处理
一、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(), timeout=30.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(last=False) self._cache[fingerprint] = CachedResult( fingerprint=fingerprint, result=result, created_at=time.time(), ttl=self.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 的处理：
如果 temperature=0.7 和 temperature=0.0 的两个相同请求去重了，前者失去了随机性。正确处理是：请求指纹包含 temperature，不同 temperature 的请求不去重。除非明确知道某个请求类型（如分类任务）的温度不影响结果。
适用边界：
最适合标准化问题回答、模板内容生成等请求模式重复度高的场景。多租户 AI SaaS 平台（多个客户运营相似场景）是去重发挥最大价值的地方。
禁用场景：
不适合创意性生成（temperature > 0.8）的去重——用户期望每次得到不同结果，去重违背了这个期望。也不适合对延迟要求极低（< 100ms）的场景。
五、总结
推理请求去重的本质是在并发窗口内以空间（缓存）换时间（GPU 计算）。指纹计算决定了"结果是否可共享"的边界——排除 temperature、保留 prompt + deterministic params。三层设计——缓存做跨窗口复用、并发去重做窗口内合并、近似去重做跨请求标记。在客服问答、模板生成等场景下，20-40% 的去重率意味着同等 GPU 成本下 20-40% 的吞吐提升。
