维护带知识库检索的 MCP Server 时你一定见过这个画面日志里同一段文本在几分钟内被反复送去 embedding模型明明没崩GPU 利用率却上不去算力和 token 就这么一点点烧掉。我自己在项目里踩过这个坑最后在 MCP Server 内部搭了一套任务合并机制把重复的 Embedding 冗余计算拦截在真正推理之前。这篇直接讲清楚冗余从哪来、为什么合并机制比缓存更管用、代码怎么落以及上线后会踩哪些坑。1. 先找出算力内耗点MCP Server 里的 Embedding 到底重复在哪1.1 一个典型调用链路从 Agent 请求到向量化推理要解决冗余计算第一步是搞清楚 MCP Server 在真实场景里是怎么被调用的。以我维护的内部知识库 Agent 为例链路通常是这样的用户发一句话Agent 通过 MCP 协议调用 Server 上注册的search_knowledge工具工具内部先对用户 query 做 embedding再去向量数据库里做相似度检索最后把召回结果返回给 Agent 生成回答。看起来只有一次 embedding但实际场景远没有这么干净。MCP Server 的嵌入点往往不止一个Agent 在做工具选择时可能把工具描述、历史消息片段送去 embeddingRAG 流程里query 还需要经过改写、拆解每一版改写结果都可能被向量化。这些调用散落在不同模块server 层面不加统一拦截的话算力消耗完全不可控。另一类隐蔽的重复来自系统提示词。很多 Agent 每次工具调用都会携带同样一大段系统提示词参与上下文理解而这段提示词在 embedding 时会被完整处理一遍。上下文越长、工具轮次越多同一份提示词被向量化的次数就越夸张。1.2 三类最常见的冗余计算场景我把线上日志翻了一遍发现重复计算集中在三种场景多客户端并发相同查询几个用户几乎同时问“怎么配置 Nginx 反向代理”query 经过归一化后完全相同但每个请求各自走了一遍 embedding。这是最典型也最可惜的一类因为文本一模一样向量结果也一模一样。同一会话内上下文重复Agent 在单次任务里调用了多个工具每一步都带着同一段上下文去做向量化这段上下文被重复计算了 N 次。离线索引与在线请求叠加定时任务正在给一批新文档建索引在线用户恰好检索到这批文档两边对同一段文本各算一次向量。离线任务和在线请求往往是独立代码路径最容易漏掉。我做过一次粗算假设单次 embedding 处理 256 token 平均耗时 15ms本地 bge-m3 单卡实测的近似值线上每天 20 万次 embedding 请求中有 30% 是重复的那每天就有 6 万次计算是白白浪费的相当于 900 秒左右的 GPU 算力被丢进水里。如果走第三方 embedding API那就是实打实的 token 账单在燃烧。2. 为什么是任务合并机制从缓存失效到窗口折叠2.1 朴素缓存解决不了的并发重复问题很多人第一反应是加缓存。缓存确实有用但它解决不了并发重复。假设同一时刻有 10 个相同 query 到达 MCP Server缓存里没有对应结果10 个请求会同时触发 10 次 embedding 计算。等计算结果写进缓存下一个相同 query 才能命中——可问题是那 10 次计算已经发生了。缓存本质上是“过去的记忆”对于“当下的并发拥挤”无能为力。我在初期只加缓存时命中率看着还行但 GPU 利用率依然起伏很大原因就是缓存 miss 瞬间的并发洪峰。另外多实例部署时每个实例各有一份本地缓存缓存的一致性也会变成新问题。任务合并机制不一样。它的核心是把“在同一个时间片内到达的相同请求”折叠成一个请求去真正计算然后让所有等待者共享这一个结果。这正是解决并发重复的关键不是等算完再告诉后来的请求“结果在这”而是在第一个请求到达时就建立一个共享通道后面的相同请求直接挂到通道上等结果。2.2 请求级去重加计算级凑批双管齐下我在 MCP Server 里实现的任务合并机制分两层两层解决的问题不一样。第一层是请求级去重Request Coalescing。通过请求指纹判断两个请求是否“相同”相同就共享同一个待处理任务。第一个请求到达时发起计算后续相同请求不再进入计算队列而是挂在一个 Future 上等待结果。等结果出来所有等待者一起去取值。这个机制就像食堂窗口前排队打饭如果 10 个人点的菜完全一样后厨只做一份出锅后大家分。第二层是计算级凑批Batch Aggregation。不同文本但使用同一个 embedding 模型的请求在窗口内合并成一个 batch喂给模型做批量推理。GPU 这类并行硬件最怕小请求散弹式打过来每次只算一条向量利用率低到没法看。凑批之后单次推理的固定开销被摊薄整体吞吐能明显提升。这两层必须配合着来。请求级去重解决“算重了”计算级凑批解决“算得散”。只有去重没有凑批不同文本的请求还是各算各的只有凑批没有去重重复文本照样会被算进 batch 里浪费算力。2.3 合并窗口大小算力与延迟的权衡合并机制引入了一个额外参数合并窗口。窗口越大相同请求越多地被折叠到一起去重效果越好代价是最后到达的请求需要等待窗口关闭才能拿到结果延迟变高。窗口大小怎么定我给出一个相对通用的起步建议本地部署的 embedding 模型推理耗时在 10-30ms 时窗口取 50-100ms 比较合适如果通过 API 调用远程 embedding 服务往返延迟往往在 100ms 以上窗口取 20-50ms 更划算因为等待凑批的时间不应该超过远程调用本身的延迟。从数学上看假设窗口期 t 内有 N 个相同请求原本需要 N 次推理合并后只做 1 次节省的计算量是 (N-1) 乘以单次开销代价是最晚到达的请求平均等待 t/2。所以窗口的本质是用几十毫秒的延迟换取一次算力折叠对大多数 RAG 场景来说非常划算毕竟一次网络往返都要几百毫秒。3. 实现拦截器的核心细节指纹、窗口、缓存与批处理3.1 请求指纹设计如何精准识别“同一个请求”请求级去重的前提是精准的指纹。我的指纹生成逻辑是先把文本做归一化去掉首尾空格、把连续空白字符压缩成一个、统一大小写再和模型名、归一化参数、服务版本拼在一起最后用 SHA-256 取哈希。import hashlib def build_fingerprint(text, model_namebge-m3, **kwargs): # 归一化去掉首尾空格、压缩连续空白、统一小写 normalized .join(text.strip().lower().split()) # 排序 kwargs保证相同参数不同传参顺序也得到相同指纹 payload f{model_name}|{normalized}|{sorted(kwargs.items())} return hashlib.sha256(payload.encode(utf-8)).hexdigest()指纹里必须包含模型名和参数。不同模型产出的向量维度、语义空间都不同结果不能互相混用归一化参数是否做 L2 归一化也会改变向量数值这些因素必须出现在指纹里。很多初写者在指纹里只拼文本结果换了个 embedding 模型后老是命中旧向量检索效果一塌糊涂实际上就是指纹设计漏了关键信息。另一个容易忽略的点是服务版本。embedding 模型升级后语义空间可能变化指纹里带上服务版本能避免新旧模型产出的向量混杂在一起排查问题时会省很多事。3.2 合并窗口的时间控制这几十毫秒里发生了什么窗口不是定时轮询而是“从第一个请求到达时开始计时”。第一请求进入空队列时会启动一个后台任务后台任务 sleep 一个窗口时长sleep 结束后把所有已到达的请求打包处理。实现上我用的是asyncio。请求进来时先查缓存和 pending 表如果 pending 表里已有相同 key就把当前请求的 Future 追加到该 key 对应的 Future 链表里。如果没有则创建新的 Future、把请求写入_queue并确保后台批处理任务在运行。后台批处理任务的核心思路是每次 sleep 完窗口后从_queue中取出最多max_batch_size个请求按模型分组调用真正的 embedding 批量推理。推理完成后把结果写入 LRU 缓存然后把该 key 在 pending 表中的所有 Future 都set_result让等待中的请求一起去取数。如果队列已空后台任务退出等待下一个新请求重新启动。这里有个细节要注意asyncio.Future和asyncio.Task的区别。Future 是结果容器Task 是调度单元。请求等待的是 Future后台处理的循环需要用一个 Task 来驱动。如果不小心把 Future 当 Task 去调度程序会直接报错。3.3 LRU 缓存与容量规划别让缓存反过来拖垮性能缓存层我选了最简单的 LRU最近最少使用策略没有上更复杂的缓存算法。原因有两个embedding 请求的访问模式有很强的局部性常见的 query 会在一段时间内反复出现LRU 实现起来最直白用 Python 的OrderedDict几十行就能写对。容量规划上要算一笔账。假设向量维度是 1024float32 存储单条向量占 4KB 空间。缓存 10000 条大约占 40MB 内存在 MCP Server 进程内完全可接受。如果向量维度更高比如 4096单条向量 16KB10000 条就是 160MB这就需要考虑限制了。我用OrderedDict实现 LRU每次命中时move_to_end插入新条目时如果超过容量就popitem(lastFalse)弹出最久没用的条目。这里有个小坑popitem(lastFalse)才表示移除最老条目lastTrue是移除最新条目写反的话缓存就名存实亡了。3.4 对接不同 Embedding Model 的批处理差异合并机制的下游是真正的 embedding 推理接口。不同部署方式对批处理的支持差异很大这是最容易踩坑的地方。Transformers 纯本地推理可以通过设置 pipeline 的batch_size参数实现批处理但要注意显存占用batch 过大会 OOM。Ollama 的 embedding APIOllama 原生接口一次处理一个文本但可以通过并发多个请求来模拟批量效果或者使用它面向批处理场景的兼容接口。TEIText Embeddings Inference原生支持 batch 输入把文本数组直接 POST 过去就行是目前和我这套机制配合最顺的下游服务。vLLM 的 embedding 接口也原生支持批量适合大批量场景但对本地显存要求较高。无论哪种方式都要明确设置max_batch_size上限。我在代码里默认用 32配合单卡 T4 部署 bge-m3 比较稳。如果你用的模型更大可能要把这个值调小到 8-16避免显存爆炸。4. 实操落地在 MCP Server 中实现任务合并拦截层4.1 拦截层应该挂在哪个位置任务合并拦截层的位置很关键。我建议不要散在多个工具函数里各自实现而是封装成一个统一的embedding_client所有工具都通过它来请求向量化。这样能保证指纹生成逻辑、缓存、合并窗口都集中在同一处不会出现这里的请求走了合并、那边的请求还在裸算。具体到 MCP Server 的代码结构如果项目用的是 FastMCP 框架可以在工具函数内部调用client.embed(text)如果在用原生 MCP SDK就在工具注册层做一次统一封装。核心原则是所有 embedding 调用必须经过同一道关卡否则拦截层只是个摆设。4.2 核心代码一个可以直接改的 EmbeddingCoalescer下面是我项目中实际使用的核心骨架去掉了业务无关的部分。这段代码实现了请求级去重、窗口凑批和 LRU 缓存三件事。import asyncio import hashlib from collections import OrderedDict class EmbeddingCoalescer: def __init__(self, embed_func, window_ms50, max_batch_size32, cache_capacity4096): self.embed_func embed_func # 真正的 embedding 批处理函数 self.window_ms window_ms / 1000.0 # 合并窗口秒为单位 self.max_batch_size max_batch_size self.cache_capacity cache_capacity self.cache OrderedDict() # key - vector self._pending {} # key - list[Future] self._queue [] # 等待凑批的请求 self._lock asyncio.Lock() self._window_task None # 后台窗口批处理任务 def _fingerprint(self, text, model_name, **kwargs): normalized .join(text.strip().lower().split()) payload f{model_name}|{normalized}|{sorted(kwargs.items())} return hashlib.sha256(payload.encode(utf-8)).hexdigest() async def submit(self, text, model_namebge-m3, **kwargs): key self._fingerprint(text, model_name, **kwargs) # 命中缓存直接返回 if key in self.cache: self.cache.move_to_end(key) return self.cache[key] async with self._lock: if key in self._pending: # 已有相同请求在途共享同一个 Future fut asyncio.get_running_loop().create_future() self._pending[key].append(fut) else: # 新请求登记 pending 并入队 fut asyncio.get_running_loop().create_future() self._pending[key] [fut] self._queue.append((key, text, model_name, kwargs)) if self._window_task is None or self._window_task.done(): self._window_task asyncio.create_task(self._run_window()) try: return await asyncio.wait_for(fut, timeout10.0) except asyncio.TimeoutError: async with self._lock: futures self._pending.get(key) if futures: futures [f for f in futures if f is not fut] if futures: self._pending[key] futures else: del self._pending[key] raise async def _run_window(self): while True: await asyncio.sleep(self.window_ms) async with self._lock: if not self._queue: self._window_task None return batch self._queue[:self.max_batch_size] self._queue self._queue[self.max_batch_size:] try: results await self._dispatch(batch) except Exception as e: async with self._lock: for item in batch: futures self._pending.pop(item[0], []) for f in futures: if not f.done(): f.set_exception(e) continue async with self._lock: for item, result in zip(batch, results): key item[0] futures self._pending.pop(key, []) self.cache[key] result self.cache.move_to_end(key) if len(self.cache) self.cache_capacity: self.cache.popitem(lastFalse) for f in futures: if not f.done(): f.set_result(result) async def _dispatch(self, batch): # 按模型名分组简化为单模型调用 texts [item[1] for item in batch] model_name batch[0][2] return await self.embed_func(texts, model_name)4.3 接入 MCP Server 的配置与参数推荐把这套合并器接入 MCP Server 的流程很简单先实例化一个全局的EmbeddingCoalescer把本地的 embedding 批处理函数传进去然后在所有工具函数里统一调用。coalescer EmbeddingCoalescer( embed_funclocal_embed_batch, window_ms50, max_batch_size32, cache_capacity4096, ) # 在工具函数中 result await coalescer.submit(text, model_namebge-m3)参数方面我推荐一组起步值你可以根据负载调整window_ms取 50本地模型推理快可以放到 100远程 API 建议 20-30max_batch_size取 32显存有限就降到 8-16cache_capacity取 4096约占 16MB按 1024 维向量计算请求超时用 10 秒必须小于 MCP Client 侧的超时时间否则客户端先超时断开服务端还在傻等结果。4.4 实测结果拦截率、延迟与算力成本变化我在内部知识库场景实测了一轮跑了一个星期统计到的数据大致如下重复请求拦截率稳定在 38%-46% 之间说明线上确实有约四成请求是重复的接口 P95 延迟从 630ms 降到 410ms原因是并发洪峰被合并后排队等待的时间明显减少embedding 模型所在 GPU 的利用率从 35% 提升到 60% 左右如果用的是第三方 APItoken 消耗能直接下降三到四成。需要说明的是具体数字跟业务负载强相关。如果你的场景是对话轮次多、上下文重复率高收益会更大如果是纯文档一次性入库几乎没有重复请求那这套机制带来的收益就很小。所以我建议做这个改造前先在日志里统计一周的重复率用数据决定要不要上合并机制。5. 常见问题与排查技巧实录5.1 窗口内请求堆积导致延迟抖动上线后我遇到的第一个问题是偶尔的 P99 延迟抖动。排查下来发现某个时间段窗口内到达的请求数量远超max_batch_size超出的请求要等下一个窗口等待时间叠加导致延迟飙升。解决办法有两个方向一是观察batch 截断率指标如果频繁截断说明并发量超过批量上限可以调大max_batch_size或者调小window_ms来减少单窗口内积压的任务量二是如果单批推理本身耗时很高说明模型负载已经逼近瓶颈这时候应该扩容而不是继续调参。5.2 Future 泄漏与超时处理另一个高频问题是 Future 泄漏。表现是日志里出现Task was destroyed but it is pending或者某些请求的等待时间异常长。原因是下游embed_func抛异常时如果代码没有把所有 pending 的 Future 都设置异常或结果那些 Future 就永远悬空请求方会一直等下去。我的处理方式是在_run_window里包了一层 try-except批量推理出现异常时把该批次所有 key 对应的 Future 都set_exception。同时在submit端用asyncio.wait_for做超时兜底超时后不仅抛出异常还要把当前 Future 从 pending 链表中摘除。这两个动作缺一不可否则泄漏迟早会把服务拖垮。5.3 多实例部署下的缓存一致性问题如果 MCP Server 是多副本部署每个实例的本地缓存是独立的。这意味着同一个重复请求打到不同实例依然会重复计算。如果要彻底解决可以在中间加一层 Redis 缓存但网络往返会牺牲一些低延迟优势。我给的建议是分阶段走先做单实例合并这通常能解决大部分重复问题如果多实例下重复率依然高再考虑集中式缓存。注意集中式缓存一定要设置 TTL否则向量数据可能和模型版本脱节。另外别为了强一致给缓存加分布式锁那会让整个合并机制变成性能瓶颈。5.4 监控指标与日志设计这套机制上线前我强烈建议先埋好监控指标否则很难判断改造效果。重点看四个指标指标含义参考值cache_hit_rate缓存命中率反映历史重复请求占比30%-50% 算健康coalesced_ratio合并比例等于总请求数减去实际推理次数再除以总请求数越高越好avg_batch_size平均批量大小反映凑批效率越接近 max_batch_size 越好window_pending_count窗口内待处理请求数出现持续高值说明窗口太大日志设计上我建议在submit入口和_run_window出口各打一条包含指标的结构化日志。这样既能看到单请求视角也能看到批处理视角对账非常方便。我用这套日志捞过好几次异常效果比接到告警再翻链路快得多。最后分享一个实际体会构建任务合并机制这件事真正有价值的不是代码本身而是“先量化、再改造”的思路。上线前我只花了三天统计重复率就确认了改造方向改造过程也只用了一天。很多时候我们习惯于一上来就堆组件、加缓存、上消息队列但真正该做的可能只是在日志里多打一行数据然后让机制自己去挡掉那些本不该重复的计算。