pydantic-ai durable_exec 引擎集成开发指南从后端分层到操作命名与配置解析的完整实现规范【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai导读durable_exec是 pydantic-ai 面向持久化执行durable execution场景的官方集成层它把 Temporal、DBOS、Prefect、Restate 等外部工作流引擎作为 Agent 语义的一等兼容目标而非外围适配器。本文以 pydantic_ai_slim/pydantic_ai/durable_exec/AGENTS.md 为骨架结合该子包的源码实现与兼容性测试完整讲解如何为新的持久化执行引擎编写集成选择后端分层、声明DurabilityEngineSpec、统一操作命名、解析操作配置以及如何在跨边界场景下保证 run context、依赖、消息历史、重试、模型/Profile 选择与工具集生命周期的语义一致性。读完本文你将掌握一套可直接照搬的引擎接入实现规范并能理解为什么操作名称是兼容性数据而必须被固定。一、durable_exec 的设计定位一等公民的兼容目标durable_exec子包的模块文档init.py明确说明每个子包通过一个可挂载到Agent上的 capability能力为一种持久化执行平台提供持久性能力目前官方内置三种pydantic_ai.durable_exec.temporal→TemporalDurabilitypydantic_ai.durable_exec.dbos→DBOSDurabilitypydantic_ai.durable_exec.prefect→PrefectDurability从 AGENTS.md 的指导原则看这些引擎必须被当作核心 Agent 语义的兼容性检查而非外围适配器。这意味着集成层必须完整保留跨持久化边界的语义run context、依赖注入deps、消息历史、重试、模型/Profile 选择、工具集生命周期都要穿越 durable 边界后依然成立避免隐藏的排序假设不允许依赖隐式的执行顺序不允许携带不可序列化的状态不允许使用仅存在于运行时的闭包除非 durable 包装器明确拥有它们优先使用通用扩展点用通用的 capabilities / toolsets / models 扩展点而不是为某个引擎开 escape hatch逃生通道改动联动检查当修改 graph / tool / output / streaming / MCP 行为时必须检查 durable 包装器是否需要同步更新并在外部运行时行为相关的场景补充工作流级测试。1.1 同步流的特殊约定AGENTS.md 特别指出一条容易被误解的约定同步流run_stream_sync通过self.run_stream继承缺少 wrapper override 是有意为之——工作流本身是异步的而SyncStreamBridge会拒绝正在运行的事件循环。因此集成者不要试图为同步流单独补一个 override。二、构建新引擎的整体路径AGENTS.md 规定新集成必须构建在公开的pydantic_ai.durable_exec表面上而不是把框架内部的模型、工具集、事件或 capability-operation 收集机制复制进集成代码。具体步骤是子类化BaseDurabilityCapability作为面向 Agent 的能力提供一个DurableOperationBackend通过get_durable_operation_backend返回只实现引擎特有的原语durable primitive框架负责参数/结果传输、缓存身份、命名、配置解析等通用机制。2.1 基类职责从源码看 BaseDurabilityCapability在 _base.py 中BaseDurabilityCapability的 docstring 说得很清楚它拥有模型注册表与模型跨 durable 边界的往返。核心原因是一个Model实例无法被序列化进 activity/step/task所以请求只携带model_id字符串None表示 Agent 默认模型或者是models注册表键或者是模型名字符串由对端按需重建模型——重建是 deps 感知的走 Agent 完整的resolve_model_id能力链注册表兜底。从源码看子类只需关注三个钩子_base.pyfor_agent中在绑定副本上调用_bind_modelsworkflow/flow 侧调用_find_model_id把Model实例映射为字符串 IDactivity/step/task 内部调用_resolve_model_for_request用字符串重建模型。基类还承担了工具集包装职责_wrap_leaf_toolset会按engine_spec.wrapped_toolset_kinds判断是否需要把叶子工具集FunctionToolset/DynamicToolset/MCPToolset包装成对应的DurableFunctionToolset/DurableDynamicToolset/DurableMCPToolset每个工具集的调用、参数校验、发现get_tools都会走 backend 绑定出的 durable operation。这正对应 AGENTS.md 中base owns parameter and result transport, cache identity, naming, and config resolution的说法。三、选择后端分层Callable 与 Registered 两大体系AGENTS.md 要求根据引擎 SDK 的执行模型选择后端层级两种层级都定义在 _operation_backend.py3.1 CallableOperationBackend调用期回调型引擎适用于SDK 在调用发生时接受一个异步回调的引擎。子类只需实现execute方法把回调作为一个命名的 durable 单元执行基类负责参数与结果传输、缓存身份、命名和配置解析。class MyCallableBackend(CallableOperationBackend[ConfigT]): async def execute( self, *, operation_id: DurableOperationId, name: str, body: Callable[[], Awaitable[object]], cache_key: tuple[object, ...], config: ConfigT, ) - object: # 把 body 作为单个命名 durable 单元提交给引擎 SDK return await engine_sdk.run_durable_unit(name, body)CallableOperationBackend.bind的内部流程_operation_backend.py展示了基类拥有的完整职责用invocation_label若有和 namer 计算持久化调用名用config.base(operation.config_role, ...)解析角色基础配置用operation.cache_identity.project(params)计算缓存键供哈希键引擎使用在execute内部执行result_codec.dump(handler(params))在execute返回后result_codec.load还原结果。3.2 RegisteredOperationBackend预注册型引擎适用于处理程序必须在 worker 启动前完成注册的引擎。子类实现register返回绑定的调用器 所有 SDK 注册句柄。基类在bind时收集这些注册registrations()在任何时候都能拿到完整集合_operation_backend.py。关键点AGENTS.md 原文强调base 在 Agent 组装期间就绑定四个模型操作request、request_stream、compact_messages、cancel_suspended_response所以registrations()在 worker 启动前就是完整的集成代码应把收集到的注册句柄一次性传给引擎 SDK 创建 worker。在 _base.py 的for_agent中可以看到这种区分被显式落实backend bound.get_durable_operation_backend() # Registered engines need the complete registration set before a worker starts. if isinstance(backend, RegisteredOperationBackend) and bound._bound_model_operations is None: bound._bound_model_operations bound._bind_model_operations(backend, model_idNone, model_namedefault)而 Callable 引擎按请求绑定因为其 durable 单元名可能依赖该请求的model_id。3.3 两个内置辅助件JournalCallableOperationBackend带标准 journal 命名约定的 Callable 后端构造时自动使用JournalOperationNamerRoleBasedOperationConfig按角色model/event/capability/tool提供默认配置并可选挂一个 per-tool 解析器。四、DurabilityEngineSpec引擎的声明式配置AGENTS.md 要求集成者只设置一次公开的engine_spec用DurabilityEngineSpec声明引擎名称、durable 单元名词、durable 容器名词、codec、不支持的运行时工具集种类、需要包装的工具集种类、工具集生命周期、工具调用结果升级策略、发现策略、顺序工具策略和工具配置键。该数据类定义在 _spec.py字段及语义如下字段类型默认值说明engine_namestr必填错误消息中展示的引擎名如Temporaldurable_unit_nounstr必填单个 durable 单元名词如activity/step/taskdurable_container_nounstr必填durable 容器名词如workflow/flowdurable_unit_pluralstr | NoneNone自动取unit_noun s复数形式codecDurabilityCodecIDENTITY_CODEC每个 durable 边界的序列化方式serialization_failureCallable | NoneNone把确定性 codec 失败映射为引擎的终态错误wrapped_toolset_kindsfrozenset[ToolsetKind]{function,mcp,dynamic}需要包装的叶子工具集种类DBOS 省略functiontoolset_lifecyclesMapping[ToolsetKind, Lifecycle]function: enter-always, mcp: enter-always, dynamic: enter-never每种工具集的生命周期配置tool_call_result_upgrade_lenientboolFalse是否宽容解码旧版未包装控制流异常前的工具结果journal_discoveryboolTrue发现get_tools/get_instructions是否运行在自己的 durable 单元中sequential_tools_in_durable_contextboolFalsedurable 容器内工具调用是否必须串行unsupported_runtime_toolset_kindsfrozenset空durable 容器内被拒绝的运行时工具集种类tool_config_keystr | NoneNone承载引擎专属 durable 配置的工具元数据键__post_init__会做防御性校验_spec.pydurable_unit_noun/durable_container_noun不能为空、复数不能为空串、被包装的工具集种类必须都有生命周期配置否则抛出UserError。4.1 选择 codecIDENTITY_CODEC 与 JSON_CODEC_codec.py 把引擎分成两个家族对象传递型引擎Temporal、DBOS、Prefect把活的 Python 对象直接交给 durable 原语由原语自己的序列化器持久化因此使用IDENTITY_CODEC——它原样返回值并忽略类型JSON-journal 型引擎Restate、AWS Lambda、Absurd必须先把值压缩成 JSON 兼容载荷再写 journal因此使用JSON_CODEC——它通过缓存的TypeAdapter(tp)做往返dump 用dump_python(modejson)load 用validate_python。注意dump(tp, value)是在 durable 单元内部调用的不可序列化载荷会在 step 内失败生产与测试行为一致load(tp, payload)在单元外部调用。4.2 工具集生命周期toolset_lifecycles的取值来自 _toolset.py 的Lifecycle包括enter-always、enter-outside-durable、enter-never。AGENTS.md 要求按engine_spec.toolset_lifecycles进入和关闭工具集资源含失败与取消路径并验证在 durable 单元内创建的资源不得逃逸出该单元。该字段被强制显式声明是有原因的——注释明确指出曾有两次真实 bug 来自默认门控spec 源码中引用了 #5477 requirement 3。五、操作身份与配置解析5.1 DurableOperationId可增长的公开联合类型_operation.py 定义了DurableOperationId联合类型包含ModelRequestId含streaming标志——模型请求/流式请求ModelCompactMessagesId——消息压缩ModelCancelSuspendedResponseId——挂起响应取消CapabilityOperationId——capability 贡献的操作EventStreamHandlerId——事件流处理器ToolsetGetToolsId/ToolsetGetInstructionsId——工具发现ToolsetValidateToolArgumentsId——工具参数校验ToolsetCallToolId——工具调用。AGENTS.md 提醒该联合类型在 minor 版本中会增长所以引擎配置代码的匹配必须保留默认分支default branch不能只处理已知变体就断言穷尽。5.2 OperationConfigRole粗粒度配置桶OperationConfigRole Literal[model, event, tool, capability]它是操作的粗粒度配置类别细粒度身份由operation_id承载。DurableOperationConfig协议提供base(role, operation_id...)与for_tool(role, operation_id..., tool..., tool_name...)两个解析入口resolve_tool_operation_config是 Callable 与 Registered 两类后端共享的工具配置解析函数。5.3 durable_operationcapability 方法的统一路径AGENTS.md 特别强调标记了durable_operation的 capability 方法与框架操作走同一个 backend 与配置解析器不要为它们维护第二条注册路径。这可以从_base.py的_bind_capability_operations看到实现通过collect_capability_operations收集每个叶子 capability 的声明用DurableOperation(CapabilityOperationId(capability_id, operationname), ...)构造操作再backend.bind并把 dispatcher 注册进RunContext。同时要求贡献 durable 操作的 capability必须有显式idcapability_id为None会抛UserError因为持久化操作身份与 worker 侧恢复必须保持稳定。六、持久化操作名称兼容性数据禁止随意改动AGENTS.md 的核心红线是持久化操作名称是兼容性数据独立于 Python 类名。改名会让进行中的工作流与已记录运行被搁浅stranded。6.1 JournalOperationNamer 的命名约定_operation_names.py 实现了默认命名策略名称格式以{agent_name}__为前缀操作持久化名称模型请求{agent}__model.request/{agent}__model.request_stream非默认模型加.model_id后缀消息压缩{agent}__model.compact_messages挂起响应取消{agent}__model.cancel_suspended_response事件流处理器{agent}__event_stream_handler工具发现function/dynamic{agent}__function_toolset__{id}.get_toolsMCP 发现{agent}__mcp_server__{id}.get_tools/{agent}__mcp_server__{id}.get_instructions工具参数校验{agent}__function_toolset__{id}.validate_args等工具调用{agent}__function_toolset__{id}.call_tool:{tool_name}非 MCP 会追加工具名 labelcapability 操作{agent}__capability__{capability_id}.{operation}如果JournalOperationNamer的约定不匹配目标运行时可以自行实现DurableOperationNamer协议含operation_name与invocation_name但必须保持输出稳定。6.2 用测试固定完整名称集AGENTS.md 明确要求在重构 agent、model、toolset 或 capability 身份之前用tests/durable_exec/test_durable_exec_compat.py的模式固定完整名称集绝不要在实现重命名时顺手更新这些固定值除非进行中执行的迁移是刻意设计并经过评审的。从 tests/durable_exec/test_durable_exec_compat.py 可以看到这套模式的实际形态test_default_journal_operation_name_matrix用JournalOperationNamer(compat)对全部操作 ID 生成名称断言等于JOURNAL_OPERATION_NAMES固定集合test_journal_operation_name_assembly_sequence构建一个真实的 Agent含 function toolset、dynamic toolset、capability、event handler运行后断言recorded_names的完整执行序列并校验 JOURNAL_OPERATION_NAMESPrefect 同样有test_prefect_operation_name_matrix固定PREFECT_OPERATION_NAMES。这些测试文件顶部维护着完整的名称固定集合JOURNAL_OPERATION_NAMES任何改名都会在测试中暴露出来。七、工具配置的解析基类配置与 per-tool 元数据7.1 配置解析器构建_base.py的_build_resolve_tool_config展示了 per-tool 配置解析的核心逻辑若引擎未声明tool_config_key为None则完全不读取工具元数据——注释强调不能把None折叠成去读空字符串键否则metadata{: False}会让工具悄悄退出 durable 单元这正是 DBOS工具元数据被忽略契约要防止的若声明了键则通过resolve_tool_durable_config读取metadataFalse表示该工具不参与 durable 单元其余情况把 per-tool 配置叠加到基类配置之上最后经_normalize_unit_config后处理例如 Prefect/Temporal 确保不可重试错误。MCP 工具例外_base.py 中 MCP 工具的 durable 配置被显式设为False会直接抛UserError因为MCP 工具执行 I/O不能脱离 durable 单元运行必须删除该元数据以保持调用可持久化。7.2 控制流异常的值包装与升级策略工具调用结果会经过wrap_tool_call_result/unwrap_tool_call_result做控制流即值control-flow-as-values的转换把异常编码进载荷再穿越边界。engine_spec.tool_call_result_upgrade_lenient只对存储了旧版 DBOS/Prefect 集成记录的引擎开启宽容解码无法区分早期原始记录与损坏载荷因此新引擎必须保持False让严格解码暴露损坏而不是把垃圾喂给模型。八、重放安全与确定性引擎作者的强制要求AGENTS.md 最后一部分规定了每个引擎作者都必须遵守的重放安全纪律假设 durable 单元可能执行不止一次进程在副作用提交之后、检查点checkpoint提交之前失败就会发生重放。引擎必须在文档中说明自己的保证并要求工作流侧代码具备幂等性或在 SDK 可用时暴露引擎原生的 at-most-once 选项保持工作流侧代码确定性不要在工具调用、事件处理等可重放路径中依赖随机性、时钟或进程内状态工具集资源进入与关闭按engine_spec.toolset_lifecycles处理包括失败与取消路径并验证 durable 单元内创建的资源不逃逸出该单元引擎自有测试套件必须覆盖重放replay、teardown、控制流异常、持久化输出升级persisted-output upgrades、以及 durable 上下文之外的行为。从源码还能看到两类针对重放安全的显式防护运行期工具集拦截_reject_runtime_toolsets会拒绝 durable 容器内每次运行新加入的工具集——构造期工具集在 capability 绑定时已注册运行期新增会绕过注册并在恢复时重复执行非执行型工具集可以放行运行期 capability 拦截_validate_runtime_capabilities仅 Registered 引擎拒绝 per-run 添加的 capability因为它们在 worker 启动前没有注册 durable 单元取消令牌拦截before_run会在 durable 容器内拒绝带CancellationToken的 run——同一进程内的令牌句柄无法穿越 durable 边界且在重放时触发会非确定性地取消 durable 任务但在 durable 容器之外带持久化能力的 Agent 仍像普通 Agent 一样接受令牌。九、从 AGENTS.md 到可运行集成的落地清单综合以上内容为 pydantic-ai 编写一个新的 durable execution 引擎集成时最终落地清单如下在pydantic_ai_slim/pydantic_ai/durable_exec/下新建子包在公开表面上导出XXXDurability能力类子类化BaseDurabilityCapability实现in_durable_context属性与get_durable_operation_backend()按引擎 SDK 执行模型选择CallableOperationBackend实现execute或RegisteredOperationBackend实现register设置唯一的engine_spec DurabilityEngineSpec(...)逐一声明第 4 节表格中的全部字段按引擎序列化能力选择IDENTITY_CODEC或JSON_CODECjournal 引擎同时配置serialization_failure映射优先使用JournalOperationNamer否则实现稳定的DurableOperationNamer参考 tests/durable_exec/test_durable_exec_compat.py 固定完整操作名称矩阵与运行序列处理DurableOperationId时保留默认分支durable_operation的 capability 方法与框架操作走同一 backend不建第二条路径在引擎自有测试套件中覆盖重放、teardown、控制流异常、持久化输出升级与 durable 上下文之外的行为并落实幂等性文档说明。遵循这套规范新引擎就能作为一等公民与 Temporal、DBOS、Prefect 等内置引擎并列复用框架统一的模型往返、工具集包装、配置解析与命名体系而不是各自为政的外围适配器。【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考