1. 项目概述:当AI Agent遇上类型系统

最近在折腾AI Agent项目,发现一个特别普遍又头疼的问题:你精心设计的Agent,在跟LLM(大语言模型)对话时,经常给你返回一些“惊喜”。比如,你让它分析用户情绪并返回一个0到10的分数,它可能给你回个“比较开心”;你让它从一段文本里提取日期,它可能返回“下周二”这种模糊表述,而不是你程序能直接处理的 datetime 对象。这些非结构化的、不可预测的输出,就像在代码里埋下了一颗颗定时炸弹,让后续的业务逻辑处理变得异常脆弱,调试起来更是让人抓狂。

这其实就是AI应用开发,特别是Agent开发中的一个核心痛点: 如何让LLM这种非确定性的“黑盒”输出,变得确定、可靠、可编程? 传统的做法是写一大堆后处理的正则表达式、写复杂的解析逻辑,或者用提示词(Prompt)反复强调格式,但效果往往差强人意,而且代码会变得又臭又长。

直到我遇到了 PydanticAI 。这个框架的核心思想,用一句话概括就是: 给AI Agent套上一个强大的类型系统 。它基于Python生态里广受好评的Pydantic库,将类型提示(Type Hints)和模型验证的能力,直接注入到与LLM的交互流程中。简单来说,你不再需要祈祷LLM会按你的想法输出,而是可以像定义函数参数和返回值类型一样,去严格定义和约束LLM的输入与输出。

我把它用在实际的几个Agent项目里,效果立竿见影。以前需要花80%时间处理的格式错误、类型转换、边界情况检查,现在几乎都被框架自动消化了。开发效率飙升,代码的可读性和健壮性也上了一个台阶。这不仅仅是少写几行代码,更是从根本上改变了我们构建可靠AI应用的方式。接下来,我就结合实战,带你彻底搞懂PydanticAI,看看它如何帮你避开AI Agent开发中那些最常见的“坑”。

2. 核心设计:类型系统如何为AI Agent保驾护航

2.1 从“提示工程”到“类型契约”

在没有类型系统介入之前,我们与LLM的协作模式可以称为“提示工程”模式。我们通过精心设计的提示词(Prompt),试图引导LLM理解我们的意图并输出特定格式。例如:

prompt = """
请分析以下用户评论的情感倾向,并严格按照JSON格式输出,包含两个字段:
- `sentiment`: 字符串,取值为 `positive`, `neutral`, `negative` 之一。
- `confidence`: 浮点数,范围0到1,表示置信度。

评论:{user_comment}
"""

然后,我们调用LLM API,拿到回复后,开始心惊胆战地解析:

import json
import re

response = llm_client.chat(prompt)
# 尝试1:直接解析JSON
try:
    result = json.loads(response)
except json.JSONDecodeError:
    # 尝试2:用正则从回复中抠出JSON
    json_match = re.search(r'\{.*\}', response, re.DOTALL)
    if json_match:
        result = json.loads(json_match.group())
    else:
        # 尝试3:手动处理...
        raise ValueError("LLM返回无法解析")
# 继续验证result里的字段和类型...

这个过程充满了不确定性。LLM可能在JSON外加了说明,可能用了不同的引号,可能字段名拼写错误,可能 confidence 返回了字符串“0.85”。每一个“可能”都需要代码去防御,这就是那“80%的坑”的来源。

PydanticAI将这种模式转变为“类型契约”模式。你不再是通过自然语言去“请求”LLM,而是通过定义一个强类型的Pydantic模型来“要求”LLM。这个模型就是你和LLM之间的契约。框架会:

  1. 自动将你的模型结构转化为LLM能理解的提示词。
  2. 在后台要求LLM(特别是支持结构化输出的模型)直接生成符合该模型的数据。
  3. 自动用Pydantic的验证器对返回结果进行解析和校验。

你的代码从复杂的字符串处理和异常捕获,变成了清晰的定义和调用:

from pydantic import BaseModel, Field
from pydantic_ai import Agent

class SentimentAnalysis(BaseModel):
    sentiment: Literal['positive', 'neutral', 'negative']
    confidence: float = Field(ge=0, le=1)

agent = Agent('openai:gpt-4-turbo-preview', result_type=SentimentAnalysis)
result = await agent.run('用户评论:这个产品太棒了!')
# result.data 已经是一个验证通过的SentimentAnalysis实例
print(result.data.sentiment) # 'positive'
print(result.data.confidence) # 0.92 (已经是float类型)

这种转变是革命性的。它将AI交互从“自然语言协商”拉回到了“编程接口调用”的范畴,极大地提升了可靠性和开发体验。

2.2 PydanticAI 的核心组件与工作流

要理解PydanticAI,需要先理清它的几个核心概念,它们共同构成了一个高效的工作流。

1. Agent(智能体) Agent是执行任务的核心单元。它封装了LLM的调用、消息历史管理、工具(Tools)的使用以及最重要的——结果类型的约束。创建Agent时,你需要指定底层的LLM(如OpenAI GPT-4、Anthropic Claude等)和 result_type (一个Pydantic模型)。

2. Model(模型) 这里特指Pydantic数据模型,即 result_type 。它定义了Agent一次运行( run )所期望的输出结构。这个模型可以非常简单,比如一个字符串字段;也可以非常复杂,包含嵌套模型、联合类型(Union)、列表等。模型中的字段描述( Field(description=...) )会直接影响生成给LLM的指令。

3. Tool(工具) Tool是Agent可以调用的外部函数。当LLM认为自己需要获取额外信息(如查询数据库、调用API、进行计算)才能完成任务时,它可以“决定”调用一个Tool。在PydanticAI中,Tool同样受益于类型系统:Tool函数的参数和返回值也推荐用Pydantic模型来定义,这使得函数签名清晰,且调用时的参数传递由框架自动处理。

4. 工作流简述 一次典型的Agent执行流程如下:

  • 初始化 :你创建一个Agent,并为其绑定一个结果模型( result_type )和可选的工具集。
  • 运行 :你调用 agent.run(prompt, ...) ,传入用户输入和一些上下文。
  • 提示构建 :框架内部将你的 result_type 模型结构、字段描述、以及你提供的工具函数签名,整合成一份高度结构化的系统提示(System Prompt),发送给LLM。
  • LLM推理 :LLM基于系统提示和用户输入进行推理。如果定义了工具,LLM可能会决定调用工具,框架会执行工具函数并将结果以结构化格式返回给LLM,继续推理。
  • 结果解析与验证 :LLM最终输出一个旨在匹配 result_type 的结构化数据(通常是JSON)。PydanticAI接收后,立即使用Pydantic对其进行解析和验证。如果验证失败(类型错误、缺少字段等),框架可以配置重试机制或直接抛出清晰的错误。
  • 返回 :你将获得一个包含已验证数据( result.data )的响应对象。

这个工作流的关键在于, 类型定义驱动了整个交互过程 ,从提示生成到结果验证,形成了一个闭环的可靠性保障。

3. 实战演练:从零构建一个类型安全的AI Agent

理论说得再多,不如亲手实现一个。我们一起来构建一个“智能会议纪要生成器”Agent。它的功能是:接收一段会议录音的文本转录,自动提取关键信息,并生成结构化的会议纪要。

3.1 定义核心数据模型

这是最重要的一步,决定了Agent能力的边界和输出质量。我们需要仔细设计会议纪要应该包含哪些信息。

from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, Field
from enum import Enum

class Attendee(BaseModel):
    """参会者"""
    name: str = Field(description="参会人姓名")
    department: Optional[str] = Field(None, description="所属部门")
    role: Optional[str] = Field(None, description="在会议中的角色,如主持人、汇报人")

class ActionItem(BaseModel):
    """行动项"""
    task: str = Field(description="具体的任务描述")
    owner: str = Field(description="负责人,通常是一个参会者姓名")
    deadline: Optional[datetime] = Field(None, description="截止日期,请尽量解析为具体日期")

class Decision(BaseModel):
    """会议决议"""
    topic: str = Field(description="决议涉及的主题")
    content: str = Field(description="决议的具体内容")
    rationale: Optional[str] = Field(None, description="做出该决议的主要理由或依据")

class MeetingSummary(BaseModel):
    """会议纪要核心模型"""
    meeting_title: str = Field(description="会议主题")
    meeting_date: datetime = Field(description="会议日期")
    attendees: List[Attendee] = Field(description="参会人员列表")
    key_points: List[str] = Field(description="会议讨论的关键要点,每条尽量简洁")
    decisions: List[Decision] = Field(description="本次会议达成的决议")
    action_items: List[ActionItem] = Field(description="会后需要跟进的行动项")
    next_meeting_time: Optional[datetime] = Field(None, description="下次会议预定时间")
    summary: str = Field(description="对本次会议的总体总结,一段话")

注意 :字段的 description 至关重要!它不仅是代码注释,更会直接作为指令的一部分传递给LLM。描述要清晰、无歧义,明确告诉LLM这个字段期望的内容是什么。例如 deadline: Optional[datetime] ,LLM会知道需要解析出日期时间,并尝试格式化为ISO字符串。

3.2 创建并配置Agent

有了模型,创建Agent就非常简单了。我们使用OpenAI的模型,并将上面定义的 MeetingSummary 模型作为 result_type

import asyncio
from pydantic_ai import Agent
from dotenv import load_dotenv
import os

load_dotenv()  # 从.env文件加载OPENAI_API_KEY

# 创建Agent,指定模型和结果类型
meeting_agent = Agent(
    'openai:gpt-4-turbo',  # 使用支持JSON模式的GPT-4 Turbo
    result_type=MeetingSummary,
    system_prompt="你是一个专业的会议秘书,擅长从混乱的对话文本中提取结构化信息,生成格式规范、内容准确的会议纪要。请严格根据提供的模型字段要求输出。"
)

# 准备一段模拟的会议转录文本
transcript = """
团队周会 - 2024年5月27日
参会人:张三(技术主管)、李四(产品经理)、王五(开发)、赵六(测试)
...
张三:我们上周完成了用户登录模块的重构,性能提升了30%。
李四:很好。关于下个季度的“智能推荐”功能,我们需要确定技术方案。王五,你调研得怎么样?
王五:我对比了A方案和B方案。A方案开发快,但扩展性差;B方案初期投入大,但长期更稳定。
李四:我认为我们应该选择B方案,为未来考虑。大家有意见吗?
(众人表示同意)
张三:那就定B方案。王五,你本周内输出一个详细的设计文档。
王五:好的,我周五前给出来。
李四:另外,新功能的UI稿,赵六你同步开始设计测试用例吧。
赵六:收到,我下周初完成第一版用例。
李四:我们下周一同一时间再同步一次进度。
"""

3.3 运行Agent并处理结果

现在,让我们运行Agent并看看结果。注意,我们使用异步(async/await)接口,这是处理LLM调用的推荐方式。

async def generate_summary():
    result = await meeting_agent.run(transcript)
    
    # result.data 已经是MeetingSummary实例,并且通过了验证
    summary: MeetingSummary = result.data
    
    print(f"会议主题: {summary.meeting_title}")
    print(f"会议日期: {summary.meeting_date.strftime('%Y-%m-%d')}")
    print(f"\n参会人员:")
    for attendee in summary.attendees:
        print(f"  - {attendee.name} ({attendee.role or '参会者'})")
    
    print(f"\n关键决议:")
    for decision in summary.decisions:
        print(f"  * {decision.topic}: {decision.content}")
    
    print(f"\n行动项:")
    for item in summary.action_items:
        deadline_str = item.deadline.strftime('%m-%d') if item.deadline else '待定'
        print(f"  - [{deadline_str}] {item.owner}: {item.task}")
    
    print(f"\n会议总结: {summary.summary}")

# 运行异步函数
if __name__ == '__main__':
    asyncio.run(generate_summary())

执行这段代码,你会得到一份完全结构化的会议摘要。所有日期字段( meeting_date , deadline )都已经是Python的 datetime 对象,可以直接用于计算或存入数据库。所有列表都已经是Python列表,可以直接迭代。 你完全省去了手动解析JSON、转换类型、处理缺失字段的步骤。

实操心得 :第一次看到这个结果时,感觉就像魔法。但背后其实是类型系统在起作用。PydanticAI在调用LLM时,会利用像OpenAI的“JSON模式”这样的特性,强制LLM以指定的JSON格式思考并输出,大大提高了输出的合规率。即使LLM偶尔输出格式稍有偏差,Pydantic强大的验证和纠错能力也能在多数情况下将其“拉回正轨”。

4. 进阶技巧:工具调用与流式输出

4.1 为Agent装备“工具”

单纯的文本生成Agent能力有限。真正的智能体现在能主动使用工具获取信息。比如,我们的会议纪要Agent可能需要查询公司日历确认时间,或从知识库拉取项目背景。PydanticAI让工具调用也变得类型安全。

假设我们有一个工具,可以根据人名查询员工的详细信息。

from pydantic_ai.models import Tool

# 1. 定义工具的输入模型
class EmployeeQuery(BaseModel):
    name: str = Field(description="需要查询的员工姓名")

# 2. 定义工具的返回模型
class EmployeeInfo(BaseModel):
    name: str
    employee_id: str
    department: str
    email: str

# 3. 实现工具函数,并用`Tool`装饰器装饰
@Tool
async def get_employee_info(query: EmployeeQuery) -> EmployeeInfo:
    """根据姓名查询员工信息。这是一个模拟函数,实际应连接HR系统。"""
    # 模拟一个数据库查询
    mock_db = {
        "张三": EmployeeInfo(name="张三", employee_id="001", department="技术部", email="zhangsan@company.com"),
        "李四": EmployeeInfo(name="李四", employee_id="002", department="产品部", email="lisi@company.com"),
        # ...
    }
    info = mock_db.get(query.name)
    if info is None:
        # 工具也可以抛出异常,Agent会处理
        raise ValueError(f"未找到员工: {query.name}")
    return info

# 4. 创建Agent时传入工具
enriched_meeting_agent = Agent(
    'openai:gpt-4-turbo',
    result_type=MeetingSummary,
    system_prompt="...",
    tools=[get_employee_info]  # 注册工具
)

现在,当你运行这个Agent处理会议转录时,LLM在识别到参会者姓名后,可能会自动决定调用 get_employee_info 工具来获取他们的部门等信息,从而填充 Attendee 模型的 department 字段。整个过程是自动的,你只需要定义好工具的函数签名(输入输出模型),框架会处理工具调用的调度、参数传递和结果整合。

注意事项 :工具函数的参数 必须 是单个Pydantic模型实例,返回值也 最好 是一个Pydantic模型或基础类型。这保证了工具接口的清晰和类型安全。工具的描述(函数的docstring)很重要,LLM依靠它来决定是否以及何时调用该工具。

4.2 处理流式输出与中间状态

对于耗时长或需要实时反馈的任务,流式输出(Streaming)很重要。PydanticAI支持以流的方式获取Agent的思考过程和最终结果。

async def run_agent_with_stream():
    agent = Agent('openai:gpt-4-turbo', result_type=MeetingSummary)
    
    async with agent.run_stream(transcript) as result_stream:
        async for chunk in result_stream:
            # chunk有不同的类型
            if chunk.type == 'tool-call':
                print(f"[Agent正在调用工具: {chunk.tool_name}]")
            elif chunk.type == 'tool-result':
                print(f"[工具返回结果: {chunk.result}]")
            elif chunk.type == 'llm-response-delta':
                # 这是LLM正在生成文本的片段(思考过程)
                # 注意:最终的结构化结果不通过这个流传递
                print(chunk.delta, end='', flush=True)
            elif chunk.type == 'success':
                # 流结束,成功拿到最终验证过的数据
                final_summary: MeetingSummary = chunk.data
                print(f"\n\n[完成] 纪要已生成,标题: {final_summary.meeting_title}")
            elif chunk.type == 'error':
                print(f"\n[出错] {chunk.message}")

# 运行流式处理
asyncio.run(run_agent_with_stream())

流式处理让你可以构建响应更快的用户界面,例如在聊天应用中实时显示“AI正在思考...”、“正在查询信息...”等状态,极大提升用户体验。

5. 避坑指南与性能优化

在实际项目中用了一段时间PydanticAI后,我积累了一些宝贵的经验教训,能帮你避开不少弯路。

5.1 模型设计中的常见陷阱

1. 字段描述不清或矛盾 这是导致输出不符合预期的最常见原因。避免使用模糊的描述。

  • content: str = Field(description="内容")
  • content: str = Field(description="决议的具体、可执行的描述,避免使用‘讨论一下’、‘再研究’等模糊词汇”)

2. 过度复杂的嵌套和联合类型 虽然Pydantic支持复杂的类型,但LLM理解起来可能困难,导致输出不稳定。尽量保持模型扁平化。如果必须使用 Union (联合类型),确保每个子类型的区别足够明显,并在描述中写清楚。

3. 忽略可选字段的默认值 对于 Optional 字段,如果LLM无法从上下文中推断出值,它可能会直接忽略该字段。这有时是符合预期的,但有时你可能希望有一个回退值。可以考虑在Pydantic模型中使用 default 参数,或者在业务逻辑层处理 None 值。

5.2 提示词与系统指令的协同

PydanticAI会自动根据你的模型生成一部分系统指令,但你提供的 system_prompt 依然关键。两者需要协同工作。

  • system_prompt 中明确角色和任务 :告诉LLM“你是什么”,比如“你是专业的数据分析师”。
  • 让模型定义负责具体结构 :在字段描述中告诉LLM“每个字段要填什么”。不要试图在 system_prompt 里重复所有字段规则,那样容易冲突。
  • 使用 system_prompt 提供全局约束 :例如,“所有日期请统一使用YYYY-MM-DD格式”,“如果信息缺失,请明确标注为‘未知’,不要编造”。

5.3 错误处理与重试策略

即使有类型系统,LLM也可能出错。PydanticAI提供了内置的重试机制。

from pydantic_ai import Agent
from pydantic_ai.retry import RetryConfig

# 配置重试策略
retry_config = RetryConfig(
    max_retries=2,  # 最大重试次数
    retry_on=(ValueError,),  # 在哪些异常时重试(如Pydantic验证失败)
    backoff_factor=1.0  # 退避因子
)

agent = Agent(
    'openai:gpt-4-turbo',
    result_type=MyModel,
    retries=retry_config
)

当LLM输出无法解析成目标模型时,Pydantic会抛出 ValidationError 。配置了重试后,Agent会自动用相同的输入重新运行,给LLM一次“改正”的机会。对于非确定性错误,这能显著提高成功率。

5.4 性能与成本考量

1. 选择合适的模型 gpt-4 系列精度高但价格贵、速度慢。 gpt-3.5-turbo 成本低、速度快,但在复杂结构化任务上可能不如GPT-4稳定。根据任务复杂度做权衡。对于极其简单的提取任务,甚至可以考虑更小的模型。

2. 利用缓存 相同的输入往往产生相同的输出。对于生产环境,可以考虑对Agent的 run 调用结果进行缓存,例如使用 functools.lru_cache (注意处理异步)或外部缓存如Redis,这能大幅降低成本和延迟。

3. 批量处理 如果需要处理大量独立文本,不要用循环一次次调用 agent.run 。应该将任务收集起来,使用 asyncio.gather 进行并发调用,充分利用异步IO的优势。

async def batch_summarize(transcripts: List[str]):
    tasks = [meeting_agent.run(t) for t in transcripts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    # 处理results,注意个别任务可能失败
    for i, r in enumerate(results):
        if isinstance(r, Exception):
            print(f"任务{i}失败: {r}")
        else:
            # 处理成功的r.data
            pass

6. 真实场景问题排查实录

在开发过程中,你肯定会遇到各种问题。下面是我遇到的一些典型情况及其解决方法。

问题现象 可能原因 排查步骤与解决方案
运行时报 ValidationError 1. LLM输出格式不符合模型定义。
2. 字段类型不匹配(如期望 int 但收到 str )。
3. 缺少必需字段。
1. 检查字段描述 :确保 Field(description=...) 清晰无歧义,明确告知LLM格式(如“请输出整数”)。
2. 启用调试 :使用 agent.run(..., debug=True) 或在创建Agent时设置 debug=True ,查看发送给LLM的实际提示词和LLM的原始返回。这能最直观地发现问题。
3. 简化模型 :如果模型太复杂,先尝试用一个极简模型(只有一个字符串字段)测试,确保基础通信正常,再逐步增加复杂度。
LLM总是忽略某个字段 1. 字段描述不够突出或重要。
2. 上下文中确实没有该信息。
1. 强化描述 :在字段描述中强调其重要性,如“ 必须 提供该字段”。
2. 修改系统提示 :在 system_prompt 中再次强调需要关注该字段对应的信息。
3. 设为可选 :如果信息可能缺失,将字段类型改为 Optional[...] ,并在业务逻辑中处理 None 值。
工具从未被调用 1. 工具函数签名(输入模型)太复杂,LLM不知道何时调用。
2. 工具描述(docstring)不清晰。
3. LLM认为不需要工具也能完成任务。
1. 优化工具描述 :确保工具的docstring用自然语言清晰说明“在什么情况下应该调用本工具”。
2. 提供示例 :在 system_prompt 中给LLM一两个需要调用该工具的场景示例。
3. 测试工具 :单独写一个测试,手动构造一个明确需要该工具的输入,看Agent是否会调用。
流式输出中看不到结构化数据 误解了流式输出的内容。 llm-response-delta 事件返回的是LLM的 自然语言思考过程 ,不是最终的结构化JSON。最终验证通过的数据只在 success 事件中通过 chunk.data 提供。如果需要实时显示结构化数据的部分字段,目前框架支持有限,可能需要等待最终结果。
处理长文本时超时或报错 输入文本超过了模型上下文长度。 1. 分而治之 :将长文本分割成多个片段,让Agent分别处理每个片段,再合并结果。这需要设计更精巧的流程。
2. 使用摘要 :先用一个简单的Agent对长文本进行摘要,再将摘要交给主Agent处理。
3. 选择上下文更长的模型

一个具体的调试案例 : 我曾让Agent从一个产品需求描述中提取功能点列表( List[str] )。但LLM经常返回一个用“1、2、3”编号的段落,而不是JSON数组。通过 debug=True 模式,我发现自动生成的系统指令中只是说“输出一个列表”。我修改了字段描述,从 features: List[str] 改为:

features: List[str] = Field(description="提取出的核心功能点,请以纯JSON数组格式输出,例如:['用户登录', '数据看板', '报告导出']。不要添加任何编号、项目符号或额外文本。")

同时,在 system_prompt 里加了一句:“你的所有输出都必须是有效的JSON,符合 result_type 的定义。” 问题立刻得到解决。

给AI Agent套上PydanticAI这套“类型系统”,本质上是在非确定性的AI世界和确定性的软件工程世界之间,架起了一座坚固可靠的桥梁。它把我们从繁琐的字符串解析和异常处理中解放出来,让我们能更专注于业务逻辑和AI能力本身的设计。虽然它不能解决所有问题(比如LLM本身的事实错误或逻辑错误),但它确实扫清了通往可靠AI应用道路上80%的障碍。如果你正在构建严肃的AI应用,我强烈建议你尝试将PydanticAI纳入你的技术栈,感受一下类型安全带来的开发愉悦和信心。

更多推荐