如果你维护过基于 MCP Server 的 AI 应用大概率见过这样的一幕一次用户提问触发了十几个工具调用每个工具都返回一大段文本随后这段文本被反复向量化被塞进上下文或回填给检索模块。日志里出现大量完全相同的 Embedding 请求——同一个模型、同一段内容、同一种参数字段却算了一遍又一遍。我是在一次性能治理中彻底意识到这个问题的。当时 MCP Server 挂在 GPU 服务前面GPU 使用率一直下不来逐条对日志才发现某份热门文档在十分钟里被向量化了两百多次而它的内容一个字都没变过。MCP Server 是模型上下文协议的服务端实现它做好的是工具注册、参数解析和消息转发它完全不会替你判断这次 Embedding 请求是不是重复的。只要外层没有去重机制重复计算就是必然的。这篇文章不绕弯子直接讲我怎么在 MCP Server 内部构建任务合并机制把重复的 Embedding 请求精准拦截掉先分析冗余计算的真实来源再讲清任务合并、结果缓存、批量请求这三者的分工然后给出一个基于 asyncio 的完整实现最后附上压测数据和我在这个过程中踩过的坑。适合正在做 Agent 性能优化、或者想控制 Embedding API 费用的团队参考。1. 算清这笔账MCP Server 里的重复 Embedding 究竟从哪来1.1 一次压测让我注意到 40% 的请求是重复的先说一个让我决定做这个机制的具体场景。我当时维护一个给内部知识库用的 MCP Server工具不多但每个工具背后都依赖 Embeddingsearch_document要把 query 和文档片段向量化build_context要把候选段落向量化还有一些辅助工具需要把对话历史向量化后做相关度重排。某次压测里我在 Embedding 客户端外面打了一行关键日志输出每个请求的内容摘要和参数指纹。10 分钟内MCP Server 一共收到 8120 次 Embedding 调用请求其中 3271 次的指纹完全一致——重复率超过 40%。这个数字让我停了一下。要知道我们的工具调用量还没有到很大的规模如果将来日均请求量再涨十倍这 40% 就是纯纯的算力黑洞。这不是个案。只要 Agent 在工具调用循环里拿到一段文本后先在搜索工具里做了一次向量化又把同一段文本传给总结工具再做一次向量化重复就发生了。特别是在 RAG 流程里同一批候选文档会被 ReRank、上下文构造、缓存更新等多个环节各自消费一次每个环节都当成新文本去请求 Embedding实际上算的都是同一个东西。1.2 三种最常见的重复请求形态在我观察到的重复请求里主要有三种形态你大概率也遇到过跨工具的中间结果被重复消费。工具 A 返回了一段文档摘要工具 B 拿到这段摘要后重新向量化工具 C 又向量化一次。中间结果在工具链里传递但每个工具内部都默认我需要自己算一遍向量。Agent 重试与上下文重建。LLM 在工具调用失败后重试时会把同样的上下文重新提交或者 Agent 对同一个请求做了多次候选推断每次都要把历史对话重新 Embedding 一遍。多用户并发命中同一热点文本。多个会话同时请求同一份文档、同一段代码、同一个标准问题。尤其在知识库场景里热门文档的并发命中率非常高。这三种形态有一个共同点都不是用户故意发起重复请求而是架构分层时没考虑语义去重。1.3 冗余消耗的不仅仅是 GPU 算力重复 Embedding 的代价最直接的是 GPU 或 CPU 算力。模型推理本身是计算密集型操作同样的文本算两次就要烧两份电。但更隐蔽的是另外三笔账第一笔是外部 API 费用。如果用的是 OpenAI 的text-embedding-3-small这类按 token 计费的接口重复计算等于直接烧钱。假设每 1000 token 是 0.02 美元一天多算 100 万 token 的重复请求一个月下来就是 600 美元的纯浪费。第二笔是延迟。Embedding 请求通常是 HTTP 调用一次 200ms如果同一个请求被拆成三次调用用户感受到的延迟就是 600ms 而不是 200ms。第三笔是监控噪音。日志被无意义的重复请求刷屏真正的问题反而看不出来了。MCP Server 本身处在协议转发层它只负责把工具调用请求路由到对应实现并不关心这些请求背后是不是同一个语义实体。所以这个去重机制必须由开发者自己加在工具实现层。这也是我会想到任务合并这个思路的原因在请求真正打到 Embedding 服务之前先看能不能把重复请求合并掉。2. 机制定位任务合并、结果缓存、批量请求分工别搞混很多人一听合并重复请求第一反应是不就是加个缓存吗。其实任务合并request coalescing和结果缓存、批量请求是三个不同的机制解决的问题不同最佳使用场景也不同。我在设计的时候花了不少时间理清这三者的边界这里展开说一下。2.1 任务合并解决并发窗口内的同键等待任务合并的核心能力是在同一个时间窗口内多个调用方拿着完全相同的参数请求同一个结果时只让其中一个请求真正去执行计算其余请求全部挂起等待这个执行结果而不是各自重新发起调用。用一个生活化的例子会议室里十个人同时问会议纪要发我一份。任务合并机制的做法是让一个人负责去整理纪要整理好后统一抄送给其他九个人而不是十个人同时去找行政要一份文件。关键点是任务合并只对并发窗口内的重复有效。如果第一个请求已经完成过了两秒钟又来一个完全相同的请求合并器已经不认识它了——因为那个正在计算的请求已经结束合并态不存在了。此时要拦下这第二次请求靠的是结果缓存。2.2 结果缓存解决跨请求的时间复用结果缓存解决的是时间维度上的重复和任务合并正好互补。缓存把已经算好的向量存下来后续相同 key 的请求直接返回结果不再触发计算。但缓存有个问题Embedding 结果对模型版本、文本内容、维度参数、normalize 开关都很敏感。任何一项变了缓存都可能失效。而且缓存存的是浮点数组多一份就是多一份内存开销所以必须有容量上限和过期策略。我之前见过一个团队把缓存 TTL 设为 7 天结果中间升级了一次 Embedding 模型所有旧向量和新向量混在一起做余弦相似度检索结果乱成一锅粥。后来他们把模型版本号直接拼进了缓存 key升级模型后旧缓存自然失效才彻底解决问题。2.3 批量请求不同文本也可以拼车合并负责相同 key批处理负责不同 key。以很多 Embedding 服务为例一个请求里可以传一个字符串数组一次调用返回多个向量。所以如果同时有几十个不同文本要编码与其循环调用几十次不如一次批量调用。批处理能减少 HTTP 往返、降低网络开销在某些网关场景下还能减少调用次数费用。但它和合并混在一起会引入复杂度如果一个批次里其中一段文本计算失败其他文本的结果怎么处理不同模型不能混在一个 batch 里而且批量调用会使得单个请求的超时控制更难做。我的建议是分层处理任务合并和缓存解决同 key 重复批处理解决多 key 拼车不要在一开始就把它们揉成一个东西。2.4 理想流程缓存查一遍合并等一等批处理发一程最终我在 MCP Server 里把三者串成了这样一条流水线接收 Embedding 请求计算请求指纹。先查结果缓存命中直接返回。未命中检查合并表发现有同 key 任务在途就挂到同一个 Future 上等待。没有在途任务也没有缓存则作为 Leader 创建一个 Future把文本投入批处理队列。批处理器把多个不同 key 的文本组合成一个请求发给模型服务。返回结果后写缓存、释放 Future。这个顺序很重要。先查缓存是为了命中后连合并表都不碰先查合并表是为了避免并发穿透时大量相同请求同时打到批处理队列。缓存和合并层层拦截到最后真正发出去的请求已经是唯一且必要的请求了。3. 核心实现基于 asyncio 的合并调度器从零写一遍下面进入正题。我用 Python 写一套可以落到 MCP Server 里的实现因为 MCP 官方 Python SDK 用起来最顺手而且 asyncio 原生支持 Future 的共享等待非常适合做合并机制。3.1 第一件事请求指纹要设计对合并的前提是能判断两个请求是不是同一个请求判断依据就是请求指纹。指纹设计不好合并机制要么误伤要么漏杀。我习惯把model、文本内容、dimensions、normalize等会直接影响返回向量的参数全部纳入指纹。但有一个优化点不要直接把整个长文本序列化进指纹。如果一段文本有十万字每次算指纹都做一次 JSON 序列化等于引入了新的算力浪费。安全的做法是对文本先做 SHA-256 摘要再把摘要作为指纹的一部分。import hashlib import json from dataclasses import dataclass def digest_text(text: str) - str: return hashlib.sha256(text.encode(utf-8)).hexdigest() dataclass(frozenTrue) class EmbeddingRequest: model: str text: str dimensions: int | None None normalize: bool False def key(self) - str: payload { model: self.model, text_digest: digest_text(self.text), dimensions: self.dimensions, normalize: self.normalize, } canonical json.dumps( payload, sort_keysTrue, ensure_asciiFalse, separators(,, :), ) return hashlib.sha256(canonical.encode(utf-8)).hexdigest()注意sort_keysTrue和separators(,, :)这两个参数。sort_keys保证字段顺序变化不影响指纹separators去掉空格保证同样的请求不会因为格式化差异产生不同指纹。ensure_asciiFalse则避免中文被转成\uXXXX的形式让指纹更稳定也不影响语义。3.2 核心调度器基于 Future 的 Leader 选举合并调度器的核心是一个非常简单的数据结构一个字典key 是请求指纹value 是asyncio.Future。逻辑也很直接第一个请求进来后发现字典里没有对应的 Future它就当选 Leader创建一个 Future 放入字典然后真正去执行 Embedding 计算后续进来的相同 key 请求看到字典里已经有 Future说明有人在算了就直接await这个 Future。Leader 计算完成后把结果写入 Future所有等待者一次性拿到结果。import asyncio import time from collections import OrderedDict from typing import Awaitable, Callable EmbeddingResult list[float] class EmbeddingCoalescer: def __init__( self, executor: Callable[[EmbeddingRequest], Awaitable[EmbeddingResult]], ttl: float 300.0, max_cache_entries: int 1024, ) - None: self._executor executor self._ttl ttl self._max_cache_entries max_cache_entries self._inflight: dict[str, asyncio.Future] {} self._cache: OrderedDict[str, tuple[float, EmbeddingResult]] OrderedDict() self._lock asyncio.Lock() async def embed(self, request: EmbeddingRequest) - EmbeddingResult: key request.key() now time.monotonic() async with self._lock: # 第一步读缓存 cached self._cache.get(key) if cached is not None and now - cached[0] self._ttl: self._cache.move_to_end(key) return cached[1] # 第二步读合并表 fut self._inflight.get(key) if fut is None: # 当前协程成为 Leader创建 Future loop asyncio.get_running_loop() fut loop.create_future() self._inflight[key] fut is_leader True else: # 已有 Leader 在算等待结果 is_leader False if not is_leader: # 不要在锁里 await释放锁后再等待 return await asyncio.shield(fut) try: result await self._executor(request) except BaseException: async with self._lock: if self._inflight.get(key) is fut: self._inflight.pop(key, None) raise # 写入缓存并释放 Leader async with self._lock: if self._inflight.get(key) is fut: self._inflight.pop(key, None) self._cache[key] (time.monotonic(), result) self._cache.move_to_end(key) self._evict_if_needed() return result def _evict_if_needed(self) - None: while len(self._cache) self._max_cache_entries: self._cache.popitem(lastFalse)这段代码有四个关键点值得展开讲第一_lock保护的是check cache、check inflight、create future这一段逻辑。这个操作必须原子否则两个协程可能同时发现没有 Leader各自创建一个 Future合并就失效了。虽然 CPython 的 GIL 让字典单次操作是原子的但检查 创建是两步操作中间如果有协程切换依然会出问题。用asyncio.Lock最稳妥。第二等待方拿到fut之后不要在async with self._lock块内 await。因为锁还没释放其他请求会卡在锁外面整个服务就串行化了。代码里把is_leader的判断放在锁内把真正的await asyncio.shield(fut)放在锁外这个细节极其重要。第三asyncio.shield(fut)的意义在于保护 Future 不被取消传播误伤。等待方所在的 Task 如果被用户取消直接await fut在某些异常情况下可能把取消信号传给 Future用shield包裹后等待方取消不会影响 Leader 的计算。否则一个请求被取消其他几百个等待方的结果就全部丢失了。第四Leader 失败时要把自己从_inflight中移除否则后续请求会一直等待一个已经失败的 Future。这里用BaseException捕获包括CancelledError便于 Leader 取消时也能正确清理。3.3 接入 MCP Server 的工具调用有了调度器接入 MCP Server 就很简单了。关键在于要把EmbeddingCoalescer做成 MCP Server 服务级的单例而不是在某个工具函数内部临时 new 一个。from mcp.server.fastmcp import FastMCP mcp FastMCP(coalesced-embedding) coalescer EmbeddingCoalescer( executorcall_embedding_model, ttl900, max_cache_entries4096, ) async def call_embedding_model(request: EmbeddingRequest) - list[float]: # 这里调用真实的 Embedding 模型例如 bge-m3、text-embedding-3-small 等 # 如果模型客户端是同步阻塞的用 await asyncio.to_thread(...) 包一层 result await embedding_client.embed( modelrequest.model, textrequest.text, dimensionsrequest.dimensions, normalizerequest.normalize, ) return result mcp.tool() async def embed_text(text: str, model: str bge-m3) - list[float]: req EmbeddingRequest(modelmodel, texttext) return await coalescer.embed(req)如果 MCP Server 里有多个工具都需要 Embedding比如search_document、build_context、rerank这些都要统一调用同一个coalescer.embed(req)而不是各自去调call_embedding_model。否则这个机制等于没做——每处独立的合并器依然各算各的。还要注意一点MCP Server 的事件循环是全局共享的。如果 Embedding 客户端是同步阻塞的工具函数是async直接在里面调用同步接口会瞬间阻塞整个服务。这种情况要把调用丢到线程池里比如用asyncio.to_thread(sync_embedding_client, request.model, request.text)避免拖垮其他工具。3.4 异常、取消与重试策略合并机制本身不是用来做重试的它只负责把重复请求聚合成一次执行。真正执行 Embedding 的executor内部需要自己处理重试逻辑但要注意重试次数不能太多否则所有等待者都会等一个不断重试的 Leader原本 200ms 的请求可能被拖成几秒。还有一个容易被忽略的现象当 Leader 执行失败时Future会保存异常所有等待者会同时拿到同一个异常。这会导致错误日志瞬间爆炸而且如果每个调用方都立刻重试相当于又发起了一批并发请求反而加剧雪崩。我在生产环境里做过一个简单处理在executor内部做最多 2 次重试重试之间加指数退避如果 Leader 最终失败等待方收到的同一个异常里带上coalesced标记日志系统看到这个标记就知道是共享异常不需要每个请求都报一遍完整堆栈。4. 实测与调优合并机制上线前后的数字对比光说原理不够我用一组模拟压测数据来说明效果。这一节的目标是让你对合并机制到底能省多少有个直观感受同时理解 TTL、容量和指纹开销的权衡。4.1 压测场景设计我在本地搭了一套模拟环境。MCP Server 使用 Python SDKEmbedding 模型模拟一个平均耗时 200ms 的外部 HTTP 接口。测试脚本用 asyncio 同时启动 50 个任务每个任务会随机请求 10 段文本中的任意一段总共发起 500 次请求。这样设计的用意很明显文本池只有 10 个唯一 key理想情况下这 500 次请求的上游调用次数应该只有 10 次。但如果不做任何去重上游就会被真实调用 500 次。为了更贴近真实环境我刻意把上游 API 的并发限制调整为 5模拟外部服务商常见的限流策略。也就是说在没有合并机制时500 个请求会排成很长的队列逐批处理合并机制生效后同一时刻最多只有 10 个唯一 key 在抢 5 个并发通道。配置无合并机制任务合并 缓存上游实际调用次数50010上游调用拦截率0%98%总墙钟耗时约 21 秒约 400ms等待方平均拿到结果时长取决于队列位置约等于 Leader 计算时长这个表看得很清楚合并机制把上游调用量从 500 次压到了 10 次98% 的冗余请求被精准拦截。墙钟耗时的改善尤其惊人——因为无合并时 500 个请求挤在 5 并发通道里光排队就排了 20 多秒合并后只有 10 个唯一请求两批就全部跑完其他 490 个等待任务几乎同时拿到结果。4.2 从日志里看拦截效果优化前日志长这样2025-01-10 11:23:01.123 embed keyab12... ok latency210ms 2025-01-10 11:23:01.125 embed keyab12... ok latency208ms 2025-01-10 11:23:01.127 embed keyab12... ok latency212ms ...三行日志三个完全一样的 key三次真实计算。如果这个 key 被重复请求 300 次日志里就刷出 300 行真正的性能瓶颈被淹没在一起。优化后日志变成2025-01-10 11:23:01.121 [coalescer] leader started keyab12... 2025-01-10 11:23:01.331 [coalescer] leader finished keyab12... cost210ms 2025-01-10 11:23:01.331 [coalescer] 211 waiter(s) resolved for keyab12...一个 key 可能对应几百个等待者但真正执行的计算只有一次。这时候你再去看 GPU 利用率和 API 账单会发现之前那条一直压不下去的曲线直接断崖式下跌。4.3 TTL、容量和指纹开销的权衡任务合并的代码不复杂但上线前要把三个参数想清楚。第一个是ttl。缓存的存活时间。如果 MCP Server 处理的是短对话用户提问后 60 秒内可能反复用到同一批上下文TTL 设 60 到 300 秒就够了。如果是知识库静态文档向量化TTL 可以设到 1 小时甚至更长。但 TTL 太长会导致一个问题模型升级后新旧向量混用。所以模型版本号必须作为缓存 key 的一部分升级时用 key 前缀的方式让旧缓存自动失效而不是依赖 TTL 硬扛。第二个是max_cache_entries。Embedding 结果是浮点数组内存开销不能无视。以 BGE-M3 为例单个向量是 1024 维的 float32一个结果大约占 4KB。如果缓存 1 万条数据就是 40MB如果不设上限热点文档稍微多一点内存就会失控。我这里用OrderedDict实现了一个简单的 LRU 淘汰容量超过max_cache_entries时从最久未使用的条目开始淘汰。第三个是指纹计算开销。对于超长文本先做 SHA-256 摘要再做 JSON 序列化能显著降低每次请求的 CPU 开销。我测试过一个 10 万字符的文本直接序列化整个字符串大约需要 1ms而先摘要再序列化只需要 0.2ms。在高 QPS 场景下这个差距会被放大。5. 踩坑记录让精准拦截真正精准我踩过的几个坑任何机制在上线过程中都会遇到实际问题。下面这几个坑我基本都踩过一遍写出来帮你避开。5.1 合并键过宽或过窄都会出事故合并键的设计必须覆盖所有影响返回结果的参数但不能包含调用方身份这类无关元数据。我犯过两个方向的错误。一次是只在 key 里放了model和text忘了放dimensions结果同一个模型下请求 512 维和请求 1024 维的两个任务被错误合并下游拿到 512 维向量时直接维度错误。另一次是反过来key 里放了调用方传入的request_id导致完全相同的文本因为 request_id 不同而无法合并拦截率骤降。正确的做法是model、text_digest、dimensions、normalize这类影响结果的参数必须进 keyrequest_id、user_id、source这类不影响向量结果的元数据一律不进 key。5.2 每个工具各自 new 一个合并器等于没做这是最隐蔽的坑。代码结构看起来没问题每个工具模块都封装得很好但每个模块里各自EmbeddingCoalescer(...)每个实例的_inflight和_cache都是独立的相同文本在不同工具里依然各算各的。解决方式很明确把EmbeddingCoalescer注册成 MCP Server 容器里的单例所有工具共享同一个实例。我后来直接在模块顶层创建了一个全局coalescer工具函数只负责把请求交给它这个简单粗暴的改变立刻让拦截率提了上来。5.3 等待方取消请求差点把 Leader 一起带走上线初期我遇到过一个诡异现象某个工具调用超时被用户取消紧接着后台冒出大量错误日志全部指向同一个 key而且这个 key 的 Leader 任务也被取消了。后来定位到是等待方直接await fut取消信号沿调用链传播把共享的 Future 污染了。改用asyncio.shield(fut)之后这个问题再没出现过。shield的语义就是我不怕被取消我等的这个 Future 不能被取消。对于合并机制来说Leader 一旦开始执行就应该让它安安静静算完不能因为一个等待方取消就前功尽弃。5.4 进程内合并拦不住多实例重复MCP Server 一旦做水平扩展进程内合并机制就失效了。假设前面挂了负载均衡同时有 3 个 MCP Server 实例每个实例里都有各自的EmbeddingCoalescer同一段文本还是会被算 3 次只是每个实例内部的拦截率看起来都很高总盘一看重复照样存在。要解决这个问题就得把合并机制提升到分布式层用 Redis 的原子锁抢 Leader用 Pub/Sub 广播结果或者更干脆一点把所有的 Embedding 请求收敛到一个独立的内部服务统一去重。任务合并机制的收益上限受限于实例数量这个认知要提前建立别等扩完容才发现白做了。5.5 缓存污染模型升级后旧向量还在最后是一个隐藏得很深的坑。MCP Server 里的缓存 key 如果不带模型版本某天你把 BGE-M3 换成了微调版本旧缓存里的向量依然会被返回。这些向量和新的模型向量不在同一个语义空间里下游做余弦相似度时结果会完全乱掉。我在代码里强制要求EmbeddingRequest必须显式传model指纹里也就自动包含了模型版本。模型升级时只需要改配置里的默认模型名所有旧 key 自动失效不需要手动清缓存。这一点一定要在机制设计之初就考虑进去不然后期排查语义漂移的问题会非常痛苦。最后说一句我自己的体会任务合并机制不是万能的它解决的是并发窗口内的同键重复要彻底治理算力内耗还得和结果缓存、批量请求、模型版本管理配合在一起。好在这个机制并不复杂把请求指纹和 Future 合并表写好再接入 MCP Server 的工具调用层就能挡掉一大批冗余计算。希望这篇记录能帮你少走我走过的弯路。