Pydantic AI 与 LangChain/LangGraph 概念映射实战指南从 Agent 循环到持久化的系统性迁移参考【免费下载链接】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 仓库中面向 LangChain/LangGraph 用户的官方迁移参考文档pydantic_ai_slim/pydantic_ai/.agents/skills/migrating-langchain-to-pydantic-ai/references/CONCEPT-MAPPING.md展开系统梳理两大框架在核心 Agent 循环、工具与运行时上下文、结构化输出、中间件、状态持久化、多智能体编排、流式/测试/可观测性等十个维度上的概念对照。文中所有目标侧 API 均已对照仓库源码确认读者读完后可以据此把已有 LangChain/LangGraph 应用逐模块映射到 Pydantic AI 原语识别state_schema、interrupt()、checkpointer、MessagesState等典型结构的对应替代并避开常见迁移陷阱。迁移原则保留行为而不是匹配类名Preserve behavior rather than matching class names。这是整份映射文档的第一原则——选择 Pydantic AI 原语时思考每个 LangChain 组件实际负责什么行为再寻找承载该行为的 Pydantic AI 概念而不是机械地找同名类。核心 Agent 循环Core Agent Loop概念对照表LangChain / LangGraphPydantic AI 默认方案迁移说明LCEL prompt/retriever/parser 管道围绕一次模型调用的普通 Python 代码或Agent确定性检索、格式化与路由保持显式源应用只做一次模型调用时不要引入 Agent 循环create_agent(model, tools, system_prompt...)Agent(model, tools..., instructions...)除非每次请求的构造确有差异否则保持一个可复用 Agentagent.invoke({messages: ...})await agent.run(prompt, deps..., message_history...)只消费result.output不要在公开边界暴露 Pydantic AI 消息对象ainvokeawait agent.run(...)Pydantic AI 以 async 为第一公民仅在同步边界使用run_syncRunnableConfig.configurable类型化依赖对象 显式 run 参数将应用配置与模型可见内容分离context_schema/ToolRuntime.contextdeps_type/RunContext.deps将 DB 客户端、认证身份、配置与服务网关放入依赖prompt templatesinstructions或system_promptinstructions表达当前 Agent 的策略仅当历史中的既有 prompt 必须存活时使用system_prompt多个 LangChain 动态 prompt 可能是替换而非组合关系init_chat_modelprovider 前缀模型字符串或模型实例显式保留 provider 设置核实已安装的 provider API一个最小翻译示例原文档给出的 LangChain 侧代码create_agentainvokecontext# ruff: noqa: F704, F821, Q000 # LangChain from langchain.agents import create_agent agent create_agent( modelprovider:model, tools[lookup_order], system_promptHelp authenticated customers with orders., ) result await agent.ainvoke( {messages: [{role: user, content: Where is order 123?}]}, contextruntime_context, )Pydantic AI 侧等价实现# ruff: noqa: F704, F821 # Pydantic AI from dataclasses import dataclass from pydantic_ai import Agent, RunContext dataclass class Deps: customer_id: str orders: OrderService agent Agent( provider:model, deps_typeDeps, instructionsHelp authenticated customers with orders., ) agent.tool async def lookup_order(ctx: RunContext[Deps], order_id: str) - str: return await ctx.deps.orders.lookup_for_customer(ctx.deps.customer_id, order_id) result await agent.run(Where is order 123?, depsdeps) print(result.output)从仓库源码看Agent.__init__agent/init.py确实同时接收instructions、deps_type、tools等关键字参数且model接受带 provider 前缀的字符串如openai:gpt-5.2见类 docstringoutput_type默认为str。agent.run、agent.run_sync、agent.run_stream等入口在 agent/abstract.py 中均有定义run在 L474-L525、run_sync在 L670-L721、run_stream在 L830-L882。依赖边界即安全边界上例的关键在于模型只能选择order_id而customer_id与服务客户端都由deps注入——依赖边界是安全边界。把经过认证的身份、租户、凭据放进工具参数让模型自由选择正是本仓库迁移陷阱清单中明确列出的反模式见下文迁移陷阱。模型传输与模型名要同时保留原文档强调LangChain 的ChatOpenAI、一个 OpenAI Responses 模型、一个 Azure 部署、一个 OpenAI 兼容本地端点可能共享同一模型标签但请求协议与设置完全不同。迁移时必须同时保留模型传输transport与模型名检查已安装的 Pydantic AI provider 构造函数在不发起真实请求的前提下测试每一个配置分支未知的自定义端点要显式标记为集成缺口integration gap。类型化不变量在不稳定接缝处Typed Invariants at Unstable Seams迁移实践中以下接缝曾真实造成歧义。文档建议用类型把观察到的迁移决策变成可执行约束而不是去重构稳定的应用代码源端歧义目标端不变量需要的证据自由格式配置/运行时上下文混入可信身份、服务与模型输入依赖 dataclass Agent[DepsT, OutputT]只有模型选择的值出现在工具参数中静态检查 捕获的工具 schema 与模型请求字符串/字典/图控制值可能代表多种终止状态如回答 vs 请求更多输入用 Pydantic 模型或显式输出联合类型覆盖源端实际到达的状态每个变体都能通过校验、每个应用分支都被执行注意联合中包含str意味着纯文本也可以终止运行图状态混入公开消息、模型协议历史、待恢复状态、owner、持久化版本区分公开 DTO、list[ModelMessage]、类型化工作流记录安装版本提供ModelMessagesTypeAdapter时用它序列化模型历史针对真实后端的存储往返、续跑、所有权与恢复测试provider 名称与设置以松散字符串/字典传递保留源端枚举或类型化配置映射到具体 Pydantic AIModel不发起真实请求构造每个配置分支探测 provider 特有设置ModelMessagesTypeAdapter在仓库 messages.py 中确实存在pydantic.TypeAdapter实例专门用于模型消息历史的序列化往返。不要仅凭静态类型宣称语义等价类型能暴露缺失的分支、校验边界但无法证明时序、重试、持久化、副作用或框架生命周期行为。工具与运行时上下文Tools and Runtime Context源模式 → 目标模式源模式目标模式tool普通函数agent.tool_plain或Tool(fn)带ToolRuntime的工具agent.tool首参数为RunContext[Deps]BaseTool子类优先普通类型化函数仅在需要动态 schema 时用Tool或Tool.from_schematoolkitFunctionToolset、其他AbstractToolset或一个聚焦的能力capabilityMCP 适配器Pydantic AI MCP toolset 或MCP能力按用户/状态过滤工具工具prepare...、PrepareTools、包装 toolset或能力按需加载运行时发现的大型工具目录延迟工具deferred tools/ 工具搜索tool search避免每轮重建 schema工具重试中间件同 handler 内本地重试用Hooks.on.tool_execute或服务客户端重试ModelRetry、工具retries...、校验器与传输层重试是相互独立的领域审批中间件DeferredToolRequests与DeferredToolResults工具产物ToolReturn区分模型内容、返回值与元数据工具注册与装饰器在仓库 agent/init.py 中实现agent.tool、agent.tool_plain装饰器以及Agent.__init__的tools参数Tool类与prepare_tool_def见 tools.pyFunctionToolset见 toolsets/function.pyPrepareTools能力见 capabilities/prepare_tools.pyDeferredToolRequests/DeferredToolResults定义于 _deferred.py并作为事件类型暴露在 messages.pyDeferredToolRequestsEvent/DeferredToolResultsEvent。迁移时务必保留工具名、描述、JSON schema、并发性、幂等性、超时、重试、审批、认证、以及错误回传给模型的语义。一次成功的 wrapper 导入并不证明工具行为等价。结构化输出Structured OutputLangChainPydantic AIresponse_formatSchemaoutput_typeSchema但当 wire 行为对齐重要时要显式选择输出传输方式provider 策略需要原生强制时用NativeOutput(Schema)工具策略需要工具传输时用ToolOutput(Schema)手动 parser / 非 JSON 文本TextOutput(parser)动态 JSON schemaStructuredDict(schema, name...)无效响应后重试输出校验 ModelRetry/ 配置的重试不要把str放进联合类型——如果运行必须以结构化输出结束纯文本会成为一个合法的终止结果。原文档特别提醒LangChain 对裸 schema 可能自动选择 provider 原生结构化输出而 Pydantic AI 的裸output_type遵循所选模型 profile 的默认值可能是 tool、native 或 prompted 输出。当 wire 行为重要时显式使用NativeOutput、ToolOutput、PromptedOutput或TextOutput并刻画所选 provider/model 的实际行为。仓库中四类输出标记类均可确认见 output.pyToolOutput(type_, *, nameNone, descriptionNone, max_retriesNone, strictNone, sequentialFalse)L77-L150输出走工具调用通道name未指定且只有一个输出时默认用final_resultNativeOutput(outputs, *, nameNone, descriptionNone, strictNone, templateNone)L153-L205使用模型原生结构化输出template中的{schema}占位符会被替换为输出 JSON schemaPromptedOutput(outputs, *, nameNone, descriptionNone, templateNone)L208-L275通过 prompt 引导模型输出TextOutput(output_function)L320-L349用解析函数处理模型纯文本输出StructuredDict(json_schema, nameNone, descriptionNone)L352-L416返回一个携带 JSON schema 的dict[str, Any]子类供动态 schema 场景使用。中间件与生命周期Middleware and Lifecycle按中间件拥有的行为逐项映射LangChain 中间件行为Pydantic AI 目标动态 system prompt动态agent.instructions模型/工具调用前后Hooks生命周期钩子wrapper 装饰器含on.model_request、on.tool_execute可复用的 prompt 工具 钩子 设置自定义AbstractCapability裁剪/摘要消息ProcessHistory显式测试摘要与配对策略过滤/重命名工具定义工具prepare、PrepareTools或 wrapper toolset校验工具参数args_validator或工具校验钩子动态模型选择/降级模型实例/wrapper 如FallbackModel或在应用边界选择模型模型调用次数限制UsageLimits与显式应用层限制工具错误转换在工具内捕获预期异常仅对模型可纠正的失败抛出ModelRetry日志/追踪Logfire 插桩或面向应用指标的钩子guardrail输入/输出校验钩子授权逻辑留在工具/服务内部运行中途注入用户消息RunContext.enqueue或AgentRun.enqueue显式复现钩子顺序把多个源中间件对象合并进一个不透明钩子会让对齐性的检查与测试变得困难。仓库中Hooks位于 capabilities/hooks.pyProcessHistory位于 capabilities/process_history.pyAbstractCapability位于 capabilities/abstract.pyUsageLimits位于 usage.pyFallbackModel位于 models/fallback.py。状态、记忆与持久化State, Memory, and PersistenceLangGraph 常常把四类概念混存在一起迁移时要拆开运行依赖Run dependencies单次运行中不可变或服务类取值 →deps_type与RunContext.deps。对话消息Conversation messages模型请求/响应历史 → 持久化序列化的 Pydantic AI 消息运行时传入message_history。工作流状态Workflow state计划、计数器、fan-out 结果、审批、领域进度 → 类型化应用状态或pydantic_graph状态。长期记忆Long-term memory跨线程事实 → 依赖中的显式仓库/服务。LangGraph 特性 → 迁移选择LangGraph 特性迁移选择MessagesStatePydantic AI 消息历史 另行类型化的工作流状态自定义 reducers普通更新函数或pydantic_graph的 joins/reducerscheckpointer / thread应用层持久化或 durable execution 集成store类型化依赖中的显式存储服务time travel / forkdurable 工作流特有实现不要从消息历史推断该能力interrupt()等待用户输入应用自有的待定会话状态pending conversational state与恢复resume保护工具周围的interrupt()/ HITLdeferred tool request 已认证、可持久化的应用关联与恢复失败后重放Temporal、DBOS、Prefect、Restate 或其他显式 durable 边界不要因为聊天消息能存活就宣布迁移完成——旧系统若同时承诺了 checkpoint 重放、未决写入、线程 fork 或 exactly-once 副作用保护这些都需一一验证。历史修复语义差异Pydantic AI 可能在模型请求前修复悬空的工具调用与孤儿工具结果而 LangGraph 的add_messages按消息 ID 合并。转换存量线程时要对比模型实际可见的历史。文档中给出ModelMessagesTypeAdapter作为序列化模型历史的手段仓库 messages.py并提醒在安装版本提供时使用它完成存储往返测试。图与多智能体系统Graphs and Multi-Agent Systems按拓扑选择实现常规模型/工具循环 → 一个 Pydantic AIAgent。固定序列、有界循环、asyncio.gatherfan-out → 普通 async Python。父 Agent 保持控制并消费子输出 → 通过工具委托delegation via tools。应用代码选择下一个专家 → 程序化交接programmatic hand-off。需要显式类型化节点、分支、join 或可检查的工作流状态 →pydantic_graph。模型选择委托 → 调用类型化子 Agent 的父工具显式定义子 Agent 的历史、依赖、用量、结果与失败传播。工作流必须扛住进程崩溃 → 增加 durable execution 集成单独的图不等于持久化。把Command(goto..., update...)翻译为显式的 next-node 值 类型化状态更新。翻译 LangGraphSend或并行分支时保留 fan-out 上限、取消、异常聚合与排序语义Pydantic Graph 的 fan-out 应返回分支局部结果、在需要保序时让 join 携带源索引且仅在提前完成确实是 reducer 契约时才使用ReducerContext.cancel_sibling_tasks()。流式、测试与可观测性Streaming, Testing, and ObservabilityLangChain 生态Pydantic 生态stream/astream的 values、updates、messagesrun_stream、run_stream_events、event_stream_handler或iterfake chat modelsTestModel或FunctionModel配合agent.override(...)trajectory / eval 数据集pydantic_evals的 cases、datasets、evaluatorsLangSmith tracesPydantic AI Logfire 插桩 应用自有的 OpenTelemetry spansLangSmith 可将源 traces 导出到同一后端做对比图状态检查类型化状态 应用持久化/图检查在 UI/API 边界定义应用自有的事件 schema迁移期间让两套实现都适配它不要让客户端直接依赖任一框架的事件类。用 Logfire 检查模型调用、工具、重试、错误、用量与耗时再用可执行测试证明公开契约。特别注意模型的首个 chunk 时间不等于客户端收到首个事件的时间详见仓库中同一迁移技能集下的 LOGFIRE-VERIFICATION.md 参考文档即文档原文引用的 Logfire-Assisted Migration Verification。不要在没有单独 trajectory 测试时用run_stream()替代run()run_stream()在流式过程中提交第一个匹配的输出同发的工具与重试可能导致与完整run()不同的终止结果。TestModel位于 models/test.py支持call_toolslist[str] | Literal[all]默认all与seed默认 0用于确定性生成工具参数配合agent.override(...)定义于 agent/abstract.py可在不发起真实模型请求的情况下完成确定性测试。过渡桥接Transitional BridgesPydantic AI 可以包装 LangChain 工具仓库在 ext/langchain.py 中实现了tool_from_langchain与LangChainToolset# ruff: noqa: F821 from pydantic_ai import Agent from pydantic_ai.ext.langchain import LangChainToolset, tool_from_langchain single_tool tool_from_langchain(existing_langchain_tool) toolset LangChainToolset(existing_toolkit.get_tools()) agent Agent(provider:model, tools[single_tool], toolsets[toolset])仅在工具内部移植期间用它打通垂直切片。wrapper 会把参数校验委托给 LangChain 工具、保留 LangChain 依赖并可能掩盖框架特有回调或运行时假设。每个桥接都要登记一个移除 issue 和一组对等测试。同时审计 LangChain 特有标志如return_directwrapper 会调用工具但不会自动保留源 Agent 的工具后停止路由。当该调用是模型选择的终止动作时用带ToolOutput的命名输出函数表示并验证一次执行与源模型的调用次数。如果重写 retriever 或 LCEL 管道会阻塞 Agent 迁移就先把它藏在窄工具/服务接口后面待 Agent 边界稳定后再移植。迁移陷阱Migration Traps文档最后列出的八类高频陷阱逐条自查把state_schema翻译成deps_type然后又把依赖当工作流状态去修改。把认证身份、租户或凭据作为模型可选的工具参数传递。把所有中间件都当成 hooks即便行为本应属于工具或模型 wrapper。在 Pydantic AI 历史中复用 LangChain 消息对象。为逃避学习pydantic_graph把确定性分支搬进 prompt。用内存消息列表冒充 checkpointer。同时保留两套可观测性 SDK却不定义 trace 归属与关联。只测试最终文本而工具轨迹、审批与副作用已经改变。参考资源本迁移技能的完整文档集位于仓库pydantic_ai_slim/pydantic_ai/.agents/skills/migrating-langchain-to-pydantic-ai/目录含本文所依据的 CONCEPT-MAPPING.md 与 Logfire 辅助验证参考 LOGFIRE-VERIFICATION.md。迁移涉及的 Pydantic AI 核心源码与配套文档还包括Agent 构造与运行入口agent/init.py、agent/abstract.py工具与工具集tools.py、toolsets/function.py输出标记类output.py能力体系capabilities/abstract.py、capabilities/hooks.py、capabilities/prepare_tools.py、capabilities/process_history.py延迟工具与审批_deferred.py、messages.py测试与降级模型models/test.py、models/function.py、models/fallback.pyLangChain 桥接ext/langchain.py用量限制usage.py官方配套文档docs/agent.md、docs/tools.md、docs/hooks.md、docs/third-party-tools.md、docs/multi-agent-applications.md、docs/durable_execution/overview.md【免费下载链接】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），仅供参考 Pydantic AI 与 LangChain/LangGraph 概念映射实战指南：从 Agent 循环到持久化的系统性迁移参考
Pydantic AI 与 LangChain/LangGraph 概念映射实战指南从 Agent 循环到持久化的系统性迁移参考【免费下载链接】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 仓库中面向 LangChain/LangGraph 用户的官方迁移参考文档pydantic_ai_slim/pydantic_ai/.agents/skills/migrating-langchain-to-pydantic-ai/references/CONCEPT-MAPPING.md展开系统梳理两大框架在核心 Agent 循环、工具与运行时上下文、结构化输出、中间件、状态持久化、多智能体编排、流式/测试/可观测性等十个维度上的概念对照。文中所有目标侧 API 均已对照仓库源码确认读者读完后可以据此把已有 LangChain/LangGraph 应用逐模块映射到 Pydantic AI 原语识别state_schema、interrupt()、checkpointer、MessagesState等典型结构的对应替代并避开常见迁移陷阱。迁移原则保留行为而不是匹配类名Preserve behavior rather than matching class names。这是整份映射文档的第一原则——选择 Pydantic AI 原语时思考每个 LangChain 组件实际负责什么行为再寻找承载该行为的 Pydantic AI 概念而不是机械地找同名类。核心 Agent 循环Core Agent Loop概念对照表LangChain / LangGraphPydantic AI 默认方案迁移说明LCEL prompt/retriever/parser 管道围绕一次模型调用的普通 Python 代码或Agent确定性检索、格式化与路由保持显式源应用只做一次模型调用时不要引入 Agent 循环create_agent(model, tools, system_prompt...)Agent(model, tools..., instructions...)除非每次请求的构造确有差异否则保持一个可复用 Agentagent.invoke({messages: ...})await agent.run(prompt, deps..., message_history...)只消费result.output不要在公开边界暴露 Pydantic AI 消息对象ainvokeawait agent.run(...)Pydantic AI 以 async 为第一公民仅在同步边界使用run_syncRunnableConfig.configurable类型化依赖对象 显式 run 参数将应用配置与模型可见内容分离context_schema/ToolRuntime.contextdeps_type/RunContext.deps将 DB 客户端、认证身份、配置与服务网关放入依赖prompt templatesinstructions或system_promptinstructions表达当前 Agent 的策略仅当历史中的既有 prompt 必须存活时使用system_prompt多个 LangChain 动态 prompt 可能是替换而非组合关系init_chat_modelprovider 前缀模型字符串或模型实例显式保留 provider 设置核实已安装的 provider API一个最小翻译示例原文档给出的 LangChain 侧代码create_agentainvokecontext# ruff: noqa: F704, F821, Q000 # LangChain from langchain.agents import create_agent agent create_agent( modelprovider:model, tools[lookup_order], system_promptHelp authenticated customers with orders., ) result await agent.ainvoke( {messages: [{role: user, content: Where is order 123?}]}, contextruntime_context, )Pydantic AI 侧等价实现# ruff: noqa: F704, F821 # Pydantic AI from dataclasses import dataclass from pydantic_ai import Agent, RunContext dataclass class Deps: customer_id: str orders: OrderService agent Agent( provider:model, deps_typeDeps, instructionsHelp authenticated customers with orders., ) agent.tool async def lookup_order(ctx: RunContext[Deps], order_id: str) - str: return await ctx.deps.orders.lookup_for_customer(ctx.deps.customer_id, order_id) result await agent.run(Where is order 123?, depsdeps) print(result.output)从仓库源码看Agent.__init__agent/init.py确实同时接收instructions、deps_type、tools等关键字参数且model接受带 provider 前缀的字符串如openai:gpt-5.2见类 docstringoutput_type默认为str。agent.run、agent.run_sync、agent.run_stream等入口在 agent/abstract.py 中均有定义run在 L474-L525、run_sync在 L670-L721、run_stream在 L830-L882。依赖边界即安全边界上例的关键在于模型只能选择order_id而customer_id与服务客户端都由deps注入——依赖边界是安全边界。把经过认证的身份、租户、凭据放进工具参数让模型自由选择正是本仓库迁移陷阱清单中明确列出的反模式见下文迁移陷阱。模型传输与模型名要同时保留原文档强调LangChain 的ChatOpenAI、一个 OpenAI Responses 模型、一个 Azure 部署、一个 OpenAI 兼容本地端点可能共享同一模型标签但请求协议与设置完全不同。迁移时必须同时保留模型传输transport与模型名检查已安装的 Pydantic AI provider 构造函数在不发起真实请求的前提下测试每一个配置分支未知的自定义端点要显式标记为集成缺口integration gap。类型化不变量在不稳定接缝处Typed Invariants at Unstable Seams迁移实践中以下接缝曾真实造成歧义。文档建议用类型把观察到的迁移决策变成可执行约束而不是去重构稳定的应用代码源端歧义目标端不变量需要的证据自由格式配置/运行时上下文混入可信身份、服务与模型输入依赖 dataclass Agent[DepsT, OutputT]只有模型选择的值出现在工具参数中静态检查 捕获的工具 schema 与模型请求字符串/字典/图控制值可能代表多种终止状态如回答 vs 请求更多输入用 Pydantic 模型或显式输出联合类型覆盖源端实际到达的状态每个变体都能通过校验、每个应用分支都被执行注意联合中包含str意味着纯文本也可以终止运行图状态混入公开消息、模型协议历史、待恢复状态、owner、持久化版本区分公开 DTO、list[ModelMessage]、类型化工作流记录安装版本提供ModelMessagesTypeAdapter时用它序列化模型历史针对真实后端的存储往返、续跑、所有权与恢复测试provider 名称与设置以松散字符串/字典传递保留源端枚举或类型化配置映射到具体 Pydantic AIModel不发起真实请求构造每个配置分支探测 provider 特有设置ModelMessagesTypeAdapter在仓库 messages.py 中确实存在pydantic.TypeAdapter实例专门用于模型消息历史的序列化往返。不要仅凭静态类型宣称语义等价类型能暴露缺失的分支、校验边界但无法证明时序、重试、持久化、副作用或框架生命周期行为。工具与运行时上下文Tools and Runtime Context源模式 → 目标模式源模式目标模式tool普通函数agent.tool_plain或Tool(fn)带ToolRuntime的工具agent.tool首参数为RunContext[Deps]BaseTool子类优先普通类型化函数仅在需要动态 schema 时用Tool或Tool.from_schematoolkitFunctionToolset、其他AbstractToolset或一个聚焦的能力capabilityMCP 适配器Pydantic AI MCP toolset 或MCP能力按用户/状态过滤工具工具prepare...、PrepareTools、包装 toolset或能力按需加载运行时发现的大型工具目录延迟工具deferred tools/ 工具搜索tool search避免每轮重建 schema工具重试中间件同 handler 内本地重试用Hooks.on.tool_execute或服务客户端重试ModelRetry、工具retries...、校验器与传输层重试是相互独立的领域审批中间件DeferredToolRequests与DeferredToolResults工具产物ToolReturn区分模型内容、返回值与元数据工具注册与装饰器在仓库 agent/init.py 中实现agent.tool、agent.tool_plain装饰器以及Agent.__init__的tools参数Tool类与prepare_tool_def见 tools.pyFunctionToolset见 toolsets/function.pyPrepareTools能力见 capabilities/prepare_tools.pyDeferredToolRequests/DeferredToolResults定义于 _deferred.py并作为事件类型暴露在 messages.pyDeferredToolRequestsEvent/DeferredToolResultsEvent。迁移时务必保留工具名、描述、JSON schema、并发性、幂等性、超时、重试、审批、认证、以及错误回传给模型的语义。一次成功的 wrapper 导入并不证明工具行为等价。结构化输出Structured OutputLangChainPydantic AIresponse_formatSchemaoutput_typeSchema但当 wire 行为对齐重要时要显式选择输出传输方式provider 策略需要原生强制时用NativeOutput(Schema)工具策略需要工具传输时用ToolOutput(Schema)手动 parser / 非 JSON 文本TextOutput(parser)动态 JSON schemaStructuredDict(schema, name...)无效响应后重试输出校验 ModelRetry/ 配置的重试不要把str放进联合类型——如果运行必须以结构化输出结束纯文本会成为一个合法的终止结果。原文档特别提醒LangChain 对裸 schema 可能自动选择 provider 原生结构化输出而 Pydantic AI 的裸output_type遵循所选模型 profile 的默认值可能是 tool、native 或 prompted 输出。当 wire 行为重要时显式使用NativeOutput、ToolOutput、PromptedOutput或TextOutput并刻画所选 provider/model 的实际行为。仓库中四类输出标记类均可确认见 output.pyToolOutput(type_, *, nameNone, descriptionNone, max_retriesNone, strictNone, sequentialFalse)L77-L150输出走工具调用通道name未指定且只有一个输出时默认用final_resultNativeOutput(outputs, *, nameNone, descriptionNone, strictNone, templateNone)L153-L205使用模型原生结构化输出template中的{schema}占位符会被替换为输出 JSON schemaPromptedOutput(outputs, *, nameNone, descriptionNone, templateNone)L208-L275通过 prompt 引导模型输出TextOutput(output_function)L320-L349用解析函数处理模型纯文本输出StructuredDict(json_schema, nameNone, descriptionNone)L352-L416返回一个携带 JSON schema 的dict[str, Any]子类供动态 schema 场景使用。中间件与生命周期Middleware and Lifecycle按中间件拥有的行为逐项映射LangChain 中间件行为Pydantic AI 目标动态 system prompt动态agent.instructions模型/工具调用前后Hooks生命周期钩子wrapper 装饰器含on.model_request、on.tool_execute可复用的 prompt 工具 钩子 设置自定义AbstractCapability裁剪/摘要消息ProcessHistory显式测试摘要与配对策略过滤/重命名工具定义工具prepare、PrepareTools或 wrapper toolset校验工具参数args_validator或工具校验钩子动态模型选择/降级模型实例/wrapper 如FallbackModel或在应用边界选择模型模型调用次数限制UsageLimits与显式应用层限制工具错误转换在工具内捕获预期异常仅对模型可纠正的失败抛出ModelRetry日志/追踪Logfire 插桩或面向应用指标的钩子guardrail输入/输出校验钩子授权逻辑留在工具/服务内部运行中途注入用户消息RunContext.enqueue或AgentRun.enqueue显式复现钩子顺序把多个源中间件对象合并进一个不透明钩子会让对齐性的检查与测试变得困难。仓库中Hooks位于 capabilities/hooks.pyProcessHistory位于 capabilities/process_history.pyAbstractCapability位于 capabilities/abstract.pyUsageLimits位于 usage.pyFallbackModel位于 models/fallback.py。状态、记忆与持久化State, Memory, and PersistenceLangGraph 常常把四类概念混存在一起迁移时要拆开运行依赖Run dependencies单次运行中不可变或服务类取值 →deps_type与RunContext.deps。对话消息Conversation messages模型请求/响应历史 → 持久化序列化的 Pydantic AI 消息运行时传入message_history。工作流状态Workflow state计划、计数器、fan-out 结果、审批、领域进度 → 类型化应用状态或pydantic_graph状态。长期记忆Long-term memory跨线程事实 → 依赖中的显式仓库/服务。LangGraph 特性 → 迁移选择LangGraph 特性迁移选择MessagesStatePydantic AI 消息历史 另行类型化的工作流状态自定义 reducers普通更新函数或pydantic_graph的 joins/reducerscheckpointer / thread应用层持久化或 durable execution 集成store类型化依赖中的显式存储服务time travel / forkdurable 工作流特有实现不要从消息历史推断该能力interrupt()等待用户输入应用自有的待定会话状态pending conversational state与恢复resume保护工具周围的interrupt()/ HITLdeferred tool request 已认证、可持久化的应用关联与恢复失败后重放Temporal、DBOS、Prefect、Restate 或其他显式 durable 边界不要因为聊天消息能存活就宣布迁移完成——旧系统若同时承诺了 checkpoint 重放、未决写入、线程 fork 或 exactly-once 副作用保护这些都需一一验证。历史修复语义差异Pydantic AI 可能在模型请求前修复悬空的工具调用与孤儿工具结果而 LangGraph 的add_messages按消息 ID 合并。转换存量线程时要对比模型实际可见的历史。文档中给出ModelMessagesTypeAdapter作为序列化模型历史的手段仓库 messages.py并提醒在安装版本提供时使用它完成存储往返测试。图与多智能体系统Graphs and Multi-Agent Systems按拓扑选择实现常规模型/工具循环 → 一个 Pydantic AIAgent。固定序列、有界循环、asyncio.gatherfan-out → 普通 async Python。父 Agent 保持控制并消费子输出 → 通过工具委托delegation via tools。应用代码选择下一个专家 → 程序化交接programmatic hand-off。需要显式类型化节点、分支、join 或可检查的工作流状态 →pydantic_graph。模型选择委托 → 调用类型化子 Agent 的父工具显式定义子 Agent 的历史、依赖、用量、结果与失败传播。工作流必须扛住进程崩溃 → 增加 durable execution 集成单独的图不等于持久化。把Command(goto..., update...)翻译为显式的 next-node 值 类型化状态更新。翻译 LangGraphSend或并行分支时保留 fan-out 上限、取消、异常聚合与排序语义Pydantic Graph 的 fan-out 应返回分支局部结果、在需要保序时让 join 携带源索引且仅在提前完成确实是 reducer 契约时才使用ReducerContext.cancel_sibling_tasks()。流式、测试与可观测性Streaming, Testing, and ObservabilityLangChain 生态Pydantic 生态stream/astream的 values、updates、messagesrun_stream、run_stream_events、event_stream_handler或iterfake chat modelsTestModel或FunctionModel配合agent.override(...)trajectory / eval 数据集pydantic_evals的 cases、datasets、evaluatorsLangSmith tracesPydantic AI Logfire 插桩 应用自有的 OpenTelemetry spansLangSmith 可将源 traces 导出到同一后端做对比图状态检查类型化状态 应用持久化/图检查在 UI/API 边界定义应用自有的事件 schema迁移期间让两套实现都适配它不要让客户端直接依赖任一框架的事件类。用 Logfire 检查模型调用、工具、重试、错误、用量与耗时再用可执行测试证明公开契约。特别注意模型的首个 chunk 时间不等于客户端收到首个事件的时间详见仓库中同一迁移技能集下的 LOGFIRE-VERIFICATION.md 参考文档即文档原文引用的 Logfire-Assisted Migration Verification。不要在没有单独 trajectory 测试时用run_stream()替代run()run_stream()在流式过程中提交第一个匹配的输出同发的工具与重试可能导致与完整run()不同的终止结果。TestModel位于 models/test.py支持call_toolslist[str] | Literal[all]默认all与seed默认 0用于确定性生成工具参数配合agent.override(...)定义于 agent/abstract.py可在不发起真实模型请求的情况下完成确定性测试。过渡桥接Transitional BridgesPydantic AI 可以包装 LangChain 工具仓库在 ext/langchain.py 中实现了tool_from_langchain与LangChainToolset# ruff: noqa: F821 from pydantic_ai import Agent from pydantic_ai.ext.langchain import LangChainToolset, tool_from_langchain single_tool tool_from_langchain(existing_langchain_tool) toolset LangChainToolset(existing_toolkit.get_tools()) agent Agent(provider:model, tools[single_tool], toolsets[toolset])仅在工具内部移植期间用它打通垂直切片。wrapper 会把参数校验委托给 LangChain 工具、保留 LangChain 依赖并可能掩盖框架特有回调或运行时假设。每个桥接都要登记一个移除 issue 和一组对等测试。同时审计 LangChain 特有标志如return_directwrapper 会调用工具但不会自动保留源 Agent 的工具后停止路由。当该调用是模型选择的终止动作时用带ToolOutput的命名输出函数表示并验证一次执行与源模型的调用次数。如果重写 retriever 或 LCEL 管道会阻塞 Agent 迁移就先把它藏在窄工具/服务接口后面待 Agent 边界稳定后再移植。迁移陷阱Migration Traps文档最后列出的八类高频陷阱逐条自查把state_schema翻译成deps_type然后又把依赖当工作流状态去修改。把认证身份、租户或凭据作为模型可选的工具参数传递。把所有中间件都当成 hooks即便行为本应属于工具或模型 wrapper。在 Pydantic AI 历史中复用 LangChain 消息对象。为逃避学习pydantic_graph把确定性分支搬进 prompt。用内存消息列表冒充 checkpointer。同时保留两套可观测性 SDK却不定义 trace 归属与关联。只测试最终文本而工具轨迹、审批与副作用已经改变。参考资源本迁移技能的完整文档集位于仓库pydantic_ai_slim/pydantic_ai/.agents/skills/migrating-langchain-to-pydantic-ai/目录含本文所依据的 CONCEPT-MAPPING.md 与 Logfire 辅助验证参考 LOGFIRE-VERIFICATION.md。迁移涉及的 Pydantic AI 核心源码与配套文档还包括Agent 构造与运行入口agent/init.py、agent/abstract.py工具与工具集tools.py、toolsets/function.py输出标记类output.py能力体系capabilities/abstract.py、capabilities/hooks.py、capabilities/prepare_tools.py、capabilities/process_history.py延迟工具与审批_deferred.py、messages.py测试与降级模型models/test.py、models/function.py、models/fallback.pyLangChain 桥接ext/langchain.py用量限制usage.py官方配套文档docs/agent.md、docs/tools.md、docs/hooks.md、docs/third-party-tools.md、docs/multi-agent-applications.md、docs/durable_execution/overview.md【免费下载链接】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），仅供参考