Agent 工程体系 · 第 47/98 篇。内容以 2026 年 9 月可验证的公开规范和稳定接口为基线；框架版本敏感能力会明确标注，不把实验行为写成通用保证。

PydanticAI：类型化依赖、工具、结构化输出、图和测试

PydanticAI 是一个面向 Python 的 Agent SDK。它把 Agent 建模为一组可组合的运行时组件：模型、指令、工具、结构化输出、依赖和可选能力；Agent 可以通过异步运行、同步运行、流式运行或图迭代等方式执行。其核心目标不是隐藏模型调用，而是让模型输入、工具参数、依赖对象和最终结果都尽可能具有明确的 Python 类型。(pydantic.dev)

本文以 2026 年 9 月的 Agent 工程基线为范围，重点解释以下机制：

• Agent 与 Agent Runtime 的生命周期；
• 类型化依赖与 RunContext；
• 函数工具、工具参数校验和工具错误；
• 结构化输出、JSON Schema、输出工具和输出校验；
• ModelRetry 驱动的修复流程；
• Pydantic Graph 的节点、状态、决策、并行与汇聚；
• TestModel、FunctionModel、消息断言和真实模型测试边界；
• 与 Microsoft AutoGen 的抽象层次差异。

文中的 API 以官方文档当前展示的接口为准。生产项目仍应锁定 pydantic-ai、pydantic-graph、模型适配器和 Python 版本，并在升级时重新运行兼容性测试。

---

一、先建立正确的心智模型：Agent 不是一个函数

普通函数通常可以表示为：

$$f : X \rightarrow Y$$

输入是 X，返回是 Y。如果函数内部需要数据库、HTTP 客户端或配置，可以通过参数显式传入。

Agent 的一次运行则更接近：

$$R = \operatorname{Loop}(M, P, D, T, O, U)$$

其中：

• $M$：模型及其配置；
• $P$：系统提示词、动态指令和对话历史；
• $D$：本次运行的依赖对象；
• $T$：模型可调用的函数工具；
• $O$：最终输出类型；
• $U$：使用量、超时、重试等运行限制；
• $R$：包含最终输出、消息历史和使用量信息的运行结果。

一次典型运行不是“调用模型一次然后返回文本”，而是一个循环：

1. 构造模型请求；
2. 模型返回文本、工具调用或最终输出调用；
3. 如果是工具调用，校验工具参数；
4. 执行 Python 工具；
5. 把工具结果追加到消息历史；
6. 再次请求模型；
7. 对最终输出做类型解析和业务校验；
8. 成功返回，或进入重试、失败和取消路径。

sequenceDiagram
    participant App as 应用代码
    participant Agent as PydanticAI Agent
    participant Model as LLM
    participant Tool as Python 工具
    participant Validator as 输出校验器

    App->>Agent: run(prompt, deps)
    Agent->>Model: system prompt + tools + output schema
    Model-->>Agent: tool call / final output

    alt 工具调用
        Agent->>Agent: 校验工具参数
        Agent->>Tool: 执行工具(ctx, arguments)
        Tool-->>Agent: 工具结果
        Agent->>Model: 追加 tool return
        Model-->>Agent: final output / another tool call
    else 最终输出
        Agent->>Validator: Pydantic 解析与业务校验
        alt 校验成功
            Validator-->>Agent: typed output
        else 可修复错误
            Validator-->>Agent: ModelRetry
            Agent->>Model: 追加修复提示
        end
    end

    Agent-->>App: RunResult[Output]

PydanticAI 的 Agent 可以被理解为一个配置容器，里面包含模型、指令、工具、结构化输出类型、依赖类型和模型设置。Agent 本身是泛型对象，例如依赖类型为 Deps、输出类型为 Output 时，可以抽象为：

Agent[Deps, Output]

这种泛型主要服务于 IDE 和静态类型检查，并不意味着模型推理本身变成了静态过程。模型仍然可能产生错误内容，只是错误会在工具参数解析、输出解析或业务校验阶段被明确暴露。(pydantic.dev)

---

二、类型化依赖：把运行时资源与模型上下文分开

2.1 依赖是什么

PydanticAI 中的依赖是应用运行时提供给 Agent 的对象。它们通常包括：

• 数据库连接；
• HTTP 客户端；
• 当前用户身份；
• 租户信息；
• 权限检查器；
• 配置；
• 时间服务；
• 外部 API 客户端；
• 测试替身。

依赖不是模型可以直接调用的工具，也不是自动注入到提示词中的文本。依赖是 Python 侧的运行时对象，通过 RunContext 传给动态系统提示词、工具和输出校验器。官方文档规定，依赖可以是任意 Python 类型；依赖较多时，使用 dataclass 作为容器通常更清晰。(pydantic.dev)

例如，一个客服 Agent 需要当前用户 ID、数据库连接和权限检查器：

from dataclasses import dataclass
from typing import Protocol

from pydantic_ai import Agent, RunContext


class CustomerDB(Protocol):
    async def get_name(self, customer_id: int) -> str: ...
    async def get_balance(self, customer_id: int) -> float: ...


class PermissionChecker(Protocol):
    def can_view_balance(self, customer_id: int) -> bool: ...


@dataclass
class SupportDeps:
    customer_id: int
    db: CustomerDB
    permissions: PermissionChecker

然后在创建 Agent 时声明依赖类型：

support_agent = Agent[SupportDeps, str](
    "test",
    deps_type=SupportDeps,
    instructions="回答客户问题，但不要泄露无权限访问的数据。",
)

这里的 deps_type=SupportDeps 是类型约束声明，不是把某个依赖实例放进 Agent。真正的依赖实例会在每次运行时通过 deps= 传入。官方文档明确区分了这两个阶段：Agent 构造函数接收依赖类型，运行方法接收依赖对象。(pydantic.dev)

deps = SupportDeps(
    customer_id=42,
    db=my_customer_db,
    permissions=my_permission_checker,
)

result = await support_agent.run(
    "我的余额是多少？",
    deps=deps,
)

2.2 RunContext 是依赖的访问边界

需要依赖的工具必须把 RunContext 作为第一个参数：

@support_agent.tool
async def customer_balance(ctx: RunContext[SupportDeps]) -> str:
    """查询当前客户余额。"""
    customer_id = ctx.deps.customer_id

    if not ctx.deps.permissions.can_view_balance(customer_id):
        return "当前账户无权查看余额。"

    balance = await ctx.deps.db.get_balance(customer_id)
    return f"当前余额为 {balance:.2f} 元。"

RunContext[SupportDeps] 有两个作用：

1. 运行时通过 ctx.deps 访问本次运行的依赖；
2. 静态类型检查器知道 ctx.deps 必须是 SupportDeps。

如果把依赖类型声明为 SupportDeps，却错误地写成：

@support_agent.tool
async def bad_tool(ctx: RunContext[str]) -> str:
    return ctx.deps.upper()

那么工具实现与 Agent 的依赖契约不一致。运行时可能尚未执行到这里，但 pyright 或 mypy 应该在开发阶段报告问题。

依赖也可以用于动态系统提示词：

@support_agent.system_prompt
async def customer_context(ctx: RunContext[SupportDeps]) -> str:
    name = await ctx.deps.db.get_name(ctx.deps.customer_id)
    return (
        f"当前客户姓名是 {name}。"
        "只有在权限检查通过后才能回答余额问题。"
    )

动态系统提示词适合放置每次运行都会变化、但需要作为模型上下文提供的信息；数据库连接本身不应该序列化到提示词中。依赖系统的边界正是：

Python 运行时对象
        │
        ├── 动态系统提示词：转换为受控文本
        ├── 工具：执行真实动作
        └── 输出校验器：检查最终结果

官方文档将依赖的使用范围定义为系统提示词、工具和输出校验器，而不是让模型直接持有这些对象。(pydantic.dev)

2.3 依赖与全局变量的区别

不推荐这样写：

db = create_database()
agent = Agent("openai:some-model")

@agent.tool_plain
def get_balance(customer_id: int) -> float:
    return db.get_balance(customer_id)

这个写法的问题不是“不能运行”，而是：

• 测试时难以替换 db；
• 多租户场景下容易误用错误连接；
• 依赖生命周期无法绑定到一次 Agent Run；
• 工具签名没有表达它需要数据库；
• 并发运行时难以隔离请求上下文。

更合理的方式是：

agent = Agent[SupportDeps, str](
    "openai:some-model",
    deps_type=SupportDeps,
)

@agent.tool
async def get_balance(ctx: RunContext[SupportDeps]) -> float:
    return await ctx.deps.db.get_balance(ctx.deps.customer_id)

这不是为了增加抽象层，而是为了让“本次运行使用了哪个数据库、哪个用户、哪个权限上下文”成为显式数据流。

2.4 异步依赖与资源生命周期

如果依赖包含异步客户端，应在请求边界或应用生命周期边界管理资源：

import httpx
from dataclasses import dataclass

from pydantic_ai import Agent, RunContext


@dataclass
class AppDeps:
    client: httpx.AsyncClient
    api_key: str


agent = Agent[AppDeps, str](
    "test",
    deps_type=AppDeps,
)


@agent.tool
async def fetch_status(ctx: RunContext[AppDeps], service: str) -> str:
    response = await ctx.deps.client.get(
        f"https://status.example.test/{service}",
        headers={"Authorization": f"Bearer {ctx.deps.api_key}"},
    )
    response.raise_for_status()
    return response.text

调用时：

async with httpx.AsyncClient(timeout=5.0) as client:
    deps = AppDeps(client=client, api_key="secret")
    result = await agent.run("查询支付服务状态", deps=deps)

这里的因果关系是：

1. AsyncClient 由应用代码创建；
2. Agent 只接收它，不负责猜测其来源；
3. 工具通过 ctx.deps 使用它；
4. async with 负责关闭资源；
5. 工具中的网络错误可以沿 Agent Runtime 的错误路径向上传递。

依赖注入不会自动解决连接池耗尽、超时、重试、幂等和熔断问题。它只是把这些资源以可测试、可类型检查的方式送入 Agent。

---

三、工具：模型可以请求，但 Python 决定是否执行

3.1 工具的语义

函数工具是模型可以调用的 Python 函数，用来：

• 查询外部信息；
• 执行数据库或业务操作；
• 调用 HTTP API；
• 计算确定性结果；
• 把不适合放进提示词的逻辑移出模型。

PydanticAI 支持三种常见注册方式：

• @agent.tool：工具需要访问 RunContext；
• @agent.tool_plain：工具不需要 Agent 上下文；
• 通过 Agent(..., tools=...) 传入函数或 Tool 对象。

工具集合还可以通过 toolset 统一注册。(pydantic.dev)

最小示例：

from pydantic_ai import Agent, RunContext

agent = Agent(
    "test",
    deps_type=str,
    instructions="你是一个猜数字游戏。先获取玩家姓名，再使用骰子结果回答。",
)


@agent.tool_plain
def roll_dice() -> int:
    """掷一个六面骰子。"""
    import random
    return random.randint(1, 6)


@agent.tool
def player_name(ctx: RunContext[str]) -> str:
    """返回当前玩家姓名。"""
    return ctx.deps

调用：

result = agent.run_sync(
    "我猜是 4",
    deps="Anne",
)

工具函数的名称、参数类型和 docstring 会共同影响模型看到的工具定义。官方文档说明，工具参数会在 Python 函数执行前被校验。(ai.pydantic.dev)

3.2 工具参数的校验顺序

假设工具定义为：

from datetime import date

@agent.tool
async def weather(
    ctx: RunContext[AppDeps],
    location: str,
    forecast_date: date,
) -> str:
    """查询指定地点和日期的天气。"""
    return await ctx.deps.weather_service.get(
        location,
        forecast_date,
    )

模型可能生成：

{
  "location": "Hangzhou",
  "forecast_date": "2026-09-02"
}

Agent Runtime 的处理顺序大致是：

1. 从模型响应中提取工具名；
2. 根据工具 schema 解析参数；
3. 把 "2026-09-02" 转换为 datetime.date；
4. 如果解析失败，工具函数不会执行；
5. 参数合法后，才调用 weather(...)；
6. 将返回值作为工具结果加入消息历史。

因此，工具参数校验与业务校验是两个层次：

参数校验：
    forecast_date 是否能解析成 date
    location 是否是字符串

业务校验：
    location 是否在服务覆盖范围内
    用户是否有权限查询
    该日期是否允许查询

Pydantic 能保证第一类约束，但不能自动知道你的业务规则。

3.3 工具返回值不是最终输出

工具返回值通常会被送回模型，模型再基于它继续推理：

模型：调用 weather(location="杭州", date="2026-09-02")
工具：返回 {"temperature": 29, "rain": true}
模型：根据工具结果生成最终答案

这与输出函数不同。函数工具是中间步骤；输出函数或结构化输出工具表示 Agent Run 的结束。官方文档明确区分了普通 function tool 和最终 output function：前者用于获取信息或执行动作，后者用于把一次运行结束在函数结果上。(pydantic.dev)

3.4 工具失败：错误、修复和副作用

一个工具可能失败：

@agent.tool
async def charge_card(
    ctx: RunContext[SupportDeps],
    amount: float,
) -> str:
    """扣款。金额必须为正数。"""
    if amount <= 0:
        raise ValueError("amount must be positive")

    return await ctx.deps.payment.charge(
        customer_id=ctx.deps.customer_id,
        amount=amount,
    )

这里要区分三种错误：

参数无法解析

例如模型传入：

{"amount": "not-a-number"}

这是结构错误，工具不会正常执行。

工具内部的业务异常

例如账户被冻结、支付网关超时或权限不足。这些通常应转化为安全的工具返回值或明确异常，而不是把底层堆栈直接暴露给模型。

已产生副作用后的失败

例如扣款已经成功，但模型请求超时，Runtime 不知道是否成功，再次重试可能造成重复扣款。

因此，涉及支付、发货、删除、发邮件等副作用工具时，必须额外设计：

• 幂等键；
• 请求状态查询；
• 明确的“已提交但未知结果”状态；
• 人工确认；
• 重试白名单。

类型化工具只能约束输入和输出形状，不能自动保证业务操作的幂等性。

---

四、结构化输出：从文本约定升级为可验证数据

4.1 结构化输出解决什么问题

如果 Agent 只返回字符串，调用方需要自己处理：

text = result.output
# 猜测 text 中是否包含 JSON
# 解析 JSON
# 检查字段
# 检查枚举值
# 处理缺失字段

结构化输出则把目标声明为 Python 类型：

from typing import Literal
from pydantic import BaseModel, Field
from pydantic_ai import Agent


class Sentiment(BaseModel):
    label: Literal["positive", "negative", "neutral"]
    score: float = Field(ge=-1, le=1)


sentiment_agent = Agent(
    "test",
    output_type=Sentiment,
)

调用后：

result = sentiment_agent.run_sync(
    "这次发布修复了我之前遇到的所有问题。"
)

sentiment: Sentiment = result.output
print(sentiment.label)
print(sentiment.score)

result.output 的类型不再是“可能包含 JSON 的字符串”，而是经过 Pydantic 解析后的 Sentiment 实例。PydanticAI 会根据输出类型生成 schema，并校验模型返回的数据。(pydantic.dev)

4.2 JSON Schema 在这里扮演什么角色

对于：

class Sentiment(BaseModel):
    label: Literal["positive", "negative", "neutral"]
    score: float = Field(ge=-1, le=1)

可以抽象出如下约束：

$$\text{Output} = \{ \text{label} \in \{\text{positive}, \text{negative}, \text{neutral}\}, -1 \leq \text{score} \leq 1 \}$$

JSON Schema 是把这些约束传递给模型适配器和 Runtime 的机器可读表示。它描述：

• 字段名称；
• 字段类型；
• 必填字段；
• 枚举；
• 数值范围；
• 嵌套对象；
• 数组元素；
• 字段说明。

但必须注意：JSON Schema 是约束描述，不是模型的数学证明。模型仍然可能生成不符合约束的内容，Runtime 需要进行解析和验证。

4.3 默认的输出工具模式

PydanticAI 默认会利用模型的工具调用能力来获得结构化输出。对于多个输出类型，通常会把每个候选类型注册为独立的输出工具，以降低单个 schema 的复杂度。(pydantic.dev)

例如：

from pydantic import BaseModel
from pydantic_ai import Agent


class Answer(BaseModel):
    text: str


class Refusal(BaseModel):
    reason: str


agent = Agent(
    "test",
    output_type=[Answer, Refusal],
)

模型最终可能选择：

{
  "text": "可以执行该操作。"
}

也可能选择：

{
  "reason": "请求缺少必要的身份信息。"
}

调用方则通过联合类型处理：

result = agent.run_sync("处理这个请求")

if isinstance(result.output, Answer):
    print(result.output.text)
else:
    print(f"拒绝：{result.output.reason}")

这里的联合类型不是“让模型随便返回两种 JSON”，而是明确声明 Agent 可以以两种终态之一结束。

4.4 标量和数组输出的包装

如果输出类型不是 JSON object，例如：

Agent("test", output_type=list[int])

Runtime 可能需要把它包装成对象形式，因为许多模型工具 schema 要求顶层是 object。官方文档说明，非 object 的输出类型会被包装为单字段对象，再由 Runtime 还原为目标 Python 类型。(pydantic.dev)

这解释了一个常见现象：模型交互层看到的 JSON 形状，未必与 Python 最终值的形状完全相同。工程代码应依赖 result.output 的类型，而不是依赖底层消息中的临时 JSON 表示。

---

五、输出校验与修复：合法不等于正确

5.1 两层校验模型

结构化输出通常有两层校验：

模型响应
   │
   ├── 结构校验：字段、类型、枚举、范围
   │       └── Pydantic 模型验证
   │
   └── 语义校验：数据库、权限、业务规则、外部状态
           └── output_validator

例如，以下输出在结构上合法：

{
  "sql_query": "DROP TABLE users"
}

如果输出模型只要求：

class SQLResult(BaseModel):
    sql_query: str

那么它可能通过 Pydantic 校验，但不符合应用的安全规则。

5.2 使用 output_validator

from pydantic import BaseModel
from pydantic_ai import Agent, ModelRetry, RunContext


class SQLResult(BaseModel):
    sql_query: str


class DatabaseDeps:
    async def explain(self, sql: str) -> None:
        # 真实实现中调用数据库 EXPLAIN
        ...


sql_agent = Agent[DatabaseDeps, SQLResult](
    "test",
    deps_type=DatabaseDeps,
    output_type=SQLResult,
    instructions="生成只读 SQL 查询。",
)


@sql_agent.output_validator
async def validate_sql(
    ctx: RunContext[DatabaseDeps],
    output: SQLResult,
) -> SQLResult:
    normalized = output.sql_query.strip().lower()

    if not normalized.startswith("select"):
        raise ModelRetry("只能生成 SELECT 查询，请修改 SQL。")

    try:
        await ctx.deps.explain(output.sql_query)
    except Exception as exc:
        raise ModelRetry(f"SQL 无法通过 EXPLAIN，请修正：{exc}") from exc

    return output

当校验器抛出 ModelRetry 时，Runtime 会把错误原因反馈给模型，让模型重新生成输出。官方文档说明，ModelRetry 会消耗输出重试预算；默认预算为 1，也可以在 Agent、单次运行或输出工具上调整。(pydantic.dev)

一个简化的状态变化是：

第 1 次：
    模型 -> {"sql_query": "DROP TABLE users"}
    Pydantic 校验：通过
    业务校验：失败
    ModelRetry("只能生成 SELECT 查询")

第 2 次：
    模型 -> {"sql_query": "SELECT * FROM users"}
    Pydantic 校验：通过
    业务校验：通过
    Agent 返回 SQLResult

5.3 ModelRetry 不是异常吞噬器

以下写法不合理：

@agent.output_validator
def bad_validator(output: Output) -> Output:
    try:
        check(output)
    except Exception:
        return output
    return output

这会把业务错误伪装成成功结果。

正确的选择应当是：

• 可由模型修复的错误：抛出 ModelRetry；
• 不可由模型修复的错误：抛出业务异常并终止；
• 外部系统暂时不可用：由应用层决定是否重试，不要无限让模型重写；
• 涉及安全边界的输出：宁可失败，也不要靠模型“自行理解”。

输出校验器可以执行异步 I/O，例如检查 SQL、查询权限或验证外部对象是否存在；但每次重试都会增加请求成本和延迟，因此必须设置上限。(pydantic.dev)

5.4 结构化输出与输出函数

输出函数适合以下场景：

• 最终动作本身就是一次函数调用；
• 不需要把函数结果再次发给模型；
• 希望把多个终态建模为不同函数或 Pydantic 类型。

例如：

from pydantic import BaseModel
from pydantic_ai import Agent, ModelRetry


class QueryRow(BaseModel):
    name: str
    country: str


def execute_query(query: str) -> list[QueryRow]:
    if query != "SELECT * FROM capital_cities":
        raise ModelRetry("当前只支持 SELECT * FROM capital_cities。")

    return [
        QueryRow(name="Amsterdam", country="Netherlands"),
        QueryRow(name="Mexico City", country="Mexico"),
    ]


class QueryFailure(BaseModel):
    explanation: str


agent = Agent(
    "test",
    output_type=[execute_query, QueryFailure],
    instructions="把用户请求转换成 SQL，并在无法执行时返回失败原因。",
)

输出函数与普通工具的关键差异是生命周期语义：普通工具执行后通常把结果送回模型；输出函数则可以直接结束 Agent Run。官方文档也提醒，不应同时把同一个输出函数注册为普通工具，否则模型会面对两个语义相近的调用入口。(pydantic.dev)

---

六、运行方式、流式输出和生命周期边界

PydanticAI 提供多种运行方式：

result = await agent.run("问题")
result = agent.run_sync("问题")

流式方式：

async with agent.run_stream("问题") as response:
    async for text in response.stream_text():
        print(text, end="")

事件方式：

async with agent.run_stream_events("问题") as events:
    async for event in events:
        print(event)

图迭代方式：

async with agent.iter("问题") as run:
    async for node in run:
        print(node)

官方文档将这些方式分别定义为完整运行、同步封装、文本或结构化输出流、事件流以及 Agent 底层图节点迭代。(pydantic.dev)

6.1 run_stream() 的一个边界

流式运行中，如果模型同时产生最终输出和额外工具调用，Runtime 可能把第一个匹配输出类型的结果视为最终输出。默认情况下，后续悬挂工具调用不一定会执行；如果需要观察所有模型事件和工具执行过程，应使用 run_stream_events() 或 iter()，而不能只依赖 stream_text()。(pydantic.dev)

因此：

• 面向用户展示文本：使用 run_stream()；
• 需要审计工具调用：使用 run_stream_events()；
• 需要控制 Agent 状态机：使用 iter()；
• 需要中途取消：设计取消后的消息历史和副作用恢复策略。

流式输出也会影响校验器。结构化输出在流式过程中可能经历多个部分结果，输出校验器应通过 ctx.partial_output 区分中间结果和最终结果，否则可能在字段尚未完整时提前失败。(pydantic.dev)

---

七、什么时候需要图：把隐式循环变成显式状态机

7.1 Agent Loop 与业务流程不是同一层

普通 Agent Loop 适合：

用户问题
  -> 模型选择工具
  -> 工具返回
  -> 模型继续
  -> 最终输出

但以下流程通常更适合显式图：

接收订单
  -> 风险评估
  -> [低风险] 自动批准
  -> [高风险] 人工审批
  -> 扣款
  -> 发货
  -> 失败补偿

这里的分支、状态、人工节点和补偿动作不应完全交给模型决定。

PydanticAI 关联的 pydantic-graph 是一个独立的异步图和状态机库，可以不依赖 pydantic-ai 单独使用。它使用类型提示定义节点和边，并支持状态、依赖、决策、广播、并行和汇聚。(pydantic.dev)

7.2 图的核心类型

一个图通常包含：

• GraphRunContext：保存图运行状态和依赖；
• BaseNode：带有 run() 方法的节点；
• End[T]：以类型 T 结束图；
• GraphBuilder：构建图；
• Graph：构建完成后执行的图对象。

节点的 run() 返回类型会参与确定可达的后继节点。图的泛型通常包括：

StateT  : 可变共享状态类型
DepsT   : 注入依赖类型
InputT  : 初始输入类型
OutputT : 最终输出类型

官方文档明确说明，GraphBuilder 正是围绕这四类类型构建的。(pydantic.dev)

7.3 一个可运行的计数图

import asyncio
from dataclasses import dataclass

from pydantic_graph import GraphBuilder, StepContext


@dataclass
class CounterState:
    value: int = 0


async def main() -> None:
    graph_builder = GraphBuilder(
        state_type=CounterState,
        output_type=int,
    )

    @graph_builder.step
    async def increment(
        ctx: StepContext[CounterState, None, None],
    ) -> int:
        ctx.state.value += 1
        return ctx.state.value

    @graph_builder.step
    async def double(
        ctx: StepContext[CounterState, None, int],
    ) -> int:
        return ctx.inputs * 2

    graph_builder.add(
        graph_builder.edge_from(graph_builder.start_node)
        .to(increment),
        graph_builder.edge_from(increment)
        .to(double),
        graph_builder.edge_from(double)
        .to(graph_builder.end_node),
    )

    graph = graph_builder.build()

    state = CounterState()
    result = await graph.run(state=state)

    print(result)
    print(state.value)


if __name__ == "__main__":
    asyncio.run(main())

预期输出：

2
1

数据流是：

start
  -> increment
       state.value: 0 -> 1
       node output: 1
  -> double
       input: 1
       output: 2
  -> end
       graph output: 2

这里有两个不同的值：

• state.value 是节点间共享、可变的状态；
• result 是图最终返回值。

不能把二者混为一谈。状态适合记录流程进度、累计结果和控制信息；最终输出适合交给调用方。

7.4 决策节点

决策节点的关键是：分支结果应具有有限、可分析的类型。

from dataclasses import dataclass
from typing import Literal

from pydantic_graph import GraphBuilder, StepContext, TypeExpression


@dataclass
class DecisionState:
    path: str | None = None


async def main() -> None:
    builder = GraphBuilder(
        state_type=DecisionState,
        output_type=str,
    )

    @builder.step
    async def choose(
        ctx: StepContext[DecisionState, None, None],
    ) -> Literal["left", "right"]:
        return "left"

    @builder.step
    async def left(
        ctx: StepContext[DecisionState, None, object],
    ) -> str:
        ctx.state.path = "left"
        return "went left"

    @builder.step
    async def right(
        ctx: StepContext[DecisionState, None, object],
    ) -> str:
        ctx.state.path = "right"
        return "went right"

    builder.add(
        builder.edge_from(builder.start_node).to(choose),
        builder.edge_from(choose).to(
            builder.decision()
            .branch(
                builder.match(
                    TypeExpression[Literal["left"]]
                ).to(left)
            )
            .branch(
                builder.match(
                    TypeExpression[Literal["right"]]
                ).to(right)
            )
        ),
        builder.edge_from(left, right).to(builder.end_node),
    )

    graph = builder.build()
    state = DecisionState()

    result = await graph.run(state=state)

    print(result)
    print(state.path)


asyncio.run(main())

决策过程不是：

模型说走左边，所以走左边

而是：

choose() 返回 Literal["left", "right"]
       ↓
Graph 决策节点按类型表达式匹配
       ↓
进入 left 或 right

Pydantic Graph 的决策构建器支持按类型和谓词匹配分支。官方示例使用 Literal 返回有限分支，再通过 TypeExpression 连接到不同节点。(pydantic.dev)

7.5 并行执行与 Reducer

如果多个分支互不依赖，可以并行执行：

from dataclasses import dataclass

from pydantic_graph import (
    GraphBuilder,
    StepContext,
    reduce_list_append,
)


@dataclass
class EmptyState:
    pass


async def main() -> None:
    builder = GraphBuilder(
        state_type=EmptyState,
        output_type=list[int],
    )

    @builder.step
    async def source(
        ctx: StepContext[EmptyState, None, None],
    ) -> int:
        return 10

    @builder.step
    async def add_one(
        ctx: StepContext[EmptyState, None, int],
    ) -> int:
        return ctx.inputs + 1

    @builder.step
    async def add_two(
        ctx: StepContext[EmptyState, None, int],
    ) -> int:
        return ctx.inputs + 2

    @builder.step
    async def add_three(
        ctx: StepContext[EmptyState, None, int],
    ) -> int:
        return ctx.inputs + 3

    collect = builder.join(
        reduce_list_append,
        initial_factory=list[int],
    )

    builder.add(
        builder.edge_from(builder.start_node).to(source),
        builder.edge_from(source).to(add_one, add_two, add_three),
        builder.edge_from(add_one, add_two, add_three).to(collect),
        builder.edge_from(collect).to(builder.end_node),
    )

    graph = builder.build()
    result = await graph.run(state=EmptyState())

    print(sorted(result))


asyncio.run(main())

三个并行节点分别产生：

add_one   -> 11
add_two   -> 12
add_three -> 13

Reducer 将它们汇聚为：

[11, 12, 13]

官方文档将广播、spread、join 和 reducer 作为 Graph Builder 的基本并行构件；广播会把同一个输入发送到多个节点，并行节点完成后再通过 join 汇聚。(pydantic.dev)

并行并不意味着可以随意共享可变状态。若多个节点同时修改同一个 state 字段，就必须明确：

• 写入是否互斥；
• 写入顺序是否重要；
• 失败时是否部分提交；
• 汇聚是否幂等；
• 是否需要 reducer 而不是共享变量。

对于外部副作用，通常应把“并行计算”和“并行提交”分开。先并行读取或计算，再由单独节点按明确策略提交，故障路径更容易推理。

---

八、Agent 与 Graph 的组合方式

PydanticAI Agent 本身也可以被看作一个有内部节点的运行循环；Graph 则把业务流程层显式化。一个常见组合是：

flowchart TD
    A[接收用户请求] --> B[路由节点]
    B -->|知识查询| C[PydanticAI 检索 Agent]
    B -->|订单操作| D[PydanticAI 订单 Agent]
    C --> E[结果校验]
    D --> F[人工审批]
    F --> G[执行副作用]
    E --> H[统一结构化响应]
    G --> H

适合交给 Agent 的部分：

• 从自然语言识别用户意图；
• 选择查询工具；
• 生成候选结构化结果；
• 根据检索内容组织回答。

适合交给 Graph 的部分：

• 哪些状态可以转移；
• 哪些操作必须人工审批；
• 失败后走哪个补偿节点；
• 哪些步骤可以并行；
• 哪些副作用只能执行一次。

不要把所有业务控制流都写成：

“请模型决定下一步该做什么”

更可靠的方式是：

Graph 决定允许的状态转移；
Agent 在某个节点内部完成受约束的语言任务。

这样可以形成两层约束：

$$\text{系统行为} = \text{Graph 状态约束} \cap \text{Agent 输入输出约束}$$

Graph 限制“能不能走到这个节点”，结构化输出限制“节点内返回什么形状”，工具权限限制“能执行什么动作”。

---

九、测试：先测运行时，再测模型能力

Agent 测试不能只断言最终文本，因为文本变化可能来自：

• 模型版本；
• provider；
• 温度或采样配置；
• 工具调用顺序；
• 提示词细微变化；
• 输出格式变化；
• 非确定性。

PydanticAI 官方提供了 TestModel 和 FunctionModel。前者用于快速离线测试工具和输出类型，后者允许测试代码精确控制模型如何调用工具、如何返回最终响应。(pydantic.dev)

9.1 TestModel：测试工具链路

一个简单 Agent：

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


class Weather(BaseModel):
    location: str
    temperature: int
    summary: str


class WeatherDeps:
    async def query(self, location: str) -> dict:
        return {
            "location": location,
            "temperature": 28,
            "summary": "sunny",
        }


weather_agent = Agent[WeatherDeps, Weather](
    "openai:some-model",
    deps_type=WeatherDeps,
    output_type=Weather,
    instructions="查询天气并返回结构化结果。",
)


@weather_agent.tool
async def weather(
    ctx: RunContext[WeatherDeps],
    location: str,
) -> dict:
    """查询天气。"""
    return await ctx.deps.query(location)

测试时覆盖模型：

import pytest

from pydantic_ai import models
from pydantic_ai.models.test import TestModel


pytestmark = pytest.mark.anyio
models.ALLOW_MODEL_REQUESTS = False


async def test_weather_agent() -> None:
    deps = WeatherDeps()

    with weather_agent.override(model=TestModel()):
        result = await weather_agent.run(
            "查询杭州天气",
            deps=deps,
        )

    assert isinstance(result.output, Weather)
    assert result.output.location
    assert isinstance(result.output.temperature, int)

TestModel 不是一个小型语言模型。它不会真正理解“杭州天气”并作出高质量推理，而是根据工具和输出 schema 生成尽量满足类型约束的数据。官方文档特别强调，它没有机器学习逻辑，生成的数据通常不具备真实业务相关性。(pydantic.dev)

因此，TestModel 适合验证：

• Agent 是否正确注册工具；
• 工具参数是否可解析；
• 依赖是否成功注入；
• 输出模型是否能构造；
• 结果是否能写入数据库；
• 运行时是否错误调用真实模型。

它不适合证明：

• 模型是否正确理解用户意图；
• 模型是否选择了最优工具；
• 提示词是否具有良好效果；
• 真实 provider 是否支持目标输出模式。

9.2 使用 ALLOW_MODEL_REQUESTS=False 防止测试泄漏

测试中设置：

models.ALLOW_MODEL_REQUESTS = False

可以在全局层面阻止意外调用非测试模型。官方示例把它作为测试安全措施，再通过 agent.override(model=TestModel()) 替换具体 Agent 的模型。(pydantic.dev)

这两个机制解决不同问题：

ALLOW_MODEL_REQUESTS = False
    防止测试误发真实请求

Agent.override(...)
    在不修改业务调用点的情况下替换模型、依赖或工具集

如果只使用 override 而没有全局保护，某个未覆盖的 Agent 仍可能访问真实模型。

9.3 FunctionModel：精确控制工具调用

当测试需要覆盖特定分支时，TestModel 往往不够。例如工具逻辑根据日期分流：

@weather_agent.tool
async def weather(
    ctx: RunContext[WeatherDeps],
    location: str,
    forecast_date: date,
) -> str:
    if forecast_date < date.today():
        return await ctx.deps.query_history(location, forecast_date)
    return await ctx.deps.query_forecast(location, forecast_date)

如果 TestModel 总是生成过去的日期，就无法测试未来天气分支。

可以使用 FunctionModel：

from datetime import date

from pydantic_ai import ModelMessage, ModelResponse, TextPart, ToolCallPart
from pydantic_ai.models.function import AgentInfo, FunctionModel


def model_behavior(
    messages: list[ModelMessage],
    info: AgentInfo,
) -> ModelResponse:
    if len(messages) == 1:
        return ModelResponse(
            parts=[
                ToolCallPart(
                    "weather",
                    {
                        "location": "Hangzhou",
                        "forecast_date": "2032-01-01",
                    },
                )
            ]
        )

    tool_return = messages[-1].parts[0]

    return ModelResponse(
        parts=[
            TextPart(
                f"工具返回：{tool_return.content}"
            )
        ]
    )


async def test_future_weather() -> None:
    with weather_agent.override(
        model=FunctionModel(model_behavior)
    ):
        result = await weather_agent.run(
            "查询 2032-01-01 的杭州天气",
            deps=WeatherDeps(),
        )

    assert "工具返回" in result.output

FunctionModel 的关键能力是：测试函数可以读取完整消息历史和 AgentInfo，然后主动决定返回工具调用还是最终文本。官方文档使用它测试特定日期、特定工具参数和多轮工具交互。(pydantic.dev)

9.4 测试消息历史，而不仅是最终结果

对于 Agent，以下断言通常比“最终文本完全相等”更稳定：

assert result.output.location == "Hangzhou"
assert "weather" in called_tools
assert saved_record.status == "success"

如果需要检查工具调用，可以使用消息捕获机制：

from pydantic_ai import capture_run_messages

with capture_run_messages() as messages:
    with weather_agent.override(model=TestModel()):
        await weather_agent.run(
            "查询天气",
            deps=WeatherDeps(),
        )

assert messages

消息级测试可以验证：

• 是否调用了正确工具；
• 工具参数是否符合预期；
• 工具返回是否被追加；
• 是否发生了重试；
• 最终输出是否在工具结果之后生成。

官方测试示例会捕获完整模型消息，并对 ToolCallPart、ToolReturnPart、请求和响应进行断言。(pydantic.dev)

9.5 测试原生工具的边界

TestModel 不能模拟由 provider 执行的原生工具，例如某些模型内置的搜索或代码执行能力。官方文档建议在测试中移除这些原生工具，除非测试目标正是验证它们是否被正确传给模型。(pydantic.dev)

这意味着测试需要区分：

应用侧 Python 工具
    可以由 TestModel / FunctionModel 驱动

provider-native 工具
    需要 provider 集成测试或专门的适配器测试

不能因为 TestModel 测试通过，就认为生产模型的原生工具一定可用。

---

十、把测试分成四层

一个较完整的测试结构可以是：

第一层：纯函数和 Pydantic 模型测试

测试：

• 字段约束；
• 枚举；
• 数值范围；
• 序列化；
• 业务校验函数。

这层不需要模型。

第二层：Agent Runtime 单元测试

使用 TestModel 测试：

• 工具注册；
• 依赖注入；
• 工具参数解析；
• 结构化输出构造；
• 错误传播；
• 消息历史。

第三层：确定性流程测试

使用 FunctionModel 或固定模型响应测试：

• 工具调用顺序；
• 重试；
• 分支；
• 失败恢复；
• 并行流程；
• Graph 状态转换。

第四层：真实模型和 provider 集成测试

测试：

• provider 是否支持目标输出模式；
• schema 是否被正确转换；
• 工具调用名称是否匹配；
• 流式事件是否符合预期；
• 模型版本升级后提示词是否仍然有效。

真实模型测试应限制调用次数并记录成本。它们不是每次提交都运行的普通单元测试，而更接近兼容性和回归测试。

---

十一、与 Microsoft AutoGen 的抽象层次差异

PydanticAI 和 Microsoft AutoGen 都可以构建 Agent 应用，但核心抽象不同。

AutoGen 官方文档将其分为：

• AgentChat：用于快速构建会话式单 Agent 和多 Agent 应用；
• Core：事件驱动的可扩展多 Agent 框架；
• Extensions：与外部服务和其他库集成的扩展层。

AutoGen 的官方示例使用 AssistantAgent 和模型客户端启动一个异步任务；Core 则强调事件驱动、可扩展和分布式多 Agent 场景。(microsoft.github.io)

可以用下表理解差异：

维度	PydanticAI	AutoGen
核心切入点	类型化 Agent Runtime	会话式和事件驱动的多 Agent 系统
主要数据流	依赖 → 工具 → 结构化输出	Agent 消息、事件、协作
类型系统	强调 Python 类型、Pydantic schema、泛型	强调 Agent、消息和运行时拓扑
工具调用	函数签名和 schema 驱动	Agent/Workbench/扩展驱动
工作流	Agent Loop + Pydantic Graph	AgentChat 或 Core 事件流
测试重点	TestModel、FunctionModel、消息与输出校验	Agent 行为、消息流和运行时协作
适合场景	单 Agent、类型化数据处理、确定性业务流程	多 Agent 对话、事件驱动和分布式协作

这不是“哪个框架更好”的结论，而是边界不同：

• 如果系统核心是“输入一段自然语言，调用若干工具，返回严格 Python 类型”，PydanticAI 的类型化模型更直接；
• 如果系统核心是“多个 Agent 通过消息协作，并且需要事件驱动或分布式运行时”，AutoGen Core 的抽象更贴近问题；
• 如果既需要多 Agent 协作，又需要严格结构化输出，可以把二者视为不同层次的设计选择，而不是简单替换关系。

框架选型应先回答：

系统的主要不确定性在哪里？

如果不确定性来自数据格式和工具参数，优先强化 schema 与校验；如果不确定性来自 Agent 拓扑和协作过程，优先强化消息、事件和状态机。

---

十二、常见误解与失败表现

误解一：声明 output_type 就等于模型一定正确

错误理解：

agent = Agent("some-model", output_type=Invoice)

因此模型一定会生成正确发票。

实际保证范围是：

• Runtime 会尝试按 Invoice 解析；
• 不符合结构的结果会失败或触发重试；
• 结构合法但业务错误的结果仍可能通过。

例如：

{
  "total": 0,
  "currency": "CNY",
  "items": []
}

可能完全符合 Pydantic 类型，却不符合“发票必须包含至少一项商品”的业务规则。后者需要模型约束、字段约束或输出校验器共同处理。

误解二：工具是普通 Python 函数，所以不会有模型风险

工具代码本身可以是确定性的，但“模型什么时候调用哪个工具、传入什么参数”仍然是不确定的。

因此工具必须：

• 验证权限；
• 限制参数范围；
• 控制资源；
• 防止越权；
• 对危险动作增加确认；
• 对外部副作用实现幂等。

不能把工具权限安全寄托在 docstring 上。

误解三：TestModel 通过代表真实 Agent 通过

TestModel 主要验证 Runtime 能否正确驱动工具和结构化输出，不验证模型理解能力。它生成的数据可能与用户输入无关。(pydantic.dev)

真实模型仍需单独测试：

• 工具选择；
• 工具参数语义；
• 提示词；
• schema 兼容性；
• provider 差异；
• 输出质量。

误解四：图越复杂，系统越可靠

图能显式表达状态和控制流，但复杂图也会带来：

• 节点数量增加；
• 状态字段增多；
• 并行汇聚困难；
• 失败恢复路径变长；
• 版本迁移成本上升。

图的价值是让复杂流程可检查，而不是把所有简单任务都改写成状态机。

误解五：重试可以解决所有失败

重试适合修复：

• 缺失字段；
• 格式错误；
• 可纠正的 SQL；
• 不满足业务约束但模型可以改正的输出。

重试不适合解决：

• API 密钥失效；
• 数据库宕机；
• 权限拒绝；
• 已发生副作用但结果未知；
• 模型没有能力完成的任务。

特别是副作用工具，盲目重试可能造成重复操作。

---

十三、生产中的组合原则

一个类型化 Agent Runtime 可以用如下边界组织：

外部请求
   │
   ▼
应用层：身份、租户、超时、幂等键
   │
   ▼
Graph：状态、分支、审批、补偿
   │
   ▼
Agent：模型指令、工具选择、结构化输出
   │
   ├── RunContext：依赖和权限上下文
   ├── Tool：受控外部动作
   └── Validator：结构和业务校验
   │
   ▼
应用层：持久化、审计、响应

其中每层负责不同问题：

• 应用层负责请求边界和安全边界；
• Graph负责确定性状态转移；
• Agent负责语言理解和受约束的决策；
• Tool负责真实世界动作；
• Pydantic 模型负责数据形状；
• Validator负责跨字段和外部状态校验；
• 测试系统负责验证每一层的可重复行为。

如果把这些职责全部交给模型，系统会变成一个不可检查的长提示词；如果把所有语言任务都写成固定状态机，又会失去 Agent 对自然语言的处理能力。

PydanticAI 的核心价值正在于中间层：它不要求把模型当作普通函数，也不要求把整个系统变成自由对话，而是提供一条可以被类型、依赖、工具、输出校验和图状态共同约束的运行路径。

---

系列导航与关联阅读

• 系列入口：Agent 工程完整路线：从运行循环、记忆与协议到安全、评测和生产交付
• 上一篇：LangChain Agent：Model、Tool、Middleware、State 与适用边界
• 下一篇：Microsoft AutoGen：Agent、Team、消息、终止、运行时和扩展
• 延伸：Agent 结构化输出：JSON Schema、严格解析、修复和版本兼容
• 延伸：Python 实现 Agent Runtime：类型、异步、工具、状态和测试

官方资料

• PydanticAI Documentation
• Microsoft AutoGen Documentation

本文依据 Agent、模型、协议与框架官方资料重新梳理；正文、示例与生产清单由 WR BLOG 编写。