Docling 开发技能指南：Pydantic AI 编排与集成（多智能体、图工作流、A2A 与持久化执行）

Docling 开发技能指南：Pydantic AI 编排与集成（多智能体、图工作流、A2A 与持久化执行）

【免费下载链接】doclingGet your documents ready for gen AI项目地址: https://gitcode.com/GitHub_Trending/do/docling

本篇技术指南以 Docling 仓库.agents/skills/目录下的开发技能参考文档ORCHESTRATION-AND-INTEGRATIONS.md为主体，系统讲解 Pydantic AI 的编排与集成能力：多智能体协作、基于pydantic_graph的状态机工作流、无 Agent 的直连模型调用、A2A 协议暴露、Temporal/DBOS/Prefect 持久化执行、RAG 嵌入、LangChain/ACI 生态桥接以及pydantic_evals评估体系。读完本文，你将掌握在真实项目中选型与落地各类 Agent 编排模式的完整依据。

文档定位：它来自 Docling 的"开发技能"体系

在展开正文前，先说明这份文档在 Docling 仓库中的位置与用途，这直接决定了本文的适用前提：

• Docling 仓库根目录的 AGENTS.md 明确规定了"技能"（Skills）的双层结构：开发技能（Development skills，服务于在 Docling 上工作的 AI 编码代理）存放于仓库根目录的.agents/skills/，其中就包括building-pydantic-ai-agents；使用技能（Usage skills，教代理如何调用 Docling 做文档转换）则随 Python 包分发，位于docling/.agents/skills/docling/，其机制详见 Agent Skills 说明。
• SKILL.md 是building-pydantic-ai-agents技能的入口路由文件，其中声明该技能要求Python 3.10+，并通过"任务路由表"将不同任务指向按需加载的 references 文件。编排相关任务在路由表中对应两条记录：
  • "Coordinate multiple agents or build graph workflows" → 本文主体文档 ORCHESTRATION-AND-INTEGRATIONS.md；
  • "Call the model directly, expose A2A, use durable execution, embeddings, evals, or third-party integrations" → 同一文档。
• 该文档开头的阅读指引（Read this file when...）与 COMMON-TASKS.md 中的任务映射表互为索引，后者把 "Coordinate Multiple Agents"、"Build Multi-Step Workflows with Graphs" 等旧式锚点链接统一转发到本文档的对应章节。

需要强调一个适用前提：pydantic_ai并非 Docling 运行时的依赖（Docling 的pyproject.toml仅依赖pydantic与pydantic-settings），该技能是面向贡献者/编码代理的知识参考，实际使用本文各示例需自行安装pydantic-ai及相关可选组件（如pydantic-graph、pydantic-evals）。

协调多智能体：用工具委托（Delegation）保留父级控制权

原文档的第一条核心模式是agent delegation：当"一个 Agent 应该调用另一个 Agent 并把结果带回来"时，把子 Agent 的调用封装成父 Agent 的一个工具。参考给出的完整示例：

from pydantic_ai import Agent, RunContext parent = Agent('openai:gpt-5.2') researcher = Agent('openai:gpt-5.2', output_type=str) @parent.tool async def research(ctx: RunContext[None], topic: str) -> str: result = await researcher.run(f'Research: {topic}', usage=ctx.usage) return result.output

对这段示例可以补充几点实现层面的观察：

• 委托即工具调用：research以@parent.tool装饰后，父 Agent 在自身运行循环中"看见"的是一个普通函数工具；子 Agent 的完整运行（researcher.run）发生在工具执行期间，其output作为工具返回值交还给父 Agent。这样父 Agent 始终掌握对话走向，子 Agent 的输出只是父级决策的输入。
• usage=ctx.usage的传递：示例把父级上下文的用量对象透传给子 Agent 的run。从源码结构看，这是为了把父子两次运行的 token/请求计数汇入同一统计口径——在多跳委托中，这是控制成本与避免超限的实用做法。
• output_type=str的作用：子 Agent 显式声明纯文本输出，保证result.output是字符串，可安全地作为父级工具返回值。参考 SKILL.md 中的注意事项：output_type的 union 中若包含str（或未设置output_type），模型可以用纯文本来结束运行——在委托场景中，子 Agent 用str是刻意保持"只返回一段文本"的简单契约。

原文档同时给出了委托之外的两种"让出控制权"的切分准则（Good split）：

• delegation via tools：父级保留控制时，用工具委托；
• output functions 或 programmatic hand-off：控制权应当转移到其他位置时，用输出函数或程序化交接。

仓库内的 ARCHITECTURE.md 提供了一个与之对应的决策树，可以把它当作本文档"Good split"的扩展判据：

Child agent returns result to parent? ├── Yes → Use agent delegation via tools └── No → Permanent hand-off to specialist? ├── Yes → Use output functions └── Application code between agents? ├── Yes → Use programmatic hand-off └── Complex state machine? └── Yes → Use Graph-based control

即：子结果要回到父级 → 工具委托；控制权永久移交专家 Agent → 输出函数；Agent 之间需要插入应用代码 → 程序化交接；出现复杂状态机 → 进入下一节的图控制。

用 pydantic_graph 构建多步工作流：状态机优于单 Agent 循环

当工作流的本质是状态机而非单一 Agent 循环时，参考文档建议使用pydantic_graph。示例是一个双节点互相推进、达到阈值后以End收尾的计数器：

from dataclasses import dataclass from pydantic_graph import BaseNode, End, Graph, GraphRunContext @dataclass class FirstNode(BaseNode[None, None, int]): value: int async def run(self, ctx: GraphRunContext) -> 'SecondNode | End[int]': if self.value >= 5: return End(self.value) return SecondNode(self.value + 1) @dataclass class SecondNode(BaseNode): value: int async def run(self, ctx: GraphRunContext) -> FirstNode: return FirstNode(self.value) graph = Graph(nodes=[FirstNode, SecondNode]) result = graph.run_sync(FirstNode(0))

从该示例的结构可以读出pydantic_graph的核心约定：

• 节点即 dataclass：每个节点用@dataclass定义并继承BaseNode，节点携带自己的状态字段（这里的value），状态随节点实例在边上传递。FirstNode(BaseNode[None, None, int])的第三个泛型参数标注了该节点"输出"的结果类型为int。
• run返回下一个节点：节点的async def run通过返回另一个节点实例来推进图，通过返回End(self.value)来终结图并携带最终结果；返回值类型注解（SecondNode | End[int]）声明了出边。
• GraphRunContext提供运行时上下文：每个run方法都接收ctx: GraphRunContext，用于访问执行期间的上下文信息。
• Graph(nodes=[...])注册节点集合，graph.run_sync(FirstNode(0))从起始节点同步驱动整个图；异步场景则对应run的异步入口。

这类模式适合"步骤之间需要显式状态、回退与终止条件"的流程。结合上一节决策树的末端分支（"Complex state machine? → Use Graph-based control"），可以把它与委托模式划清边界：委托解决的是"父子协作"，图解决的是"控制流"。

不使用 Agent 直连模型：Direct API

当只需要一次模型请求、不需要工具调用、重试或 Agent 循环状态时，参考文档建议使用 direct API：

from pydantic_ai import ModelRequest from pydantic_ai.direct import model_request_sync response = model_request_sync( 'openai:gpt-5.2', [ModelRequest.user_text_prompt('Summarize this in one sentence.')], )

要点：

• model_request_sync接收模型字符串（沿用 SKILL.md 速查表中的provider:model-name约定，如openai:gpt-5.2）与一个请求列表；ModelRequest.user_text_prompt(...)构造用户文本提示，返回列表形式意味着可以批量组织多条请求消息。
• 参考文档给出的使用判据非常明确：没有工具、没有重试、没有 Agent 循环状态需求时才走这条路；一旦需要这些能力，应回到Agent抽象。

以 A2A 协议把 Agent 暴露为 HTTP 服务

当 Agent 需要被其他系统以服务化方式调用时，参考文档建议使用 A2A（Agent-to-Agent）集成：agent.to_a2a()会把 Agent 暴露为一个说 A2A 协议的 ASGI 应用：

from pydantic_ai import Agent agent = Agent('openai:gpt-5.2') app = agent.to_a2a()

从示例结构看，to_a2a()的产物app是标准 ASGI 应用对象，因此可以用任意 ASGI 服务器（如 uvicorn）承载与部署，无需手写协议层。这与 Docling 自身的 API Server 文档 描述的服务化思路是同构的：Docling 通过docling-serve把转换能力暴露为 REST 服务，而 Pydantic AI 通过 A2A 把"Agent"这一更高层抽象暴露给其他 Agent 调用，两者分别对应"工具服务化"与"智能体服务化"两个层次。

持久化执行（Durable Execution）：让运行跨越崩溃与长时任务

对于必须存活于崩溃、重试或长生命周期工作流的运行，参考文档建议使用持久化执行集成，并给出 Temporal 的三个入口：

• TemporalAgent
• PydanticAIWorkflow
• PydanticAIPlugin

同时说明存在面向DBOS与Prefect的平行集成。三类 Temporal 入口从命名结构可以推断出各自的接入层次：TemporalAgent面向"把一个 Pydantic AI Agent 放进 Temporal 活动"的场景；PydanticAIWorkflow面向"在 Temporal Workflow 中编排 Agent 运行"的场景；PydanticAIPlugin则更像是 SDK 级的插件集成方式。选型时应以所用版本的官方文档为准，本文档只负责指明入口存在及其面向的问题（durable execution）。

用 Embedder 构建 RAG 检索

构建检索或语义搜索时，参考文档建议直接使用Embedder生成查询/文档嵌入：

from pydantic_ai import Embedder embedder = Embedder('openai:text-embedding-3-small')

与直连模型 API 一致，Embedder同样接受provider:model-name格式的模型字符串，这里使用 OpenAI 的text-embedding-3-small。这一能力与 Docling 生态的衔接点是：Docling 负责把 PDF/Office/HTML 等文档转换为结构化的DoclingDocument（分块与序列化能力见 chunking 概念文档 与 serialization 概念文档），Pydantic AI 的Embedder负责把分块后的文本向量化，两者组合即构成完整的 RAG 数据通路；Docling 文档目录中的多个 RAG 集成示例（如 rag_langchain.ipynb、rag_llamaindex.ipynb）也印证了"Docling 产出 → 检索框架消费"是项目预设的典型链路。

接入 LangChain 或 ACI.dev 工具生态

当用户明确希望复用 LangChain 或 ACI.dev 生态的工具、而非 Pydantic AI 原生工具时，参考文档列出四个桥接入口：

• tool_from_langchain
• LangChainToolset
• tool_from_aci
• ACIToolset

从命名结构看，每个生态各提供两种粒度：tool_from_*用于把单个第三方工具转换后挂到 Agent 上，*Toolset用于把一组工具作为工具集批量接入。原文档给出的使用边界同样明确——仅当用户显式希望使用这些生态时才用它们，否则优先 Pydantic AI 原生工具。这一取向与 SKILL.md 的"Common Gotchas"一致：原生装饰器（@agent.tool/@agent.tool_plain）有严格的第一参数约定，混用会触发运行时错误，能不走桥接就不走桥接可以降低出错面。

用 pydantic_evals 系统化验证 Agent 行为

当需要可重复的评估数据集与评估器而非临时测试时，参考文档建议使用pydantic_evals，常见入口：

• Case
• Dataset
• pydantic_evals.evaluators中的各类 evaluator

即：用Case描述单个评估用例，用Dataset组织用例集合，再用evaluators中现成的评估器对 Agent 输出打分。这与 Docling 自身的测试实践是同构的——Docling 的 tests/ 目录采用"输入样本 + groundtruth 文件"的模式（如tests/data/html/下每个源文件都有对应的.json/.md/.itxt期望输出）来固化回归预期；pydantic_evals则是把同样的"数据驱动、可重复"思想搬到 LLM Agent 行为验证上，区别在于评估对象从确定性转换输出变成了模型输出。

扩展点：自建 Toolset、Model、Agent 与 Capability

参考文档最后列出 Pydantic AI 的扩展性入口，并强调只有当内置原语确实不足时才使用：

• AbstractToolset/WrapperToolset—— 自定义工具集，或以包装方式改造现有工具集行为；
• Model/WrapperModel—— 自定义模型后端，或包装既有模型（如加缓存、限流、路由）；
• AbstractAgent/WrapperAgent—— 自定义 Agent，或包装既有 Agent；
• AbstractCapability—— 自定义能力单元（组合工具、钩子、指令与模型设置的可复用行为包）。

这组入口与 ARCHITECTURE.md 中"Choosing How to Extend Agent Behavior"决策树的结论闭环：跨 Agent 复用行为 → 子类化AbstractCapability；仅拦截生命周期事件 → 用Hooks能力；从配置文件定义 Agent →Agent.from_file()；单纯加工具 →@agent.tool或 Toolset。也就是说，扩展点是为"原语不够"准备的最后手段。

落地前提与延伸阅读

汇总本文各节引用的仓库证据，便于读者继续深入：

最后重申适用前提与限制：

1. 本文所有代码示例均出自该技能参考文档，目标环境为Python 3.10+（见 SKILL.md front matter 的compatibility声明）；
2. 模型字符串需带 provider 前缀（openai:gpt-5.2而非gpt-5.2），否则无法解析 provider——这是 SKILL.md 列出的高频错误之一；
3. pydantic_ai、pydantic_graph、pydantic_evals及其 A2A、Temporal/DBOS/Prefect 集成均为独立组件，需按所用版本单独安装与核对 API 细节；本文对TemporalAgent/PydanticAIWorkflow/PydanticAIPlugin三者分工的描述属于基于命名的推断，落地前请以对应版本文档为准。

【免费下载链接】doclingGet your documents ready for gen AI项目地址: https://gitcode.com/GitHub_Trending/do/docling