构建由 AI 驱动的应用常常会遇到非结构化输出、类型不匹配以及生产可用性问题。将 LLM 集成到 Python 应用的传统方法缺乏生产系统所需的结构和校验。Pydantic AI 通过将 Pydantic 的数据验证与用于 LLM 交互的代理框架结合起来，解决了这些问题。
在本教程中，您将学习如何创建结构化的 AI 代理，使其产出经过验证、并保持类型安全的结果。我们将介绍如何使用系统提示设置代理、构建对话中可由 LLM 调用的函数工具，以及实现结构化输出验证。您还将了解如何在代理组件间管理依赖，以及为实时应用进行响应流式传输。到最后，您将亲手完成能处理复杂工作流的 AI 应用的构建。
让我们从最简单的代理开始——能够回答有关销售数据的基础问题。在 Pydantic AI 中，代理是进行 LLM 交互的主要接口（在 Python 脚本中运行以下代码）：
from pydantic_ai import Agent# Create a basic sales analysis agentsales_agent = Agent( 'openai:gpt-4o', system_prompt=( "You are a data analyst specializing in sales performance. " "Provide clear, actionable insights based on the data provided." ))# Run the agent with a simple questionresult = sales_agent.run_sync("What are the key metrics I should track for sales performance? Answer in two sentences.")print(result.output)技术支持
输出：
Key metrics to track for sales performance include conversion rate, which measures the percentage of leads that result in sales, and average deal size, indicating the average revenue earned per transaction. Additionally, monitoring customer acquisition cost (CAC) and customer lifetime value (CLV) helps evaluate the efficiency and profitability of sales efforts.技术支持
result =await agent.run("...")print(result.output)技术支持
更换 LLM 及其设置
Pydantic AI 支持多个 LLM 提供商，并允许您自定义模型行为。以下是如何尝试不同选项：
# Creating agents with different modelsclaude_agent = Agent('anthropic:claude-3-5-sonnet-20241022', system_prompt="You are a sales analyst.")gpt_mini_agent = Agent('openai:gpt-4o-mini', system_prompt="You are a sales analyst.")技术支持
内置工具是由 LLM 提供商提供的即用型函数。Pydantic AI 支持网页搜索与代码执行工具，它们直接在提供商的基础设施上运行。
先从网页搜索开始，它非常适合获取最新的市场信息：
from pydantic_ai import Agent, WebSearchTool# Create an agent with web search abilitiesmarket_research_agent = Agent( 'anthropic:claude-3-5-sonnet-20241022', builtin_tools=[WebSearchTool()], system_prompt=( "You are a sales analyst. Use web search to find current information. " "Keep responses to 3-4 sentences maximum." ))# Get current market informationresult = market_research_agent.run_sync( "Search for 2025 laptop sales trends and average selling prices. Give me the main trends.")print(result.output)技术支持
Let me search for current laptop sales trends.Based on the search results, here are the key laptop market trends for 2025:…技术支持
from pydantic_ai import Agent, CodeExecutionTool# Agent that can run calculations and data analysiscalculation_agent = Agent( 'anthropic:claude-sonnet-4-0', builtin_tools=[CodeExecutionTool()], system_prompt=( "You are a data analyst. Use code to perform calculations and create visualizations. " "Show your work and provide clear explanations." ))# Analyze sales data with calculationsresult = calculation_agent.run_sync("""I have sales data for Q1 2025:- North region: 150 laptops at $1,500 each, 320 phones at $500 each- South region: 89 laptops at $1,500 each, 245 phones at $500 each - East region: 201 laptops at $1,500 each, 180 phones at $500 each- West region: 167 laptops at $1,500 each, 290 phones at $500 eachCalculate total revenue by region and create a simple chart showing the results.""")print(result.output)```plaintext## SummaryBased on the Q1 2025 sales data analysis, here are the key findings:**Total Revenue by Region:**- **West Region**: $395,500 (highest)- **East Region**: $391,500- **North Region**: $385,000- **South Region**: $256,000 (lowest)技术支持
from pydantic_ai import Agent# Create an agent with custom business toolssales_agent = Agent( 'anthropic:claude-sonnet-4-0', system_prompt=( "You are a sales analyst. Use tools to calculate metrics and analyze business data. " "Give concise answers in 3-4 sentences maximum." ))@sales_agent.tool_plaindef calculate_conversion_rate(leads: int, sales: int) -> str: """Calculate conversion rate from leads to sales.""" if leads == 0: return "Cannot calculate conversion rate: no leads provided" conversion_rate = (sales / leads) * 100 return f"Conversion rate: {conversion_rate:.2f}% ({sales} sales from {leads} leads)"@sales_agent.tool_plaindef calculate_average_order_value(total_revenue: float, total_orders: int) -> str: """Calculate average order value.""" if total_orders == 0: return "Cannot calculate AOV: no orders provided" aov = total_revenue / total_orders return f"Average Order Value: ${aov:.2f} (${total_revenue:,.0f} revenue from {total_orders} orders)"# Use the custom toolsresult = sales_agent.run_sync( "Calculate conversion rate for 500 leads that resulted in 75 sales, and AOV for $180,000 revenue from 120 orders")print(result.output)技术支持
Your conversion rate is 15.00%, meaning you're successfully converting 75 out of every 500 leads into sales. The average order value is $1,500.00, indicating strong revenue per transaction. These metrics suggest effective sales processes with high-value customers, though there's room to improve lead conversion rates.技术支持
from pydantic import BaseModelfrom pydantic_ai import Agentclass SalesInsight(BaseModel): total_revenue:float best_performing_region:str worst_performing_region:str recommendation:str# Create an agent that returns structured analysisanalysis_agent = Agent( 'anthropic:claude-sonnet-4-0', output_type=SalesInsight, system_prompt="Analyze sales data and provide structured insights with clear recommendations.")# Get structured analysisresult = analysis_agent.run_sync("""Q1 2025 sales data:- North: $385,000 revenue (470 units)- South: $256,000 revenue (334 units) - East: $391,500 revenue (381 units)- West: $395,500 revenue (457 units)Analyze this data and provide insights.""")print(f"Total Revenue: ${result.output.total_revenue:,.0f}")print(f"Best Region: {result.output.best_performing_region}")print(f"Recommendation: {result.output.recommendation}")技术支持
Total Revenue: $1,428,000Best Region: WestRecommendation: Focus on improving South region performance through targeted marketing and sales training, as it significantly underperforms with only $256,000 revenue. The West region's success ($395,500) should be analyzed and replicated across other regions. Additionally, investigate why East region has the lowest average selling price per unit ($1,028) compared to West's highest ($865) - there may be pricing strategy opportunities.技术支持
每次运行代理，Pydantic AI 都会记录完整对话。您可以访问这段历史来了解过程，或在之后继续对话：
from pydantic_ai import Agent# Create our sales analysis agentsales_agent = Agent( 'anthropic:claude-sonnet-4-0', system_prompt="You are a sales analyst. Provide clear, concise analysis.")# First questionresult1 = sales_agent.run_sync("What are the main KPIs I should track for Q1 2025 sales?")print(result1.output)# Access the conversation historyall_messages = result1.all_messages()print(f"Total messages in conversation: {len(all_messages)}")# Just the new messages from this runnew_messages = result1.new_messages()print(f"New messages from this run: {len(new_messages)}")技术支持
Here are the essential KPIs to track for Q1 2025 sales:...Total messages in conversation: 2New messages from this run: 2技术支持
# Continue the conversation from where we left offresult2 = sales_agent.run_sync( "How should I calculate conversion rates for each of those KPIs?", message_history=result1.all_messages())print(result2.output)# The agent now has context from both interactionsprint(f"Full conversation length: {len(result2.all_messages())}")# Ask a follow-up that references the entire conversationresult3 = sales_agent.run_sync( "Which of these metrics would be most important for a monthly executive report?", message_history=result2.all_messages())print(result3.output)技术支持
import json# Get the conversation as JSONconversation_json = result3.all_messages_json()# Save to filewith open('sales_analysis_conversation.json', 'w') as f: f.write(conversation_json.decode('utf-8'))print("Conversation saved to JSON file")# Load it back laterwith open('sales_analysis_conversation.json', 'r') as f: loaded_conversation = f.read()# You can now use this loaded conversation in a new agent run# (Note: You'd need to convert back from JSON to message objects for actual use)print(f"Loaded conversation size: {len(loaded_conversation)} characters")技术支持
all_messages_json() 会以 JSON 字节的形式返回对话，适合保存到数据库、通过 API 发送或归档分析会话。若只想保存最近一段对话，可使用 new_messages_json()。
真实的数据分析不仅涉及文本和数字。您可能需要分析报告中的图表、从 PDF 中提取数据，或解读图形与可视化。Pydantic AI 支持图像与文档输入，让您的代理可直接处理这些富媒体数据源。
分析来自 URL 的图像
当您的图表、图形或可视化托管在线上时，可以直接让代理访问：
from pydantic_ai import Agent, ImageUrl# Create an agent that can analyze visual datavisual_agent = Agent( 'anthropic:claude-sonnet-4-0', system_prompt="You are a data analyst who can interpret charts, graphs, and business documents.")# Analyze a sales chart from a URLresult = visual_agent.run_sync([ "What are the main trends shown in this sales chart? Give me 3 key takeaways.", ImageUrl(url='https://example.com/q1-sales-chart.png')])print(result.output)技术支持
代理能够识别图表类型、读取坐标轴数值、发现趋势，并像人类分析师一样给出业务洞察。
处理本地图像
对于本地存储的图像——如来自 BI 工具的截图或从 Excel 导出的图表——使用 BinaryContent：
from pathlib import Pathfrom pydantic_ai import Agent, BinaryContent# Load a local sales dashboard screenshotdashboard_path = Path('sales_dashboard_jan_2025.png')result = visual_agent.run_sync([ "Analyze this sales dashboard. What metrics need attention based on what you see?", BinaryContent( data=dashboard_path.read_bytes(), media_type='image/png' )])print(result.output)技术支持
from pydantic_ai import Agent, DocumentUrl# Create an agent for document analysisdoc_agent = Agent( 'anthropic:claude-sonnet-4-0', system_prompt="Extract and summarize business-relevant insights from documents.")# Analyze a quarterly report PDFresult = doc_agent.run_sync([ "Read this quarterly earnings report and summarize the revenue trends and outlook.", DocumentUrl(url='https://company.com/reports/q1-2025-earnings.pdf')])print(result.output)技术支持
代理可以通读完整文档，提取相关数据点，并提供人类需要更长时间才能整理出的摘要。
处理本地文档
对于本地系统中的文档——如导出的报告、合同或内部演示文稿——使用 BinaryContent：
from pathlib import Pathfrom pydantic_ai import Agent, BinaryContent# Load a local Excel export saved as PDFreport_path = Path('monthly_sales_analysis_jan_2025.pdf')result = doc_agent.run_sync([ "Extract the key performance metrics from this monthly sales report and identify any concerning trends.", BinaryContent( data=report_path.read_bytes(), media_type='application/pdf' )])print(result.output)技术支持
import asynciofrom pydantic_ai import Agent# Create our sales analysis agentsales_agent = Agent( 'anthropic:claude-sonnet-4-0', system_prompt=( "You are a data analyst specializing in sales performance. " "Provide detailed, actionable insights based on the data provided." ))async def demo_streaming(): question = """ Analyze this sales scenario: Our company sold 1,200 laptops and 2,300 phones last quarter across 4 regions. Laptops average $1,500 each, phones $500 each. What are the key insights and recommendations? """ async with sales_agent.run_stream(question) as result: async for text in result.stream_text(): print(text, end='', flush=True)# Run the streaming analysisasyncio.run(demo_streaming())技术支持
理解流式语法：
async with 会创建到 LLM 的连接，并在接收文本分片时保持打开。可将其看作打电话——需要先建立连接、进行通话、然后正确挂断。
而 async for 循环会在每个文本分片到达时进行处理。请注意 stream_text() 每次都会再生成完整文本——因此您先得到“Hello”，然后是“Hello world”，再到“Hello world, how”。这能提供完整上下文，但用于展示时可能显得重复。
from pydantic_ai.messages import PartDeltaEvent, TextPartDeltaasync defstream_agent_response(agent: Agent, task:str):""" Stream a response from any Pydantic AI agent showing just new text deltas. """asyncdefevent_handler(ctx, event_stream):asyncfor event in event_stream:ifisinstance(event, PartDeltaEvent)andisinstance( event.delta, TextPartDelta ):print(event.delta.content_delta, end="", flush=True)# Use agent.run() with event_stream_handler for proper streaming result = await agent.run(task, event_stream_handler=event_handler) print() # Add a newline when streaming is complete return result# Usage exampleresponse = asyncio.run(stream_agent_response( sales_agent, "What are the top 3 metrics I should track for quarterly sales analysis?"))技术支持
%22%3E%3Cpath%20d%3D%22m-369%200h408v79h-408z%22%20fill%3D%22%237933ff%22%2F%3E%3Cpath%20d%3D%22m-148%200h306v45h-306z%22%20fill%3D%22%23974dff%22%2F%3E%3C%2Fg%3E%3C%2Fsvg%3E)
%22%2F%3E%3C%2FclipPath%3E%3Cg%20clip-path%3D%22url(%23a)%22%3E%3Cpath%20d%3D%22m76%200h484v36h-484z%22%20fill%3D%22%23974dff%22%2F%3E%3Cpath%20d%3D%22m0%200h384v24h-384z%22%20fill%3D%22%2306bdfc%22%20transform%3D%22matrix(1%200%200%20-1%200%2060)%22%2F%3E%3C%2Fg%3E%3C%2Fsvg%3E)
构建由 AI 驱动的应用常常会遇到非结构化输出、类型不匹配以及生产可用性问题。将 LLM 集成到 Python 应用的传统方法缺乏生产系统所需的结构和校验。Pydantic AI 通过将 Pydantic 的数据验证与用于 LLM 交互的代理框架结合起来，解决了这些问题。
在本教程中，您将学习如何创建结构化的 AI 代理，使其产出经过验证、并保持类型安全的结果。我们将介绍如何使用系统提示设置代理、构建对话中可由 LLM 调用的函数工具，以及实现结构化输出验证。您还将了解如何在代理组件间管理依赖，以及为实时应用进行响应流式传输。到最后，您将亲手完成能处理复杂工作流的 AI 应用的构建。
如果您在寻找动手实践的代理式 AI 系统课程，推荐 Building Multi-Agent Systems with LangGraph 课程。
什么是 Pydantic AI？
Pydantic AI 是一个 Python 代理框架，为 LLM 应用带来结构与类型安全。不同于只返回原始文本的基础集成方式，Pydantic AI 将 AI 交互视为具有经验证输入输出的结构化对话，从而更容易构建可靠的应用。
主要特性与优势：
这种方法将 AI 从不可预测的文本生成，转变为可依赖的、具备类型约束的交互，能够自然融入现有 Python 应用。与其苦苦应对提示工程和输出解析，不如把精力放在构建用户真正需要的功能上。
前置条件
在开始本教程之前，您需要准备以下内容：
必备知识：
有帮助但非必需：
环境要求：
如果您是 Pydantic 新手，他们的 官方文档对基础内容有很好的覆盖。至于 LLM 基础，任何 ChatGPT 或类似模型的入门指南都足以让您顺利跟进。
开始使用 Pydantic AI
在整个教程中，我们将构建一个数据分析代理，帮助我们理解不同地区与产品的销售表现。这个代理将从简单入手，随着我们加入自定义工具、结构化输出与流式能力而逐步增强。
安装与设置
在编写代码之前，先安装 Pydantic AI 并完成环境设置：
您还需要 OpenAI 与 Anthropic 的 API 密钥。将它们设置为环境变量：
> 我们会同时使用 OpenAI 与 Anthropic 的模型，因为 Pydantic AI 的部分特性在其中一家支持、而另一家不支持。
在数据分析示例中，我们还将使用 pandas 进行数据处理：
创建您的第一个代理
让我们从最简单的代理开始——能够回答有关销售数据的基础问题。在 Pydantic AI 中，代理是进行 LLM 交互的主要接口（在 Python 脚本中运行以下代码）：
输出：
这将创建一个理解自身为销售分析师角色、并能提供相关建议的代理。
system_prompt为 LLM 提供所需响应类型的上下文，而run_sync()用于执行代理并返回结果。这里，
run_sync()仅适用于 Python 脚本。若要在 Jupyter Notebook 中运行代理，您可以使用以下语法：更换 LLM 及其设置
Pydantic AI 支持多个 LLM 提供商，并允许您自定义模型行为。以下是如何尝试不同选项：
每个模型都会为您的数据分析工作流带来不同能力。Claude 模型擅长细致推理与复杂分析，在您需要从销售数据中获取深入洞察时非常理想。GPT-4o-mini 则提供更快的响应与更低成本，适合快速问答或批量处理大量查询。
除了选择不同模型外，您还可以通过模型设置微调其行为：
可自定义的模型设置：
对于数据分析任务，较低的 temperature 值（0.1–0.3）通常效果更好，因为您希望获得一致、客观的响应而非创意发挥。设置
max_tokens有助于控制响应长度，这在需要精炼摘要或关注 API 成本时很有用。在代理中使用工具
到目前为止，我们的销售分析代理只能使用我们在提示中提供的信息。但真实的数据分析往往需要访问外部信息、执行计算或连接业务系统。这正是工具发挥作用的地方——它们让您的代理在对话中具备行动与信息采集能力。
内置工具
内置工具是由 LLM 提供商提供的即用型函数。Pydantic AI 支持 网页搜索与代码执行工具，它们直接在提供商的基础设施上运行。
先从网页搜索开始，它非常适合获取最新的市场信息：
这让您的数据分析从仅依赖旧信息，转变为纳入实时市场数据。不再只是单独分析数据，您的代理还能将公司表现与当前行业趋势对照。
现在我们加入代码执行，用于数学分析和数据处理：
代码执行使响应不再局限于文本，而是进入真正的计算分析。您的代理如今可以执行复杂计算、创建可视化，并处理单靠对话无法完成的数据。结果对象包含代理的分析以及代码执行过程中生成的任何文件，您可以下载以便后续使用。
编写自定义工具
尽管内置工具很有用，但您经常需要与自身业务系统配套的自定义工具。 自定义工具 是在对话中可被代理调用的 Python 函数。
让我们创建用于计算常见销售指标的工具：
这些自定义工具负责处理您的业务经常使用的特定销售计算。转化率工具有助于衡量营销效果，平均订单价值工具则用于跟踪客户消费水平。自定义工具让代理与您的业务逻辑和系统紧密连接。
若想了解更高级的工具模式，包括依赖注入与错误处理，请参阅 完整工具文档。
使用结构化输出
到目前为止，我们的代理都返回纯文本响应，这在探索阶段还算可行，但在生产应用中会带来问题。当您需要以编程方式处理代理响应、解析特定数据点或将 AI 输出与其他系统集成时，就需要有保证的结构。
假设您想分析销售数据，并始终拿到特定指标。若使用纯文本，AI 每次可能返回不同格式；而使用结构化输出，您可以精确定义所需内容：
请注意
output_type=SalesInsight参数。它告诉代理必须返回一个包含且仅包含那四个字段的SalesInsight对象。结果以result.output形式返回，并会被自动验证与类型化。结构化输出通过将您的 Pydantic 模型转换成 LLM 遵循的 JSON schema 来实现。AI 无法偏离该结构——如果尝试返回其它内容，Pydantic AI 会要求其重试。
在更复杂的分析场景中，您可以嵌套模型并使用列表：
这种方式将 AI 从不可预测的文本生成转变为可靠的数据处理。您可以将各地区指标存入数据库、基于表现评级触发告警，或自动生成管理层看板。无需正则解析，也不用处理各异的响应格式——每次都能得到干净、已验证的数据。
output_type接受任意 Pydantic 模型、Python 数据类、TypedDict，甚至int或list[str]等简单类型。对于数据分析工作流，这意味着您可以构建可靠的管道，让 AI 分析直接馈入现有业务系统。Pydantic 代理中的消息与聊天历史
在使用数据分析代理时，您常常需要跨多次交互的对话。也许您先询问销售趋势，然后深入某些地区，或对比不同时间段。 消息与聊天历史可让您在这些对话中保持上下文。
访问消息历史
每次运行代理，Pydantic AI 都会记录完整对话。您可以访问这段历史来了解过程，或在之后继续对话：
all_messages()会返回所有内容——系统提示、用户提问与代理响应。new_messages()只返回当前这次运行产生的消息。当您构建较长对话并想跟踪变更时，这很有用。使用消息历史续接对话
要在此前对话基础上继续交流，将消息历史传给下一次代理运行。这样代理就能记住您之前讨论的内容：
每次新运行都会在先前上下文上叠加。代理不仅记住您的问题，也记住自己此前的回答，从而形成连贯的分析性对话。
将消息存入与加载为 JSON
为了长期存储或在系统间共享对话，您可以将消息历史序列化为 JSON：
all_messages_json()会以 JSON 字节的形式返回对话，适合保存到数据库、通过 API 发送或归档分析会话。若只想保存最近一段对话，可使用new_messages_json()。这种方式能帮助您构建持久的分析工作流。您可以保存每位客户的分析会话、在多天内继续复杂的数据调查，或在同一分析上将对话上下文共享给不同团队成员。
处理图像与文档
真实的数据分析不仅涉及文本和数字。您可能需要分析报告中的图表、从 PDF 中提取数据，或解读图形与可视化。 Pydantic AI 支持图像与文档输入，让您的代理可直接处理这些富媒体数据源。
分析来自 URL 的图像
当您的图表、图形或可视化托管在线上时，可以直接让代理访问：
代理能够识别图表类型、读取坐标轴数值、发现趋势，并像人类分析师一样给出业务洞察。
处理本地图像
对于本地存储的图像——如来自 BI 工具的截图或从 Excel 导出的图表——使用 BinaryContent：
这种方法非常适合自动化报告工作流。您可以以编程方式截取仪表盘截图，再让 AI 提供配套的文字分析。
分析来自 URL 的文档
许多业务文档通过 URL 共享——公司内网的报告、云存储中的 PDF，或为您的分析提供信息的公开研究论文：
代理可以通读完整文档，提取相关数据点，并提供人类需要更长时间才能整理出的摘要。
处理本地文档
对于本地系统中的文档——如导出的报告、合同或内部演示文稿——使用
BinaryContent：这为强大的自动化打开了大门。您可以批量处理报告、从发票中提取数据，或分析合同条款——这些通常需要人工审阅的任务都可自动化。
重要提示：
不同模型支持的文件格式不同。大多数能处理常见图像格式（PNG、JPEG）和 PDF，但请查阅所用模型的文档以了解细节。有些模型（如 Google 的 Vertex AI）可以直接访问云存储 URL，另一些则会先下载内容。
对于数据分析工作流，这种多模态能力弥合了可视化报告与 AI 洞察之间的鸿沟，使您能自动化分析日常处理的各类业务文档。
在 Pydantic AI 中启用流式传输
在处理复杂数据分析任务时，您希望在生成过程中就能看到结果，而不是等待完整响应。流式传输让用户可以实时观看分析展开，这在深入询问销售表现时尤其有价值。
Pydantic AI 通过
run_stream()提供内置流式能力。其工作方式如下：理解流式语法：
async with会创建到 LLM 的连接，并在接收文本分片时保持打开。可将其看作打电话——需要先建立连接、进行通话、然后正确挂断。而
async for循环会在每个文本分片到达时进行处理。请注意stream_text()每次都会再生成完整文本——因此您先得到“Hello”，然后是“Hello world”，再到“Hello world, how”。这能提供完整上下文，但用于展示时可能显得重复。为何要同时使用 async 与 asyncio：
当您向 LLM 提问时，生成响应需要时间——通常 3–10 秒。在此期间，模型会每 50–200 毫秒返回一小段文本。
如果不使用 async，程序会在等待每段文本时完全阻塞。有了 async，Python 的事件循环（管理这些等待的系统）可以在任务间切换。这意味着程序保持响应，甚至可以同时运行多段对话。
asyncio.run()是启动这一切的入口——它创建协调各项操作的事件循环。当您调用asyncio.run(demo_streaming())时，它会搭建起让异步函数在等待 LLM 响应时暂停与恢复的运行环境。现在让我们创建一个可复用函数，只展示新的文本增量，以获得更好的流式体验：
分解该流式函数：
stream_agent_response()通过只显示新到达的文本，营造出更自然的流式体验。其工作原理如下：event_handler()会在代理处理期间监听特定事件。它会筛选带有TextPartDelta的PartDeltaEvent——这些表示新生成的文本片段。event.delta.content_delta只给出新增的词或字符。这样就能看到“打字机”效果，响应像逐字输出一样呈现。event_stream_handler参数的agent.run()。这会让代理在每个流式事件上调用自定义处理器，从而对展示内容进行精细控制。print(event.delta.content_delta, end="", flush=True)会立刻显示每个新增片段且不换行，形成流畅的文本。这种方式让您感受到 AI 正在实时“思考”分析，而无需等待完整响应，也不会看到重复的重生成文本。接下来我们会在数据分析示例中使用该函数展示流式响应。
结语
通过本教程中的实践数据分析示例，您已系统掌握 Pydantic AI。您学会了如何使用系统提示创建代理、添加内置与自定义工具、用 Pydantic 模型构建结构化输出，并通过流式传输获取实时反馈。这些组件组合在一起，让 AI 系统更像分析伙伴，而非简单的聊天机器人。
应用场景远不止我们演示的销售分析。您现在可以构建读取特定数据源、遵循业务规则、并以兼容现有系统格式返回结果的 AI 代理。无论是自动生成报告、分析客户反馈，还是处理财务数据，Pydantic AI 都能为生产级应用提供所需的结构与可靠性。
如果您对 使用 LangGraph 构建多代理系统感兴趣，欢迎查看我们的动手课程。