Giới thiệu
Các mô hình ngôn ngữ lớn (LLM) thường chỉ tạo ra dữ liệu ở dạng văn bản, chưa có cấu trúc rõ ràng. Dù bạn yêu cầu LLM trả về kết quả dưới dạng JSON hay định dạng nào đó, thực chất nó chỉ trả về một chuỗi văn bản giống như JSON hợp lệ. Vì vậy, đầu ra có thể bị sai tên trường, thiếu trường, sai kiểu dữ liệu hoặc bị chèn thêm các câu chú thích, tiêu đề, v.v. Nếu không kiểm tra và xác thực kỹ, những lỗi này dễ dẫn đến lỗi phần mềm và khó phát hiện nguyên nhân.
Pydantic là một thư viện Python giúp kiểm tra, xác thực dữ liệu đầu vào/đầu ra dựa trên kiểu dữ liệu bạn định nghĩa sẵn. Pydantic sẽ tự động kiểm tra dữ liệu trả về từ LLM có khớp với schema bạn yêu cầu không, chuyển đổi kiểu dữ liệu nếu có thể và trả về lỗi rõ ràng khi xác thực không thành công. Nhờ đó, bạn có thể thiết lập một “cam kết” rõ ràng giữa dữ liệu LLM cung cấp và yêu cầu của ứng dụng.
Trong bài viết này, bạn sẽ biết cách dùng Pydantic để xác thực kết quả từ LLM, từ việc xây dựng schema, xử lý phản hồi lỗi định dạng, làm việc với dữ liệu lồng nhau, tích hợp vào API LLM, đến việc thử lại khi xác thực thất bại.
🔗 Xem mã nguồn ví dụ tại GitHub. Trước khi thử nghiệm, hãy cài đặt Pydantic phiên bản 2.x với phần mở rộng email: `pip install pydantic[email]`.
Bắt đầu với ví dụ cơ bản
Giả sử bạn muốn xây dựng một công cụ tự động trích xuất thông tin liên hệ từ văn bản tự do. LLM sẽ nhận một đoạn văn bản và trả về dữ liệu có cấu trúc, bạn sẽ xác thực dữ liệu này bằng Pydantic:
from pydantic import BaseModel, EmailStr, field_validator
from typing import Optional
class ContactInfo(BaseModel):
name: str
email: EmailStr
phone: Optional[str] = None
company: Optional[str] = None
@field_validator('phone')
@classmethod
def validate_phone(cls, v):
if v is None:
return v
cleaned = ''.join(filter(str.isdigit, v))
if len(cleaned) < 10:
raise ValueError('Phone number must have at least 10 digits')
return cleaned
Các lớp kế thừa từ BaseModel sẽ tự động xác thực dữ liệu đầu vào. Ví dụ, trường email dùng EmailStr để kiểm tra định dạng email, còn trường phone dùng validator để kiểm tra và chuẩn hoá số điện thoại tối thiểu 10 chữ số.
Cách sử dụng model này để xác thực dữ liệu trả về từ LLM như sau:
import json
llm_response = '''
{
"name": "Sarah Johnson",
"email": "[email protected]",
"phone": "(555) 123-4567",
"company": "TechCorp Industries"
}
'''
data = json.loads(llm_response)
contact = ContactInfo(**data)
print(contact.name)
print(contact.email)
print(contact.model_dump())
Chỉ cần truyền dữ liệu vào ContactInfo, Pydantic sẽ kiểm tra toàn bộ nội dung. Nếu có lỗi, bạn sẽ nhận được báo lỗi chi tiết để dễ xác định vấn đề.
Xử lý đầu ra LLM không đúng chuẩn
Không phải lúc nào LLM cũng trả về kết quả JSON chuẩn xác. Đôi khi, nó còn chèn thêm tiêu đề, nhận xét hoặc định dạng markdown. Để xử lý các trường hợp này, bạn có thể tách phần JSON bằng regex và xác thực như sau:
from pydantic import BaseModel, ValidationError, field_validator
import json
import re
class ProductReview(BaseModel):
product_name: str
rating: int
review_text: str
would_recommend: bool
@field_validator('rating')
@classmethod
def validate_rating(cls, v):
if not 1 <= v <= 5:
raise ValueError('Rating must be an integer between 1 and 5')
return v
def extract_json_from_llm_response(response: str) -> dict:
"""Tách phần JSON từ phản hồi của LLM (nếu bị lẫn chữ thừa)."""
json_match = re.search(r'\{.*\}', response, re.DOTALL)
if json_match:
return json.loads(json_match.group())
raise ValueError("Không tìm thấy JSON trong phản hồi")
def parse_review(llm_output: str) -> ProductReview:
"""Phân tích và xác thực dữ liệu LLM trả về."""
try:
data = extract_json_from_llm_response(llm_output)
review = ProductReview(**data)
return review
except json.JSONDecodeError as e:
print(f"Lỗi đọc JSON: {e}")
raise
except ValidationError as e:
print(f"Lỗi xác thực dữ liệu: {e}")
raise
except Exception as e:
print(f"Lỗi khác: {e}")
raise
Cách này dùng regex tìm đoạn JSON đầu tiên trong chuỗi, loại bỏ các phần thừa đầu/cuối. Mỗi loại lỗi đều được phân biệt rõ ràng: JSON sai định dạng, dữ liệu không hợp lệ với schema, hoặc lỗi khác.
Ví dụ thực tế với phản hồi có thêm chữ thừa:
messy_response = '''
Here's the review in JSON format:
{
"product_name": "Wireless Headphones X100",
"rating": 4,
"review_text": "Great sound quality, comfortable for long use.",
"would_recommend": true
}
Hope this helps!
'''
review = parse_review(messy_response)
print(f"Sản phẩm: {review.product_name}")
print(f"Đánh giá: {review.rating}/5")
Trình phân tích sẽ tự động tách JSON và xác thực đúng theo schema định nghĩa.
Làm việc với dữ liệu lồng nhau
Trong thực tế, dữ liệu thường có cấu trúc lồng nhau, ví dụ sản phẩm có nhiều đánh giá, thông số kỹ thuật. Pydantic hỗ trợ xác thực dữ liệu lồng nhau rất thuận tiện:
from pydantic import BaseModel, Field, field_validator
from typing import List
class Specification(BaseModel):
key: str
value: str
class Review(BaseModel):
reviewer_name: str
rating: int = Field(..., ge=1, le=5)
comment: str
verified_purchase: bool = False
class Product(BaseModel):
id: str
name: str
price: float = Field(..., gt=0)
category: str
specifications: List[Specification]
reviews: List[Review]
average_rating: float = Field(..., ge=1, le=5)
@field_validator('average_rating')
@classmethod
def check_average_matches_reviews(cls, v, info):
reviews = info.data.get('reviews', [])
if reviews:
calculated_avg = sum(r.rating for r in reviews) / len(reviews)
if abs(calculated_avg - v) > 0.1:
raise ValueError(
f'Average rating {v} does not match calculated average {calculated_avg:.2f}'
)
return v
Ở đây, model Product có các trường là danh sách các đối tượng con Specification, Review. Validator cho average_rating giúp kiểm tra điểm trung bình có khớp với các đánh giá thực tế không. Khi truyền dữ liệu vào, Pydantic sẽ tự động xác thực tất cả các lớp con bên trong.
Ví dụ xác thực dữ liệu lồng nhau:
llm_response = {
"id": "PROD-2024-001",
"name": "Smart Coffee Maker",
"price": 129.99,
"category": "Kitchen Appliances",
"specifications": [
{"key": "Capacity", "value": "12 cups"},
{"key": "Power", "value": "1000W"},
{"key": "Color", "value": "Stainless Steel"}
],
"reviews": [
{
"reviewer_name": "Alex M.",
"rating": 5,
"comment": "Makes excellent coffee every time!",
"verified_purchase": True
},
{
"reviewer_name": "Jordan P.",
"rating": 4,
"comment": "Good but a bit noisy",
"verified_purchase": True
}
],
"average_rating": 4.5
}
product = Product(**llm_response)
print(f"{product.name}: ${product.price}")
print(f"Đánh giá trung bình: {product.average_rating}")
print(f"Số lượng đánh giá: {len(product.reviews)}")
Chỉ với một lần xác thực, toàn bộ dữ liệu lồng nhau đều được kiểm tra đúng chuẩn.
Kết hợp Pydantic với API LLM và các framework phổ biến
Khi bạn làm việc với API của OpenAI hoặc các framework như LangChain và LlamaIndex, việc xác thực dữ liệu với Pydantic cũng rất dễ dàng.
Xác thực dữ liệu LLM với OpenAI API
Bạn có thể yêu cầu LLM trả về dữ liệu đúng mẫu JSON mong muốn, sau đó xác thực với Pydantic:
from openai import OpenAI
from pydantic import BaseModel
from typing import List
import os
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
class BookSummary(BaseModel):
title: str
author: str
genre: str
key_themes: List[str]
main_characters: List[str]
brief_summary: str
recommended_for: List[str]
def extract_book_info(text: str) -> BookSummary:
"""
Gửi yêu cầu tới LLM để trích xuất thông tin sách về dưới dạng JSON,
sau đó xác thực với Pydantic.
"""
prompt = f"""
Extract book information from the following text and return it as JSON.
Required format:
{{
"title": "book title",
"author": "author name",
"genre": "genre",
"key_themes": ["theme1", "theme2"],
"main_characters": ["character1", "character2"],
"brief_summary": "summary in 2-3 sentences",
"recommended_for": ["audience1", "audience2"]
}}
Text: {text}
Return ONLY the JSON, no additional text.
"""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[
{"role": "system", "content": "You are a helpful assistant that extracts structured data."},
{"role": "user", "content": prompt}
],
temperature=0
)
llm_output = response.choices[0].message.content
import json
data = json.loads(llm_output)
return BookSummary(**data)
Bạn đưa mẫu JSON cụ thể vào prompt để LLM trả về đúng định dạng. Sau đó chỉ cần xác thực với schema đã xây dựng.
Ví dụ:
book_text = """
'The Midnight Library' by Matt Haig is a contemporary fiction novel that explores
themes of regret, mental health, and the infinite possibilities of life. The story
follows Nora Seed, a woman who finds herself in a library between life and death,
where each book represents a different life she could have lived. Through her journey,
she encounters various versions of herself and must decide what truly makes a life worth living.
The book resonates with readers dealing with depression, anxiety, or life transitions.
"""
try:
book_info = extract_book_info(book_text)
print(f"Tiêu đề: {book_info.title}")
print(f"Tác giả: {book_info.author}")
print(f"Chủ đề: {', '.join(book_info.key_themes)}")
except Exception as e:
print(f"Lỗi khi trích xuất thông tin sách: {e}")
Sử dụng LangChain cùng Pydantic
LangChain hỗ trợ sẵn việc trích xuất dữ liệu cấu trúc theo schema Pydantic, với hai cách nổi bật:
Dùng PydanticOutputParser
Bạn chỉ cần định nghĩa model và truyền vào PydanticOutputParser, LangChain sẽ tự động sinh hướng dẫn định dạng cho LLM dựa trên schema:
from langchain_openai import ChatOpenAI
from langchain.output_parsers import PydanticOutputParser
from langchain.prompts import PromptTemplate
from pydantic import BaseModel, Field
from typing import List, Optional
class Restaurant(BaseModel):
"""Thông tin về nhà hàng."""
name: str = Field(description="Tên nhà hàng")
cuisine: str = Field(description="Loại ẩm thực phục vụ")
price_range: str = Field(description="Mức giá: $, $$, $$$, hoặc $$$$")
rating: Optional[float] = Field(default=None, description="Điểm đánh giá trên 5.0")
specialties: List[str] = Field(description="Món đặc sắc, nổi bật")
def extract_restaurant_with_parser(text: str) -> Restaurant:
parser = PydanticOutputParser(pydantic_object=Restaurant)
prompt = PromptTemplate(
template="Extract restaurant information from the following text.\n{format_instructions}\n{text}\n",
input_variables=["text"],
partial_variables={"format_instructions": parser.get_format_instructions()}
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
chain = prompt | llm | parser
result = chain.invoke({"text": text})
return result
Parser sẽ tự động tạo hướng dẫn định dạng và kiểm tra dữ liệu trả về.
Dùng with_structured_output (function calling)
Nếu LLM hỗ trợ function calling, bạn chỉ cần truyền schema, LangChain sẽ xử lý parsing tự động:
def extract_restaurant_structured(text: str) -> Restaurant:
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Restaurant)
prompt = PromptTemplate.from_template(
"Extract restaurant information from the following text:\n\n{text}"
)
chain = prompt | structured_llm
result = chain.invoke({"text": text})
return result
Cách này ngắn gọn, dễ dùng, độ chính xác cao.
Ví dụ:
restaurant_text = """
Mama's Italian Kitchen is a cozy family-owned restaurant serving authentic
Italian cuisine. Rated 4.5 stars, it's known for its homemade pasta and
wood-fired pizzas. Prices are moderate ($$), and their signature dishes
include lasagna bolognese and tiramisu.
"""
try:
restaurant_info = extract_restaurant_structured(restaurant_text)
print(f"Nhà hàng: {restaurant_info.name}")
print(f"Ẩm thực: {restaurant_info.cuisine}")
print(f"Món đặc sắc: {', '.join(restaurant_info.specialties)}")
except Exception as e:
print(f"Lỗi: {e}")
Kết hợp với LlamaIndex
LlamaIndex cũng có nhiều cách trích xuất dữ liệu cấu trúc với Pydantic, phù hợp cho cả ứng dụng nhỏ lẫn hệ thống lớn.
Cách đơn giản nhất dùng LLMTextCompletionProgram:
from llama_index.core.program import LLMTextCompletionProgram
from pydantic import BaseModel, Field
from typing import List, Optional
class Product(BaseModel):
"""Thông tin sản phẩm."""
name: str = Field(description="Tên sản phẩm")
brand: str = Field(description="Thương hiệu hoặc nhà sản xuất")
category: str = Field(description="Danh mục sản phẩm")
price: float = Field(description="Giá USD")
features: List[str] = Field(description="Tính năng nổi bật")
rating: Optional[float] = Field(default=None, description="Đánh giá khách hàng trên 5")
def extract_product_simple(text: str) -> Product:
prompt_template_str = """
Extract product information from the following text and structure it properly:
{text}
"""
program = LLMTextCompletionProgram.from_defaults(
output_cls=Product,
prompt_template_str=prompt_template_str,
verbose=False
)
result = program(text=text)
return result
Nếu muốn kiểm soát chi tiết quá trình parsing, bạn có thể dùng PydanticOutputParser:
from llama_index.core.program import LLMTextCompletionProgram
from llama_index.core.output_parsers import PydanticOutputParser
from llama_index.llms.openai import OpenAI
def extract_product_with_parser(text: str) -> Product:
prompt_template_str = """
Extract product information from the following text:
{text}
{format_instructions}
"""
llm = OpenAI(model="gpt-4o-mini", temperature=0)
program = LLMTextCompletionProgram.from_defaults(
output_parser=PydanticOutputParser(output_cls=Product),
prompt_template_str=prompt_template_str,
llm=llm,
verbose=False
)
result = program(text=text)
return result
Ví dụ:
product_text = """
The Sony WH-1000XM5 wireless headphones feature industry-leading noise cancellation,
exceptional sound quality, and up to 30 hours of battery life. Priced at $399.99,
these premium headphones include Adaptive Sound Control, multipoint connection,
and speak-to-chat technology. Customers rate them 4.7 out of 5 stars.
"""
try:
product_info = extract_product_with_parser(product_text)
print(f"Sản phẩm: {product_info.name}")
print(f"Thương hiệu: {product_info.brand}")
print(f"Giá: ${product_info.price}")
print(f"Tính năng: {', '.join(product_info.features)}")
except Exception as e:
print(f"Lỗi: {e}")
Thử lại (retry) khi dữ liệu trả về không hợp lệ
Nếu dữ liệu trả về từ LLM chưa đúng mong muốn, bạn có thể thử lại với prompt bổ sung, gửi kèm thông báo lỗi cụ thể từ lần xác thực trước để LLM hiểu cần sửa gì:
from pydantic import BaseModel, ValidationError
from typing import Optional
import json
class EventExtraction(BaseModel):
event_name: str
date: str
location: str
attendees: int
event_type: str
def extract_with_retry(llm_call_function, max_retries: int = 3) -> Optional[EventExtraction]:
"""Tự động thử lại khi xác thực thất bại, mỗi lần gửi kèm thông báo lỗi cho LLM."""
last_error = None
for attempt in range(max_retries):
try:
response = llm_call_function(last_error)
data = json.loads(response)
return EventExtraction(**data)
except ValidationError as e:
last_error = str(e)
print(f"Lần thử {attempt + 1} thất bại: {last_error}")
if attempt == max_retries - 1:
print("Đạt số lần thử tối đa, dừng lại")
return None
except json.JSONDecodeError:
print(f"Lần thử {attempt + 1}: JSON không hợp lệ")
last_error = "The response was not valid JSON. Please return only valid JSON."
if attempt == max_retries - 1:
return None
return None
Ví dụ mô phỏng LLM cải thiện dần qua từng lần thử:
def mock_llm_call(previous_error: Optional[str] = None) -> str:
"""Giả lập LLM trả về kết quả khác nhau dựa trên lỗi nhận được."""
if previous_error is None:
return '{"event_name": "Tech Conference 2024", "date": "2024-06-15", "location": "San Francisco"}'
elif "attendees" in previous_error.lower():
return '{"event_name": "Tech Conference 2024", "date": "2024-06-15", "location": "San Francisco", "attendees": "about 500", "event_type": "Conference"}'
else:
return '{"event_name": "Tech Conference 2024", "date": "2024-06-15", "location": "San Francisco", "attendees": 500, "event_type": "Conference"}'
result = extract_with_retry(mock_llm_call)
if result:
print(f"\nThành công! Sự kiện trích xuất: {result.event_name}")
print(f"Số người dự kiến: {result.attendees}")
else:
print("Không trích xuất được dữ liệu hợp lệ")
Lần đầu thiếu trường, lần hai sai kiểu, lần ba trả về đúng yêu cầu.
Kết luận
Pydantic giúp bạn chuyển đổi kết quả tự do, không đồng nhất từ LLM thành dữ liệu có cấu trúc, được kiểm tra kỹ càng về kiểu và giá trị. Khi kết hợp với các công cụ hiện đại như OpenAI API, LangChain, LlamaIndex, bạn sẽ vừa tận dụng sức mạnh AI vừa bảo đảm tính ổn định, an toàn cho ứng dụng.
Tóm lại, bạn nên:
- Định nghĩa schema rõ ràng, sát với nhu cầu thực tế
- Luôn xác thực dữ liệu, xử lý lỗi hợp lý (có thể thử lại khi cần)
- Dùng type hint, validator để bảo vệ tính toàn vẹn của dữ liệu
- Đưa ví dụ schema vào prompt để tăng độ chính xác đầu ra từ LLM
Hãy bắt đầu từ model đơn giản, bổ sung kiểm tra khi phát hiện các trường hợp dữ liệu đặc biệt từ LLM. Chúc bạn áp dụng thành công!
Tham khảo
- Function Calling Program for Structured Extraction | Tài liệu LlamaIndex Python
- Pydantic – LlamaIndex
- Structured output – Tài liệu LangChain
- How to Use Pydantic for LLMs – Pydantic
