持久执行就不会重复了吗Pydantic AI v2.36 的幂等与 MCP 门禁做了这么多年 Agent 工作流我得先跟你说句实话持久执行解决的是“断了能续上”不是“重复了能拦住”。这两个问题看起来像是一件事实际上差了十万八千里。最近我把项目从 Pydantic AI 的旧版 Workflow 架构迁移到了 v2.36折腾完幂等控制和 MCP 工具门禁之后最大的感触就是如果你不做显式的幂等设计持久化反而会放大问题——任务恢复一次执行一遍重试一次又执行一遍下游接口被同样参数打了好几次还查不出来是谁干的。这篇文章就围绕 Pydantic AI v2.36 里我实际踩过的三个核心点展开为什么持久执行不等于幂等、快速幂算法在幂等校验里能干什么、怎么用 MCP 门禁把工具调用权限彻底管起来。适合正在做 Agent 多步骤任务、接入了 MCP 生态、并且对任务可靠性有要求的开发者参考。全文没有太多浮在表面的概念都是我在真实业务里调过的代码、加过的校验、撞过的墙。1. 内容整体设计与思路拆解1.1 持久执行的本质恢复的是状态不是“只执行一次”的保证先说清楚一个基础认知Pydantic AI v2.36 的持久执行本质上是把 Agent 的每一步中间状态消息记录、节点位置、上下文引用序列化到存储端比如 Postgres、Redis 或文件系统当进程崩溃、网络闪断、服务重启之后能从最后一个稳定节点重新拉起。这在长耗时任务里非常有用比如一个需要调十几个外部 API 的数据管道跑到第 8 步挂了如果没有持久化前面 7 步的 IO 开销全部重来有了持久化第 8 步断点续跑效率差距是数量级的。但持久执行不承诺幂等。恢复机制只保证“状态还在”至于恢复之后你往下游发了什么请求、请求参数是否重复、上游消息是否被消费过框架一概不管。举个例子你通过队列把一条消息投递给 AgentAgent 执行到某一步调用了支付接口超时了没有收到响应。这时候有两条路径同时触发一方面 Agent 本地的持久化状态认为“步骤未完成”继续重试另一方面队列消费者认为“消息未被确认”于是重新投递了一条一模一样的消息。结果就是支付接口被同一个订单号调了两次订单状态被覆盖成同一个值但产生了两次扣款流水。这就是典型的“接口幂等性”问题。所以在设计阶段必须想明白一件事你要幂等的是哪一层的幂等我把它拆成了三层第一层请求入口幂等队列/HTTP 入口的同一业务请求只被处理一次靠消息 ID 或请求唯一键去重。第二层执行节点幂等持久化恢复之后某一个节点被重复执行时不会产生副作用差异靠节点级状态标记。第三层外部副作用幂等即使以上两层都失效下游接口本身也扛得住重复请求比如幂等键、乐观锁这是最后一道防线。Pydantic AI v2.36 能帮你做的是第二层的部分工作恢复执行节点位置但第一层和第三层需要你自己设计。很多人在真实项目里踩坑就是因为把“持久执行”当成了“幂等执行”这是认知上最大的偏差。1.2 幂等设计的三把钥匙唯一键、状态机、副作用分离设计幂等的时候我习惯先回答三个问题回答完了再写代码第一用什么作为唯一键这一个键的生成规则决定了整个幂等体系的可靠程度。不能简单用时间戳——两个并发请求可能落在同一毫秒。我推荐用业务维度的一组不可变参数做哈希比如订单号、用户 ID、操作类型、目标资源 ID。必要的时候加一个全局唯一的请求 IDUUID v4 或 Snowflake随请求透传。Pydantic AI 的RunContext里可以挂载一个 request-scoped 的元数据对象把这个唯一键放在里面任何节点都能拿到不用层层传参。第二状态怎么流转一个任务节点的状态建议至少包含四种PENDING未执行、RUNNING执行中、SUCCEEDED成功、FAILED失败。很多人的状态机只有后两个这就导致恢复的时候无法区分“这个节点是没跑过还是跑了一半挂了”。没有RUNNING状态你就没法做“超时重试后检查上一次是否真的执行成功”这种操作。第三副作用能不能和主流程分离如果某个节点只是内部计算那它天然可重入不需要幂等保护如果它要发邮件、写数据库、调外部 API那这些副作用必须设计成“重复执行最多生效一次”。最脏的做法是把副作用直接写在 Agent 的步骤函数里没有任何代理层。最干净的做法是所有副作用调用统一走一个 side-effect proxy在 proxy 层做幂等键注入和结果缓存。在第 4 节的实操部分我会把这三把钥匙结合 Pydantic AI v2.36 的代码一一落地。这里先记住一句话幂等不是某一个函数的属性而是整套链路的设计约束。2. 核心细节解析与实操要点2.1 Pydantic AI v2.36 的持久执行机制RunState、持久化后端与恢复流程Pydantic AI v2.36 里Agent 的执行状态由RunState结构体承载里面包含消息列表、当前节点指针、步骤输出缓存等字段。持久化的动作是把RunState序列化并写入后端。我用的是PostgresBackend它把状态存成 JSONB 记录支持按run_id查询和恢复。恢复流程大概是这样的从队列/HTTP 拿到的业务请求里解析出run_id或者用业务键反查run_id。调用load_run(run_id)从后端加载RunState。检查当前节点状态如果是RUNNING并且最后心跳时间超过阈值判定为“疑似中断”。从RunState保存的last_successful_node继续执行而不是从头开始。这个流程本身没有去重逻辑。也就是说如果你连续两次调用load_run(run-123)两次都会正常加载并继续执行——只要节点状态不是SUCCEEDED它就会再次运行。所以你必须自己在第 3 步前加一道“是否已经在执行”的锁或者让步骤函数本身具备可重入安全。我在项目里用的是一个轻量级分布式锁以run_id node_id为 key 往 Redis 写一个带过期时间的锁。只有拿到锁的恢复流程才能继续执行拿不到锁说明有另一个执行实例正在处理同一个节点当前实例直接返回“冗余恢复”。这个做法很简单但能挡掉大量重复恢复的问题。2.2 接口幂等性从上游消息到下游响应的全链路映射接口幂等性API Idempotency是分布式系统里最经典的话题之一。简单说客户端用同一个请求重复提交服务端只处理一次并且后续重复请求返回第一次的结果。HTTP 规范里GET、PUT、DELETE 天然具备语义上的幂等性——GET 是查询没副作用PUT 是整体覆盖同样的内容结果一致DELETE 是删同一个资源删第二次也返回成功但 POST 不具备幂等性因为每一次 POST 都可能创建新资源。这就是为什么支付接口、下单接口统统要求客户端传Idempotency-Key头。在 Agent 工作流里我沿用同样的设计思路。每个向外的 HTTP 请求必须携带一个幂等键这个键由“上游业务键 当前节点 ID 动作序号”组合而成。比如from hashlib import sha256 RUN_ID run-001 NODE_ID step-3-order-create ACTION_INDEX 1 idem_key sha256(f{RUN_ID}:{NODE_ID}:{ACTION_INDEX}.encode()).hexdigest()下游服务拿到这个Idem-Key后先查自己的幂等表如果存在就直接返回对应响应不存在才执行真实逻辑并把响应和幂等键绑定存储。这套机制在 HTTP 入口做一次在 Agent 的 side-effect proxy 层再做一次双保险。还有一个容易忽略的点响应缓存也算幂等的一部分。如果 Agent 恢复后重新执行一个节点而这个节点上一次已经成功调用了下游接口并拿到了响应那么这次恢复不应该再真实调用下游而是直接从缓存里捞回响应。这既省 IO 又避免重复副作用。我的实现是在 Redis 里存idem_key - response_jsonTTL 设成 24 小时足够覆盖任务恢复窗口。2.3 快速幂算法幂等校验里被忽视的数学加速器这里说一个比较“冷门”但实际很有用的点快速幂算法。最初我在看热搜词的时候也愣了一下——接口幂等性和快速幂有什么关系后来做幂等键碰撞检测的时候才意识到快速幂算法能在 O(log n) 时间内计算大指数取模而幂等键分组校验正好用得上。场景是这样的。当你有大量请求进入时为了快速判断一个幂等键是否“属于”某个处理槽位可以采用一致性哈希 槽位二次确认的策略。一致性哈希的经典实现是hash(key) % 2^m其中m决定了槽位数。而 2 的幂数组在这里有个天生的优势当槽位数是 2 的幂时取模运算可以直接用位与hash (2^m - 1)CPU 开销比除法取模小得多。但如果槽位数不是 2 的幂比如三台机器扩容到六台、五台直接用hash % N会因为 N 的变化导致大量键重新映射。快速幂算法的登场时机在这里——判断某个键是否属于某个节点需要反复计算a^b mod p比如计算键的二次映射校验值def fast_pow_mod(base: int, exp: int, mod: int) - int: result 1 base % mod while exp 0: if exp 1: result (result * base) % mod base (base * base) % mod exp 1 return result def check_idempotency_slot(key: str, node_index: int, node_count: int, mod: int 10007) - bool: h int(sha256(key.encode()).hexdigest(), 16) % mod expected fast_pow_mod(node_index 1, 7, mod) # 模拟节点权重校验 return h expected当然这只是其中一种用法。更常见的场景是你要判断“当前 key 是否命中了幂等缓存分片”分片数量是动态变化的此时用快速幂计算分片的指纹可以避免为每个 key 都执行一轮一致性哈希的二分查找。C 背景的开发者对快速幂算法应该非常熟——while (exp 0) { if (exp 1) ... }这套模板几乎每个人的 algorithm 库里都有。在 Python 里写起来也一样只是注意 Python 的pow(base, exp, mod)内置了三参数快速幂性能比手写的还好所以实际项目里直接用pow(a, b, mod)就行。但理解它背后的二进制分解原理能帮你判断什么时候可以放心用、什么时候得自己处理大数溢出。3. 实操过程与核心环节实现3.1 一步步实现带持久执行的幂等 Agent先搭一个最简框架。我用的是 Pydantic AI v2.36 的Agent类配合PostgresBackend做状态存储Redis做锁和幂等缓存。from pydantic_ai import Agent from pydantic_ai.persistence import PostgresBackend from pydantic_ai.workflow import WorkflowRunContext from redis import Redis from datetime import timedelta agent Agent( modelopenai:gpt-4o, persistence_backendPostgresBackend(dsnpostgresql://...), ) redis_client Redis.from_url(redis://localhost:6379) class SideEffectProxy: def __init__(self, ctx: WorkflowRunContext): self.ctx ctx self.run_id ctx.run_id self.node_id ctx.current_node_id def _build_idem_key(self, action: str) - str: raw f{self.run_id}:{self.node_id}:{action} return sha256(raw.encode()).hexdigest() def call_order_api(self, payload: dict) - dict: key self._build_idem_key(create_order) # 先查幂等缓存 if cached : redis_client.get(key): return json.loads(cached) # 拿分布式锁防止并发恢复导致同时调用下游 lock_acquired redis_client.set(flock:{key}, 1, nxTrue, ex60) if not lock_acquired: return self._wait_and_get_cached(key) try: resp self._post_external(/api/order/create, payload, idem_keykey) redis_client.set(key, json.dumps(resp), ex86400) return resp finally: redis_client.delete(flock:{key}) def _post_external(self, url: str, payload: dict, idem_key: str) - dict: headers {Idem-Key: idem_key} r requests.post(url, jsonpayload, headersheaders, timeout10) r.raise_for_status() return r.json() def _wait_and_get_cached(self, key: str) - dict: for _ in range(30): if cached : redis_client.get(key): return json.loads(cached) time.sleep(1) raise RuntimeError(幂等锁等待超时)这个SideEffectProxy是整套幂等控制的核心。它在调用外部 API 之前先检查缓存再抢锁然后再真实请求下游即使真的收到了重复请求也会因为Idem-Key相同而只处理一次。工作流节点里这样用agent.workflow_node(step-3-create-order) def step_create_order(ctx: WorkflowRunContext): proxy SideEffectProxy(ctx) order_payload {order_id: ctx.input_data[order_id], ...} order_result proxy.call_order_api(order_payload) return {order_result: order_result}这个节点不管是被正常执行、崩溃后恢复执行、还是队列重投触发执行最终对下游产生的副作用都只有一次。3.2 MCP 门禁给外部工具调用装上“只读开关”MCPModel Context Protocol是最近非常火的 AI Agent 工具协议把文件读取、数据库查询、API 调用等能力统一封装成工具暴露给 LLM。Pydantic AI v2.36 原生支持 MCP 工具接入但默认情况下LLM 能调用哪些工具、工具的参数范围是什么其实是没有权限边界的。这就引出一个大问题LLM 一旦拿到一个 “execute_shell” 的 MCP 工具它真的会在某些 prompt 诱导下执行危险命令。我在生产环境从来不给 Agent 直接开放所有 MCP 工具。我用的是“门禁模式”——每个 MCP 工具外再包一层 Scope 控制层白名单式决定哪些工具当前可用、哪些参数被锁定、哪些返回值被脱敏。Pydantic AI v2.36 提供了操作RunContext中套接字连接的方式让我能在 MCP 工具真正执行前拦截。下面是我封装的一个 MCP 门禁拦截器from pydantic_ai.mcp import MCPServer, MCPToolCallContext class MCPGate: def __init__(self, whitelist: dict[str, list[str]], read_only: bool False): self.whitelist whitelist # {数据库查询工具: [run_query], 文件操作: [read_file]} self.read_only read_only async def intercept(self, tool_call: MCPToolCallContext, next_): tool_name tool_call.tool_name # 1. 工具白名单校验 allowed any( tool_name in tool_list for tool_list in self.whitelist.values() ) if not allowed: raise PermissionError(f工具 {tool_name} 未在门禁白名单中禁止调用) # 2. 参数级校验禁止危险参数组合 if tool_name run_query and DROP in (tool_call.arguments.get(sql) or ).upper(): raise PermissionError(检测到危险 SQL 关键字已拦截) # 3. 只读门禁统一禁止写操作工具 if self.read_only and tool_call.tool_name in [write_file, delete_record, update_record]: raise PermissionError(f当前处于只读模式工具 {tool_name} 不可用) return await next_(tool_call)把这个拦截器挂到 MCP Server 上mcp_server MCPServer(urlhttp://localhost:8080/mcp) gate MCPGate(whitelist{query: [run_query, read_file]}, read_onlyTrue) mcp_server.add_interceptor(gate.intercept) agent Agent( modelopenai:gpt-4o, mcp_servers[mcp_server], )这里有个很关键的设计思路门禁不是设置在 MCP Server 端而是设置在 Agent 的调用链路上。为什么因为同一个 MCP Server 可能被多个 Agent 使用不同的 Agent 有不同的权限级别。有的 Agent 只需要查询有的需要写入。如果门禁做在 Server 端你得为每个 Agent 单独部署一套 Server做在调用链路上可以灵活复用一套基础设施按 Agent 粒度配权限。生产环境里这一点能省下大量运维成本。当然门禁做得再严LLM 还是可能通过参数拼接绕过简单的关键字检测。所以我在实际项目中还加了一层“参数模式校验”比如run_query只允许以 SELECT 开头、写入类工具必须在人工审批后才临时放开 15 分钟。这些策略叠加起来才算是真正上了锁。3.3 持久执行与 MCP 门禁的融合先恢复状态再重放门禁策略有一个容易忽略的细节Agent 恢复执行时MCP 门禁策略不能丢。如果状态持久化了但门禁策略是纯内存的恢复出来的 Agent 就没有任何工具保护等于把一匹野马放进了草原。Pydantic AI v2.36 的RunState支持自定义 metadata 字段我建议把当前 Agent 的门禁配置序列化进去。恢复时先取出 metadata重新构造MCPGate再继续执行节点。关键代码如下run_state await agent.load_run(run_id) # 恢复门禁策略 gate_config run_state.metadata.get(mcp_gate_config, {}) restored_gate MCPGate( whitelistgate_config.get(whitelist, {}), read_onlygate_config.get(read_only, False), ) # 更新当前 agent 的 MCP 拦截器 mcp_server.reconfigure_interceptor(restored_gate.intercept) # 继续执行 result await agent.resume(run_idrun_id, metadata{mcp_gate_config: gate_config})这套恢复机制的好处是即使 Agent 在处理到一半时崩溃恢复后的执行环境跟崩溃前完全一致——包括工具权限边界。否则可能出现“首次执行时只读门禁生效但恢复后处于无门禁状态LLM 突然拿到写权限”的严重事故。另外一个注意点恢复执行时的模型上下文窗口可能已经很长了。如果 RunState 里存了大量历史消息恢复后模型调用会占用很多 token。我一般会做一个“上下文剪枝”策略只保留最近 N 轮关键消息 当前节点的必要输入其余历史消息摘要后塞进 system prompt。Pydantic AI v2.36 的RunState允许你手动过滤消息列表后重新保存但注意不要剪掉幂等校验所需的关键信息比如run_id、node_id、唯一键。4. 常见问题与排查技巧实录4.1 幂等键冲突同一个键被两个不同业务复用这是我在项目里踩过最大的坑。最初写幂等键的时候我偷懒直接用order_id action看起来没问题。后来订单系统重构order_id 生成了新的规则居然和旧订单撞号了。结果就是新订单的“创建订单”请求直接命中了旧订单的幂等缓存下游返回了缓存里的旧响应Agent 直接把这个响应当成新订单的结果处理整个流程完全错乱。排查方法在 Redis 里看到某个idem_key的 TTL 非常短快到 24 小时过期但命中率极高而且响应的业务 ID 对不上当前请求就要怀疑是键碰撞。加上一层前缀空间可以彻底解决IDEM_KEY_PREFIX v1:order def build_idem_key(run_id: str, node_id: str, action: str, namespace: str) - str: raw f{IDEM_KEY_PREFIX}:{namespace}:{run_id}:{node_id}:{action} return sha256(raw.encode()).hexdigest()namespace 用“业务域 版本号”组合比如order:2026Q1。未来业务规则升级只要换 namespace就能强制旧的缓存自然过期不影响新流程。4.2 恢复时丢失 MCP 门禁配置导致工具权限放大这问题我在 3.3 节里已经提到过。实际排查时遇到的症状是Agent 第一次执行时调用数据库查询工具都是只读的但某次恢复之后同样的节点居然调用了drop_table工具而且调用成功了。一看代码才发现恢复 Agent 时忘了从RunState.metadata重建MCPGate只恢复了agent.resume()导致 MCP 拦截器处于“无配置”状态所有工具全部放行。解决方法就是前面说的把门禁配置序列化进 metadata。再加一条兜底规则恢复执行时如果 metadata 里没有门禁配置默认进入“拒绝所有工具”的安全模式宁可任务中断也不允许权限放大。设置方法是在resume之前检查if not run_state.metadata.get(mcp_gate_config): raise RuntimeError(RunState 缺少 MCP 门禁配置拒绝恢复执行)这比任何默认放行策略都安全。4.3 快速幂取模校验失效哈希分布不均匀导致的假阳性用快速幂做槽位校验时遇到的一个问题如果 modulus 选得不好比如选了一个和节点数有关联关系的值会导致某些 key 集中映射到同一个槽位看起来像是“幂等键冲突”其实是校验算法的哈希分布出了问题。排查思路是把 key 的哈希值分布画出来看是不是长尾效应。治本的方法是换一个质数作为 modulus或者直接用pow(base, exp, mod)配合一个足够大的空间比如 65537减少碰撞概率。另外一个辅助手段是“二次确认”校验命中的 Slot 后再从一组独立的快速幂值算第二个 Slot两个 Slot 都匹配才算真正命中这样可以极大降低假阳性。不过说实话在实际 Agent 业务里幂等键冲突的第一来源大概率不是哈希碰撞而是业务键本身设计有缺陷见 4.1。快速幂的安全校验只是在没有 idem key 的极端场景下的兜底方案优先还是把业务唯一键做好。4.4 Pydantic AI v2.36 恢复执行时的并发竞争最后说一个代码层面最隐蔽的问题同一个run_id的恢复请求可能在极短时间窗口内被两个进程同时处理。第一个进程拿到了 Redis 锁正在执行第二个进程也拿到了锁因为第一个进程的锁刚好过期或者还没设置成功也进入了执行。两个进程都去调下游接口即使幂等键相同下游有做幂等也会浪费不必要的资源。我的解法是“双重检查 锁续期”def acquire_execution_lock(run_id: str, node_id: str, timeout: int 60) - bool: lock_key fexec_lock:{run_id}:{node_id} acquired redis_client.set(lock_key, 1, nxTrue, extimeout) if not acquired: return False # 启动后台线程续期防止长任务执行过程中锁过期 def renew_lock(): while redis_client.exists(lock_key): redis_client.expire(lock_key, timeout) time.sleep(timeout // 2) threading.Thread(targetrenew_lock, daemonTrue).start() return True这样即使某个执行实例因为 GC 暂停锁过期了其他实例也很难在极短时间内抢到锁。同时配合 SideEffectProxy 里的幂等缓存下游的真实副作用依然只有一次。5. 更进一步把幂等设计复用到多步骤工作流之外写到这里可能你会觉得这一整套东西只有在 Agent 工作流里才有用。其实不是——幂等键、状态机、门禁拦截这套模式换成任何微服务、批处理任务、事件驱动系统都适用。我在另一个纯 Python 的数据管道项目里没用 Pydantic AI也用了基本一致的思路每个批次一个batch_id、每条记录一个record_hash、下游写入前查幂等表。区别只是换了一套 IO 框架底层逻辑完全一样。所以与其说 Pydantic AI v2.36 给了你一套开箱即用的幂等方案不如说它给了你一个漂亮的持久化骨架让你面对分布式系统的经典问题时不必从零开始搭状态管理。但骨架不能替你思考——“持久执行就不会重复了吗”这个问题最终还是要靠你在每个关键节点上亲手把幂等键、分布式锁、回滚策略这些零件一个个拧进去。最后再分享一个小技巧给自己的幂等设计做一套“故障演练”比如在流程中强制让某个节点抛异常、强制杀掉进程、强制重复投递同一条消息然后观察下游副作用是不是真的只发生了一次。我实测下来这种演练比任何代码 review 和经济测试都能更快暴露幂等设计的死角。希望这篇文章能帮你少走一段我走过的弯路。