字数 1233，阅读大约需 7 分钟

Python公开课：学习Pydantic AI 的提纲总结和心得

Pydantic AI 是一个由 Pydantic 团队开发的 Python 框架，用于构建生产级的 AI 应用程序。它是在 2024 年推出的相对较新的框架.
无论是调用 OpenAI 的接口，还是使用 LangChain 构建应用，Pydantic 都在背后默默发挥着关键作用。诸如 OpenAI、Anthropic、LangChain、LlamaIndex 等明星项目，都将 Pydantic 作为核心组件，负责数据验证、序列化等重要功能。


    
    
    
  from pydantic_ai import Agent

# 创建一个简单的 agent
agent = Agent('openai:gpt-4')

# 同步调用
result = agent.run_sync('What is the capital of France?')
print(result.data)

# 异步调用
result = await agent.run('Tell me a joke')

主要功能

核心概念

agent


    
    
    
  roulette_agent = Agent(  
    'openai:gpt-4o',
    deps_type=int,
    output_type=bool,
    system_prompt=(
        'Use the `roulette_wheel` function to see if the '
        'customer has won based on the number they provide.'
    ),
)

result = roulette_agent.run_sync('Put my money on square eighteen', deps=success_number)

工具


    
    
    
  @roulette_agent.tool
async def roulette_wheel(ctx: RunContext[int], square: int) -> str:  
    """check if the square is a winner"""
    return 'winner' if square == ctx.deps else 'loser'

用量


    
    
    
  from pydantic_ai import Agent, ModelSettings
from pydantic_ai.models.openai import OpenAIChatModel

# 1. Model-level defaults
model = OpenAIChatModel(
    'gpt-4o',
    settings=ModelSettings(temperature=0.8, max_tokens=500)  # Base defaults
)

# 2. Agent-level defaults (overrides model defaults by merging)
agent = Agent(model, model_settings=ModelSettings(temperature=0.5))

# 3. Run-time overrides (highest priority)
result_sync = agent.run_sync(
    'What is the capital of Italy?',
    model_settings=ModelSettings(temperature=0.0)  # Final temperature: 0.0
)
print(result_sync.output)
#> The capital of Italy is Rome.

依赖项

Pydantic AI 使用依赖注入系统，为（agent）的系统提示、工具和输出验证器提供数据和服务。（依赖项可以是任何 Python 类型）.
依赖项通过 RunContext 类型进行访问。


    
    
    
  from dataclasses import dataclass

import httpx

from pydantic_ai import Agent, RunContext


@dataclass
class MyDeps:
    api_key: str
    http_client: httpx.AsyncClient


agent = Agent(
    'openai:gpt-4o',
    deps_type=MyDeps,
)


@agent.system_prompt  
async def get_system_prompt(ctx: RunContext[MyDeps]) -> str:  
    response = await ctx.deps.http_client.get(  
        'https://example.com',
        headers={'Authorization': f'Bearer {ctx.deps.api_key}'},  
    )
    response.raise_for_status()
    return f'Prompt: {response.text}'


async def main():
    async with httpx.AsyncClient() as client:
        deps = MyDeps('foobar', client)
        result = await agent.run('Tell me a joke.', deps=deps)
        print(result.output)
        #> Did you hear about the toothpaste scandal? They called it Colgate.

工具

函数工具为模型提供了一种机制，使其能够执行操作和检索额外信息。

输出

“输出”指的是运行代理后返回的最终值。这可以是纯文本、结构化数据，或者是使用模型提供的参数调用的函数的结果。

输出被包装在 AgentRunResult 或 StreamedRunResult 中


    
    
    
  from pydantic import BaseModel

from pydantic_ai import Agent


class CityLocation(BaseModel):
    city: str
    country: str


agent = Agent('google-gla:gemini-1.5-flash', output_type=CityLocation)
result = agent.run_sync('Where were the olympics held in 2012?')
print(result.output)
#> city='London' country='United Kingdom'
print(result.usage())
#> RunUsage(input_tokens=57, output_tokens=8, requests=1)

消息和历史

可以用来继续连贯的对话，也可以用来了解智能体的执行情况。


    
    
    
  from pydantic_ai import Agent

agent = Agent('openai:gpt-4o', system_prompt='Be a helpful assistant.')

result = agent.run_sync('Tell me a joke.')
print(result.output)
#> Did you hear about the toothpaste scandal? They called it Colgate.

# all messages from the run
print(result.all_messages())

模型直接请求


    
    
    
  model_request：向模型发起非流式的异步请求
model_request_sync：向模型发起非流式的同步请求
model_request_stream：向模型发起流式的异步请求
model_request_stream_sync：向模型发起流式的同步请求

from pydantic_ai.direct import model_request_sync
from pydantic_ai.messages import ModelRequest

# Make a synchronous request to the model
model_response = model_request_sync(
    'anthropic:claude-3-5-haiku-latest',
    [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_tokens=56, output_tokens=7)

类型安全和验证

利用 Pydantic 的强大数据验证能力
提供完整的类型提示支持
确保 AI 应用的输入输出类型安全


    
    
    
  # Pydantic的数据验证模式
class User(BaseModel):
    name: str
    age: int

# Pydantic AI的Agent类型化
class SupportOutput(BaseModel):
    advice: str
    risk_level: int

agent: Agent[SupportDeps, SupportOutput] = Agent(
    'openai:gpt-4o',
    deps_type=SupportDeps,
    output_type=SupportOutput
)

Agent 架构

基于pydantic-graph构建的状态机执行引擎
支持构建 AI Agent（代理）
可以定义工具和函数供 AI 调用
支持多轮对话和上下文管理
内置agent状态和图执行引擎架构


    
    
    
  graph TD
    A[UserPromptNode] --> B[ModelRequestNode]
    B --> C[CallToolsNode]
    C --> D{需要更多工具调用?}
    D -->|是| B
    D -->|否| E[End]


    
    
    
  @dataclasses.dataclass(kw_only=True)
class GraphAgentState:
    message_history: list[ModelMessage]  # 消息历史
    usage: RunUsage                      # 使用统计
    retries: int                         # 重试次数
    run_step: int                       # 运行步骤

模型无关

支持多种 LLM 提供商（OpenAI、Anthropic、Gemini 等）
统一的 API 接口
易于切换不同的模型


    
    
    
  class Model(ABC):
    @abstractmethod
    async def request(
        self, 
        messages: list[ModelMessage],
        model_settings: ModelSettings,
        parameters: ModelRequestParameters,
        run_context: RunContext[Any],
    ) -> tuple[ModelResponse, RequestUsage]:
        """统一的模型请求接口"""

依赖注入

内置依赖注入系统
方便管理应用状态和资源
提高代码的可测试性


    
    
    
  T = TypeVar('T')  # 依赖类型
S = TypeVar('S')  # 输出类型

@dataclasses.dataclass(init=False)
class Agent(AbstractAgent[T, S]):
    """Agent[依赖类型, 输出类型]的泛型设计"""

彩蛋结尾

嘿，别滑了！手指停一停，听我说句悄悄话👇
🌟 关注我：下次更新，系统会自动弹窗提醒你；
📌 收藏本文：点个收藏，让它成为你的知识库，随时挖宝；
❤️ 点赞在看：你的每个赞都是我熬夜写文的“鸡血”；

目录CONTENT

Python公开课：学习Pydantic AI 的提纲总结和心得

Python公开课：学习Pydantic AI 的提纲总结和心得

主要功能

核心概念

agent

工具

用量

依赖项

工具

输出

消息和历史

模型直接请求

类型安全和验证

Agent 架构

模型无关

依赖注入

彩蛋结尾

更多内容

评论区