2026/9/8 17:08:02
FEATURED · 精选文章 
持久执行就不会重复了吗这个问题我第一次在项目群里看到的时候脑子里是有点懵的。因为当时我们刚把业务从普通的同步调用改成 durable execution 模式心里默认了既然能恢复执行那总该稳了吧。结果上线第一周我就被支付回调的重复通知打脸了。同一笔订单用户只付了一次钱我们的 Agent 却把发货流程跑了两遍还差点给同一个客户发了两张优惠券。后来我才彻底想明白一件事所有号称持久的执行框架保证的都只是任务不会丢挂了能重放它默认给你的是 at-least-once 语义而不是 exactly-once。想做到业务层面的只发生一次光靠持久执行远远不够必须在应用层做幂等设计。而这一步恰恰是很多做 AI Agent 工程化的人最容易忽略的。这篇文章就围绕我最近在 Pydantic AI v2.36 环境里落地的一套方案展开把 Agent 的持久化执行、幂等键Idempotency Key和 MCP 门禁组合在一起解决重复执行和工具乱调两个老大难问题。内容会比较偏实战适合正在做 Agent 工程化、接 MCP 工具、或者被重试和并发搞得焦头烂额的朋友。1. 先搞清楚一件事持久执行到底保证什么1.1 持久执行是什么不是不重复是不丢失首先得把概念掰清楚。持久执行Durable Execution这套东西近几年随着 Temporal、AWS Step Functions 还有各种工作流引擎火起来本质上是把任务的执行状态、事件日志、中间结果全部落盘再配合重放机制让一个跑了一半的任务在进程崩溃后能从中断点继续跑而不是从头来。这就像你写文档的时候开了自动保存。Word 崩溃了重启之后它会问你要不要恢复未保存的版本你点恢复文档回来了。但注意恢复的可能是你十分钟前的版本里面对你刚才敲的几段话可能还是旧内容。持久执行也是这个道理——它保存的是执行轨迹不是执行结果。它能保证任务不会因为崩溃就永远消失但它不承诺每个副作用只执行一次。在 Pydantic AI 这种 Agent 框架里持久执行通常体现在一个 Agent 的 run 过程可以被快照、被暂停、被恢复多步工具调用的中间状态可以保存。比如一个 Agent 正在执行查库存 - 下单 - 发通知三步如果跑完第二步、还没跑第三步的时候进程崩了恢复之后它会从第三步重新开始而不是把整个流程重来一遍。这已经比纯内存执行强太多了但它依然不解决第二步重复执行的问题。1.2 三种投递语义at-most-once / at-least-once / exactly-once这是分布式系统里最经典的一组概念放到 Agent 里一样适用At-most-once最多一次发一次就完事丢了不补。性能最好但消息丢失不负责。适合日志、埋点这种丢了也无所谓的场景。At-least-once至少一次只要没确认成功就不断重试、重放保证消息不丢但可能重复。绝大多数消息队列、任务队列、持久执行框架默认都是这个语义。Exactly-once精确一次不丢不重理想状态但分布式环境下极难真正实现通常只能通过实现幂等 至少一次投递来达到效果。注意最后这句几乎没人能在分布式系统里真正做 exactly-once 的投递大家做的都是至少一次投递 业务幂等。因为没法保证网络不重试、进程不在提交前崩溃所以只能在接收方做去重。这也是为什么我后来养成了一个习惯所有被 Agent 调用的业务接口第一件事就是查幂等键而不是直接开干。1.3 重复的高危场景崩溃、超时、重试、消息队列到底哪些场景最容易触发重复执行我列几个真实踩过的进程崩溃在副作用完成之后。Agent 调用工具 A 成功工具 A 返回结果但结果还没来得及写进持久化存储进程崩了。恢复之后框架从快照重放它不知道工具 A 已经执行过了于是又调了一次。HTTP 超时导致的自动重试。Pydantic AI 调用 MCP 工具内网抖动导致响应超时框架自动重试。但服务端其实已经处理完了重试意味着同样的操作被处理两遍。消息队列的重投。Agent 的请求是从消息队列拉出来的消费成功但没来得及 ack或者 ack 丢失消息被重新投递整个 Agent 流程又跑了一遍。用户手动重试。前端看不到结果用户狂点重试按钮同一个请求进来好几次。并行触发。两个上游服务同时回调同一个 webhook都触发了同一个 Agent 流程。这些场景的共同特点是框架层面看起来只执行了一次但实际系统里执行了多次因为存在执行成功和被确认之间的时间缝隙。1.4 为什么持久执行解决不了重复一句话总结持久执行解决的是中断后恢复的问题它给你的保障是任务不会丢断了会接着跑但它无法知道你执行过程中的每个外部副作用是否已经生效。比如说你的 Agent 在步骤二调用了一个发送短信的 HTTP 接口。这个接口执行成功但网络返回超时。持久执行框架捕获到的是调用超时这个结果它不知道短信其实已经发出去了。到了恢复重放的时候它自然会再次调用发送短信。这时候你指望框架帮你判断框架做不到因为它没有业务语义——它并不知道发送短信这个操作对于业务来说是幂等的还是非幂等的。所以结论非常明确持久执行是基础设施层面的保障幂等是业务层面的保障两者缺一不可。只有把持久执行和幂等设计叠在一起你才能在实践中逼近 exactly-once 的效果。2. Pydantic AI v2.36 里的组合拳持久Run 幂等键2.1 v2.36 带来的变化我是在把项目从 v2.28 升到 v2.36 的时候重新审视这个问题的。Pydantic AI 的版本更新一直比较快v2.36 这个版本有几个点对我很有吸引力Agent 的运行状态管理更细了对 MCP 协议的支持也更成熟而且可以通过 hooks 拦截工具调用流程。这些能力正好是我做幂等和门禁的抓手。这里要说明一下我下面聊的方案不完全依赖框架自带的某个开关而是基于 v2.36 提供的能力做了一层自研封装。因为幂等这种东西框架不可能帮你全做完——它不知道你的业务哪个操作是天然幂等的哪个操作需要有业务去重键。所以框架能做的是提供更可靠的持久化、更灵活的拦截点、以及更方便的上下文管理而幂等策略本身得你自己设计。在 v2.36 里我主要用了这几个能力Agent 的 run_id 和上下文透传每个 run 都有一个唯一 ID可以把它作为幂等键的基础。持久化状态存储Agent 的运行状态可以保存到外部存储Redis、PostgreSQL 等崩溃后能恢复。工具调用拦截 hooks可以在工具执行前、执行后插入自定义逻辑这就是门禁和幂等检查的关键位置。MCP Server 接入可以挂载 MCP Server 作为工具源用标准协议调用外部工具。2.2 幂等键设计用 run_id 还是业务键幂等键听起来高大上其实就是个字符串。客户端在发起请求时生成一个全局唯一的键通常用 UUID服务端收到带幂等键的请求后先查一下这个键之前有没有处理过没处理过正常执行把结果存下来键和结果绑定。处理过直接把之前的结果返回不再执行业务逻辑。这个先查再执行的过程要保证原子性否则两个并发请求同时查到没处理过又同时执行还是会重复。所以我在项目里的做法是先往数据库插一条幂等键记录利用唯一索引抢占插入成功才继续执行插入失败说明之前已经处理过直接返回历史结果。在 Pydantic AI 的环境里我通常会把 Agent 的一次完整执行一个 run作为一个幂等单元。幂等键的生成规则有两种选择使用 Agent 自带的 run_id适合一次用户请求对应一次 Agent 执行的场景比如用户问一个问题Agent 跑一轮回复。同一次请求的重试run_id 保持一致。使用业务自定义键适合一个业务事件可能触发多次 Agent 执行的场景比如支付回调用 order_id payment_event_id 拼一个唯一键保证同一笔支付即使回调十次也只执行一次 Agent 流程。我比较推荐第二种。因为 run_id 只代表这一次执行而业务幂等关心的是这个业务事件是否已处理。同一个业务事件可能因为队列重投产生多个 run_id这时候用 run_id 做幂等键就失效了。2.3 唯一约束与结果缓存双保险幂等设计光靠查缓存是不够的必须双管齐下第一层是唯一约束。在存储层面对幂等键加 unique index这是最终防线。两个请求同时进来只有一个能插入成功。我用 PostgreSQL 的表结构大致是这样的CREATE TABLE agent_idempotency ( id BIGSERIAL PRIMARY KEY, idempotency_key VARCHAR(128) NOT NULL UNIQUE, request_payload JSONB NOT NULL, response_payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), completed_at TIMESTAMPTZ );第二层是结果缓存。把幂等键对应执行结果缓存到 Redis设置过期时间。这样即使同一个键在短时间内被查询很多次也不需要每次都打数据库可以大大减轻压力。import hashlib import json import uuid from redis.asyncio import Redis class IdempotencyManager: def __init__(self, redis: Redis, ttl_seconds: int 86400): self.redis redis self.ttl_seconds ttl_seconds def generate_key(self, *parts) - str: raw |.join(str(p) for p in parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest() async def get_cached(self, key: str) - dict | None: data await self.redis.get(fidem:{key}) return json.loads(data) if data else None async def cache_result(self, key: str, result: dict) - None: await self.redis.set( fidem:{key}, json.dumps(result), exself.ttl_seconds, )这段代码看着简单但它解决的是重复请求导致重复执行的核心问题。每次 Agent 要跑一个带副作用的业务流程之前先查一次这个管理器命中就直接返回没命中就继续。注意这里缓存只是用来加速读取真正的唯一性保证必须依赖数据库的唯一索引Redis 的过期和并发竞争都不足以承担最终防线。2.4 什么时候必须自己写幂等很多人会问Pydantic AI 自带 Agent 持久化是不是就不需要我管幂等了我的答案是分情况如果你的 Agent 只是做读操作——查个天气、搜个资料、读一下文档——那不做幂等问题不大因为读操作天然幂等执行一百次结果都一样。但如果你的 Agent 会触发写操作——下单、转账、发消息、改数据库——那就必须做幂等。哪怕这个操作后面接的是一个号称幂等的接口比如数据库 UPDATE 语句也未必安全因为 Agent 发出的指令可能不是纯 UPDATE而是读余额 - 算优惠 - 改余额这样多步操作中间任意一步重复执行都可能导致资金错误。我的经验法则很简单只要 Agent 的某个步骤有外部副作用就必须把它包进幂等单元。副作用越大越要谨慎。比如发邮件这种操作重复发一封可能只是让用户觉得烦但如果是调支付接口、扣库存、写账单重复一次可能就是事故。3. MCP 门禁把工具调用关进笼子里3.1 MCP 的价值与失控风险MCPModel Context Protocol最近几个月在圈内确实火各种某某工具已支持 MCP的新闻层出不穷。它的本质是给 AI 提供一套标准化的工具调用协议模型通过 MCP 客户端暴露出来的工具列表去调用外部能力相当于给 Agent 装上了万能遥控器。权力越大风险越大。没打补丁之前我们项目里的 Agent 可以自由调用挂载的任何 MCP 工具。低危操作还好但像发邮件、删除文件、修改数据库这类高危操作一旦被模型的幻觉或者 prompt 注入带偏后果不堪设想。做 AI Agent 的人应该都听说过那个经典例子让 Agent 帮我整理一下文档结果它把整个目录删了。这就是工具调用没有门禁的后果。MCP 门禁本质上就是在模型要调用工具和工具真正执行之间加一道闸。开闸还是关闸由你的业务策略说了算而不是由模型的意图说了算。3.2 门禁三层白名单、参数校验、审批流我在项目里把门禁拆成了三层每层都可以独立配置策略第一层工具白名单。不是 MCP Server 暴露的所有工具都能被 Agent 调用。每个 Agent 只允许调用白名单里的工具其他一律拒绝。这一层最简单但过滤掉了大多数误调用。比如一个只做客服问答的 Agent白名单里只能有查订单查物流绝不能有修改库存删除商品。第二层参数校验。工具能调但参数得合规。比如发邮件工具收件人必须是当前用户邮件内容长度必须限制附件路径必须在允许的目录里。这一层主要防的是 prompt injection——模型被恶意上下文影响把不该传的参数传进去。参数校验规则用 JSON Schema 写最合适因为 MCP 工具本身就支持 JSON Schema 描述参数。第三层审批流。最高危的操作比如付款、批量删除、对外发送公告必须走人工审批。Agent 调用这种工具时门禁不是直接拒绝或放行而是先把请求挂起通知相关审批人等审批通过后才真正执行。下面是门禁校验的简化示例from typing import Callable from pydantic import BaseModel, ValidationError class PolicyResult(BaseModel): allowed: bool reason: str class MCPGate: def __init__(self): self._policies: dict[str, list[Callable]] {} self._approvers: dict[str, list[str]] {} self._audit_log: list[dict] [] def allow(self, tool_name: str, policy: Callable): self._policies.setdefault(tool_name, []).append(policy) def require_approval(self, tool_name: str, approvers: list[str]): self._approvers[tool_name] approvers async def check(self, tool_name: str, arguments: dict, context: dict) - PolicyResult: # 1. 工具必须已注册 if tool_name not in self._policies: return PolicyResult(allowedFalse, reasonftool {tool_name} is not allowed) # 2. 执行参数校验策略 for policy in self._policies[tool_name]: result await policy(arguments, context) if not result.allowed: return result # 3. 需要审批的先挂起 if tool_name in self._approvers: return PolicyResult( allowedFalse, reasonftool {tool_name} requires manual approval: {self._approvers[tool_name]}, ) self._audit_log.append({tool: tool_name, args: arguments, context: context}) return PolicyResult(allowedTrue)3.3 审计日志与超时熔断门禁除了拦截还要有留痕能力。每次 Agent 调用工具不管被拦截还是放行都要记录日志调用了哪个工具、参数是什么、是谁发起的、结果如何。一旦后续出问题翻审计日志就能还原整个过程。审计日志还有个重要作用它能帮你发现异常的调用模式。比如某一天开始Agent 频繁调用下载文件工具下载地址全是外部域名这就很可能是遭了 prompt injection通过日志能第一时间发现。超时熔断也很关键。MCP 工具都是外部服务外部服务可能卡住。如果 Agent 调一个工具一直等不到响应它就会一直挂在那里甚至触发重试风暴。所以我的门禁还有一个职责给所有工具调用设置超时上限和并发上限。某个工具调用超时次数超过阈值就自动熔断一段时间让服务喘口气。3.4 门禁的两种部署形态有人会问MCP 门禁到底放在哪里是放在 Agent 服务端还是放在 MCP Server 那一侧我的回答是两层都要放但职责不同。Agent 服务端门禁拦截模型发出的工具调用请求。这是第一道防线处理 prompt injection 和权限控制。MCP Server 端门禁拦截所有来源的 MCP 调用请求。这是第二道防线处理的是认证、限流、审计。因为 MCP Server 不止被一个 Agent 调用可能被多个客户端调用所以 Server 端必须有独立的防护。实际落地时如果 MCP Server 是第三方提供的没法改它的代码那你只能在 Agent 服务端加门禁如果 MCP Server 是你自己写的那两边都加双保险。4. 实操搭一个重复也不怕、乱调也不行的 Agent4.1 环境准备先交代一下环境。以下步骤基于我本机的 Python 3.11 环境安装 Pydantic AI 和相关依赖pip install pydantic-ai2.36 pydantic-mcp redis asyncpg我这里用了 pydantic-mcp它提供了 MCP 协议的 Python 客户端和服务端支持。另外mock 一下模型接口方便没有真实 API Key 的朋友也能把整个流程跑通。需要说明的是下面的代码做了适当精简重点展示幂等和门禁的骨架而不是把生产级项目代码全贴出来。你在自己的项目里需要根据自己的业务结构调整数据结构。4.2 定义一个 MCP Server 和两个工具先写一个最简单的 MCP Server暴露两个工具一个是低危的查询快递一个是高危的发送优惠券。# mcp_server.py from pydantic_mcp import MCPServer, MCPTool class CourierService(MCPServer): MCPTool(description查询快递物流信息) async def query_courier(self, order_id: str) - dict: # 模拟查询实际项目里这里会调业务接口 return {order_id: order_id, status: in_transit, location: 转运中心} MCPTool(description给用户发送优惠券) async def send_coupon(self, user_id: str, amount: float) - dict: # 注意这里是非幂等操作重复调用会导致重复发券 print(f[COUPON] send to {user_id}, amount{amount}) return {coupon_id: C12345, user_id: user_id, amount: amount} server CourierService() server.run()注意这个send_coupon工具它天然不是幂等的。重放一次就多发一张券。这正是我们要在最外层用幂等键保护的地方。4.3 接入幂等中间件然后写一个幂等中间件包在 Agent 执行路径的外面确保同一个业务请求即使被触发多次Agent 也只真正跑一次。import json import uuid from dataclasses import dataclass from redis.asyncio import Redis class DuplicateRequestError(Exception): pass dataclass class IdempotentExecutor: redis: Redis async def run_with_idempotency( self, idempotency_key: str, func, ): # 1. 先查缓存避免频繁打库 cached await self.redis.get(fidem:{idempotency_key}) if cached: return json.loads(cached) # 2. 尝试抢占SET NX 模拟唯一索引 acquired await self.redis.set( flock:{idempotency_key}, running, nxTrue, ex3600, ) if not acquired: # 说明另一个并发请求正在处理同一个 key # 这里可以等待轮询缓存也可以直接抛错引发重试 raise DuplicateRequestError(idempotency_key) try: result await func() # 3. 成功后把结果写入幂等缓存 await self.redis.set( fidem:{idempotency_key}, json.dumps(result), ex86400, ) return result finally: # 释放抢占锁 await self.redis.delete(flock:{idempotency_key})为了简单这里我用 Redis 的 SET NX 模拟了幂等键的抢占。生产环境建议还是用数据库唯一索引因为 Redis 的锁在某些极端情况下可能因为过期而失效最终防线还得靠数据库。4.4 给 Agent 挂上 MCP 门禁接下来是重头戏创建一个 Pydantic AI Agent挂载刚才的 MCP Server同时接入门禁。import asyncio from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import MCPServerClient from mcp_server import CourierService from mcp_gate import MCPGate, PolicyResult class AgentWithGate: def __init__(self): self.gate MCPGate() # 配置门禁策略 self.gate.allow( query_courier, lambda args, ctx: PolicyResult( allowedTrue, reasonquery is safe, ), ) # 高危操作允许调用但必须校验金额上限 self.gate.allow( send_coupon, lambda args, ctx: PolicyResult( allowedargs.get(amount, 0) 20, reasonamount exceeds 20, ), ) # 创建 Agent挂载 MCP Server self.agent Agent( openai:gpt-4o-mini, system_prompt你是客服助手使用 MCP 工具帮助用户查询快递和发放优惠券。, ) # 这里假设 pydantic-ai 支持通过 CustomClient 挂载本地 MCP Server self.agent.mcp_servers [MCPServerClient.from_stdio(python, [mcp_server.py])] # 给 Agent 的工具调用注入门禁检查 self.agent.tool_call_hook self._gate_hook async def _gate_hook(self, ctx: RunContext, tool_name: str, args: dict): result await self.gate.check(tool_name, args, ctxctx) if not result.allowed: raise PermissionError(fgate denied: {tool_name}, reason: {result.reason}) return args async def run(self, user_message: str, idempotency_key: str): # 这里把整个 Agent 执行包进幂等执行器 return await idempotent_executor.run_with_idempotency( idempotency_key, lambda: self.agent.run_sync(user_message), )这段代码简化了 Pydantic AI 真实 API 的细节但核心思想是一致的Agent 每次跑起来之前先过一个 idempotency key 的检查Agent 每次要调用工具前先过一个 gate 的检查。两道闸门一个管重不重复一个管能不能调。4.5 反向验证模拟重放和未授权调用代码写完最爽的时刻就是反着来验证。第一组测试模拟同一个请求重放两次。import asyncio import uuid from redis.asyncio import Redis async def test_idempotency(): redis Redis.from_url(redis://localhost:6379/0) agent AgentWithGate() uid str(uuid.uuid4()) # 用户请求了两次 result1 await agent.run(帮我查订单 O10001 的物流, idempotency_keyftrack:{uid}) result2 await agent.run(帮我查订单 O10001 的物流, idempotency_keyftrack:{uid}) print(两次结果是否一致:, result1 result2) # 预期输出True因为幂等器直接把第一次的结果缓存返回了第二组测试让 Agent 调用未授权的工具。async def test_gate_deny(): agent AgentWithGate() try: await agent.run(帮我调用 delete_coupon 把优惠券删掉, idempotency_keytest-gate-1) except PermissionError as e: print(门禁拦截成功:, e)第一次跑的时候我挺兴奋的因为真的拦下来了。但注意第二组有个坑如果 Agent 的模型比较聪明它被拒绝一次之后可能会换个方式绕过。比如门禁不允许 delete_coupon它可能会说我先调用 list_coupons 查出优惠券再用 update_coupon 修改状态为已删除。这就是为什么门禁不仅要做工具级白名单还要做参数级校验和上下文关联分析。5. 常见问题与排查技巧实录5.1 幂等键到底存多久这是所有做幂等的人一定会问的问题。存太短重试晚到的请求就穿过了门禁存太长数据库里全是历史数据越积越多。我的建议是按业务流程的最大重试窗口来定。比如你的消息队列最长重投时间是 24 小时那幂等键至少保留 48 小时留足余量。如果涉及支付、退款这类资金操作建议至少保留 30 天因为你无法预测用户的退款纠纷会在什么时间点冒出来。我见过一个项目把幂等表的数据保留 90 天配合定期归档任务既保证了安全又不至于数据爆炸。5.2 被门禁拒绝后 Agent 为什么疯狂重试这个问题我踩过不少次。一开始门禁实现很简单工具被拒绝就抛异常结果 Agent 框架以为工具调用失败自动进入重试逻辑连着重试五六次把审计日志刷得满满当当而且每次都调同一个被拒绝的高危工具。排查思路是门禁拒绝时要区分这是可重试错误还是不可重试错误。参数不合规、工具不在白名单、权限不足这些都是确定性错误重试一万次结果都一样应该直接终止该路径而不是重试。只有网络超时、服务端 5xx 这类错误才值得重试。在代码里的落地方式自定义一个GateDeniedError不在工具调用层触发框架的自动重试机制。你可以把这种错误转换成对模型的工具结果反馈工具已被策略禁用请换一个方式或直接回答用户。这样一来模型会更快意识到这条路走不通而不是傻乎乎地在原地打转。5.3 MCP 工具返回结构不稳定怎么办接第三方 MCP Server 的时候我最头疼的不是认证而是工具返回的数据结构经常变。今天返回{status: success}明天变成{code: 0, message: ok}。模型解析不了就会反复尝试同一个工具或者编造错误的结果。我现在的做法是在门禁之后、模型看到结果之前加一层结果规范器。把 MCP 工具返回的各种结构统一转换成模型友好、字段稳定的标准结构。这一步虽然听着繁琐但对降低 Agent 的胡言乱语概率极其有效。本质上门禁管能不能调规范器管怎么让模型看懂结果两者配合才是一个完整的 Agent 工具治理链路。5.4 问题速查表现象可能原因排查方法解决方案同一条处理逻辑执行了两次持久执行重放 业务非幂等查看 Agent 运行日志确认是否有重放日志在业务入口加幂等键数据库唯一索引兜底幂等键存储后仍重复执行并发请求同时抢到锁检查 Redis SET NX 或数据库唯一索引是否生效用数据库 unique index 替换或配合 Redis 锁工具被门禁拒绝后疯狂重试门禁拒绝被当作可重试错误查看审计日志中同一工具的调用频率自定义不可重试异常禁用该工具路径Agent 调用了未注册的工具模型自由发挥了查看工具调用日志加固白名单检查系统提示词是否被注入MCP 工具响应超时外部服务故障查看外部服务监控和超时配置设置超时阈值增加熔断机制恢复执行后重复发通知副作用发生在崩溃前检查持久化快照的粒度将副作用操作独立成幂等子任务最后再分享一个小技巧。我在实际项目里会给每个 Agent 的 run 分配一个 request_id这个 request_id 会和 idempotency_key 一起打进所有日志里。这样排查问题的时候只要拿到用户反馈的时间点就能用 request_id 把所有日志串起来从入口到门禁到工具结果一条链路看得清清楚楚。这个习惯帮我省了很多排查的时间。持久执行和 MCP 门禁听起来是两个独立的技术名词但当它们真正叠在同一个 Agent 工程里时解决的就是同一个问题让 AI 系统既扛得住故障又守得住边界。希望这篇实践记录对你有用。
持久执行≠不重复？用幂等键与MCP门禁打造可靠Agent
持久执行就不会重复了吗这个问题我第一次在项目群里看到的时候脑子里是有点懵的。因为当时我们刚把业务从普通的同步调用改成 durable execution 模式心里默认了既然能恢复执行那总该稳了吧。结果上线第一周我就被支付回调的重复通知打脸了。同一笔订单用户只付了一次钱我们的 Agent 却把发货流程跑了两遍还差点给同一个客户发了两张优惠券。后来我才彻底想明白一件事所有号称持久的执行框架保证的都只是任务不会丢挂了能重放它默认给你的是 at-least-once 语义而不是 exactly-once。想做到业务层面的只发生一次光靠持久执行远远不够必须在应用层做幂等设计。而这一步恰恰是很多做 AI Agent 工程化的人最容易忽略的。这篇文章就围绕我最近在 Pydantic AI v2.36 环境里落地的一套方案展开把 Agent 的持久化执行、幂等键Idempotency Key和 MCP 门禁组合在一起解决重复执行和工具乱调两个老大难问题。内容会比较偏实战适合正在做 Agent 工程化、接 MCP 工具、或者被重试和并发搞得焦头烂额的朋友。1. 先搞清楚一件事持久执行到底保证什么1.1 持久执行是什么不是不重复是不丢失首先得把概念掰清楚。持久执行Durable Execution这套东西近几年随着 Temporal、AWS Step Functions 还有各种工作流引擎火起来本质上是把任务的执行状态、事件日志、中间结果全部落盘再配合重放机制让一个跑了一半的任务在进程崩溃后能从中断点继续跑而不是从头来。这就像你写文档的时候开了自动保存。Word 崩溃了重启之后它会问你要不要恢复未保存的版本你点恢复文档回来了。但注意恢复的可能是你十分钟前的版本里面对你刚才敲的几段话可能还是旧内容。持久执行也是这个道理——它保存的是执行轨迹不是执行结果。它能保证任务不会因为崩溃就永远消失但它不承诺每个副作用只执行一次。在 Pydantic AI 这种 Agent 框架里持久执行通常体现在一个 Agent 的 run 过程可以被快照、被暂停、被恢复多步工具调用的中间状态可以保存。比如一个 Agent 正在执行查库存 - 下单 - 发通知三步如果跑完第二步、还没跑第三步的时候进程崩了恢复之后它会从第三步重新开始而不是把整个流程重来一遍。这已经比纯内存执行强太多了但它依然不解决第二步重复执行的问题。1.2 三种投递语义at-most-once / at-least-once / exactly-once这是分布式系统里最经典的一组概念放到 Agent 里一样适用At-most-once最多一次发一次就完事丢了不补。性能最好但消息丢失不负责。适合日志、埋点这种丢了也无所谓的场景。At-least-once至少一次只要没确认成功就不断重试、重放保证消息不丢但可能重复。绝大多数消息队列、任务队列、持久执行框架默认都是这个语义。Exactly-once精确一次不丢不重理想状态但分布式环境下极难真正实现通常只能通过实现幂等 至少一次投递来达到效果。注意最后这句几乎没人能在分布式系统里真正做 exactly-once 的投递大家做的都是至少一次投递 业务幂等。因为没法保证网络不重试、进程不在提交前崩溃所以只能在接收方做去重。这也是为什么我后来养成了一个习惯所有被 Agent 调用的业务接口第一件事就是查幂等键而不是直接开干。1.3 重复的高危场景崩溃、超时、重试、消息队列到底哪些场景最容易触发重复执行我列几个真实踩过的进程崩溃在副作用完成之后。Agent 调用工具 A 成功工具 A 返回结果但结果还没来得及写进持久化存储进程崩了。恢复之后框架从快照重放它不知道工具 A 已经执行过了于是又调了一次。HTTP 超时导致的自动重试。Pydantic AI 调用 MCP 工具内网抖动导致响应超时框架自动重试。但服务端其实已经处理完了重试意味着同样的操作被处理两遍。消息队列的重投。Agent 的请求是从消息队列拉出来的消费成功但没来得及 ack或者 ack 丢失消息被重新投递整个 Agent 流程又跑了一遍。用户手动重试。前端看不到结果用户狂点重试按钮同一个请求进来好几次。并行触发。两个上游服务同时回调同一个 webhook都触发了同一个 Agent 流程。这些场景的共同特点是框架层面看起来只执行了一次但实际系统里执行了多次因为存在执行成功和被确认之间的时间缝隙。1.4 为什么持久执行解决不了重复一句话总结持久执行解决的是中断后恢复的问题它给你的保障是任务不会丢断了会接着跑但它无法知道你执行过程中的每个外部副作用是否已经生效。比如说你的 Agent 在步骤二调用了一个发送短信的 HTTP 接口。这个接口执行成功但网络返回超时。持久执行框架捕获到的是调用超时这个结果它不知道短信其实已经发出去了。到了恢复重放的时候它自然会再次调用发送短信。这时候你指望框架帮你判断框架做不到因为它没有业务语义——它并不知道发送短信这个操作对于业务来说是幂等的还是非幂等的。所以结论非常明确持久执行是基础设施层面的保障幂等是业务层面的保障两者缺一不可。只有把持久执行和幂等设计叠在一起你才能在实践中逼近 exactly-once 的效果。2. Pydantic AI v2.36 里的组合拳持久Run 幂等键2.1 v2.36 带来的变化我是在把项目从 v2.28 升到 v2.36 的时候重新审视这个问题的。Pydantic AI 的版本更新一直比较快v2.36 这个版本有几个点对我很有吸引力Agent 的运行状态管理更细了对 MCP 协议的支持也更成熟而且可以通过 hooks 拦截工具调用流程。这些能力正好是我做幂等和门禁的抓手。这里要说明一下我下面聊的方案不完全依赖框架自带的某个开关而是基于 v2.36 提供的能力做了一层自研封装。因为幂等这种东西框架不可能帮你全做完——它不知道你的业务哪个操作是天然幂等的哪个操作需要有业务去重键。所以框架能做的是提供更可靠的持久化、更灵活的拦截点、以及更方便的上下文管理而幂等策略本身得你自己设计。在 v2.36 里我主要用了这几个能力Agent 的 run_id 和上下文透传每个 run 都有一个唯一 ID可以把它作为幂等键的基础。持久化状态存储Agent 的运行状态可以保存到外部存储Redis、PostgreSQL 等崩溃后能恢复。工具调用拦截 hooks可以在工具执行前、执行后插入自定义逻辑这就是门禁和幂等检查的关键位置。MCP Server 接入可以挂载 MCP Server 作为工具源用标准协议调用外部工具。2.2 幂等键设计用 run_id 还是业务键幂等键听起来高大上其实就是个字符串。客户端在发起请求时生成一个全局唯一的键通常用 UUID服务端收到带幂等键的请求后先查一下这个键之前有没有处理过没处理过正常执行把结果存下来键和结果绑定。处理过直接把之前的结果返回不再执行业务逻辑。这个先查再执行的过程要保证原子性否则两个并发请求同时查到没处理过又同时执行还是会重复。所以我在项目里的做法是先往数据库插一条幂等键记录利用唯一索引抢占插入成功才继续执行插入失败说明之前已经处理过直接返回历史结果。在 Pydantic AI 的环境里我通常会把 Agent 的一次完整执行一个 run作为一个幂等单元。幂等键的生成规则有两种选择使用 Agent 自带的 run_id适合一次用户请求对应一次 Agent 执行的场景比如用户问一个问题Agent 跑一轮回复。同一次请求的重试run_id 保持一致。使用业务自定义键适合一个业务事件可能触发多次 Agent 执行的场景比如支付回调用 order_id payment_event_id 拼一个唯一键保证同一笔支付即使回调十次也只执行一次 Agent 流程。我比较推荐第二种。因为 run_id 只代表这一次执行而业务幂等关心的是这个业务事件是否已处理。同一个业务事件可能因为队列重投产生多个 run_id这时候用 run_id 做幂等键就失效了。2.3 唯一约束与结果缓存双保险幂等设计光靠查缓存是不够的必须双管齐下第一层是唯一约束。在存储层面对幂等键加 unique index这是最终防线。两个请求同时进来只有一个能插入成功。我用 PostgreSQL 的表结构大致是这样的CREATE TABLE agent_idempotency ( id BIGSERIAL PRIMARY KEY, idempotency_key VARCHAR(128) NOT NULL UNIQUE, request_payload JSONB NOT NULL, response_payload JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), completed_at TIMESTAMPTZ );第二层是结果缓存。把幂等键对应执行结果缓存到 Redis设置过期时间。这样即使同一个键在短时间内被查询很多次也不需要每次都打数据库可以大大减轻压力。import hashlib import json import uuid from redis.asyncio import Redis class IdempotencyManager: def __init__(self, redis: Redis, ttl_seconds: int 86400): self.redis redis self.ttl_seconds ttl_seconds def generate_key(self, *parts) - str: raw |.join(str(p) for p in parts) return hashlib.sha256(raw.encode(utf-8)).hexdigest() async def get_cached(self, key: str) - dict | None: data await self.redis.get(fidem:{key}) return json.loads(data) if data else None async def cache_result(self, key: str, result: dict) - None: await self.redis.set( fidem:{key}, json.dumps(result), exself.ttl_seconds, )这段代码看着简单但它解决的是重复请求导致重复执行的核心问题。每次 Agent 要跑一个带副作用的业务流程之前先查一次这个管理器命中就直接返回没命中就继续。注意这里缓存只是用来加速读取真正的唯一性保证必须依赖数据库的唯一索引Redis 的过期和并发竞争都不足以承担最终防线。2.4 什么时候必须自己写幂等很多人会问Pydantic AI 自带 Agent 持久化是不是就不需要我管幂等了我的答案是分情况如果你的 Agent 只是做读操作——查个天气、搜个资料、读一下文档——那不做幂等问题不大因为读操作天然幂等执行一百次结果都一样。但如果你的 Agent 会触发写操作——下单、转账、发消息、改数据库——那就必须做幂等。哪怕这个操作后面接的是一个号称幂等的接口比如数据库 UPDATE 语句也未必安全因为 Agent 发出的指令可能不是纯 UPDATE而是读余额 - 算优惠 - 改余额这样多步操作中间任意一步重复执行都可能导致资金错误。我的经验法则很简单只要 Agent 的某个步骤有外部副作用就必须把它包进幂等单元。副作用越大越要谨慎。比如发邮件这种操作重复发一封可能只是让用户觉得烦但如果是调支付接口、扣库存、写账单重复一次可能就是事故。3. MCP 门禁把工具调用关进笼子里3.1 MCP 的价值与失控风险MCPModel Context Protocol最近几个月在圈内确实火各种某某工具已支持 MCP的新闻层出不穷。它的本质是给 AI 提供一套标准化的工具调用协议模型通过 MCP 客户端暴露出来的工具列表去调用外部能力相当于给 Agent 装上了万能遥控器。权力越大风险越大。没打补丁之前我们项目里的 Agent 可以自由调用挂载的任何 MCP 工具。低危操作还好但像发邮件、删除文件、修改数据库这类高危操作一旦被模型的幻觉或者 prompt 注入带偏后果不堪设想。做 AI Agent 的人应该都听说过那个经典例子让 Agent 帮我整理一下文档结果它把整个目录删了。这就是工具调用没有门禁的后果。MCP 门禁本质上就是在模型要调用工具和工具真正执行之间加一道闸。开闸还是关闸由你的业务策略说了算而不是由模型的意图说了算。3.2 门禁三层白名单、参数校验、审批流我在项目里把门禁拆成了三层每层都可以独立配置策略第一层工具白名单。不是 MCP Server 暴露的所有工具都能被 Agent 调用。每个 Agent 只允许调用白名单里的工具其他一律拒绝。这一层最简单但过滤掉了大多数误调用。比如一个只做客服问答的 Agent白名单里只能有查订单查物流绝不能有修改库存删除商品。第二层参数校验。工具能调但参数得合规。比如发邮件工具收件人必须是当前用户邮件内容长度必须限制附件路径必须在允许的目录里。这一层主要防的是 prompt injection——模型被恶意上下文影响把不该传的参数传进去。参数校验规则用 JSON Schema 写最合适因为 MCP 工具本身就支持 JSON Schema 描述参数。第三层审批流。最高危的操作比如付款、批量删除、对外发送公告必须走人工审批。Agent 调用这种工具时门禁不是直接拒绝或放行而是先把请求挂起通知相关审批人等审批通过后才真正执行。下面是门禁校验的简化示例from typing import Callable from pydantic import BaseModel, ValidationError class PolicyResult(BaseModel): allowed: bool reason: str class MCPGate: def __init__(self): self._policies: dict[str, list[Callable]] {} self._approvers: dict[str, list[str]] {} self._audit_log: list[dict] [] def allow(self, tool_name: str, policy: Callable): self._policies.setdefault(tool_name, []).append(policy) def require_approval(self, tool_name: str, approvers: list[str]): self._approvers[tool_name] approvers async def check(self, tool_name: str, arguments: dict, context: dict) - PolicyResult: # 1. 工具必须已注册 if tool_name not in self._policies: return PolicyResult(allowedFalse, reasonftool {tool_name} is not allowed) # 2. 执行参数校验策略 for policy in self._policies[tool_name]: result await policy(arguments, context) if not result.allowed: return result # 3. 需要审批的先挂起 if tool_name in self._approvers: return PolicyResult( allowedFalse, reasonftool {tool_name} requires manual approval: {self._approvers[tool_name]}, ) self._audit_log.append({tool: tool_name, args: arguments, context: context}) return PolicyResult(allowedTrue)3.3 审计日志与超时熔断门禁除了拦截还要有留痕能力。每次 Agent 调用工具不管被拦截还是放行都要记录日志调用了哪个工具、参数是什么、是谁发起的、结果如何。一旦后续出问题翻审计日志就能还原整个过程。审计日志还有个重要作用它能帮你发现异常的调用模式。比如某一天开始Agent 频繁调用下载文件工具下载地址全是外部域名这就很可能是遭了 prompt injection通过日志能第一时间发现。超时熔断也很关键。MCP 工具都是外部服务外部服务可能卡住。如果 Agent 调一个工具一直等不到响应它就会一直挂在那里甚至触发重试风暴。所以我的门禁还有一个职责给所有工具调用设置超时上限和并发上限。某个工具调用超时次数超过阈值就自动熔断一段时间让服务喘口气。3.4 门禁的两种部署形态有人会问MCP 门禁到底放在哪里是放在 Agent 服务端还是放在 MCP Server 那一侧我的回答是两层都要放但职责不同。Agent 服务端门禁拦截模型发出的工具调用请求。这是第一道防线处理 prompt injection 和权限控制。MCP Server 端门禁拦截所有来源的 MCP 调用请求。这是第二道防线处理的是认证、限流、审计。因为 MCP Server 不止被一个 Agent 调用可能被多个客户端调用所以 Server 端必须有独立的防护。实际落地时如果 MCP Server 是第三方提供的没法改它的代码那你只能在 Agent 服务端加门禁如果 MCP Server 是你自己写的那两边都加双保险。4. 实操搭一个重复也不怕、乱调也不行的 Agent4.1 环境准备先交代一下环境。以下步骤基于我本机的 Python 3.11 环境安装 Pydantic AI 和相关依赖pip install pydantic-ai2.36 pydantic-mcp redis asyncpg我这里用了 pydantic-mcp它提供了 MCP 协议的 Python 客户端和服务端支持。另外mock 一下模型接口方便没有真实 API Key 的朋友也能把整个流程跑通。需要说明的是下面的代码做了适当精简重点展示幂等和门禁的骨架而不是把生产级项目代码全贴出来。你在自己的项目里需要根据自己的业务结构调整数据结构。4.2 定义一个 MCP Server 和两个工具先写一个最简单的 MCP Server暴露两个工具一个是低危的查询快递一个是高危的发送优惠券。# mcp_server.py from pydantic_mcp import MCPServer, MCPTool class CourierService(MCPServer): MCPTool(description查询快递物流信息) async def query_courier(self, order_id: str) - dict: # 模拟查询实际项目里这里会调业务接口 return {order_id: order_id, status: in_transit, location: 转运中心} MCPTool(description给用户发送优惠券) async def send_coupon(self, user_id: str, amount: float) - dict: # 注意这里是非幂等操作重复调用会导致重复发券 print(f[COUPON] send to {user_id}, amount{amount}) return {coupon_id: C12345, user_id: user_id, amount: amount} server CourierService() server.run()注意这个send_coupon工具它天然不是幂等的。重放一次就多发一张券。这正是我们要在最外层用幂等键保护的地方。4.3 接入幂等中间件然后写一个幂等中间件包在 Agent 执行路径的外面确保同一个业务请求即使被触发多次Agent 也只真正跑一次。import json import uuid from dataclasses import dataclass from redis.asyncio import Redis class DuplicateRequestError(Exception): pass dataclass class IdempotentExecutor: redis: Redis async def run_with_idempotency( self, idempotency_key: str, func, ): # 1. 先查缓存避免频繁打库 cached await self.redis.get(fidem:{idempotency_key}) if cached: return json.loads(cached) # 2. 尝试抢占SET NX 模拟唯一索引 acquired await self.redis.set( flock:{idempotency_key}, running, nxTrue, ex3600, ) if not acquired: # 说明另一个并发请求正在处理同一个 key # 这里可以等待轮询缓存也可以直接抛错引发重试 raise DuplicateRequestError(idempotency_key) try: result await func() # 3. 成功后把结果写入幂等缓存 await self.redis.set( fidem:{idempotency_key}, json.dumps(result), ex86400, ) return result finally: # 释放抢占锁 await self.redis.delete(flock:{idempotency_key})为了简单这里我用 Redis 的 SET NX 模拟了幂等键的抢占。生产环境建议还是用数据库唯一索引因为 Redis 的锁在某些极端情况下可能因为过期而失效最终防线还得靠数据库。4.4 给 Agent 挂上 MCP 门禁接下来是重头戏创建一个 Pydantic AI Agent挂载刚才的 MCP Server同时接入门禁。import asyncio from pydantic_ai import Agent, RunContext from pydantic_ai.mcp import MCPServerClient from mcp_server import CourierService from mcp_gate import MCPGate, PolicyResult class AgentWithGate: def __init__(self): self.gate MCPGate() # 配置门禁策略 self.gate.allow( query_courier, lambda args, ctx: PolicyResult( allowedTrue, reasonquery is safe, ), ) # 高危操作允许调用但必须校验金额上限 self.gate.allow( send_coupon, lambda args, ctx: PolicyResult( allowedargs.get(amount, 0) 20, reasonamount exceeds 20, ), ) # 创建 Agent挂载 MCP Server self.agent Agent( openai:gpt-4o-mini, system_prompt你是客服助手使用 MCP 工具帮助用户查询快递和发放优惠券。, ) # 这里假设 pydantic-ai 支持通过 CustomClient 挂载本地 MCP Server self.agent.mcp_servers [MCPServerClient.from_stdio(python, [mcp_server.py])] # 给 Agent 的工具调用注入门禁检查 self.agent.tool_call_hook self._gate_hook async def _gate_hook(self, ctx: RunContext, tool_name: str, args: dict): result await self.gate.check(tool_name, args, ctxctx) if not result.allowed: raise PermissionError(fgate denied: {tool_name}, reason: {result.reason}) return args async def run(self, user_message: str, idempotency_key: str): # 这里把整个 Agent 执行包进幂等执行器 return await idempotent_executor.run_with_idempotency( idempotency_key, lambda: self.agent.run_sync(user_message), )这段代码简化了 Pydantic AI 真实 API 的细节但核心思想是一致的Agent 每次跑起来之前先过一个 idempotency key 的检查Agent 每次要调用工具前先过一个 gate 的检查。两道闸门一个管重不重复一个管能不能调。4.5 反向验证模拟重放和未授权调用代码写完最爽的时刻就是反着来验证。第一组测试模拟同一个请求重放两次。import asyncio import uuid from redis.asyncio import Redis async def test_idempotency(): redis Redis.from_url(redis://localhost:6379/0) agent AgentWithGate() uid str(uuid.uuid4()) # 用户请求了两次 result1 await agent.run(帮我查订单 O10001 的物流, idempotency_keyftrack:{uid}) result2 await agent.run(帮我查订单 O10001 的物流, idempotency_keyftrack:{uid}) print(两次结果是否一致:, result1 result2) # 预期输出True因为幂等器直接把第一次的结果缓存返回了第二组测试让 Agent 调用未授权的工具。async def test_gate_deny(): agent AgentWithGate() try: await agent.run(帮我调用 delete_coupon 把优惠券删掉, idempotency_keytest-gate-1) except PermissionError as e: print(门禁拦截成功:, e)第一次跑的时候我挺兴奋的因为真的拦下来了。但注意第二组有个坑如果 Agent 的模型比较聪明它被拒绝一次之后可能会换个方式绕过。比如门禁不允许 delete_coupon它可能会说我先调用 list_coupons 查出优惠券再用 update_coupon 修改状态为已删除。这就是为什么门禁不仅要做工具级白名单还要做参数级校验和上下文关联分析。5. 常见问题与排查技巧实录5.1 幂等键到底存多久这是所有做幂等的人一定会问的问题。存太短重试晚到的请求就穿过了门禁存太长数据库里全是历史数据越积越多。我的建议是按业务流程的最大重试窗口来定。比如你的消息队列最长重投时间是 24 小时那幂等键至少保留 48 小时留足余量。如果涉及支付、退款这类资金操作建议至少保留 30 天因为你无法预测用户的退款纠纷会在什么时间点冒出来。我见过一个项目把幂等表的数据保留 90 天配合定期归档任务既保证了安全又不至于数据爆炸。5.2 被门禁拒绝后 Agent 为什么疯狂重试这个问题我踩过不少次。一开始门禁实现很简单工具被拒绝就抛异常结果 Agent 框架以为工具调用失败自动进入重试逻辑连着重试五六次把审计日志刷得满满当当而且每次都调同一个被拒绝的高危工具。排查思路是门禁拒绝时要区分这是可重试错误还是不可重试错误。参数不合规、工具不在白名单、权限不足这些都是确定性错误重试一万次结果都一样应该直接终止该路径而不是重试。只有网络超时、服务端 5xx 这类错误才值得重试。在代码里的落地方式自定义一个GateDeniedError不在工具调用层触发框架的自动重试机制。你可以把这种错误转换成对模型的工具结果反馈工具已被策略禁用请换一个方式或直接回答用户。这样一来模型会更快意识到这条路走不通而不是傻乎乎地在原地打转。5.3 MCP 工具返回结构不稳定怎么办接第三方 MCP Server 的时候我最头疼的不是认证而是工具返回的数据结构经常变。今天返回{status: success}明天变成{code: 0, message: ok}。模型解析不了就会反复尝试同一个工具或者编造错误的结果。我现在的做法是在门禁之后、模型看到结果之前加一层结果规范器。把 MCP 工具返回的各种结构统一转换成模型友好、字段稳定的标准结构。这一步虽然听着繁琐但对降低 Agent 的胡言乱语概率极其有效。本质上门禁管能不能调规范器管怎么让模型看懂结果两者配合才是一个完整的 Agent 工具治理链路。5.4 问题速查表现象可能原因排查方法解决方案同一条处理逻辑执行了两次持久执行重放 业务非幂等查看 Agent 运行日志确认是否有重放日志在业务入口加幂等键数据库唯一索引兜底幂等键存储后仍重复执行并发请求同时抢到锁检查 Redis SET NX 或数据库唯一索引是否生效用数据库 unique index 替换或配合 Redis 锁工具被门禁拒绝后疯狂重试门禁拒绝被当作可重试错误查看审计日志中同一工具的调用频率自定义不可重试异常禁用该工具路径Agent 调用了未注册的工具模型自由发挥了查看工具调用日志加固白名单检查系统提示词是否被注入MCP 工具响应超时外部服务故障查看外部服务监控和超时配置设置超时阈值增加熔断机制恢复执行后重复发通知副作用发生在崩溃前检查持久化快照的粒度将副作用操作独立成幂等子任务最后再分享一个小技巧。我在实际项目里会给每个 Agent 的 run 分配一个 request_id这个 request_id 会和 idempotency_key 一起打进所有日志里。这样排查问题的时候只要拿到用户反馈的时间点就能用 request_id 把所有日志串起来从入口到门禁到工具结果一条链路看得清清楚楚。这个习惯帮我省了很多排查的时间。持久执行和 MCP 门禁听起来是两个独立的技术名词但当它们真正叠在同一个 Agent 工程里时解决的就是同一个问题让 AI 系统既扛得住故障又守得住边界。希望这篇实践记录对你有用。 
