如何不用 Agent 循环直接用 Pydantic AI 的 direct API 调用模型【免费下载链接】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当你只需要给模型一条消息、拿回一个响应不想引入Agent的工具执行、重试和结构化输出解析等完整运行时Pydantic AI 提供了pydantic_ai.direct模块。它是对各Model实现的薄封装唯一的抽象是输入和输出 schema 的转换让你用同一套 API 调用 OpenAI、Anthropic、Google 等所有支持的模型。本文按安装 → 发起一次同步请求 → 换用异步/流式 → 挂工具调用 → 验证返回的顺序给出可直接执行的代码路径并说明 direct API 与 Agent API 的适用边界。准备条件Python 3.10 及以上。安装 Pydantic AI安装文档pip/uv-add pydantic-ai这会安装pydantic_ai包、核心依赖以及使用 OpenAI、Anthropic 和 Google 模型所需的库。如果只用某一个模型也可以改装pydantic-ai-slim加对应 extras例如pip/uv-add pydantic-ai-slim[openai,google,logfire]避免装多余依赖。四个入口函数怎么选direct 文档定义了四个函数区别只在同步/异步和是否流式函数行为model_request非流式异步请求model_request_sync非流式同步请求model_request_stream流式异步请求model_request_stream_sync流式同步请求选同步还是异步取决于你的代码形态在顶层脚本里用_sync版本最省事在async函数里FastAPI 处理器、已有事件循环的代码必须用异步版本因为model_request_sync内部用loop.run_until_complete(...)驱动不能在有活动事件循环的 async 代码中使用见 direct 源码 的 docstring 说明。发起一次同步请求下面这段是文档给出的完整示例可以按原样运行from pydantic_ai import ModelRequest from pydantic_ai.direct import model_request_sync # 发起同步请求 model_response model_request_sync( anthropic:claude-haiku-4-5, [ModelRequest.user_text_prompt(What is the capital of France?)] ) print(model_response.parts[0].content) # The capital of France is Paris. print(model_response.usage) # RequestUsage(input_tokens56, output_tokens7)调用要点第一个参数是模型标识格式为provider:model如anthropic:claude-haiku-4-5。文档说明允许传str因为支持的模型列表经常变化也可以用KnownModelName或Model实例。第二个参数是消息列表。ModelRequest.user_text_prompt(text)是最简构造方式它创建一个只含一条用户文本的ModelRequest还可选传instructions参数附加指令。返回值是ModelResponse通过.parts访问响应内容这里第一个 part 的.content是文本通过.usage访问本次请求的 token 用量。上面#两行是文档给出的示例输出实际 token 数会随请求内容变化不要把它当作固定预期。换用异步或流式请求异步版本model_request与同步版参数一致在async函数里await即可from pydantic_ai import ModelRequest from pydantic_ai.direct import model_request async def main(): model_response await model_request( anthropic:claude-haiku-4-5, [ModelRequest.user_text_prompt(What is the capital of France?)] ) print(model_response)流式请求用async with进入上下文管理器逐块消费stream。model_request_stream的文档示例openai:gpt-5-mini会依次产出PartStartEvent、PartDeltaEvent、PartEndEvent等事件其中PartEndEvent携带该 part 的完整内容from pydantic_ai import ModelRequest from pydantic_ai.direct import model_request_stream async def main(): messages [ModelRequest.user_text_prompt(Who was Albert Einstein?)] async with model_request_stream(openai:gpt-5-mini, messages) as stream: chunks [] async for chunk in stream: chunks.append(chunk) print(chunks)同步代码用model_request_stream_sync写法相同只是async with换成withfrom pydantic_ai import ModelRequest from pydantic_ai.direct import model_request_stream_sync messages [ModelRequest.user_text_prompt(Who was Albert Einstein?)] with model_request_stream_sync(openai:gpt-5-mini, messages) as stream: chunks [] for chunk in stream: chunks.append(chunk) print(chunks)注意model_request_stream_sync的限制它必须作为上下文管理器使用且返回的同步流必须在创建它的同一线程上使用并关闭退出with块会立即取消底层请求并关闭连接而不是等整个响应收完。给请求挂上工具定义direct API 也能做函数/工具调用。通过model_request_parameters传入ModelRequestParameters用function_tools给出工具定义JSON schema 可以直接由 Pydantic 模型生成from typing import Literal from pydantic import BaseModel from pydantic_ai import ModelRequest, ToolDefinition from pydantic_ai.direct import model_request from pydantic_ai.models import ModelRequestParameters class Divide(BaseModel): Divide two numbers. numerator: float denominator: float on_inf: Literal[error, infinity] infinity async def main(): model_response await model_request( openai:gpt-5-nano, [ModelRequest.user_text_prompt(What is 123 / 456?)], model_request_parametersModelRequestParameters( function_tools[ ToolDefinition( nameDivide.__name__.lower(), descriptionDivide.__doc__, parameters_json_schemaDivide.model_json_schema(), ) ], allow_text_outputTrue, # 允许模型用工具或直接文本回答 ), ) print(model_response)运行这段需要import asyncio并追加asyncio.run(main())其余无需改动文档原文说明。文档示例的返回值是一个ModelResponseparts中为ToolCallPart含tool_namedivide和解析出的参数usage为RequestUsage(input_tokens55, output_tokens7)——同样是示例输出实际 token 数不要按此校验。allow_text_outputTrue表示模型既可以发起工具调用也可以直接文本回答文档示例的注释即此含义。验证与结果判断direct 调用的成功路径就是拿到ModelResponse对象文本响应检查model_response.parts[0].content是否为模型文本。工具调用响应检查parts中是否为ToolCallPart并读取其tool_name与args。token 用量读model_response.usage类型为RequestUsageModelResponse还带model_name、timestamp等字段可用于核对请求确实发到了预期的模型。如果要在生产代码里观测每次请求可以开启 OpenTelemetry/Logfire 插桩。这是可选分支两种用法来自 direct 文档更多细节见 logfire 文档全局插桩import logfire from pydantic_ai import ModelRequest from pydantic_ai.direct import model_request_sync logfire.configure() logfire.instrument_pydantic_ai() model_response model_request_sync( anthropic:claude-haiku-4-5, [ModelRequest.user_text_prompt(What is the capital of France?)], )按单次调用插桩不全局 instrument只在调用时传instrumentTrue。四个 direct 函数都接受instrument参数不传None时沿用logfire.instrument_pydantic_ai设置的默认值。限制与何时该回到 Agent API直接来自文档的边界条件instructions 不累积如果消息历史中有多条带instructions的ModelRequestdirect API 只使用最近一条而不是把历次指令叠加起来。model_request_sync不能在 async 代码里用它在有活动事件循环的上下文中无法工作请改用model_request。model_request_stream_sync有线程约束流必须在创建它的线程上使用并关闭。direct API 不执行工具它只负责发请求、收响应工具由模型返回、由你的代码处理。文档给出的选择标准是当你需要更直接地控制模型交互、想在模型请求周围实现自定义行为、或要在模型交互之上构建自己的抽象时用 direct API对大多数应用来说AgentAPI 更方便因为它额外提供工具执行、重试、结构化输出解析等能力。API 参考入口是 pydantic_ai.direct四个函数的签名、参数model、messages、model_settings、model_request_parameters、instrument和返回值的完整说明在 direct 源码 的 docstring 中。【免费下载链接】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），仅供参考