如何用 Kitaru 适配器把普通 Pydantic AI Agent 包装成可恢复的 flow【免费下载链接】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如果你的 Pydantic AI agent 跑长任务——先请求模型、再调工具、中途可能还等人审批——进程一旦崩溃重启后通常会重放已经完成的模型请求甚至重复工具调用的副作用。Kitaru 是 Pydantic AI 文档列出的外部 SDK 集成之一官方一方支持的是 Temporal、DBOS、Prefect、Restate 等Kitaru 属于 external SDK 集成见 durable execution 概览。它把 agent 的每次调用记录成一个可恢复的flowflow 内的模型请求、工具调用、MCP 调用和人工等待会被记成checkpoint进程重启后Kitaru 重新执行 Python 函数遇到已有 checkpoint 的操作直接返回保存的结果从第一个未完成的操作继续。本文的目标把一个普通的pydantic_ai.Agent用KitaruAgent包装起来本地跑通一次可恢复的调用再换成生产环境需要的显式kitaru.flow写法。前提是你已经有一个可用的 Pydantic AI agent。安装 Kitaru 并验证连接Kitaru 独立于 Pydantic AI 安装适配器位于kitaru包的kitaru.adapters.pydantic_ai模块中而不是pydantic_ai.durable_execuv add kitaru[pydantic-ai]本地开发要连接 Kitaru 的 local server 时再加localextrauv add kitaru[pydantic-ai,local]初始化项目并确认连接可用再运行后面的示例kitaru init kitaru login kitaru statuskitaru status是文档给出的连接验证方式建议在跑任何示例前执行。把 Agent 包装成 KitaruAgent最小可运行的持久化 agent 如下文档示例代码openai:gpt-5-nano是文档使用的模型名from pydantic_ai import Agent from kitaru.adapters.pydantic_ai import KitaruAgent agent Agent(openai:gpt-5-nano, nameresearcher) durable_agent KitaruAgent(agent) result durable_agent.run_sync(Summarize quantum error correction.) print(result.output)关键点KitaruAgent不替换原来的 agent 对象。它把实际的 Pydantic AI 运行委托给被包装的 agent在运行过程中记录可恢复的操作模型请求、Pydantic AI 工具调用、MCP 工具调用、hitl_tool人工等待。它暴露常规的运行方法包括Agent.run和Agent.run_sync。原来的Agent在 Kitaru 之外仍可照常使用。运行结果通过result.output取回示例中打印它即是文档展示的验证方式。文档没有给出run_sync的成功输出值实际输出取决于模型响应不要把它当作固定预期。文档中自动创建 flow 的用法即上面这种直接调用durable_agent.run_sync(...)而不显式声明 flow 的写法是面向本地开发的注意它不是生产路径生产写法见下文。生产环境把 durable 调用放进显式 kitaru.flow远端 stack 和生产服务中要把 durable agent 调用放进显式kitaru.flow让 Kitaru 有一个稳定的 flow 入口点来提交、重放和检查import kitaru from pydantic_ai import Agent from kitaru.adapters.pydantic_ai import KitaruAgent agent Agent(openai:gpt-5-nano, nameresearcher) durable_agent KitaruAgent(agent) kitaru.flow def research_topic(topic: str) - str: result durable_agent.run_sync(fSummarize {topic}.) return result.output文档给出的选择标准很直接第一次实验用短 wrapper上一节的写法当运行需要离开本地进程时用显式 flow。选择 checkpoint 策略KitaruAgent支持两种 checkpoint 策略策略默认持久化内容适用calls是可安全重放的模型请求、工具调用、MCP 调用和人工等待各自作为独立 checkpoint大多数 agent尤其是单次调用昂贵或有副作用时turn否一个 checkpoint 包整个 agent 运行不需要按调用打 checkpoint 的简单运行或流式约束要求整轮 checkpoint 的场景默认calls策略下有一个边界Kitaru 不能在用户自定义的kitaru.checkpoint体内创建嵌套 checkpoint。如果durable_agent.run_sync(...)运行在用户kitaru.checkpoint内部Kitaru 会把整个 agent 轮次记录在那个外层 checkpoint 下而不是生成单独的模型、工具或 MCP checkpoint 行。流式调用的处理Kitaru 支持 Pydantic AI 流式但有约束事件流优先使用Agent.run的event_stream_handler参数说明见 Streaming All Events。当某次运行使用了event_stream_handler时Kitaru 对该次调用回退到整轮turncheckpoint。如果用的是run_stream()或iter()把流式调用包进显式kitaru.checkpoint这样 Kitaru 只需重放一个 durable 操作而不是尝试逐个持久化流式事件。必须遵守的约束文档在 Requirements and Constraints 中列出的硬性要求逐条核对agent 构造时用具体模型定义如Agent(openai:gpt-5-nano, ...)每个 durable agent 给一个稳定的nameKitaru 用它跨运行识别持久化的工作使用KitaruAgent时不要通过model按运行覆盖模型远端 stack 和生产服务用显式kitaru.flow自动创建 flow 仅限本地避免在用户定义的kitaru.checkpoint体内嵌套 Kitaru checkpoint。已知边界与延伸阅读Kitaru 的人工审批场景优先使用其hitl_tool等待会被转成可恢复的工具调用进程可以在等人响应时停止恢复后继续。普通 Pydantic AI 工具体内的同步等待需要额外的 Kitaru 配置文档将其指向 Kitaru 官方的 Pydantic AI 适配器指南外部文档仓库内未附正文。更深入的 checkpoint 配置同样在 Kitaru 官方适配器指南中本仓库文档未展开。Pydantic AI 自身的 deferred tool 模式含 human-in-the-loop 工具审批见 deferred tools这是 Pydantic AI 内置机制与 Kitaru 的hitl_tool是两条独立路径。完成验证方式与上文一致本地跑通最小 wrapper 并确认result.output有输出生产路径下把调用包进kitaru.flow后flow 即成为 Kitaru 提交、重放和检查的稳定入口。【免费下载链接】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），仅供参考 
长河君 · 2026/9/14 12:23:07 · 阅读 1.2k 如何用 Kitaru 适配器把普通 Pydantic AI Agent 包装成可恢复的 flow
如何用 Kitaru 适配器把普通 Pydantic AI Agent 包装成可恢复的 flow【免费下载链接】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如果你的 Pydantic AI agent 跑长任务——先请求模型、再调工具、中途可能还等人审批——进程一旦崩溃重启后通常会重放已经完成的模型请求甚至重复工具调用的副作用。Kitaru 是 Pydantic AI 文档列出的外部 SDK 集成之一官方一方支持的是 Temporal、DBOS、Prefect、Restate 等Kitaru 属于 external SDK 集成见 durable execution 概览。它把 agent 的每次调用记录成一个可恢复的flowflow 内的模型请求、工具调用、MCP 调用和人工等待会被记成checkpoint进程重启后Kitaru 重新执行 Python 函数遇到已有 checkpoint 的操作直接返回保存的结果从第一个未完成的操作继续。本文的目标把一个普通的pydantic_ai.Agent用KitaruAgent包装起来本地跑通一次可恢复的调用再换成生产环境需要的显式kitaru.flow写法。前提是你已经有一个可用的 Pydantic AI agent。安装 Kitaru 并验证连接Kitaru 独立于 Pydantic AI 安装适配器位于kitaru包的kitaru.adapters.pydantic_ai模块中而不是pydantic_ai.durable_execuv add kitaru[pydantic-ai]本地开发要连接 Kitaru 的 local server 时再加localextrauv add kitaru[pydantic-ai,local]初始化项目并确认连接可用再运行后面的示例kitaru init kitaru login kitaru statuskitaru status是文档给出的连接验证方式建议在跑任何示例前执行。把 Agent 包装成 KitaruAgent最小可运行的持久化 agent 如下文档示例代码openai:gpt-5-nano是文档使用的模型名from pydantic_ai import Agent from kitaru.adapters.pydantic_ai import KitaruAgent agent Agent(openai:gpt-5-nano, nameresearcher) durable_agent KitaruAgent(agent) result durable_agent.run_sync(Summarize quantum error correction.) print(result.output)关键点KitaruAgent不替换原来的 agent 对象。它把实际的 Pydantic AI 运行委托给被包装的 agent在运行过程中记录可恢复的操作模型请求、Pydantic AI 工具调用、MCP 工具调用、hitl_tool人工等待。它暴露常规的运行方法包括Agent.run和Agent.run_sync。原来的Agent在 Kitaru 之外仍可照常使用。运行结果通过result.output取回示例中打印它即是文档展示的验证方式。文档没有给出run_sync的成功输出值实际输出取决于模型响应不要把它当作固定预期。文档中自动创建 flow 的用法即上面这种直接调用durable_agent.run_sync(...)而不显式声明 flow 的写法是面向本地开发的注意它不是生产路径生产写法见下文。生产环境把 durable 调用放进显式 kitaru.flow远端 stack 和生产服务中要把 durable agent 调用放进显式kitaru.flow让 Kitaru 有一个稳定的 flow 入口点来提交、重放和检查import kitaru from pydantic_ai import Agent from kitaru.adapters.pydantic_ai import KitaruAgent agent Agent(openai:gpt-5-nano, nameresearcher) durable_agent KitaruAgent(agent) kitaru.flow def research_topic(topic: str) - str: result durable_agent.run_sync(fSummarize {topic}.) return result.output文档给出的选择标准很直接第一次实验用短 wrapper上一节的写法当运行需要离开本地进程时用显式 flow。选择 checkpoint 策略KitaruAgent支持两种 checkpoint 策略策略默认持久化内容适用calls是可安全重放的模型请求、工具调用、MCP 调用和人工等待各自作为独立 checkpoint大多数 agent尤其是单次调用昂贵或有副作用时turn否一个 checkpoint 包整个 agent 运行不需要按调用打 checkpoint 的简单运行或流式约束要求整轮 checkpoint 的场景默认calls策略下有一个边界Kitaru 不能在用户自定义的kitaru.checkpoint体内创建嵌套 checkpoint。如果durable_agent.run_sync(...)运行在用户kitaru.checkpoint内部Kitaru 会把整个 agent 轮次记录在那个外层 checkpoint 下而不是生成单独的模型、工具或 MCP checkpoint 行。流式调用的处理Kitaru 支持 Pydantic AI 流式但有约束事件流优先使用Agent.run的event_stream_handler参数说明见 Streaming All Events。当某次运行使用了event_stream_handler时Kitaru 对该次调用回退到整轮turncheckpoint。如果用的是run_stream()或iter()把流式调用包进显式kitaru.checkpoint这样 Kitaru 只需重放一个 durable 操作而不是尝试逐个持久化流式事件。必须遵守的约束文档在 Requirements and Constraints 中列出的硬性要求逐条核对agent 构造时用具体模型定义如Agent(openai:gpt-5-nano, ...)每个 durable agent 给一个稳定的nameKitaru 用它跨运行识别持久化的工作使用KitaruAgent时不要通过model按运行覆盖模型远端 stack 和生产服务用显式kitaru.flow自动创建 flow 仅限本地避免在用户定义的kitaru.checkpoint体内嵌套 Kitaru checkpoint。已知边界与延伸阅读Kitaru 的人工审批场景优先使用其hitl_tool等待会被转成可恢复的工具调用进程可以在等人响应时停止恢复后继续。普通 Pydantic AI 工具体内的同步等待需要额外的 Kitaru 配置文档将其指向 Kitaru 官方的 Pydantic AI 适配器指南外部文档仓库内未附正文。更深入的 checkpoint 配置同样在 Kitaru 官方适配器指南中本仓库文档未展开。Pydantic AI 自身的 deferred tool 模式含 human-in-the-loop 工具审批见 deferred tools这是 Pydantic AI 内置机制与 Kitaru 的hitl_tool是两条独立路径。完成验证方式与上文一致本地跑通最小 wrapper 并确认result.output有输出生产路径下把调用包进kitaru.flow后flow 即成为 Kitaru 提交、重放和检查的稳定入口。【免费下载链接】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），仅供参考