--- title: "Pydantic AI" description: "Pydantic AI 用 tool、依赖和 output type 定义有类型的 Agent。Dex adapter 把 Pydantic AI Agent 放进 durable session 中运行。它支持 Python。" canonical_url: "https://docs.superdurable.io/zh-Hans/agent-frameworks/pydantic-ai/" source: "docs/i18n/zh-Hans/docusaurus-plugin-content-docs/current/agent-frameworks/pydantic-ai.mdx" locale: "zh-Hans" --- # Pydantic AI Pydantic AI 用 tool、依赖和 output type 定义有类型的 Agent。Dex adapter 把 Pydantic AI Agent 放进 durable session 中运行。它支持 Python。 ## 何时使用 Agent 已经用 Pydantic AI 编写时，使用它。保留 Agent、它的 tool 和 capability，由 Dex 管理 session。这个 adapter 取代 Pydantic AI 自带的 durable execution 集成。 ## Session 如何运行 每个 **RunModel** Step 运行 Agent 时，会加上一个 capability，把每个 function tool 标记为 external tool。Pydantic AI 从不运行 external tool，因此模型请求 tool 之后，run 立即结束。 每个 tool call 在自己的 Step 中并行运行：read 或 idempotent write 用 **CallTool**，其他写操作用 **CallSingleAttemptTool**。需要审批的调用先在 **AwaitToolApproval** 中等待。Tool Step 通过 Agent 运行 tool 一次，因此 Agent 的 tool hook 在这里运行。下一个 **RunModel** 用记录下来的结果恢复 run。 查询订单的 turn 依次运行 **RunModel**、**CallTool**、**ResumeTurn** 和 **RunModel**。 ## 安装 把 adapter 与它的 framework 一起安装。每个 adapter 都会带上 Dex Agents core。 **Python** ```bash pip install dex-agents-pydantic-ai "pydantic-ai-slim[openai]" ``` 例子: [agents/python/pydantic-ai/README.md](https://github.com/superdurable/dex/tree/main/agents/python/pydantic-ai/README.md) ## Example 这个 example 就是每个 framework 页面共用的[订单客服 Agent](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#the-examples)。 Flow type 是 **PydanticAIOrderSupport**（Python）。完整 example 位于 [examples/python/agent-harnesses](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses)。 ### Agent 与 tool Tool 是 Agent 上普通的 Pydantic AI tool。Tool policy 指定每个 tool。Agent 推迟了模型检查，因此 Client 进程不需要 provider key。 **Python** ```python def build_pydantic_ai_order_support_flow( order_desk: OrderDesk, model: str | Model, slack_connection: slack.Connection | None ) -> AgentSessionFlow: """Builds the session Flow; ``model`` is a Pydantic AI model name, such as "openai:gpt-5.2".""" # The model is resolved at the first run, so a Client process needs no provider key. agent = Agent(model, instructions=ORDER_SUPPORT_INSTRUCTIONS, defer_model_check=True) # The tools are closures, so they call the OrderDesk passed in here. @agent.tool_plain def look_up_order(order_id: str) -> str: """Look up one order by its order ID and return its items and status.""" return order_desk.describe_order(order_id) @agent.tool_plain def cancel_order(order_id: str, reason: str) -> str: """Cancel one order that has not shipped yet.""" return order_desk.cancel_order(order_id, reason) return pydantic_ai_session_flow( agent, flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), tool_policies={ "look_up_order": ToolPolicy(ToolEffect.READ), "cancel_order": ToolPolicy(ToolEffect.NON_IDEMPOTENT_WRITE), }, ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py) ### Worker 一个 Worker 为 example 中所有 harness 的 session Flow 提供服务。Agent session 的 Step 是 async 的，因此 Python Registry 允许 async handler。每个启动 session 或调用其 RPC 的进程，都要构建同一个 session Flow。 **Python** ```python # Agent session Steps are async, so the Registry allows async handlers. self.registry = Registry( tuple(example.session_flow for example in self.examples), allow_async_handlers=True ) config.blob_cache_dir.mkdir(parents=True, exist_ok=True) self.blob_cache = open_blob_cache(BlobCacheConfig(str(config.blob_cache_dir), 1 << 30)) self.worker = AsyncWorker( self.registry, self.blob_cache, WorkerOptions( server_address=config.server_address, pull=PullWorkerOptions(max_concurrent_tasks=MAX_CONCURRENT_TASKS), ), ) self.client = AsyncClient( self.registry, self.blob_cache, ClientOptions(server_address=config.server_address) ) # ... def _new_pydantic_ai_example( config: AgentHarnessExamplesConfig, slack_connection: slack.Connection | None ) -> AgentHarnessExample: order_desk = OrderDesk() model = _pydantic_ai_model(config.pydantic_ai_model) flow = build_pydantic_ai_order_support_flow(order_desk, model, slack_connection) return AgentHarnessExample("pydantic-ai", flow, order_desk) # ... def _pydantic_ai_model(model: str) -> str | PydanticAIModel: return build_pydantic_ai_demo_model() if is_demo_model(model) else model ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/app.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/app.py) ### Session route 每个 harness 都提供这同一组 route。这些 route 和 session token 的说明见 [Session route](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#session-routes)。 **Python** ```python @blueprint.post("/sessions") async def start_session() -> Response: # The server picks the session ID, so a caller cannot claim an existing session. session_id = f"{harness_name}-order-support-{uuid.uuid4()}" await sessions.start_session(session_id) session_token = sign_session_id(app_state.config.session_key, session_id) return jsonify(session_id=session_id, session_token=session_token) @blueprint.post("/messages") async def send_user_message() -> Response: body, session_id = await session_request() accepted = await sessions.send_user_message(session_id, _required_string(body, "text")) return jsonify(accepted=accepted) @blueprint.post("/interrupts/resolve") async def resolve_interrupt() -> Response: body, session_id = await session_request() interrupt_id = _required_string(body, "interrupt_id") if not isinstance(approved := body.get("approved"), bool): raise BadRequest("approved must be a boolean") rejection_message = body.get("rejection_message") if rejection_message is not None and not isinstance(rejection_message, str): raise BadRequest("rejection_message must be a string") accepted = await sessions.resolve_interrupt( session_id, interrupt_id, approved=approved, rejection_message=rejection_message ) return jsonify(accepted=accepted) @blueprint.get("/snapshot") async def get_agent_snapshot() -> Response: session_id = request.args.get("session_id", "") if not session_id: raise BadRequest("session_id is required") authorize_session(session_id) try: snapshot = await read_agent_event_stream_snapshot( app_state.client, example.session_flow, session_id ) except FlowNotActiveOrNotFoundError as error: raise NotFound("the agent session is missing or no longer retained") from error return jsonify(snapshot.to_json()) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/http_routes.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/http_routes.py) ## 在本地运行 ### 启动 Dex 与 example 在第一个终端中，先渲染 session graph，再启动 Dex。Dex Web 启动时读取这个 graph，用它显示 session 和审批控件。 **Python** ```bash cd examples/python/agent-harnesses uv sync --locked mkdir -p definitions uv run --frozen dex-agents-session-graph --flow-type PydanticAIOrderSupport \ --permission order-support.operate --out definitions/pydantic-ai.json dexcli dev --flow-rendering-dir ./definitions ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) **dexcli dev** 会一直在前台运行。在第二个终端中启动 example： **Python** ```bash cd examples/python/agent-harnesses uv run --frozen python -m agent_harness_examples ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) 应用监听 **127.0.0.1:8081**。Dex Web 位于 **127.0.0.1:8802**。 没有设置 **DEX_PYDANTIC_AI_MODEL** 时，Agent 使用 **mock/order-support**，这是一个不需要 key 的本地 demo model。收到 "look up order 1001" 时它调用 **look_up_order**，收到 "cancel order 1002" 时它调用 **cancel_order**。 | 变量 | 默认值 | 含义 | | ------------------------- | ---------------------- | --------------------------------------- | | **DEX_PYDANTIC_AI_MODEL** | **mock/order-support** | Pydantic AI 模型名称，例如 **openai:gpt-5.2**。 | | **OPENAI_API_KEY** | 无 | Provider key，使用 OpenAI 模型时需要。 | | **DEX_AGENT_SESSION_KEY** | 随机 key | 签发 session token 的 key。 | ### 调用 session 启动一个 session，让 Agent 取消订单 1002，然后批准这次调用： **Python** ```bash base=http://127.0.0.1:8081/products/agent-harnesses/pydantic-ai session=$(curl -s -X POST "$base/sessions" -H 'Content-Type: application/json' -d '{}') id=$(echo "$session" | jq -r .session_id) token=$(echo "$session" | jq -r .session_token) curl -s -X POST "$base/messages" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" -d "{\"session_id\": \"$id\", \"text\": \"Cancel order 1002\"}" until interrupt=$(curl -s "$base/snapshot?session_id=$id" -H "X-Agent-Session-Token: $token" \ | jq -re '.snapshot.pending_interrupts[0].interrupt_id'); do sleep 1; done curl -s -X POST "$base/interrupts/resolve" -H 'Content-Type: application/json' \ -H "X-Agent-Session-Token: $token" \ -d "{\"session_id\": \"$id\", \"interrupt_id\": \"$interrupt\", \"approved\": true}" ``` 例子: [examples/python/agent-harnesses/README.md](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/README.md) 循环会反复读取 snapshot，直到 Agent 请求运行 **cancel_order**。这次调用先等待审批，批准后只运行一次。再读一次 snapshot 即可看到 Agent 的回答。 ### 在 Dex Web 中审批 也可以不执行最后一条命令，改在 Dex Web 中审批，步骤见 [在 Dex Web 中审批](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#approve-in-dex-web)。 ## 连接 Connector Agent 可以把 [Connector](https://docs.superdurable.io/zh-Hans/concepts/connector.md) 的 operation 当作 tool 调用，见 [把 Connector 用作 tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#connectors-as-tools)。这个 example 加入了 Slack tool **slack_post_thread_reply**，用来在 Slack thread 中回复。 ### 提供 tool 安装 **dex-agents** 的 **connectors** extra，以及 Connector 包，例如 **dex-connector-slack**。安装命令见 [Connector tools](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#install)。 Worker 从项目配置中打开 Slack connection **support-slack**。**connector_tool** 把 operation 变成 tool，由 **ConnectorToolSource** 持有。没有这个 connection 时，agent 不会得到 Slack tool。 **Python** ```python def build_slack_tool_sources(connection: slack.Connection | None) -> list[ToolSource]: """Returns the Slack tool source, or no source when Slack is not configured.""" if connection is None: return [] return [ ConnectorToolSource([ # The manifest does not ask for approval; this agent asks before every post. connector_tool( connection.post_thread_reply(), connection.reference, requires_approval=True ), ]), ] # ... async def open_slack_project_connection( environment: Mapping[str, str] = os.environ, ) -> SlackProjectConnection | None: # ... # An incomplete or invalid configuration raises here, so the Worker fails at startup. try: project = await load_from_environment(environment) except FileNotFoundError: _log_slack_not_set_up("the connector configuration file does not exist") return None # ... try: connection = slack.new_project_connection(project, SLACK_CONNECTION_NAME) except BaseException: await project.close() raise return SlackProjectConnection(project, connection) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/slack_connector_tools.py) Worker 在启动时打开一次 connection，每个 agent 都使用它： **Python** ```python async def main() -> None: config = AgentHarnessExamplesConfig.from_env() slack_setup = await open_slack_project_connection() try: app_state = AgentHarnessExamplesApp(config, slack_setup.connection if slack_setup else None) # ... finally: if slack_setup is not None: await slack_setup.aclose() ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/\_\_main\_\_.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/__main__.py) Agent 把 tool source 和自己的 tool 一起传入： **Python** ```python return pydantic_ai_session_flow( agent, flow_type=PYDANTIC_AI_ORDER_SUPPORT_FLOW_TYPE, tool_sources=build_slack_tool_sources(slack_connection), # ... ) ``` 例子: [examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py](https://github.com/superdurable/dex/tree/main/examples/python/agent-harnesses/agent_harness_examples/pydantic_ai_order_support.py) ### 调用如何运行 每次调用 **slack_post_thread_reply** 都要先经人批准，然后在自己的 **CallSingleAttemptTool** Step 中只运行一次。原因见 [Slack Connector tool](https://docs.superdurable.io/zh-Hans/agent-frameworks.md#slack-connector-tool)。 Adapter 把 Connector tool 作为 **ExternalToolset** 传给每次 run。Pydantic AI 从不运行 external tool，所以 Connector tool call 会结束这次 run，调用在它的 tool Step 中运行。 ### 配置 connection Worker 从哪里找到 connection **support-slack**、Slack app 需要什么，见 [Connection](https://docs.superdurable.io/zh-Hans/reference/agents/connector-tools.md#connector-tools-connection)。 试用时，先设置 **DEX_CONNECTOR_CONFIG_FILE** 再启动 example。然后让 agent 在 Slack thread 中回复，并给出 channel ID 和 thread timestamp： ```text Reply in Slack thread C0123456789 1710000000.000100: Order 1002 ships on Thursday. ``` 这次发帖和 **cancel_order** 一样，会先等待审批。 ## 限制 - Pydantic AI 自己发给模型的重试，会在同一个 **RunModel** Step 中再调用一次模型。例如 tool 参数无效，或 output 校验失败。 - Output function 在 **RunModel** 内运行，至少运行一次。请把写操作放进 tool。 - 由 provider 运行的 tool，例如 **WebSearchTool**，会在 **RunModel** 重跑时重复执行。 - Deferred tool 返回给模型的是文本。图片和文件会丢失。 - 只在运行时才请求审批的 tool，需要一个带 **requires_approval=True** 的 **ToolPolicy**。 - **UsageLimits** 和 tool 重试预算按 Step 计算，而不是按 turn 计算。 Adapter README 列出了全部限制。Server 的限制参见 [Limits for long sessions](https://docs.superdurable.io/zh-Hans/reference/agents/session-limits.md)。 ## 相关内容 - [Agent Framework](https://docs.superdurable.io/zh-Hans/agent-frameworks.md) - [Agents reference](https://docs.superdurable.io/zh-Hans/reference/agents.md) - [Python adapter README](https://github.com/superdurable/dex/tree/main/agents/python/pydantic-ai)