Pydantic AI Agent开发实战：用类型安全构建生产级AI应用的5个核心模式

2026年，AI Agent已经从实验性项目走向生产级部署。然而，Python生态中的AI开发长期面临一个核心矛盾：LLM的输出是不可控的字符串，而生产系统需要的是类型安全的数据结构。Pydantic AI应运而生——它将Pydantic的类型验证能力与AI Agent的灵活性深度结合，让开发者用Python类型注解就能定义Agent的行为契约。

本文从一线开发实践出发，提炼出5个核心模式，覆盖从Agent定义到流式响应的完整开发链路。每个模式都配有可直接运行的代码，帮助你在项目中快速落地。

核心概念速览

概念	说明	核心价值
Agent	Pydantic AI核心抽象，封装LLM交互逻辑	统一Agent定义接口
RunContext	运行时上下文，携带依赖注入对象	类型安全的依赖管理
Tool	函数工具，Agent可调用的外部能力	Function Calling标准化
ResultType	结构化输出类型，约束LLM返回格式	输出类型安全保障
StreamedRunResult	流式响应结果，支持逐token输出	实时交互体验

问题分析：为什么传统AI开发不够好？

1. 类型缺失：LLM返回原始字符串，需要手动解析和验证，运行时错误频发
2. 依赖混乱：数据库连接、API密钥等依赖在函数间层层传递，耦合严重
3. 工具注册繁琐：Function Calling的JSON Schema手写易错，与代码逻辑脱节
4. 输出不可控：LLM可能返回不符合预期的格式，缺乏编译时保障
5. 流式处理复杂：SSE/WebSocket流式响应的解析和错误处理代码冗长

模式一：Agent基础定义与类型安全模型

Pydantic AI的Agent定义将LLM交互抽象为类型安全的函数调用。通过Python类型注解，你在编写代码时就定义了Agent的输入输出契约。

"""模式一：Agent基础定义与类型安全模型"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent


# 1. 定义结构化输出模型
class CodeReviewResult(BaseModel):
    """代码审查结果 - LLM的输出将被约束为这个结构"""

    quality_score: float = Field(
        ...,
        ge=0.0,
        le=10.0,
        description="代码质量评分，0-10分",
    )
    issues: list[str] = Field(
        default_factory=list,
        description="发现的问题列表",
    )
    suggestions: list[str] = Field(
        default_factory=list,
        description="改进建议列表",
    )
    summary: str = Field(
        ...,
        min_length=10,
        max_length=500,
        description="审查摘要",
    )
    language: str = Field(
        ...,
        description="代码语言",
    )


# 2. 定义Agent - 通过泛型参数指定输出类型
code_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="""你是一位资深代码审查专家。
请对用户提交的代码进行专业审查，给出质量评分、问题列表、改进建议和审查摘要。
评分标准：代码可读性、性能、安全性、最佳实践。""",
)


# 3. 使用Agent
async def review_code(code: str) -> CodeReviewResult:
    """审查代码并返回结构化结果"""
    result = await code_review_agent.run(code)
    return result.data


# 4. 同步运行方式
def review_code_sync(code: str) -> CodeReviewResult:
    """同步方式审查代码"""
    result = code_review_agent.run_sync(code)
    return result.data


# 5. 带重试的Agent定义
robust_review_agent = Agent(
    "openai:gpt-4o",
    result_type=CodeReviewResult,
    system_prompt="你是一位资深代码审查专家。请严格按照JSON Schema输出。",
    retries=3,  # 输出验证失败时自动重试
    model_settings={
        "temperature": 0.1,  # 低温度确保输出稳定
        "max_tokens": 2000,
    },
)


# 6. 运行并查看使用信息
async def review_with_usage(code: str) -> tuple[CodeReviewResult, dict]:
    """审查代码并返回使用统计"""
    result = await robust_review_agent.run(code)
    usage = result.usage()
    return result.data, {
        "request_tokens": usage.request_tokens,
        "response_tokens": usage.response_tokens,
        "total_tokens": usage.total_tokens,
    }


# 运行示例
if __name__ == "__main__":
    import asyncio

    sample_code = """
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        total = total + items[i]['price'] * items[i]['qty']
    return total
    """

    result = asyncio.run(review_code(sample_code))
    print(f"质量评分: {result.quality_score}")
    print(f"问题: {result.issues}")
    print(f"建议: {result.suggestions}")
    print(f"摘要: {result.summary}")
    print(f"语言: {result.language}")

关键要点：

• result_type参数让LLM输出直接绑定到Pydantic模型，验证自动执行
• retries参数在输出验证失败时自动重试，提升鲁棒性
• model_settings控制LLM参数，低temperature确保结构化输出稳定

模式二：依赖注入与系统提示

生产级Agent需要访问数据库、缓存、配置等外部依赖。Pydantic AI的依赖注入系统让这些依赖通过类型注解自动注入，彻底消除全局变量和参数传递的混乱。

"""模式二：依赖注入与系统提示"""
from __future__ import annotations

from dataclasses import dataclass, field
from typing import Optional

from pydantic import BaseModel
from pydantic_ai import Agent, RunContext


# 1. 定义依赖类型
@dataclass
class DatabaseDep:
    """数据库依赖"""

    connection_string: str

    async def query(self, sql: str) -> list[dict]:
        """模拟数据库查询"""
        # 实际项目中替换为真实数据库连接
        print(f"[DB] 执行查询: {sql}")
        return [{"id": 1, "name": "张三", "score": 95}]


@dataclass
class CacheDep:
    """缓存依赖"""

    backend: str = "redis"
    _cache: dict = field(default_factory=dict)

    async def get(self, key: str) -> Optional[str]:
        return self._cache.get(key)

    async def set(self, key: str, value: str, ttl: int = 3600) -> None:
        self._cache[key] = value
        print(f"[Cache] 设置 {key} (TTL={ttl}s)")


@dataclass
class UserContext:
    """用户上下文依赖"""

    user_id: str
    username: str
    role: str = "user"
    preferences: dict = field(default_factory=dict)


# 2. 组合依赖
@dataclass
class AgentDeps:
    """Agent的所有依赖"""

    db: DatabaseDep
    cache: CacheDep
    user: UserContext


# 3. 定义输出模型
class AnalysisReport(BaseModel):
    title: str
    findings: list[str]
    confidence: float
    recommendation: str


# 4. 创建带依赖的Agent
analysis_agent = Agent(
    "openai:gpt-4o",
    result_type=AnalysisReport,
    deps_type=AgentDeps,
)


# 5. 动态系统提示 - 根据依赖动态生成
@analysis_agent.system_prompt
async def build_system_prompt(ctx: RunContext[AgentDeps]) -> str:
    """根据用户角色和偏好动态生成系统提示"""
    role_instructions = {
        "admin": "你拥有管理员权限，可以访问所有数据和执行所有操作。",
        "analyst": "你拥有分析师权限，可以查看数据但不可修改。",
        "user": "你拥有普通用户权限，只能查看自己的数据。",
    }

    base_prompt = f"""你是一个数据分析助手。
当前用户: {ctx.deps.user.username} (角色: {ctx.deps.user.role})
{role_instructions.get(ctx.deps.user.role, role_instructions['user'])}"""

    # 根据用户偏好定制
    if ctx.deps.user.preferences.get("detail_level") == "verbose":
        base_prompt += "\n请提供详细的分析过程和推理步骤。"
    else:
        base_prompt += "\n请简洁明了地给出分析结论。"

    return base_prompt


# 6. 在工具函数中使用依赖
@analysis_agent.tool
async def query_user_data(
    ctx: RunContext[AgentDeps],
    query: str,
) -> str:
    """查询用户相关数据"""
    # 先查缓存
    cache_key = f"user:{ctx.deps.user.user_id}:{query}"
    cached = await ctx.deps.cache.get(cache_key)
    if cached:
        return f"[缓存命中] {cached}"

    # 查数据库
    sql = f"SELECT * FROM user_data WHERE user_id = '{ctx.deps.user.user_id}' AND query LIKE '%{query}%'"
    results = await ctx.deps.db.query(sql)

    # 写入缓存
    await ctx.deps.cache.set(cache_key, str(results))

    return str(results)


@analysis_agent.tool
async def get_user_preferences(
    ctx: RunContext[AgentDeps],
) -> str:
    """获取用户偏好设置"""
    return str(ctx.deps.user.preferences)


# 7. 运行Agent
async def run_analysis(user_question: str) -> AnalysisReport:
    """运行分析Agent"""
    deps = AgentDeps(
        db=DatabaseDep(connection_string="postgresql://localhost/mydb"),
        cache=CacheDep(backend="redis"),
        user=UserContext(
            user_id="u_001",
            username="张三",
            role="analyst",
            preferences={"detail_level": "verbose", "language": "zh-CN"},
        ),
    )

    result = await analysis_agent.run(user_question, deps=deps)
    return result.data


# 运行示例
if __name__ == "__main__":
    import asyncio

    report = asyncio.run(run_analysis("分析最近一个月的销售数据趋势"))
    print(f"标题: {report.title}")
    print(f"发现: {report.findings}")
    print(f"置信度: {report.confidence}")
    print(f"建议: {report.recommendation}")

关键要点：

• deps_type声明Agent的依赖类型，运行时自动注入到RunContext
• @agent.system_prompt装饰器支持动态生成系统提示，根据依赖内容定制
• 工具函数通过ctx: RunContext[AgentDeps]访问依赖，类型安全且有IDE补全

模式三：工具注册与Function Calling

Function Calling是AI Agent连接外部世界的桥梁。Pydantic AI将工具注册简化为装饰器+类型注解，自动生成JSON Schema，让LLM能精确调用你的Python函数。

"""模式三：工具注册与Function Calling"""
from __future__ import annotations

import json
from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, Field
from pydantic_ai import Agent, RunContext, Tool


# 1. 定义输出模型
class TravelPlan(BaseModel):
    """旅行规划结果"""

    destination: str
    start_date: str
    end_date: str
    estimated_cost: float
    itinerary: list[str]
    tips: list[str]


# 2. 创建Agent
travel_agent = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    system_prompt="你是一个旅行规划助手。根据用户需求规划旅行，使用提供的工具查询实时信息。",
)


# 3. 基础工具注册 - 装饰器方式
@travel_agent.tool
async def search_flights(
    ctx: RunContext[None],
    origin: str = Field(description="出发城市"),
    destination: str = Field(description="目的城市"),
    date: str = Field(description="出发日期，格式YYYY-MM-DD"),
) -> str:
    """搜索航班信息"""
    # 模拟API调用
    flights = [
        {"flight": "CA1234", "price": 1280, "departure": "08:00"},
        {"flight": "MU5678", "price": 980, "departure": "14:30"},
        {"flight": "CZ9012", "price": 1580, "departure": "19:00"},
    ]
    return json.dumps(flights, ensure_ascii=False)


@travel_agent.tool
async def get_weather(
    ctx: RunContext[None],
    city: str = Field(description="城市名称"),
    date: Optional[str] = Field(default=None, description="日期，不填则为今天"),
) -> str:
    """查询天气信息"""
    weather_data = {
        "city": city,
        "date": date or datetime.now(timezone.utc).strftime("%Y-%m-%d"),
        "temperature": "25°C",
        "condition": "晴转多云",
        "humidity": "65%",
    }
    return json.dumps(weather_data, ensure_ascii=False)


@travel_agent.tool
async def search_hotels(
    ctx: RunContext[None],
    city: str = Field(description="城市名称"),
    check_in: str = Field(description="入住日期"),
    check_out: str = Field(description="退房日期"),
    budget: Optional[float] = Field(default=None, description="预算上限（元/晚）"),
) -> str:
    """搜索酒店信息"""
    hotels = [
        {"name": "市中心大酒店", "price": 580, "rating": 4.5, "location": "市中心"},
        {"name": "湖景度假酒店", "price": 880, "rating": 4.8, "location": "湖畔"},
        {"name": "经济型连锁酒店", "price": 280, "rating": 4.0, "location": "火车站旁"},
    ]
    if budget:
        hotels = [h for h in hotels if h["price"] <= budget]
    return json.dumps(hotels, ensure_ascii=False)


# 4. 高级工具注册 - Tool类方式（更多控制）
async def calculate_exchange_rate(
    amount: float,
    from_currency: str,
    to_currency: str,
) -> str:
    """货币换算"""
    rates = {"CNY": 1.0, "USD": 0.14, "EUR": 0.13, "JPY": 21.5}
    if from_currency not in rates or to_currency not in rates:
        return f"不支持的货币: {from_currency} -> {to_currency}"
    cny_amount = amount / rates[from_currency]
    result = cny_amount * rates[to_currency]
    return f"{amount} {from_currency} = {result:.2f} {to_currency}"


exchange_tool = Tool(
    name="calculate_exchange_rate",
    description="货币换算工具，支持CNY、USD、EUR、JPY",
    function=calculate_exchange_rate,
)

travel_agent.register_tool(exchange_tool)


# 5. 带依赖的工具
@dataclass
class TravelDeps:
    user_id: str
    membership_level: str


travel_agent_with_deps = Agent(
    "openai:gpt-4o",
    result_type=TravelPlan,
    deps_type=TravelDeps,
    system_prompt="你是一个旅行规划助手。",
)


@travel_agent_with_deps.tool
async def get_member_discount(
    ctx: RunContext[TravelDeps],
) -> str:
    """获取会员折扣信息"""
    discounts = {
        "gold": "8折优惠 + 免费升舱机会",
        "silver": "9折优惠",
        "bronze": "95折优惠",
        "regular": "无折扣",
    }
    level = ctx.deps.membership_level
    return f"会员等级: {level}, 优惠: {discounts.get(level, '无折扣')}"


# 6. 工具执行追踪
class ToolCallLogger:
    """工具调用日志记录器"""

    def __init__(self):
        self.calls: list[dict] = []

    def log(self, tool_name: str, args: dict, result: str) -> None:
        self.calls.append(
            {
                "tool": tool_name,
                "args": args,
                "result_preview": result[:100],
                "timestamp": datetime.now(timezone.utc).isoformat(),
            }
        )


logger = ToolCallLogger()


# 7. 运行Agent
async def plan_travel(user_request: str) -> TravelPlan:
    """规划旅行"""
    result = await travel_agent.run(user_request)

    # 查看工具调用历史
    for msg in result.all_messages():
        if hasattr(msg, "tool_calls") and msg.tool_calls:
            for tc in msg.tool_calls:
                print(f"工具调用: {tc.function.name}({tc.function.arguments})")

    return result.data


# 运行示例
if __name__ == "__main__":
    import asyncio
    from dataclasses import dataclass

    plan = asyncio.run(plan_travel("我想下个月从北京去东京玩5天，预算1万元"))
    print(f"目的地: {plan.destination}")
    print(f"日期: {plan.start_date} ~ {plan.end_date}")
    print(f"预算: {plan.estimated_cost}元")
    print(f"行程: {plan.itinerary}")
    print(f"贴士: {plan.tips}")

关键要点：

• @agent.tool装饰器自动从函数签名和类型注解生成JSON Schema
• Field(description=...)为LLM提供参数说明，提升Function Calling准确率
• Tool类方式提供更细粒度的控制，如自定义工具名称和描述
• 工具函数可以是async或sync，Pydantic AI自动处理

模式四：结构化输出与结果验证

结构化输出是Pydantic AI的核心价值。当LLM返回的JSON不符合模型定义时，Pydantic会自动验证并报错，配合重试机制确保最终输出符合预期。

"""模式四：结构化输出与结果验证"""
from __future__ import annotations

import json
from enum import Enum
from typing import Literal, Optional, Union

from pydantic import BaseModel, Field, field_validator, model_validator
from pydantic_ai import Agent


# 1. 基础结构化输出
class SentimentType(str, Enum):
    POSITIVE = "positive"
    NEGATIVE = "negative"
    NEUTRAL = "neutral"


class SentimentResult(BaseModel):
    """情感分析结果"""

    text: str = Field(description="被分析的文本")
    sentiment: SentimentType = Field(description="情感类型")
    confidence: float = Field(ge=0.0, le=1.0, description="置信度")
    keywords: list[str] = Field(description="关键词列表")

    @field_validator("keywords")
    @classmethod
    def keywords_not_empty(cls, v: list[str]) -> list[str]:
        if not v:
            raise ValueError("关键词列表不能为空")
        return v


# 2. 嵌套结构化输出
class Address(BaseModel):
    province: str
    city: str
    district: str
    street: Optional[str] = None


class BusinessInfo(BaseModel):
    name: str
    category: str
    address: Address
    rating: float = Field(ge=0.0, le=5.0)
    price_range: str


class BusinessExtraction(BaseModel):
    """商家信息提取结果"""

    businesses: list[BusinessInfo] = Field(description="提取的商家列表")
    total_count: int = Field(ge=0, description="商家总数")
    source_text_summary: str = Field(description="原文摘要")

    @model_validator(mode="after")
    def validate_count(self) -> "BusinessExtraction":
        if self.total_count != len(self.businesses):
            raise ValueError(
                f"total_count({self.total_count})与实际商家数({len(self.businesses)})不匹配"
            )
        return self


# 3. 联合类型输出 - 让LLM选择最合适的输出类型
class TextSummary(BaseModel):
    type: Literal["summary"] = "summary"
    content: str
    word_count: int


class TextOutline(BaseModel):
    type: Literal["outline"] = "outline"
    sections: list[str]
    total_sections: int


class TextAnalysis(BaseModel):
    type: Literal["analysis"] = "analysis"
    key_points: list[str]
    tone: str
    audience: str


# 使用Union让LLM选择输出格式
TextOutput = Union[TextSummary, TextOutline, TextAnalysis]


# 4. 创建多个专业Agent
sentiment_agent = Agent(
    "openai:gpt-4o",
    result_type=SentimentResult,
    system_prompt="你是情感分析专家。分析文本的情感倾向，给出分类、置信度和关键词。",
    retries=3,
)

extraction_agent = Agent(
    "openai:gpt-4o",
    result_type=BusinessExtraction,
    system_prompt="你是信息提取专家。从文本中提取商家信息，包括名称、类别、地址和评分。",
    retries=3,
)

flexible_agent = Agent(
    "openai:gpt-4o",
    result_type=TextOutput,
    system_prompt="你是文本分析助手。根据文本特点选择最合适的分析方式：摘要(summary)、大纲(outline)或深度分析(analysis)。",
    retries=3,
)


# 5. 结果验证与后处理
class ValidatedResult(BaseModel):
    """带验证逻辑的结果包装器"""

    raw_result: BaseModel
    is_valid: bool = True
    validation_errors: list[str] = []

    @classmethod
    def from_agent_result(cls, result: BaseModel) -> "ValidatedResult":
        """从Agent结果创建验证结果"""
        errors = []
        is_valid = True

        # 自定义验证逻辑
        if hasattr(result, "confidence") and result.confidence < 0.5:
            errors.append("置信度过低，结果可能不可靠")
            is_valid = False

        if hasattr(result, "keywords") and len(result.keywords) < 2:
            errors.append("关键词过少，分析可能不够深入")
            is_valid = False

        return cls(
            raw_result=result,
            is_valid=is_valid,
            validation_errors=errors,
        )


# 6. 批量处理与结果聚合
async def batch_sentiment_analysis(texts: list[str]) -> list[ValidatedResult]:
    """批量情感分析"""
    results = []
    for text in texts:
        try:
            result = await sentiment_agent.run(text)
            validated = ValidatedResult.from_agent_result(result.data)
            results.append(validated)
        except Exception as e:
            error_result = ValidatedResult(
                raw_result=SentimentResult(
                    text=text,
                    sentiment=SentimentType.NEUTRAL,
                    confidence=0.0,
                    keywords=["error"],
                ),
                is_valid=False,
                validation_errors=[str(e)],
            )
            results.append(error_result)

    return results


# 7. 结构化输出与JSON Schema导出
def export_result_schema() -> dict:
    """导出结果模型的JSON Schema"""
    return SentimentResult.model_json_schema()


# 运行示例
if __name__ == "__main__":
    import asyncio

    # 情感分析
    text = "这款产品真的太棒了！性能强劲，设计精美，性价比超高！"
    result = asyncio.run(sentiment_agent.run(text))
    print(f"情感: {result.data.sentiment.value}")
    print(f"置信度: {result.data.confidence}")
    print(f"关键词: {result.data.keywords}")

    # 商家提取
    business_text = """
    味千拉面位于北京市朝阳区三里屯路19号，日式拉面，评分4.2，人均80元。
    海底捞火锅在北京市海淀区中关村大街1号，火锅，评分4.8，人均150元。
    """
    extraction = asyncio.run(extraction_agent.run(business_text))
    print(f"商家数: {extraction.data.total_count}")
    for b in extraction.data.businesses:
        print(f"  {b.name} - {b.category} - {b.rating}分")

    # 导出Schema
    schema = export_result_schema()
    print(f"JSON Schema: {json.dumps(schema, indent=2, ensure_ascii=False)[:200]}...")

关键要点：

• Pydantic的field_validator和model_validator在LLM输出验证中发挥关键作用
• Union类型让LLM根据输入选择最合适的输出格式
• retries参数配合验证器，在验证失败时自动重试LLM调用
• 批量处理时用ValidatedResult包装器统一错误处理

模式五：流式响应与多轮对话

流式响应让用户实时看到AI的生成过程，大幅提升交互体验。Pydantic AI的流式API与结构化输出完美兼容，同时支持多轮对话的上下文管理。

"""模式五：流式响应与多轮对话"""
from __future__ import annotations

from typing import Optional

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


# 1. 定义输出模型
class ChatResponse(BaseModel):
    """聊天响应"""

    content: str = Field(description="回复内容")
    mood: str = Field(default="neutral", description="情绪状态")
    follow_up_questions: list[str] = Field(
        default_factory=list, description="追问建议"
    )


class CodeExplanation(BaseModel):
    """代码解释结果"""

    explanation: str
    complexity: str
    key_concepts: list[str]
    related_topics: list[str]


# 2. 创建流式Agent
stream_agent = Agent(
    "openai:gpt-4o",
    result_type=ChatResponse,
    system_prompt="你是一个友好的AI助手，善于用简洁的语言解释复杂概念。",
)


# 3. 流式文本输出（逐token）
async def stream_text_response(user_message: str) -> None:
    """流式输出文本响应"""
    print("用户:", user_message)
    print("助手: ", end="", flush=True)

    async with stream_agent.run_stream(user_message) as stream:
        # 方式1：逐token输出
        async for token in stream.stream_text():
            print(token, end="", flush=True)

    print("\n")


# 4. 流式结构化输出（逐字段）
async def stream_structured_response(user_message: str) -> ChatResponse:
    """流式输出结构化响应"""
    async with stream_agent.run_stream(user_message) as stream:
        # 方式2：流式构建结构化对象
        async for partial in stream.stream_structured():
            # partial是逐步构建的ChatResponse对象
            if partial.content:
                print(f"\r内容: {partial.content[:50]}...", end="", flush=True)

        # 获取最终完整结果
        result = await stream.get_result()
        return result.data


# 5. 多轮对话管理
class ConversationManager:
    """多轮对话管理器"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="你是一个友好的AI助手。记住之前的对话内容，保持上下文连贯。",
        )
        self.message_history: list = []

    async def chat(self, user_input: str) -> ChatResponse:
        """进行一轮对话"""
        result = await self.agent.run(
            user_input,
            message_history=self.message_history,
        )

        # 更新消息历史
        self.message_history = result.all_messages()

        return result.data

    async def chat_stream(self, user_input: str) -> None:
        """流式多轮对话"""
        print(f"用户: {user_input}")
        print("助手: ", end="", flush=True)

        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

        # 更新消息历史
        result = await stream.get_result()
        self.message_history = result.all_messages()
        print("\n")

    def reset(self) -> None:
        """重置对话"""
        self.message_history = []

    def get_history_summary(self) -> dict:
        """获取对话历史摘要"""
        return {
            "total_messages": len(self.message_history),
            "user_messages": sum(
                1 for m in self.message_history if hasattr(m, "content") and m.role == "user"
            ),
        }


# 6. 带工具的流式Agent
from pydantic_ai import Tool


class StreamingTravelAgent:
    """流式旅行助手"""

    def __init__(self):
        self.agent = Agent(
            "openai:gpt-4o",
            result_type=ChatResponse,
            system_prompt="你是一个旅行助手。使用工具查询实时信息，给出专业建议。",
        )
        self.message_history: list = []

        # 注册工具
        self._register_tools()

    def _register_tools(self) -> None:
        @self.agent.tool
        async def search_flights(
            ctx: RunContext[None],
            origin: str,
            destination: str,
        ) -> str:
            """搜索航班"""
            return f"找到3个从{origin}到{destination}的航班: CA1234(¥1280), MU5678(¥980), CZ9012(¥1580)"

        @self.agent.tool
        async def get_weather(ctx: RunContext[None], city: str) -> str:
            """查询天气"""
            return f"{city}今天晴转多云，25°C，适合出行"

    async def chat_stream(self, user_input: str) -> ChatResponse:
        """流式对话"""
        async with self.agent.run_stream(
            user_input,
            message_history=self.message_history,
        ) as stream:
            async for token in stream.stream_text():
                print(token, end="", flush=True)

            result = await stream.get_result()
            self.message_history = result.all_messages()
            print()
            return result.data


# 7. 流式输出与进度追踪
class StreamProgress:
    """流式输出进度追踪器"""

    def __init__(self):
        self.token_count = 0
        self.tool_calls = 0
        self.start_time: Optional[float] = None

    def report(self) -> dict:
        """生成进度报告"""
        import time

        elapsed = time.time() - self.start_time if self.start_time else 0
        return {
            "token_count": self.token_count,
            "tool_calls": self.tool_calls,
            "elapsed_seconds": round(elapsed, 2),
            "tokens_per_second": round(self.token_count / elapsed, 1) if elapsed > 0 else 0,
        }


# 运行示例
if __name__ == "__main__":
    import asyncio

    # 流式文本输出
    print("=== 流式文本输出 ===")
    asyncio.run(stream_text_response("什么是量子计算？"))

    # 多轮对话
    print("\n=== 多轮对话 ===")
    manager = ConversationManager()

    async def demo_conversation():
        await manager.chat_stream("我想学习Python，从哪里开始？")
        await manager.chat_stream("那数据结构呢？有什么推荐的学习路径？")
        await manager.chat_stream("能给我一个快速排序的例子吗？")
        print(f"对话统计: {manager.get_history_summary()}")

    asyncio.run(demo_conversation())

    # 流式旅行助手
    print("\n=== 流式旅行助手 ===")
    travel = StreamingTravelAgent()
    asyncio.run(travel.chat_stream("我想从北京去上海，天气怎么样？"))

关键要点：

• run_stream()返回异步上下文管理器，支持stream_text()和stream_structured()两种模式
• message_history参数传递历史消息，实现多轮对话上下文
• 流式模式下工具调用自动执行，LLM在获取工具结果后继续生成
• ConversationManager封装对话管理逻辑，适合生产环境使用

踩坑指南

坑1：未指定result_type导致输出为字符串

# ❌ 错误：没有result_type，输出是原始字符串
agent = Agent("openai:gpt-4o")
result = await agent.run("分析这段代码")
print(type(result.data))  # <class 'str'>，无法进行类型安全操作

# ✅ 正确：指定result_type，输出自动验证为结构化对象
class AnalysisResult(BaseModel):
    score: float
    issues: list[str]

agent = Agent("openai:gpt-4o", result_type=AnalysisResult)
result = await agent.run("分析这段代码")
print(type(result.data))  # <class 'AnalysisResult'>，类型安全

坑2：依赖类型不匹配导致运行时错误

# ❌ 错误：deps_type与实际传入的依赖类型不匹配
@dataclass
class DepsA:
    api_key: str

agent = Agent("openai:gpt-4o", deps_type=DepsA)

@dataclass
class DepsB:
    token: str  # 不同的字段

result = await agent.run("hello", deps=DepsB(token="xxx"))  # 运行时类型错误

# ✅ 正确：确保deps_type与传入的依赖类型一致
result = await agent.run("hello", deps=DepsA(api_key="sk-xxx"))

坑3：工具函数缺少类型注解

# ❌ 错误：工具函数没有类型注解，LLM无法理解参数含义
@agent.tool
async def search(query):  # 缺少类型注解和描述
    return "results"

# ✅ 正确：完整的类型注解和Field描述
@agent.tool
async def search(
    ctx: RunContext[None],
    query: str = Field(description="搜索关键词"),
    limit: int = Field(default=10, description="返回结果数量上限"),
) -> str:
    return "results"

坑4：流式响应中忘记处理异常

# ❌ 错误：流式响应没有异常处理，网络断开时程序崩溃
async with agent.run_stream(prompt) as stream:
    async for token in stream.stream_text():
        print(token, end="")

# ✅ 正确：添加异常处理和超时控制
import asyncio

try:
    async with asyncio.timeout(30):
        async with agent.run_stream(prompt) as stream:
            async for token in stream.stream_text():
                print(token, end="")
except asyncio.TimeoutError:
    print("\n[超时] 响应时间过长，请重试")
except Exception as e:
    print(f"\n[错误] {type(e).__name__}: {e}")

坑5：message_history未正确传递导致上下文丢失

# ❌ 错误：每次调用都不传历史，Agent无法记住之前的对话
result1 = await agent.run("我叫张三")
result2 = await agent.run("我叫什么名字？")  # Agent不知道你叫张三

# ✅ 正确：传递message_history保持上下文
result1 = await agent.run("我叫张三")
history = result1.all_messages()
result2 = await agent.run("我叫什么名字？", message_history=history)  # Agent知道你叫张三

错误排查表

错误信息	原因	解决方案
ValidationError: field required	LLM输出缺少必填字段	检查模型定义，确保字段有默认值或增加retries
UnexpectedModelBehavior	LLM输出无法解析为result_type	增加system_prompt中的格式说明，降低temperature
ModelHTTPError: 429	API调用频率超限	添加请求间隔，使用指数退避重试
ToolRunError	工具函数执行异常	检查工具函数逻辑，添加try/except
RunContextError: deps not provided	运行时未提供依赖	确保调用run()时传入deps参数
StreamCompleteError	流式响应中断	检查网络连接，添加重连逻辑
SchemaGenerationError	result_type包含不支持的类型	避免使用复杂嵌套Union，简化模型定义
TimeoutError	LLM响应超时	增加timeout设置，优化prompt减少token消耗
AuthenticationError	API密钥无效或过期	检查环境变量中的API密钥配置
JSONDecodeError in tool args	LLM生成的工具参数不是有效JSON	在工具描述中明确参数格式，增加示例

进阶优化

1. 模型降级策略：为Agent配置fallback模型，当主模型不可用时自动切换

from pydantic_ai.models import Model

# 主模型不可用时降级到备用模型
agent = Agent(
    "openai:gpt-4o",  # 主模型
    result_type=AnalysisResult,
    model_settings={"fallback": "openai:gpt-4o-mini"},  # 备用模型
)

2. 工具调用缓存：对幂等工具的调用结果进行缓存，减少重复API请求

import hashlib
import json

tool_cache: dict[str, str] = {}

async def cached_tool_call(tool_name: str, args: dict) -> str:
    cache_key = hashlib.md5(
        json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()
    ).hexdigest()
    if cache_key in tool_cache:
        return tool_cache[cache_key]
    result = await execute_tool(tool_name, args)
    tool_cache[cache_key] = result
    return result

3. 结构化输出Schema优化：精简Schema描述，降低LLM生成错误格式的概率

class OptimizedResult(BaseModel):
    # 使用简洁的Field描述，避免冗长说明导致LLM困惑
    score: float = Field(ge=0, le=10, description="0-10分")
    issues: list[str] = Field(max_length=5, description="最多5个问题")
    # 避免使用Optional，给明确默认值
    category: str = Field(default="general", description="分类")

4. 并发Agent调用：使用asyncio.gather并行执行多个Agent任务

import asyncio

async def parallel_analysis(texts: list[str]) -> list[SentimentResult]:
    tasks = [sentiment_agent.run(text) for text in texts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [
        r.data if isinstance(r, AgentRunResult) else SentimentResult(text=t, sentiment=SentimentType.NEUTRAL, confidence=0, keywords=[])
        for r, t in zip(results, texts)
    ]

5. 消息历史裁剪：对过长的对话历史进行摘要裁剪，控制token消耗

def trim_message_history(
    messages: list,
    max_messages: int = 20,
) -> list:
    """裁剪消息历史，保留最近的max_messages条"""
    if len(messages) <= max_messages:
        return messages
    # 保留系统消息 + 最近的对话
    system_msgs = [m for m in messages if hasattr(m, "role") and m.role == "system"]
    recent_msgs = messages[-max_messages:]
    return system_msgs + recent_msgs

方案对比

特性	原生OpenAI SDK	LangChain	Pydantic AI
类型安全	❌ 无	⚠️ 部分	✅ 完整Pydantic验证
依赖注入	❌ 手动管理	⚠️ 全局变量	✅ 类型注解自动注入
结构化输出	⚠️ 需手动解析	⚠️ 需OutputParser	✅ 自动验证+重试
工具注册	⚠️ 手写JSON Schema	⚠️ 装饰器+手动Schema	✅ 类型注解自动生成
流式响应	⚠️ 需手动处理	✅ 支持	✅ 原生支持+结构化流
学习曲线	低	高	中
代码量	多	中	少
生产就绪	⚠️ 需大量封装	⚠️ 抽象过重	✅ 开箱即用

总结

Pydantic AI将Python类型系统的力量带入AI Agent开发。通过result_type约束输出、deps_type管理依赖、@agent.tool注册工具、run_stream()流式响应这5个核心模式，开发者可以用类型安全的方式构建生产级AI应用。记住：类型注解不是文档，而是契约——Pydantic AI让这份契约在LLM交互中得到严格执行。

在线工具推荐

• JSON格式化 - 格式化和验证Agent输出的JSON数据
• cURL转代码 - 将API调试cURL命令转换为Python代码
• 哈希计算 - 计算API密钥或数据的哈希值，用于缓存键生成