Python 基础体系 · 第 101/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python Agent 框架选型：原生循环、Agents SDK、LangGraph 和 PydanticAI

Agent 并不是某个固定的类，也不是“调用一次大模型”的别名。工程上，一个 Agent 通常至少包含四个部分：

1. 模型调用：向大语言模型发送上下文并接收输出；
2. 工具调用：模型决定调用 Python 函数或外部服务；
3. 状态管理：保存消息、工具结果、中间产物和业务状态；
4. 控制流：决定继续调用模型、执行工具、切换角色，还是结束。

因此，所谓“Agent 框架选型”，本质上是在选择谁来拥有这四部分的控制权。

• 原生循环：应用代码拥有全部控制权；
• OpenAI Agents SDK：SDK 接管单 Agent 运行循环、工具、交接、守卫和会话；
• LangGraph：开发者把流程建模为带状态的图，由图运行时执行；
• PydanticAI：以类型、依赖注入和结构化结果为中心，把一次 Agent 运行封装成类型安全的调用。

这四者不是简单的“从轻到重”关系。它们解决的问题边界不同，甚至可以组合使用。

---

一、先定义 Agent：模型输出不是控制流

设一次 Agent 运行的状态为：

$$S_t = (M_t, C_t, O_t, E_t)$$

其中：

• $M_t$：当前消息或模型上下文；
• $C_t$：应用依赖，例如数据库、用户身份、配置；
• $O_t$：已经产生的中间结果；
• $E_t$：错误、重试次数、审批状态等执行元数据。

模型调用可以抽象为：

$$L(M_t, C_t) \rightarrow \begin{cases} \text{Final}(x) \\ \text{ToolCall}(n, a) \\ \text{Error}(e) \end{cases}$$

其中：

• $n$ 是工具名称；
• $a$ 是工具参数；
• $x$ 是最终答案或结构化结果。

如果模型返回 ToolCall，应用就需要执行：

$$T_n(a) \rightarrow r$$

然后把工具结果 $r$ 写回上下文：

$$M_{t+1} = M_t \mathbin{+\!\!+} [\text{tool result } r]$$

再调用模型。这个过程通常称为 Agent loop，即 Agent 循环。

关键点是：模型只产生候选动作，应用或框架决定这些动作是否执行、如何执行以及执行失败后怎么办。

一个最小控制流

flowchart TD
    A[用户输入] --> B[调用模型]
    B --> C{模型输出类型}
    C -->|最终答案| D[返回结果]
    C -->|工具调用| E[校验参数]
    E --> F[执行工具]
    F --> G[写入工具结果]
    G --> B
    C -->|拒绝或错误| H[错误处理]

这张图中，真正构成 Agent 的不是“调用模型”这一个节点，而是从模型输出回到模型输入的闭环。

---

二、原生循环：把 Agent 当作一个显式状态机

原生循环是指直接使用模型 SDK，例如 OpenAI Python SDK 的 Responses API，自行处理：

• 请求和响应；
• 工具 schema；
• 工具调用分发；
• 消息历史；
• 重试和超时；
• 最终结果解析；
• 日志、指标和取消操作。

OpenAI Python SDK 当前把 Responses API 作为主要模型交互接口，同时提供同步和异步客户端、流式响应、工具调用、结构化输出和请求级配置。SDK 默认会对连接错误、408、409、429 以及部分 5xx 错误进行有限次数的重试，也允许通过 max_retries 修改行为。(github.com)

1. 一个可运行的原生循环

下面的例子实现一个“查询订单状态”的 Agent。模型可以选择调用 get_order_status，也可以直接回答。

from __future__ import annotations

import json
import os
from typing import Any

from openai import AsyncOpenAI


client = AsyncOpenAI(
    api_key=os.environ["OPENAI_API_KEY"],
    max_retries=2,
    timeout=30.0,
)

TOOLS = [
    {
        "type": "function",
        "name": "get_order_status",
        "description": "根据订单号查询订单当前状态。",
        "parameters": {
            "type": "object",
            "properties": {
                "order_id": {
                    "type": "string",
                    "description": "订单号，例如 ORD-1001",
                }
            },
            "required": ["order_id"],
            "additionalProperties": False,
        },
        "strict": True,
    }
]


async def get_order_status(order_id: str) -> dict[str, str]:
    # 实际代码中这里应访问领域服务或仓储，而不是直接把数据库暴露给模型。
    fake_orders = {
        "ORD-1001": "已发货",
        "ORD-1002": "待支付",
    }
    status = fake_orders.get(order_id)
    if status is None:
        raise ValueError(f"订单不存在: {order_id}")
    return {"order_id": order_id, "status": status}


async def execute_tool(name: str, arguments: str) -> dict[str, Any]:
    if name != "get_order_status":
        raise ValueError(f"未知工具: {name}")

    data = json.loads(arguments)
    return await get_order_status(data["order_id"])


async def run_agent(user_input: str) -> str:
    input_items: list[dict[str, Any]] = [
        {
            "role": "developer",
            "content": (
                "你是订单客服。需要订单状态时调用工具。"
                "不要猜测订单状态。"
            ),
        },
        {
            "role": "user",
            "content": user_input,
        },
    ]

    for turn in range(8):
        response = await client.responses.create(
            model="gpt-4.1-mini",
            input=input_items,
            tools=TOOLS,
            tool_choice="auto",
        )

        # 保留模型产生的所有 output item。
        # 工具调用的 call_id 需要在后续 tool output 中对应回来。
        input_items.extend(
            item.model_dump(exclude_none=True)
            for item in response.output
        )

        function_calls = [
            item
            for item in response.output
            if item.type == "function_call"
        ]

        if not function_calls:
            return response.output_text

        for call in function_calls:
            try:
                result = await execute_tool(call.name, call.arguments)
                tool_output = {
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": json.dumps(
                        result,
                        ensure_ascii=False,
                    ),
                }
            except (ValueError, json.JSONDecodeError) as exc:
                tool_output = {
                    "type": "function_call_output",
                    "call_id": call.call_id,
                    "output": json.dumps(
                        {
                            "error": "tool_failed",
                            "message": str(exc),
                        },
                        ensure_ascii=False,
                    ),
                }

            input_items.append(tool_output)

    raise RuntimeError("Agent 超过最大轮数，可能陷入工具调用循环")

运行：

import asyncio

print(asyncio.run(run_agent("帮我查一下 ORD-1001 的状态")))

可能得到：

订单 ORD-1001 当前状态为：已发货。

2. 每一步为什么成立

第一次模型请求包含用户输入和工具 schema。模型并没有直接执行 Python 函数，而是返回类似：

{
  "type": "function_call",
  "name": "get_order_status",
  "call_id": "call_123",
  "arguments": "{\"order_id\":\"ORD-1001\"}"
}

应用执行以下步骤：

1. 根据 name 查找允许的工具；
2. 对 arguments 做 JSON 解析；
3. 对参数进行业务校验；
4. 调用领域服务；
5. 使用同一个 call_id 生成 function_call_output；
6. 把工具结果重新交给模型；
7. 模型产生最终自然语言答案。

call_id 不能被业务代码随意替换，因为它承担的是一次模型工具请求与工具结果之间的关联关系。

3. 原生循环的核心优势

原生循环最重要的特征不是“代码少”，而是控制流透明。

例如，你可以明确规定：

if order.status == "已退款":
    return "退款订单不能重复发货"

if tool_call.name == "delete_user":
    raise PermissionError("当前会话不允许执行删除操作")

这些规则可以在模型调用前后执行，而不依赖提示词。

原生循环适合：

• 单 Agent；
• 少量工具；
• 流程主要是“模型—工具—模型”；
• 需要严格控制依赖方向；
• 希望将来替换模型供应商；
• 需要把重试、审计和权限放在自己的应用层。

4. 原生循环的反例

下面这种写法看起来简单，但实际上把不可控状态藏进了消息字符串：

history = f"""
用户问题：{question}
上一次工具结果：{tool_result}
请决定下一步。
"""

它的问题包括：

• 工具结果和用户内容难以区分；
• 无法可靠识别当前是否已经执行过某个副作用操作；
• 失败后无法恢复到明确步骤；
• 并发工具调用的结果顺序容易混乱；
• 消息历史和业务状态耦合。

更好的方式是把消息、业务状态和执行元数据分开：

@dataclass
class RunState:
    messages: list[dict[str, Any]]
    order_id: str | None = None
    payment_authorized: bool = False
    retry_count: int = 0

---

三、Responses、流式、结构化输出、工具和重试的关系

这些能力经常被混为一谈，但它们处在不同层次。

Responses API

Responses API 是模型交互层。它负责：

• 输入消息；
• 模型输出；
• 工具调用 item；
• 工具结果 item；
• 结构化文本；
• 流式事件。

它本身不是 Agent 编排框架。OpenAI Agents SDK 的官方文档也明确区分了这两层：Agents SDK 默认使用 Responses API，但额外提供 Agent、Runner、工具、守卫、handoff 和 session 等编排能力；如果希望自己拥有循环，就直接使用 Responses API。(openai.github.io)

流式输出

流式输出改变的是传输方式，不改变控制流语义：

stream = await client.responses.create(
    model="gpt-4.1-mini",
    input="写一句欢迎语",
    stream=True,
)

async for event in stream:
    print(event)

流式适合：

• 用户界面逐步显示文本；
• 长答案降低首字节等待；
• 实时观察工具调用事件。

但流式并不意味着可以在每个文本片段上执行副作用。只有当工具调用事件被完整解析、参数校验通过，并且应用决定允许执行时，才应真正调用工具。

因此，以下策略是危险的：

async for event in stream:
    if "delete" in str(event):
        delete_user()

文本片段可能不完整，事件也可能是普通文本、工具参数、状态通知或错误。应以 SDK 提供的结构化事件类型和完整工具调用为边界。

结构化输出

结构化输出要求模型最终结果满足一个 schema，例如：

from pydantic import BaseModel


class OrderAnswer(BaseModel):
    order_id: str
    status: str
    needs_human_review: bool

OpenAI Python SDK 提供将 Pydantic 类型转换为结构化响应格式并解析结果的能力；当前 Responses 相关解析代码支持 Pydantic 模型和部分 Pydantic v2 可适配类型。(github.com)

但结构化输出只保证格式约束，不保证事实正确。例如：

{
  "order_id": "ORD-1001",
  "status": "已签收",
  "needs_human_review": false
}

这可能完全符合 schema，但业务事实仍然是错的。因此：

• schema 负责形状；
• 领域服务负责事实；
• 权限策略负责是否允许动作；
• Agent 循环负责下一步。

重试的三种含义

“重试”至少有三种不同含义。

网络重试

由 SDK 针对临时网络或服务错误重发同一个请求。

工具重试

工具本身失败，例如数据库连接短暂断开：

for attempt in range(3):
    try:
        return await repository.get_order(order_id)
    except TemporaryDatabaseError:
        if attempt == 2:
            raise
        await asyncio.sleep(2**attempt)

模型纠错重试

工具参数或结构化输出校验失败，把错误作为上下文交给模型，让模型重新生成。

三者不能无条件叠加。假设 SDK 自动重试 3 次，应用层重试 3 次，工具内部又重试 3 次，一次失败最多可能造成：

$$3 \times 3 \times 3 = 27$$

次实际请求或副作用尝试。

对查询操作，这通常只是延迟增加；对“扣款”“发货”“删除”这类副作用操作，则可能造成重复执行。副作用工具必须使用幂等键、事务边界或人工确认。

---

四、OpenAI Agents SDK：把一次 Agent 运行交给 Runner

OpenAI Agents SDK 的核心抽象是：

• Agent：模型、指令、工具、handoff、guardrail 和输出类型；
• Runner：执行 Agent 运行；
• RunResult：保存最终输出、最后一个 Agent 和运行项；
• tool：可由模型调用的 Python 函数；
• handoff：把当前任务交给另一个 Agent；
• guardrail：在输入或输出阶段进行检查。

官方快速入门使用 Agent 定义角色，再通过 Runner.run 执行；Runner 负责处理工具调用、handoff 和多轮运行。(openai.github.io)

一个最小 Agents SDK 示例

from __future__ import annotations

import asyncio

from agents import Agent, Runner, function_tool


@function_tool
async def get_order_status(order_id: str) -> str:
    """查询订单状态。"""
    orders = {
        "ORD-1001": "已发货",
        "ORD-1002": "待支付",
    }
    try:
        return orders[order_id]
    except KeyError:
        raise ValueError(f"订单不存在: {order_id}")


agent = Agent(
    name="订单客服",
    instructions=(
        "你负责查询订单。"
        "需要状态时调用 get_order_status，不能猜测订单状态。"
    ),
    tools=[get_order_status],
)


async def main() -> None:
    result = await Runner.run(
        agent,
        "查询 ORD-1001 的状态",
    )
    print(result.final_output)
    print(result.last_agent.name)


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

工具函数的 docstring 和类型签名会参与工具定义。SDK 支持本地函数工具、OpenAI 托管工具、Agent-as-tool 等工具类型。(openai.github.io)

Agent 和 Runner 的边界

Agent = “这个角色能做什么”
Runner = “这次运行如何推进”

Agent 通常是相对稳定的配置；Runner 处理一次具体运行中的：

• 模型轮次；
• 工具执行；
• handoff；
• guardrail；
• session；
• 流式事件；
• 结果收集。

这种设计减少了手写循环的重复代码，但也意味着控制流的一部分进入 SDK。发生问题时，不能只打印最终答案，还需要检查 run items、工具调用和 trace。

Handoff 与 Agent-as-tool

假设有一个分诊 Agent 和两个专科 Agent：

history_agent = Agent(
    name="历史专家",
    handoff_description="处理历史问题",
    instructions="回答历史问题，给出必要背景。",
)

math_agent = Agent(
    name="数学专家",
    handoff_description="处理数学问题",
    instructions="逐步推导数学问题。",
)

triage_agent = Agent(
    name="分诊助手",
    instructions="判断问题类型并交给合适的专家。",
    handoffs=[history_agent, math_agent],
)

handoff 的语义是：当前 Agent 把会话控制权交给另一个 Agent。最终负责回答的 Agent 会发生变化。

Agent-as-tool 的语义是：当前 Agent 仍然是管理者，只把另一个 Agent 当作一个工具调用。适合“主 Agent 汇总多个子结果”的场景。

二者的差异不是命名差异，而是控制权差异：

模式	最终控制者	适合场景
handoff	被交接的 Agent	路由到专家后由专家继续对话
Agent-as-tool	原 Agent	管理者调用专家并综合结果

结构化输出与守卫

Agents SDK 可以通过 output_type 指定结果类型，例如 Pydantic 模型、dataclass、TypedDict 或列表类型。指定后，Agent 不再只返回普通字符串，而是要求模型生成结构化结果。(openai.github.io)

from pydantic import BaseModel
from agents import Agent


class TriageResult(BaseModel):
    category: str
    confidence: float
    needs_human: bool


triage_agent = Agent(
    name="问题分诊",
    instructions="判断用户请求属于哪一类，并识别是否需要人工处理。",
    output_type=TriageResult,
)

守卫则用于在 Agent 运行前或最终输出后检查条件。输入守卫可以拦截越权请求，输出守卫可以阻止不符合业务规则的结果；如果守卫触发，SDK 会抛出对应异常。(openai.github.io)

需要注意，Agent 级 output guardrail 不是每个工具调用前的授权系统。对于删除、付款等敏感工具，应在工具自身或工具级守卫中再次检查权限。否则，模型可能在最终输出之前已经完成了副作用。

---

五、LangGraph：把 Agent 运行建模为有状态图

LangGraph 的核心不是“更强的提示词”，而是：

用节点、边和状态表示一个可恢复的多步骤执行流程。

官方参考将 LangGraph 定义为构建有状态、多参与者语言应用的框架，并提供图、checkpoint、SQLite/PostgreSQL 持久化等组件。(reference.langchain.com)

1. 图模型

设图为：

$$G = (V, E)$$

其中：

• $V$ 是节点集合，例如 call_model、run_tool、human_review；
• $E$ 是边集合，表示节点之间的转移；
• $S$ 是共享状态；
• 条件边根据状态决定下一节点。

一个典型流程是：

stateDiagram-v2
    [*] --> call_model
    call_model --> run_tools: 存在工具调用
    call_model --> finish: 产生最终答案
    run_tools --> call_model: 工具结果写回状态
    run_tools --> human_review: 需要人工审批
    human_review --> call_model: 审批通过
    human_review --> finish: 审批拒绝
    finish --> [*]

与原生循环相比，循环边被显式表示为 run_tools -> call_model。人工审批不是隐藏在异常处理中的分支，而是一个可持久化的状态。

2. 最小 LangGraph 示例

from typing import Annotated
from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    question: str
    answer: str
    approved: bool


def call_model(state: State) -> dict[str, str]:
    # 生产代码中这里调用模型。
    return {
        "answer": f"模型建议：处理问题「{state['question']}」",
    }


def review(state: State) -> dict[str, bool]:
    # 生产代码中这里通常由人工系统异步完成。
    return {"approved": True}


def route_after_review(state: State) -> str:
    return "finish" if state["approved"] else "call_model"


def finish(state: State) -> dict[str, str]:
    return {"answer": state["answer"]}


builder = StateGraph(State)
builder.add_node("call_model", call_model)
builder.add_node("review", review)
builder.add_node("finish", finish)

builder.add_edge(START, "call_model")
builder.add_edge("call_model", "review")
builder.add_conditional_edges(
    "review",
    route_after_review,
    {
        "call_model": "call_model",
        "finish": "finish",
    },
)
builder.add_edge("finish", END)

graph = builder.compile()

result = graph.invoke(
    {
        "question": "是否可以退款？",
        "answer": "",
        "approved": False,
    }
)

print(result["answer"])

这个例子没有体现 LangGraph 的全部能力，但展示了核心机制：

1. 输入被装入 State；
2. 节点只读取状态并返回状态更新；
3. 条件函数根据状态选择后续节点；
4. 图编译后才能执行；
5. END 表示流程结束。

3. checkpoint 与恢复

LangGraph 的持久化层会把图状态保存为 checkpoint。checkpoint 按 thread 组织，因此同一个 thread_id 可以承载多次交互、暂停、恢复和历史状态。官方文档还区分了 super-step checkpoint 与节点级 pending writes：同一 super-step 中已经完成的节点写入可以在恢复时避免重复执行。(docs.langchain.com)

这解决的是原生循环中的一个困难：

try:
    await charge_card()
    await ship_order()
except Exception:
    # 这里很难判断扣款是否成功、程序失败在哪一步
    ...

图运行时可以把流程拆成：

authorize_payment
    ↓
charge_card
    ↓
ship_order
    ↓
notify_user

如果 ship_order 失败，恢复逻辑可以从已保存的状态继续，而不是从头执行扣款。

但 checkpoint 不是事务系统。它不能自动撤销已经提交的外部副作用。因此：

• 节点应尽量设计为幂等；
• 外部调用应保存幂等键；
• 恢复时必须区分“未执行”“执行成功但未写回”“执行结果未知”；
• 对未知状态不能简单重试。

4. 并发与 super-step

如果图中一个节点同时分叉到多个节点：

retrieve_a ─┐
             ├── merge
retrieve_b ─┘

两个检索节点可能并行执行，之后在下一个 super-step 汇合。并行提高吞吐，但引入两个问题：

1. 两个节点是否会写同一个状态字段；
2. 汇合时结果是否满足交换律和幂等性。

例如，下面的状态更新存在竞争风险：

state["summary"] = result_a
state["summary"] = result_b

最终结果取决于完成顺序。更可靠的状态设计是追加独立结果：

{
    "evidence": [
        {"source": "a", "text": "..."},
        {"source": "b", "text": "..."},
    ]
}

然后由明确的 merge 节点生成摘要。

Python 3.14 的 asyncio.TaskGroup 对结构化并发提供了更清晰的生命周期和异常传播语义；其任务创建接口也在 Python 3.14 有变化。(docs.python.org) 但 LangGraph 的图级并发、checkpoint 和恢复仍然是框架层语义，不能简单等同于手写 asyncio.gather()。

5. LangGraph 适合什么

LangGraph 的优势在于：

• 长流程；
• 明确的状态转换；
• 人工审批；
• 失败恢复；
• 多分支、汇合和循环；
• 确定性步骤与 Agent 步骤混合；
• 需要查看某一步之后的完整状态。

它的代价是开发者需要理解：

• 状态 schema；
• 节点边界；
• checkpoint；
• thread；
• 中断和恢复；
• 并发写入；
• 节点幂等性。

如果应用只是“用户问问题，模型偶尔调用两个查询工具”，直接引入图通常会增加认知成本。

---

六、PydanticAI：把 Agent 看作带输入、依赖和输出类型的函数

PydanticAI 的中心抽象是一个泛型 Agent：

$$\text{Agent}[D, O]$$

其中：

• $D$ 是依赖类型；
• $O$ 是输出类型。

官方文档将 Agent 描述为包含 instructions、tools、structured output、dependency type constraint 和 model 的容器；工具参数验证和结构化输出验证失败时，可以把错误交还给模型进行纠正。(pydantic.dev)

1. 依赖注入

from dataclasses import dataclass

from pydantic_ai import Agent, RunContext
from pydantic import BaseModel


class OrderResult(BaseModel):
    order_id: str
    status: str


@dataclass
class AppDeps:
    order_service: "OrderService"


class OrderService:
    async def status(self, order_id: str) -> str:
        return {
            "ORD-1001": "已发货",
            "ORD-1002": "待支付",
        }.get(order_id, "不存在")


agent = Agent(
    "openai:gpt-4.1-mini",
    deps_type=AppDeps,
    output_type=OrderResult,
    instructions="查询订单状态，不要猜测不存在的订单。",
)


@agent.tool
async def get_order_status(
    ctx: RunContext[AppDeps],
    order_id: str,
) -> str:
    """查询订单状态。"""
    return await ctx.deps.order_service.status(order_id)

运行：

import asyncio


async def main() -> None:
    deps = AppDeps(order_service=OrderService())
    result = await agent.run(
        "请查询 ORD-1001",
        deps=deps,
    )
    print(result.output)


asyncio.run(main())

这里的关键不是装饰器，而是依赖方向：

Agent → RunContext[AppDeps] → OrderService

Agent 不需要导入数据库驱动、HTTP 客户端或具体仓储实现。测试时可以注入 fake service：

class FakeOrderService:
    async def status(self, order_id: str) -> str:
        return "测试状态"

这正好对应 Python 应用架构中的“领域层”和“可替换适配器”：

接口层
  ↓
Agent 用例层
  ↓
领域服务接口
  ↓
数据库 / HTTP / 消息队列适配器

2. 结构化结果不是普通 JSON

class RefundDecision(BaseModel):
    allowed: bool
    reason: str
    amount: float

如果 Agent 的 output_type 是 RefundDecision，最终结果应当是经过 schema 验证的 Python 对象，而不是未经验证的 dict[str, Any]。

PydanticAI 默认可以通过输出工具要求模型生成结构化结果，并对输出使用独立的重试预算；工具调用和最终输出可以分别配置重试次数。(pydantic.dev)

这使下面的错误可以被区分：

工具参数错误：
    get_order_status(order_id=123)

最终输出错误：
    allowed="yes" 而不是 bool

业务错误：
    订单已过期但模型仍建议退款

前两类可以交给 schema 或模型纠正；第三类必须由领域规则处理。

3. ModelRetry 的边界

工具内部可以主动要求模型修正调用：

from pydantic_ai import ModelRetry, RunContext


@agent.tool
async def get_order_status(
    ctx: RunContext[AppDeps],
    order_id: str,
) -> str:
    """查询订单状态。"""
    status = await ctx.deps.order_service.status(order_id)

    if status == "不存在":
        raise ModelRetry(
            f"订单 {order_id} 不存在，请检查订单号后再试。"
        )

    return status

这适合“模型传错参数”的情况，不适合所有异常。

不应把数据库宕机写成 ModelRetry：

try:
    return await database.query(...)
except ConnectionError:
    raise ModelRetry("请重新生成")

数据库连接失败不是模型可以修正的内容。正确做法通常是：

• 工具层进行有限次瞬时故障重试；
• 超过预算后抛出基础设施异常；
• 由应用层返回暂时不可用或进入任务队列；
• 不消耗模型纠错预算。

---

七、四种方案的真正差异：谁拥有状态和控制流

可以从三个问题判断框架层级。

问题一：下一步由谁决定？

原生循环：
    你的 Python if/for/try 决定

Agents SDK：
    Runner 根据 Agent、tool、handoff 决定

LangGraph：
    图的边和条件路由决定

PydanticAI：
    Agent 内部运行循环决定，业务代码控制更高层调用

问题二：状态保存在哪里？

原生循环：
    变量、数据库、消息队列，由你设计

Agents SDK：
    RunResult、session、运行上下文，以及你自己的持久化

LangGraph：
    graph state + thread + checkpointer

PydanticAI：
    单次 run 的消息历史与依赖；跨运行持久化通常由应用负责

问题三：失败后从哪里恢复？

原生循环：
    由 try/except 和业务状态决定

Agents SDK：
    通常重新运行或利用 SDK 提供的运行状态能力，
    具体恢复策略仍需应用设计

LangGraph：
    从 checkpoint 或中断点恢复图状态

PydanticAI：
    在单次运行内进行工具/输出纠错重试，
    长事务恢复仍需外围架构

---

八、依赖方向：不要让框架反向污染领域层

一个可替换的 Python Agent 应用可以设计为：

app/
├── domain/
│   ├── order.py
│   └── ports.py
├── application/
│   └── order_agent.py
├── adapters/
│   ├── openai_agent.py
│   ├── langgraph_workflow.py
│   └── pydanticai_agent.py
└── api/
    └── http.py

其中：

• domain 不依赖任何 Agent 框架；
• ports.py 定义订单服务、支付服务等接口；
• application 编排业务用例；
• adapters 负责把框架工具映射到领域服务；
• api 只处理 HTTP、认证和序列化。

例如领域接口：

from typing import Protocol


class OrderReader(Protocol):
    async def get_status(self, order_id: str) -> str:
        ...

Agents SDK、LangGraph 和 PydanticAI 都可以调用同一个 OrderReader，差异只存在于适配器层。

这比把业务逻辑写进工具函数更稳定：

@function_tool
async def refund(order_id: str) -> str:
    # 不建议在这里同时写权限、库存、支付、审计和模型提示
    ...

工具应该是边界适配器，而不是领域层本身。

---

九、选型决策：按控制流复杂度，而不是按流行度

选择原生循环

满足以下条件时优先使用：

• Agent 只有一个；
• 工具数量有限；
• 没有长时间暂停；
• 没有复杂的人工审批；
• 需要严格控制每一步；
• 需要最大限度减少框架依赖。

原生循环的成本是你要自己维护工具分发、历史管理、重试、流式事件和可观测性。

选择 OpenAI Agents SDK

适合：

• 主要使用 OpenAI 模型和工具；
• 需要快速获得工具、handoff、guardrail 和 tracing；
• 多 Agent 路由结构相对直接；
• 不希望重复实现常见 Agent loop；
• 希望保留 Python 函数作为工具入口。

Agents SDK 的主要边界是：它的编排模型围绕 Agent 运行设计。如果业务流程本质上是复杂工作流，而不是对话型 Agent，图模型可能更自然。

选择 LangGraph

适合：

• 流程本身比对话更重要；
• 有人工审批、中断和恢复；
• 需要 checkpoint；
• 需要长时间运行；
• 有多个并发分支和汇合；
• Agent 与确定性业务节点混合。

LangGraph 的核心收益来自显式状态和持久化，而不是自动让模型更聪明。

选择 PydanticAI

适合：

• Python 类型是主要设计语言；
• 需要强约束的输入、依赖和输出；
• 领域服务希望通过依赖注入提供；
• 工具参数和最终结果都需要验证；
• 希望用较少的框架概念封装一次 Agent 调用；
• 需要对工具重试和输出重试分别控制。

PydanticAI 并不自动替代长事务编排。它适合类型安全的 Agent 单元；复杂恢复仍然需要数据库、任务系统或工作流框架。

---

十、一个更实用的组合方案

实际项目不必四选一。常见组合是：

LangGraph
    ├── 确定性节点：权限、数据库、人工审批
    └── Agent 节点：PydanticAI 或 Agents SDK

领域服务
    └── 通过 Protocol 注入，不依赖上述框架

例如：

START
  ↓
解析用户请求
  ↓
PydanticAI：提取 RefundRequest
  ↓
领域层：计算是否允许退款
  ↓
LangGraph：人工审批
  ↓
领域层：执行退款
  ↓
通知用户

此时：

• PydanticAI 负责“从自然语言得到可靠类型”；
• 领域层负责业务事实；
• LangGraph 负责暂停、恢复和审批；
• OpenAI SDK 负责具体模型通信；
• Agents SDK 可用于某个独立的多 Agent 子流程。

这种组合避免让一个框架同时承担所有职责。

---

十一、生产环境中的故障路径

模型返回普通文本而不是工具调用

原因可能是：

• 工具 schema 不正确；
• 工具描述不清；
• tool_choice 配置不符合模型能力；
• 模型不支持当前工具类型；
• 输入上下文让模型认为无需调用工具。

诊断时应记录：

request_id
model
工具 schema 版本
模型原始 output item 类型
最终 output_text

不要只记录最终答案，因为最终答案无法解释模型为什么没有调用工具。

工具参数能解析，但业务上无效

例如：

{"order_id": "ORD-9999"}

JSON 和 schema 都正确，但订单不存在。这是业务错误，不是 JSON 错误。

处理方式可以是：

参数格式错误 → schema 拒绝
资源不存在 → 工具返回结构化业务错误
权限不足 → 工具直接拒绝
服务暂时不可用 → 基础设施重试或转异步任务

不能把所有错误都包装成一段字符串“请重试”，否则模型无法区分错误类别。

Agent 无限循环

典型表现是：

模型调用工具 A
工具返回错误
模型再次调用工具 A
工具返回相同错误
...

至少需要两个限制：

MAX_TURNS = 8
MAX_SAME_TOOL_FAILURES = 2

同时记录工具名、规范化参数和错误指纹：

$$f = \operatorname{hash}(\text{name}, \operatorname{normalize}(args), error)$$

如果相同 $f$ 连续出现，应停止让模型继续尝试，转入人工或确定性错误路径。

流式连接中断

流式输出可能已经向用户展示了一部分文本，但最终响应并未完成。此时不能简单把“已显示文本”当成成功结果。

应用应区分：

started
  → partial_output
  → completed
  → interrupted
  → failed

如果涉及结构化输出，只有完整响应并通过验证后，才能将其作为业务结果提交。流式更适合作为展示层事件；业务提交应以最终完成事件为边界。

自动重试与副作用

假设模型第一次生成：

调用 charge_card(amount=100)

HTTP 请求超时，但支付服务实际上已经扣款。应用无法判断结果时，直接重试可能重复扣款。

正确的工具接口应包含幂等键：

async def charge_card(
    *,
    order_id: str,
    amount: int,
    idempotency_key: str,
) -> ChargeResult:
    ...

幂等性必须由支付适配器或领域服务保证，不能依赖模型“记得自己调用过”。

---

十二、测试方式：测试控制流，不只测试答案文本

Agent 的测试至少分三层。

单元测试：不调用模型

测试工具和领域规则：

async def test_refund_expired_order():
    service = RefundService(...)
    result = await service.decide(order_status="expired")
    assert result.allowed is False

合同测试：固定模型输出

给 Agent loop 注入假的模型响应：

class FakeModel:
    responses = [
        ToolCall(
            name="get_order_status",
            arguments={"order_id": "ORD-1001"},
        ),
        FinalAnswer("订单已发货"),
    ]

验证：

• 工具是否调用一次；
• 工具结果是否写回；
• call id 是否匹配；
• 错误是否终止或重试；
• 最大轮数是否生效。

集成测试：真实模型但限制副作用

真实模型测试应：

• 使用沙箱工具；
• 禁止真实扣款、删除和发货；
• 固定输入样本；
• 检查结构化结果；
• 记录模型版本和 schema；
• 对关键路径设置回归集。

模型输出测试不应只写：

assert "已发货" in answer

还应验证：

assert tool_calls == [
    {
        "name": "get_order_status",
        "arguments": {"order_id": "ORD-1001"},
    }
]

因为“答案看起来正确”不代表 Agent 按照允许的路径工作。

---

十三、最终判断标准

可以用下面的决策顺序，而不是先问“哪个框架最好”。

第一步：是否真的需要 Agent？

如果流程可以写成：

读取输入 → 调用确定性 API → 格式化结果

就不需要 Agent。

只有当下一步需要根据自然语言、非结构化上下文或动态工具选择时，才引入 Agent loop。

第二步：谁应该拥有控制流？

• 希望完全自己掌控：原生循环；
• 希望快速获得 OpenAI Agent 能力：Agents SDK；
• 流程有持久状态和恢复：LangGraph；
• 重点是类型安全的 Agent 接口：PydanticAI。

第三步：状态是否需要跨进程、跨天恢复？

如果答案是否定的，原生循环、Agents SDK 或 PydanticAI 通常足够。

如果答案是肯定的，需要进一步设计：

• thread 标识；
• 状态序列化；
• checkpoint；
• 外部副作用幂等；
• 恢复后的补偿逻辑；
• 人工审批后的继续执行。

第四步：把模型依赖锁在适配器边界

无论选择哪种方案，都不应让领域层依赖：

from agents import Agent
from langgraph.graph import StateGraph
from pydantic_ai import Agent

这些导入应停留在应用编排层或适配器层。这样未来替换 OpenAI 模型、切换 Agent 框架，或者将 Agent 改为普通工作流时，领域规则不需要重写。

Agent 框架的核心差异，最终可以归纳为一句话：

原生循环控制每一个状态变化；Agents SDK 控制一次 Agent 运行；LangGraph 控制可恢复的流程图；PydanticAI 控制类型安全的 Agent 输入、依赖和输出。

选型时应先确定业务的状态、控制流和故障恢复要求，再选择框架，而不是从框架的装饰器或示例代码开始。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Python OpenAI SDK：Responses、流式、结构化输出、工具和重试
• 下一篇：Python Tkinter GUI：事件循环、布局、控件、线程和打包
• 延伸：Python 应用架构：模块边界、依赖方向、领域层和可替换适配器

官方资料

• PyTorch Documentation
• OpenAI Python SDK

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。