Pydantic AI: Hướng dẫn cho người mới bắt đầu với ví dụ thực tiễn

Việc xây dựng ứng dụng vận hành bằng AI thường dẫn đến đầu ra không có cấu trúc, sai lệch kiểu dữ liệu và các vấn đề về độ tin cậy trong sản xuất. Cách tiếp cận truyền thống để tích hợp LLM vào ứng dụng Python thiếu cấu trúc và xác thực cần thiết cho hệ thống sản xuất. Pydantic AI giải quyết điều này bằng cách kết hợp khả năng xác thực dữ liệu của Pydantic với một khung tác nhân cho tương tác với LLM.

Trong hướng dẫn này, bạn sẽ học cách tạo các tác nhân AI có cấu trúc, cho ra đầu ra đã được xác thực và duy trì an toàn kiểu dữ liệu. Chúng tôi sẽ hướng dẫn thiết lập tác nhân với system prompt, xây dựng các công cụ hàm mà LLM có thể gọi trong cuộc hội thoại và triển khai xác thực đầu ra có cấu trúc. Bạn cũng sẽ khám phá cách quản lý phụ thuộc giữa các thành phần của tác nhân và phát trực tuyến phản hồi cho ứng dụng thời gian thực. Sau khi hoàn thành, bạn sẽ có trải nghiệm thực hành để xây dựng ứng dụng AI xử lý các quy trình phức tạp.

Nếu bạn đang tìm một khóa học thực hành về xây dựng hệ thống AI dạng tác nhân, tôi gợi ý khóa Building Multi-Agent Systems with LangGraph. 

Pydantic AI là gì?

Pydantic AI là một khung tác nhân Python mang lại cấu trúc và an toàn kiểu dữ liệu cho các ứng dụng LLM. Khác với tích hợp LLM cơ bản chỉ trả về văn bản thô, Pydantic AI xem tương tác AI như các cuộc trò chuyện có cấu trúc với đầu vào và đầu ra được xác thực, giúp bạn dễ dàng xây dựng ứng dụng đáng tin cậy.

Tính năng và lợi ích chính:

• Xác thực đầu ra có cấu trúc — Phản hồi AI của bạn tự động tuân theo các mô hình Pydantic, ngăn lỗi phân tích cú pháp và đảm bảo tính nhất quán dữ liệu
• Công cụ hàm (Function tools) — LLM có thể gọi các hàm Python của bạn trong cuộc trò chuyện, giúp truy cập dữ liệu thực và tính toán
• An toàn kiểu dữ liệu — Hỗ trợ typing đầy đủ giúp giảm lỗi khi chạy và hỗ trợ IDE tốt hơn khi xây dựng quy trình AI phức tạp
• System prompt — Định nghĩa chỉ dẫn rõ ràng cho tác nhân AI của bạn, luôn nhất quán trong mọi tương tác
• Dependency injection — Chia sẻ ngữ cảnh, kết nối cơ sở dữ liệu và tùy chọn người dùng xuyên suốt các thành phần tác nhân một cách gọn gàng
• Nhiều chế độ thực thi — Chạy tác nhân đồng bộ, bất đồng bộ, hoặc phát trực tuyến phản hồi theo thời gian thực tùy nhu cầu
• Tái sử dụng tác nhân — Tạo tác nhân một lần và tái sử dụng khắp ứng dụng của bạn, tương tự như router của FastAPI

Cách tiếp cận này biến AI từ việc tạo văn bản khó đoán thành các tương tác được định kiểu, đáng tin cậy, phù hợp tự nhiên với ứng dụng Python hiện có. Thay vì vật lộn với prompt engineering và phân tích đầu ra, bạn có thể tập trung xây dựng chức năng thực sự mà người dùng cần.

Yêu cầu tiên quyết

Trước khi bắt đầu, bạn sẽ cần vài thứ:

Kiến thức bắt buộc:

• Kiến thức nền tảng về Python : Thành thạo với lớp và hàm
• Cơ bản về lập trình bất đồng bộ trong Python
• Kiến thức cơ bản về Pydantic : Hiểu các model, xác thực trường và chú thích kiểu
• Quen thuộc với LLM :  Kiến thức cơ bản về cách mô hình ngôn ngữ hoạt động và khái niệm prompt

Hữu ích nhưng không bắt buộc:

• Khái niệm về tác nhân :  Một chút trải nghiệm với tác nhân AI và việc gọi công cụ, dù nội dung này sẽ được đề cập

Yêu cầu cài đặt:

• Python 3.9 trở lên
• API key của OpenAI và Anthropic
• Kỹ năng cơ bản về terminal/dòng lệnh để cài đặt gói

Nếu bạn mới làm quen với Pydantic, tài liệu chính thức của họ bao quát rất tốt phần cơ bản. Với kiến thức nền tảng về LLM, bất kỳ hướng dẫn nhập môn nào về ChatGPT hoặc các mô hình tương tự đều đủ để bạn theo dõi.

Bắt đầu với Pydantic AI

Xuyên suốt hướng dẫn, chúng ta sẽ xây dựng một tác nhân phân tích dữ liệu giúp hiểu hiệu suất bán hàng theo khu vực và sản phẩm. Tác nhân này sẽ bắt đầu đơn giản và dần trở nên tinh vi hơn khi chúng ta bổ sung các tính năng như công cụ tùy chỉnh, đầu ra có cấu trúc và khả năng phát trực tuyến.

Cài đặt và thiết lập

Trước khi viết mã, hãy cài đặt Pydantic AI và thiết lập môi trường:

pip install pydantic-ai

Bạn cũng sẽ cần cả OpenAI và Anthropic API key. Đặt chúng dưới dạng biến môi trường:

export OPENAI_API_KEY="your-api-key-here"export ANTHROPIC_API_KEY="your-api-key-here"

> Chúng ta sẽ dùng cả mô hình OpenAI và Anthropic vì một số tính năng trong Pydantic AI được hỗ trợ ở mô hình này nhưng không có ở mô hình kia.

Cho các ví dụ phân tích dữ liệu, chúng ta cũng sẽ dùng pandas để xử lý dữ liệu:

pip install pandas

Tạo tác nhân đầu tiên của bạn

Hãy bắt đầu với tác nhân đơn giản nhất — trả lời câu hỏi cơ bản về dữ liệu bán hàng. Trong Pydantic AI, một tác nhân là giao diện chính của bạn để tương tác với LLM (chạy đoạn mã sau trong một script 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)

Đầu ra:

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.

Điều này tạo ra một tác nhân hiểu vai trò của mình là nhà phân tích bán hàng và có thể đưa ra lời khuyên phù hợp. system_prompt cung cấp ngữ cảnh cho LLM về kiểu phản hồi bạn muốn, trong khi run_sync() thực thi tác nhân và trả về kết quả.

Ở đây, hàm run_sync() chỉ hoạt động trong script Python. Để chạy tác nhân trong Jupyter Notebook, bạn có thể dùng cú pháp sau:

result = await agent.run("...")print(result.output)

Thay đổi LLM và thiết lập của nó

Pydantic AI hỗ trợ nhiều nhà cung cấp LLM và cho phép bạn tùy chỉnh hành vi của mô hình. Dưới đây là cách thử nghiệm các tùy chọn khác nhau:

# 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.")

Mỗi mô hình mang đến khả năng khác nhau cho quy trình phân tích dữ liệu của bạn. Các mô hình Claude giỏi lập luận chi tiết và phân tích phức tạp, rất phù hợp khi bạn cần insight chuyên sâu từ dữ liệu bán hàng. GPT-4o-mini cho phản hồi nhanh hơn và chi phí thấp hơn, lý tưởng cho các câu hỏi nhanh hoặc khi xử lý khối lượng lớn truy vấn.

Ngoài việc chọn mô hình, bạn có thể tinh chỉnh hành vi của chúng bằng các thiết lập mô hình:

# Add model-specific settingsconfigured_agent = Agent(   'openai:gpt-4o',   system_prompt="You are a sales analyst.",   model_settings={       'temperature': 0.3,  # More focused, less creative responses       'max_tokens': 500,   # Limit response length   })

Các thiết lập mô hình bạn có thể tùy chỉnh:

• temperature — Kiểm soát mức sáng tạo (0.0 = tập trung, 1.0 = sáng tạo)
• max_tokens — Giới hạn độ dài phản hồi
• top_p — Thay thế cho temperature để kiểm soát độ ngẫu nhiên

Với tác vụ phân tích dữ liệu, giá trị temperature thấp (0.1–0.3) thường hiệu quả hơn vì bạn cần phản hồi nhất quán, có cơ sở thay vì sáng tạo. Thiết lập max_tokens giúp kiểm soát độ dài phản hồi, hữu ích khi bạn cần tóm tắt ngắn gọn hoặc lo ngại chi phí API.

Sử dụng công cụ trong tác nhân

Cho đến giờ, tác nhân phân tích bán hàng của chúng ta chỉ có thể làm việc với thông tin được cung cấp trong prompt. Nhưng phân tích dữ liệu thực tế thường cần truy cập thông tin bên ngoài, chạy tính toán, hoặc kết nối hệ thống doanh nghiệp. Đây là lúc công cụ phát huy tác dụng — chúng trao cho tác nhân khả năng hành động và thu thập thông tin trong cuộc trò chuyện.

Công cụ dựng sẵn

Công cụ dựng sẵn là các chức năng sẵn dùng do nhà cung cấp LLM cung cấp. Pydantic AI hỗ trợ tìm kiếm web và công cụ thực thi mã chạy trực tiếp trên hạ tầng của nhà cung cấp.

Bắt đầu với tìm kiếm web, rất phù hợp để nắm thông tin thị trường hiện tại:

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:…

Điều này thay đổi phân tích dữ liệu của bạn từ làm việc với thông tin cũ sang bao gồm dữ liệu thị trường theo thời gian thực. Thay vì chỉ phân tích dữ liệu nội bộ, tác nhân giờ có thể so sánh hiệu suất công ty bạn với xu hướng ngành hiện tại.

Giờ hãy thêm thực thi mã để phân tích toán học và xử lý dữ liệu:

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)

Thực thi mã vượt ra khỏi phản hồi văn bản đơn thuần để tiến tới phân tích tính toán thực sự. Tác nhân của bạn giờ có thể thực hiện phép tính phức tạp, tạo trực quan hóa và xử lý dữ liệu vốn không thể làm chỉ qua hội thoại. Đối tượng kết quả chứa cả phần phân tích của tác nhân và mọi tệp được tạo trong quá trình thực thi mã, bạn có thể tải xuống để dùng tiếp.

Viết công cụ tùy chỉnh

Dù công cụ dựng sẵn hữu ích, bạn thường cần công cụ tùy chỉnh hoạt động với hệ thống doanh nghiệp cụ thể của mình. Công cụ tùy chỉnh là các hàm Python mà tác nhân của bạn có thể gọi trong cuộc trò chuyện.

Hãy tạo các công cụ tính toán chỉ số bán hàng phổ biến:

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.

Các công cụ tùy chỉnh này xử lý phép tính bán hàng cụ thể mà doanh nghiệp bạn thường dùng. Công cụ tỷ lệ chuyển đổi giúp đo lường hiệu quả marketing, trong khi công cụ giá trị đơn hàng trung bình theo dõi hành vi chi tiêu của khách hàng. Công cụ tùy chỉnh kết nối tác nhân với logic kinh doanh và hệ thống riêng của bạn.

Để biết thêm các mẫu công cụ nâng cao, bao gồm dependency injection và xử lý lỗi, hãy xem tài liệu công cụ đầy đủ.

Sử dụng đầu ra có cấu trúc

Đến giờ các tác nhân của chúng ta trả về văn bản thuần, phù hợp để khám phá nhưng gây vấn đề trong ứng dụng sản xuất. Khi bạn cần xử lý phản hồi của tác nhân bằng chương trình, trích xuất điểm dữ liệu cụ thể, hoặc tích hợp đầu ra AI với hệ thống khác, bạn cần cấu trúc được đảm bảo.

Giả sử bạn muốn phân tích dữ liệu bán hàng và luôn nhận lại những chỉ số cụ thể. Với văn bản thuần, AI có thể trả về định dạng khác nhau mỗi lần. Với đầu ra có cấu trúc, bạn định nghĩa chính xác điều mình muốn:

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.

Lưu ý tham số output_type=SalesInsight. Điều này yêu cầu tác nhân phải trả về một đối tượng SalesInsight với chính xác bốn trường đó. Không hơn, không kém. Kết quả trả về trong result.output đã được tự động xác thực và định kiểu.

Đầu ra có cấu trúc hoạt động bằng cách chuyển mô hình Pydantic của bạn thành JSON schema mà LLM tuân theo. AI không thể lệch khỏi cấu trúc — nếu nó cố trả về thứ khác, Pydantic AI sẽ yêu cầu thử lại.

Với các kịch bản phân tích phức tạp hơn, bạn có thể lồng mô hình và dùng danh sách:

from typing import Listfrom pydantic import BaseModelclass RegionalMetrics(BaseModel):   region: str   revenue: float   units_sold: int   average_price: float   performance_rating: str  # "excellent", "good", "needs_improvement"class ComprehensiveAnalysis(BaseModel):   total_revenue: float   total_units: int   regional_breakdown: List[RegionalMetrics]   top_performer: str   areas_for_improvement: List[str]   quarterly_grade: str# Agent with detailed structured outputdetailed_agent = Agent(   'anthropic:claude-sonnet-4-0',   output_type=ComprehensiveAnalysis,   system_prompt=(       "Provide comprehensive sales analysis with detailed regional metrics. "       "Rate each region's performance and give an overall quarterly grade (A-F)."   ))result = detailed_agent.run_sync("""Analyze Q1 2025 performance:- North: $385k revenue, 470 units- South: $256k revenue, 334 units- East: $391.5k revenue, 381 units - West: $395.5k revenue, 457 units""")# Access structured datafor region in result.output.regional_breakdown:   print(f"{region.region}: {region.performance_rating} (${region.average_price:.2f} avg price)")print(f"Quarterly Grade: {result.output.quarterly_grade}")

North: B ($819.15 avg price)South: C ($766.47 avg price)East: A ($1027.56 avg price)West: A- ($865.43 avg price)Quarterly Grade: B+

Cách làm này biến AI từ tạo văn bản khó đoán thành xử lý dữ liệu đáng tin cậy. Bạn có thể lưu từng chỉ số theo khu vực vào cơ sở dữ liệu, kích hoạt cảnh báo dựa trên đánh giá hiệu suất, hoặc tự động tạo bảng điều khiển cho lãnh đạo. Không cần regex, không cần xử lý nhiều định dạng phản hồi — chỉ có dữ liệu sạch, đã xác thực mỗi lần.

output_type chấp nhận bất kỳ mô hình Pydantic, dataclass Python, TypedDict, hoặc các kiểu đơn giản như int hoặc list[str]. Với quy trình phân tích dữ liệu, điều này cho phép bạn xây dựng pipeline tin cậy, nơi phân tích AI cấp dữ liệu trực tiếp vào hệ thống kinh doanh hiện có.

Tin nhắn và lịch sử chat trong tác nhân Pydantic

Khi làm việc với tác nhân phân tích dữ liệu, bạn thường cần các cuộc trò chuyện trải dài nhiều lượt tương tác. Có thể bạn bắt đầu bằng việc hỏi xu hướng bán hàng, rồi muốn đi sâu vào khu vực cụ thể, hoặc so sánh các giai đoạn khác nhau. Tin nhắn và lịch sử chat cho phép bạn duy trì ngữ cảnh xuyên suốt các cuộc trò chuyện này.

Truy cập lịch sử tin nhắn

Mỗi lần bạn chạy tác nhân, Pydantic AI sẽ ghi nhớ toàn bộ cuộc trò chuyện. Bạn có thể truy cập lịch sử này để hiểu những gì đã xảy ra hoặc tiếp tục cuộc trò chuyện sau:

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

Phương thức all_messages() trả về tất cả – system prompt, câu hỏi của người dùng và phản hồi của tác nhân. Phương thức new_messages() chỉ trả về những gì diễn ra trong lần chạy hiện tại. Điều này hữu ích khi bạn xây dựng cuộc trò chuyện dài và muốn theo dõi phần thay đổi.

Tiếp tục cuộc trò chuyện với lịch sử tin nhắn

Để xây dựng tiếp dựa trên cuộc trò chuyện trước đó, hãy truyền lịch sử tin nhắn vào lần chạy kế tiếp. Điều này giúp tác nhân nhớ những gì đã thảo luận:

# 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)

Mỗi lần chạy mới sẽ xây dựng dựa trên ngữ cảnh trước đó. Tác nhân nhớ không chỉ câu hỏi của bạn, mà cả phản hồi trước đó của chính nó, tạo nên cuộc hội thoại phân tích mạch lạc.

Lưu trữ và tải tin nhắn thành JSON

Để lưu trữ dài hạn hoặc chia sẻ cuộc trò chuyện giữa các hệ thống, bạn có thể tuần tự hóa lịch sử tin nhắn thành JSON:

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")

Phương thức all_messages_json() trả về cuộc trò chuyện dưới dạng bytes JSON, rất phù hợp để lưu vào cơ sở dữ liệu, gửi qua API, hoặc lưu trữ phiên phân tích. Cũng có new_messages_json() nếu bạn chỉ muốn lưu phần mới nhất của cuộc trò chuyện.

Cách này cho phép bạn xây dựng quy trình phân tích bền vững. Bạn có thể lưu phiên phân tích của từng khách hàng, tiếp tục các cuộc điều tra dữ liệu phức tạp qua nhiều ngày, hoặc chia sẻ ngữ cảnh cuộc trò chuyện giữa các thành viên trong nhóm làm cùng một phân tích.

Làm việc với hình ảnh và tài liệu

Phân tích dữ liệu thực tế thường không chỉ có văn bản và con số. Bạn có thể cần phân tích biểu đồ từ báo cáo, trích xuất dữ liệu từ PDF, hoặc diễn giải đồ thị và trực quan hóa. Pydantic AI hỗ trợ đầu vào hình ảnh và tài liệu, cho phép tác nhân của bạn làm việc trực tiếp với các nguồn dữ liệu phong phú này.

Phân tích hình ảnh từ URL

Khi bạn có biểu đồ, đồ thị, hoặc trực quan hóa được lưu trực tuyến, bạn có thể trỏ tác nhân đến trực tiếp chúng:

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)

Tác nhân có thể nhận diện loại biểu đồ, đọc giá trị trên trục, phát hiện xu hướng và đưa ra insight kinh doanh giống như một nhà phân tích khi nhìn vào cùng trực quan hóa.

Làm việc với hình ảnh cục bộ

Với hình ảnh lưu cục bộ — như ảnh chụp màn hình từ công cụ BI hoặc biểu đồ xuất từ Excel — hãy dùng 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)

Cách này rất phù hợp cho quy trình báo cáo tự động. Bạn có thể tự động chụp màn hình dashboard rồi để AI cung cấp phần phân tích bằng văn bản đi kèm trực quan.

Phân tích tài liệu từ URL

Nhiều tài liệu doanh nghiệp được chia sẻ qua URL — báo cáo trên intranet công ty, PDF trên lưu trữ đám mây, hoặc các nghiên cứu công khai phục vụ phân tích của bạn:

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)

Tác nhân có thể đọc toàn bộ tài liệu, trích xuất điểm dữ liệu liên quan và cung cấp bản tóm tắt vốn sẽ tốn nhiều thời gian nếu do con người thực hiện.

Xử lý tài liệu cục bộ

Với tài liệu trên hệ thống cục bộ — như báo cáo xuất ra, hợp đồng, hay bản trình bày nội bộ — hãy dùng 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)

Điều này mở ra khả năng tự động hóa mạnh mẽ. Bạn có thể xử lý hàng loạt báo cáo, trích xuất dữ liệu từ hóa đơn, hoặc phân tích điều khoản hợp đồng — những tác vụ vốn đòi hỏi rà soát thủ công.

Lưu ý quan trọng:

Các mô hình khác nhau hỗ trợ định dạng tệp khác nhau. Phần lớn xử lý được định dạng hình ảnh phổ biến (PNG, JPEG) và PDF, nhưng hãy kiểm tra tài liệu của mô hình bạn dùng để biết cụ thể. Một số mô hình như Vertex AI của Google có thể truy cập trực tiếp URL lưu trữ đám mây, trong khi mô hình khác sẽ tải nội dung về trước.

Với quy trình phân tích dữ liệu, khả năng đa phương thức này bắc cầu giữa báo cáo trực quan và insight do AI cung cấp, giúp tự động hóa phân tích toàn bộ dải tài liệu doanh nghiệp bạn làm việc hàng ngày.

Bật phát trực tuyến (Streaming) trong Pydantic AI

Khi làm việc với tác vụ phân tích dữ liệu phức tạp, bạn muốn xem kết quả ngay khi chúng được tạo thay vì chờ phản hồi hoàn chỉnh. Streaming cho phép người dùng quan sát quá trình phân tích theo thời gian thực, rất hữu ích khi đặt câu hỏi chi tiết về hiệu suất bán hàng.

Pydantic AI cung cấp streaming tích hợp qua phương thức run_stream(). Cách hoạt động như sau:

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())

Hiểu cú pháp streaming:

Từ khóa async with tạo một kết nối tới LLM và giữ mở trong lúc nhận các mảnh văn bản. Hãy hình dung như mở một cuộc gọi điện thoại – bạn cần thiết lập kết nối, trò chuyện, rồi cúp máy đúng cách.

Vòng lặp async for xử lý từng mảnh văn bản khi chúng đến. Lưu ý stream_text() tái tạo toàn bộ văn bản mỗi lần – nên bạn sẽ nhận "Hello", rồi "Hello world", rồi "Hello world, how". Điều này cung cấp đầy đủ ngữ cảnh nhưng có thể lặp khi hiển thị.

Vì sao async và asyncio phối hợp với nhau:

Khi bạn hỏi LLM, nó cần thời gian để tạo phản hồi — thường 3–10 giây. Trong thời gian này, mô hình gửi lại những mẩu văn bản nhỏ mỗi 50–200 mili giây.

Không có async, chương trình của bạn sẽ bị treo hoàn toàn khi chờ từng mẩu. Với async, vòng lặp sự kiện của Python (hệ thống quản lý các giai đoạn chờ) có thể chuyển đổi giữa các tác vụ. Điều này giúp chương trình vẫn phản hồi và thậm chí chạy nhiều cuộc trò chuyện cùng lúc.

asyncio.run() là thứ khởi động toàn bộ quá trình – nó tạo vòng lặp sự kiện điều phối mọi thứ. Khi bạn gọi asyncio.run(demo_streaming()), nó thiết lập môi trường để các hàm async có thể tạm dừng và tiếp tục khi cần trong lúc chờ phản hồi từ LLM.

Giờ hãy tạo một hàm tái sử dụng hiển thị chỉ phần văn bản delta mới để có trải nghiệm streaming tốt hơn:

from pydantic_ai.messages import PartDeltaEvent, TextPartDeltaasync def stream_agent_response(agent: Agent, task: str):   """   Stream a response from any Pydantic AI agent showing just new text deltas.   """   async def event_handler(ctx, event_stream):       async for event in event_stream:           if isinstance(event, PartDeltaEvent) and isinstance(               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?"))

Phân tích hàm streaming:

Hàm stream_agent_response() tạo trải nghiệm streaming tự nhiên hơn bằng cách chỉ hiển thị văn bản mới khi xuất hiện. Cách hoạt động như sau:

1. Thiết lập trình xử lý sự kiện: Hàm event_handler() nội bộ lắng nghe các sự kiện trong quá trình tác nhân xử lý. Nó lọc PartDeltaEvent có TextPartDelta – đại diện cho phần văn bản mới được tạo.
2. Nội dung delta: Thay vì nhận toàn bộ văn bản tái tạo mỗi lần, event.delta.content_delta chỉ cung cấp những từ hoặc ký tự mới. Điều này tạo hiệu ứng máy đánh chữ, nơi bạn thấy phản hồi được viết ra từng chữ.
3. Thực thi tác nhân: Chúng ta dùng agent.run() với tham số event_stream_handler. Điều này yêu cầu tác nhân gọi trình xử lý tùy chỉnh của chúng ta cho mỗi sự kiện streaming, cho phép kiểm soát chi tiết những gì được hiển thị.
4. Xuất ra theo thời gian thực: Lệnh print(event.delta.content_delta, end="", flush=True) hiển thị ngay từng phần mới mà không xuống dòng, tạo dòng chữ mượt mà.

Cách tiếp cận này mang lại cảm giác tự nhiên như đang xem AI suy nghĩ theo thời gian thực, thay vì chờ phản hồi hoàn chỉnh hoặc thấy văn bản tái tạo lặp lại. Chúng ta sẽ dùng hàm này ở các phần sau để hiển thị phản hồi phát trực tuyến trong những ví dụ phân tích dữ liệu.

Kết luận

Xuyên suốt hướng dẫn này, bạn đã xây dựng hiểu biết toàn diện về Pydantic AI qua các ví dụ phân tích dữ liệu thực tiễn. Bạn đã học cách tạo tác nhân với system prompt, thêm cả công cụ dựng sẵn lẫn tùy chỉnh, cấu trúc đầu ra bằng mô hình Pydantic, và phát trực tuyến phản hồi để nhận phản hồi thời gian thực. Các thành phần này kết hợp lại để tạo hệ thống AI hoạt động như đối tác phân tích hơn là chatbot đơn thuần.

Ứng dụng vượt xa các ví dụ phân tích bán hàng mà chúng ta đã đề cập. Giờ bạn có thể xây dựng tác nhân AI đọc nguồn dữ liệu riêng của bạn, tuân theo quy tắc kinh doanh của bạn, và trả kết quả ở định dạng tương thích với hệ thống hiện có. Dù bạn đang tự động hóa việc tạo báo cáo, phân tích phản hồi khách hàng, hay xử lý dữ liệu tài chính, Pydantic AI mang lại cấu trúc và độ tin cậy cần thiết cho ứng dụng sản xuất.

Nếu bạn quan tâm đến xây dựng hệ thống đa tác nhân với LangGraph, hãy xem khóa học thực hành của chúng tôi.