生成式AI领域是人类历史上增长最快的领域之一。 随着技术的快速迭代，2025 年被广泛认为将是AI 代理与多 AI 代理 彻底崛起的标志性年份。这标志着我们将从单一的 LLM（大型语言模型）应用模式，转向更加动态和复杂的 AI 代理生态系统。
这一趋势已经初见端倪，大型科技公司如AWS、OpenAI、微软 等纷纷抢滩布局，接连发布 AI 代理框架。AI 的未来，不再只是单一模型的强大，而是多个智能代理协同工作的效率与创造力的结合。
Pydantic 是一个强大的 Python 库，用于轻松验证和解析数据。它能够确保数据的准确性并符合预期结构，尤其在处理诸如 JSON 文件、用户输入或 API 响应等外部数据时非常有用。
通过使用模型，Pydantic 能够自动完成字段验证（例如检查“是否为整数”或“字符串是否过长”），避免了手动编写繁琐的检查代码，从而极大提升开发效率。
from pydantic import BaseModel, EmailStr# Define the modelclass User(BaseModel):name: strage: intemail: EmailStr# Example inputuser_data = {"name": "Alice","age": 25,"email": "alice@example.com"}# Validate the inputuser = User(**user_data)print(user.name) # Aliceprint(user.age) # 25print(user.email)
PydanticAI 的关键特性与优势
结构化响应处理
利用 Pydantic 对静态和流式响应进行验证，确保数据处理的高可靠性，避免错误输入对系统造成影响。多模型兼容性
原生支持 OpenAI、Gemini 和 Groq 等主流模型，开发者可通过简洁的接口快速集成其他模型，灵活适配各种场景需求。深厚技术根基
PydanticAI 由 Pydantic 的开发者打造。Pydantic 是 LangChain、OpenAI SDK 等热门框架的核心技术，为开发者提供强大且成熟的工具支持。简化依赖管理
引入类型安全的依赖注入系统，大幅优化测试与迭代开发流程，提升代码可维护性和开发效率。Pythonic 设计理念
采用符合 Python 编程习惯的设计，提供直观的代理组合和控制流程操作，开发者能够快速上手，降低学习成本。高效监控与调试
集成 Logfire 工具，为 AI 驱动的应用程序提供性能监控与问题定位功能，助力开发者快速排查问题并优化系统性能。类型安全的工作流
提供强类型检查功能，确保开发过程中的工作流严谨性，最大限度减少运行时错误，提升系统的稳定性。活跃测试阶段
当前处于 Beta 测试阶段，不断引入社区反馈进行优化，为用户提供更多功能和更好的使用体验。
接下来，让我们开始安装 PydanticAI。
通过简单的pip install命令，即可快速完成安装过程：
pip install 'pydantic-ai-slim[openai,vertexai,logfire]'from dataclasses import dataclassfrom pydantic import BaseModel, Fieldfrom pydantic_ai import Agent, RunContextfrom bank_database import DatabaseConn@dataclassclass SupportDependencies:customer_id: intdb: DatabaseConnclass SupportResult(BaseModel):support_advice: str = Field(description='Advice returned to the customer')block_card: bool = Field(description="Whether to block the customer's card")risk: int = Field(description='Risk level of query', ge=0, le=10)support_agent = Agent('openai:gpt-4o',deps_type=SupportDependencies,result_type=SupportResult,system_prompt=('You are a support agent in our bank, give the ''customer support and judge the risk level of their query.'),)@support_agent.system_promptasync def add_customer_name(ctx: RunContext[SupportDependencies]) -> str:customer_name = await ctx.deps.db.customer_name(id=ctx.deps.customer_id)return f"The customer's name is {customer_name!r}"@support_agent.toolasync def customer_balance(ctx: RunContext[SupportDependencies], include_pending: bool) -> float:"""Returns the customer's current account balance."""return await ctx.deps.db.customer_balance(id=ctx.deps.customer_id,include_pending=include_pending,)async def main():deps = SupportDependencies(customer_id=123, db=DatabaseConn())result = await support_agent.run('What is my balance?', deps=deps)print(result.data)"""support_advice='Hello John, your current account balance,including pending transactions, is $123.45.' block_card=False risk=1"""result = await support_agent.run('I just lost my card!', deps=deps)print(result.data)"""support_advice="I'm sorry to hear that, John. We are temporarily blockingyour card to prevent unauthorized transactions." block_card=True risk=8"""
