Trong thế giới phát triển ứng dụng AI, việc lấy dữ liệu có cấu trúc từ LLM là yêu cầu bắt buộc. Bài viết này sẽ hướng dẫn bạn cách kết hợp Python Pydantic với thư viện Instructor để đạt được độ chính xác gần như tuyệt đối khi parse response từ AI.

Tại Sao Cần Structured Output?

Khi làm việc với các API AI như GPT-4.1, Claude Sonnet 4.5 hay Gemini 2.5 Flash, việc trả về JSON thuần túy thường gặp vấn đề:

Bảng So Sánh Chi Phí API AI 2026

Trước khi đi vào code, hãy xem xét chi phí thực tế cho 10 triệu token/tháng:

ModelOutput Cost ($/MTok)10M Tokens/Tháng
GPT-4.1$8.00$80
Claude Sonnet 4.5$15.00$150
Gemini 2.5 Flash$2.50$25
DeepSeek V3.2$0.42$4.20

Với HolySheep AI, bạn được hưởng tỷ giá ¥1=$1 (tiết kiệm 85%+), hỗ trợ WeChat/Alipay, độ trễ dưới 50ms và tín dụng miễn phí khi đăng ký.

Cài Đặt Môi Trường

pip install instructor pydantic openai httpx

Ví Dụ Thực Chiến: Trích Xuất Thông Tin Sản Phẩm

Từ kinh nghiệm thực chiến của tôi khi xây dựng hệ thống e-commerce, việc trích xuất thông tin sản phẩm từ mô tả text là một use-case phổ biến. Dưới đây là cách implement hoàn chỉnh:

import instructor
from pydantic import BaseModel, Field, field_validator
from openai import OpenAI
from typing import List, Optional
import os


Khởi tạo client với base_url của HolySheep AI

client = instructor.from_openai(
    OpenAI(
        base_url="https://api.holysheep.ai/v1",
        api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
        timeout=30.0,
        max_retries=3,
    ),
    mode=instructor.Mode.TOOLS,
)


class Price(BaseModel):
    """Model cho giá sản phẩm với validation"""
    amount: float = Field(..., gt=0, description="Giá tiền (phải lớn hơn 0)")
    currency: str = Field(default="VND", pattern="^[A-Z]{3}$")
    
    @field_validator('currency')
    @classmethod
    def validate_currency(cls, v: str) -> str:
        valid_currencies = ["VND", "USD", "EUR", "CNY", "JPY"]
        if v.upper() not in valid_currencies:
            raise ValueError(f"Tiền tệ phải là một trong: {valid_currencies}")
        return v.upper()


class ProductInfo(BaseModel):
    """Model trích xuất thông tin sản phẩm"""
    name: str = Field(..., min_length=1, max_length=200)
    brand: Optional[str] = Field(default=None, max_length=100)
    price: Price
    category: str = Field(..., description="Danh mục sản phẩm")
    features: List[str] = Field(..., min_length=1, max_length=10)
    rating: float = Field(..., ge=0.0, le=5.0)
    in_stock: bool = Field(default=True)
    
    @field_validator('features')
    @classmethod
    def validate_features(cls, v: List[str]) -> List[str]:
        return [f.strip() for f in v if f.strip()]


def extract_product_info(text: str) -> ProductInfo:
    """Trích xuất thông tin sản phẩm từ mô tả text"""
    return client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {
                "role": "system",
                "content": """Bạn là chuyên gia trích xuất thông tin sản phẩm.
                Trích xuất chính xác các thông tin từ mô tả và trả về JSON hợp lệ."""
            },
            {
                "role": "user",
                "content": f"Trích xuất thông tin sản phẩm từ: {text}"
            }
        ],
        response_model=ProductInfo,
        temperature=0.1,
    )



Test với dữ liệu mẫu

if __name__ == "__main__":
    product_text = """
    iPhone 15 Pro Max của Apple có giá 34.990.000 VND.
    Điện thoại flagship với chip A17 Pro, màn hình 6.7 inch Super Retina XDR.
    Camera 48MP, hỗ trợ ProRAW. Pin 4422mAh.
    Được đánh giá 4.8/5 sao bởi người dùng. Còn hàng.
    """
    
    result = extract_product_info(product_text)
    print(f"Tên: {result.name}")
    print(f"Thương hiệu: {result.brand}")
    print(f"Giá: {result.price.amount} {result.price.currency}")
    print(f"Đánh giá: {result.rating}/5")

Ví Dụ Nâng Cao: Batch Processing Với Retry Logic

Trong production, bạn cần handle edge cases và implement retry logic. Đây là pattern tôi đã áp dụng cho hệ thống xử lý 100K+ requests/ngày:

import instructor
from pydantic import BaseModel, Field, field_validator
from openai import OpenAI
from typing import List, Optional, Dict, Any
import time
import logging
from dataclasses import dataclass
from enum import Enum

logger = logging.getLogger(__name__)


class ProcessingStatus(Enum):
    SUCCESS = "success"
    RETRY = "retry"
    FAILED = "failed"


@dataclass
class ProcessingResult:
    status: ProcessingStatus
    data: Optional[Any] = None
    error: Optional[str] = None
    attempts: int = 0


class UserFeedback(BaseModel):
    """Model phân tích feedback người dùng"""
    sentiment: str = Field(..., description="Tình cảm: positive, negative, neutral")
    score: float = Field(..., ge=1.0, le=5.0, description="Điểm từ 1-5")
    categories: List[str] = Field(..., min_length=1, max_length=5)
    key_issues: List[str] = Field(default_factory=list, max_length=5)
    recommended_action: str = Field(..., description="Hành động được khuyến nghị")
    
    @field_validator('sentiment')
    @classmethod
    def validate_sentiment(cls, v: str) -> str:
        valid = ['positive', 'negative', 'neutral']
        v_lower = v.lower()
        if v_lower not in valid:
            raise ValueError(f"Sentiment phải là một trong: {valid}")
        return v_lower
    
    @field_validator('categories')
    @classmethod
    def validate_categories(cls, v: List[str]) -> List[str]:
        allowed = ['quality', 'price', 'service', 'delivery', 'usability', 'design', 'performance']
        return [c.lower() for c in v if c.lower() in allowed]


class InstructorClient:
    """Wrapper class với retry logic và error handling"""
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        max_retries: int = 3,
        timeout: float = 30.0,
    ):
        self.client = instructor.from_openai(
            OpenAI(
                base_url=base_url,
                api_key=api_key,
                timeout=timeout,
                max_retries=0,  # Handle retries ourselves
            ),
            mode=instructor.Mode.TOOLS,
        )
        self.max_retries = max_retries
    
    def analyze_feedback(
        self,
        text: str,
        model: str = "gpt-4.1",
    ) -> ProcessingResult:
        """Phân tích feedback với retry logic"""
        
        for attempt in range(1, self.max_retries + 1):
            try:
                result = self.client.chat.completions.create(
                    model=model,
                    messages=[
                        {
                            "role": "system",
                            "content": """Phân tích feedback khách hàng một cách chính xác.
                            Trả về JSON hợp lệ với các trường đã định nghĩa."""
                        },
                        {
                            "role": "user",
                            "content": f"Phân tích feedback sau: {text}"
                        }
                    ],
                    response_model=UserFeedback,
                    temperature=0.2,
                )
                
                return ProcessingResult(
                    status=ProcessingStatus.SUCCESS,
                    data=result,
                    attempts=attempt,
                )
                
            except instructor.InstructorError as e:
                error_msg = str(e)
                logger.warning(f"Attempt {attempt}/{self.max_retries} failed: {error_msg}")
                
                if attempt == self.max_retries:
                    return ProcessingResult(
                        status=ProcessingStatus.FAILED,
                        error=error_msg,
                        attempts=attempt,
                    )
                
                # Exponential backoff
                time.sleep(2 ** attempt * 0.5)
                
            except Exception as e:
                return ProcessingResult(
                    status=ProcessingStatus.FAILED,
                    error=f"Unexpected error: {str(e)}",
                    attempts=attempt,
                )
        
        return ProcessingResult(
            status=ProcessingStatus.FAILED,
            error="Max retries exceeded",
            attempts=self.max_retries,
        )
    
    def batch_analyze(
        self,
        feedbacks: List[str],
        model: str = "gpt-4.1",
    ) -> List[ProcessingResult]:
        """Xử lý batch với rate limiting"""
        results = []
        
        for i, feedback in enumerate(feedbacks):
            result = self.analyze_feedback(feedback, model)
            results.append(result)
            
            # Rate limiting: 50ms delay giữa các request
            if i < len(feedbacks) - 1:
                time.sleep(0.05)
            
            if (i + 1) % 100 == 0:
                logger.info(f"Processed {i + 1}/{len(feedbacks)} feedbacks")
        
        return results



Sử dụng

if __name__ == "__main__":
    # Khởi tạo client
    analyzer = InstructorClient(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_retries=3,
    )
    
    # Test single feedback
    test_feedback = """
    Sản phẩm chất lượng tốt nhưng giao hàng chậm 5 ngày.
    Đóng gói cẩn thận, nhân viên tư vấn nhiệt tình.
    Giá cả hợp lý so với mặt bằng chung.
    """
    
    result = analyzer.analyze_feedback(test_feedback)
    
    if result.status == ProcessingStatus.SUCCESS:
        print(f"Sentiment: {result.data.sentiment}")
        print(f"Score: {result.data.score}")
        print(f"Categories: {result.data.categories}")
        print(f"Action: {result.data.recommended_action}")
        print(f"Attempts: {result.attempts}")
    else:
        print(f"Failed: {result.error}")

Validation Nâng Cao Với Pydantic

Pydantic không chỉ là validation thông thường. Với các model phức tạp, bạn có thể tạo custom validators và computed fields:

from pydantic import BaseModel, Field, field_validator, model_validator, computed_field
from typing import List, Optional, Dict, Any
from datetime import datetime
import re


class InvoiceItem(BaseModel):
    """Item trong hóa đơn"""
    product_id: str = Field(..., pattern=r"^PROD-\d{6}$")
    name: str = Field(..., min_length=1, max_length=200)
    quantity: int = Field(..., gt=0, le=1000)
    unit_price: float = Field(..., gt=0)
    
    @computed_field
    @property
    def subtotal(self) -> float:
        """Tính tổng tiền cho item"""
        return round(self.quantity * self.unit_price, 2)


class Invoice(BaseModel):
    """Model hóa đơn với validation phức tạp"""
    invoice_id: str = Field(..., pattern=r"^INV-\d{4}-\d{6}$")
    customer_id: str = Field(..., pattern=r"^CUST-\d{8}$")
    items: List[InvoiceItem] = Field(..., min_length=1)
    tax_rate: float = Field(default=0.1, ge=0.0, le=1.0)
    discount_code: Optional[str] = Field(default=None, max_length=20)
    created_at: datetime = Field(default_factory=datetime.now)
    
    @model_validator(mode='after')
    def validate_discount(self) -> 'Invoice':
        """Validate mã giảm giá"""
        if self.discount_code:
            # Mã giảm giá phải bắt đầu bằng SAVE hoặc PROMO
            if not re.match(r'^(SAVE|PROMO)[A-Z0-9]{4,}$', self.discount_code):
                raise ValueError("Mã giảm giá không hợp lệ")
        return self
    
    @computed_field
    @property
    def subtotal(self) -> float:
        """Tổng phụ trước thuế"""
        return round(sum(item.subtotal for item in self.items), 2)
    
    @computed_field
    @property
    def tax_amount(self) -> float:
        """Tiền thuế"""
        return round(self.subtotal * self.tax_rate, 2)
    
    @computed_field
    @property
    def total(self) -> float:
        """Tổng cộng sau thuế"""
        return round(self.subtotal + self.tax_amount, 2)
    
    @computed_field
    @property
    def total_items(self) -> int:
        """Tổng số lượng items"""
        return sum(item.quantity for item in self.items)


def extract_invoice_from_text(text: str) -> Invoice:
    """Trích xuất hóa đơn từ text bằng AI"""
    import instructor
    from openai import OpenAI
    import os
    
    client = instructor.from_openai(
        OpenAI(
            base_url="https://api.holysheep.ai/v1",
            api_key=os.environ.get("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
        ),
        mode=instructor.Mode.TOOLS,
    )
    
    return client.chat.completions.create(
        model="gpt-4.1",
        messages=[
            {
                "role": "system",
                "content": """Bạn là chuyên gia OCR và trích xuất hóa đơn.
                Trích xuất tất cả items từ hóa đơn, tính toán chính xác.
                Format: INV-YYYY-NNNNNN, PROD-NNNNNN, CUST-NNNNNNNN"""
            },
            {
                "role": "user",
                "content": f"Trích xuất hóa đơn: {text}"
            }
        ],
        response_model=Invoice,
        temperature=0.1,
    )



Test

if __name__ == "__main__":
    test_invoice = """
    Hóa đơn INV-2024-123456 cho khách hàng CUST-00001234
    
    Items:
    1. PROD-000001 - iPhone 15 - 5 cái - 25.000.000 VND/cái
    2. PROD-000002 - AirPods Pro - 3 cái - 7.500.000 VND/cái
    
    Mã giảm giá: SAVE2024
    Thuế suất: 10%
    """
    
    invoice = extract_invoice_from_text(test_invoice)
    
    print(f"Invoice ID: {invoice.invoice_id}")
    print(f"Customer: {invoice.customer_id}")
    print(f"Tổng số items: {invoice.total_items}")
    print(f"Tổng phụ: {invoice.subtotal:,.0f} VND")
    print(f"Thuế (10%): {invoice.tax_amount:,.0f} VND")
    print(f"Tổng cộng: {invoice.total:,.0f} VND")
    
    for item in invoice.items:
        print(f"  - {item.name}: {item.subtotal:,.0f} VND")

Performance Benchmark: So Sánh Độ Trễ Thực Tế

Qua quá trình thử nghiệm với HolyShehe AI, đây là độ trễ trung bình tôi đo được:

Lỗi Thường Gặp Và Cách Khắc Phục

1. Lỗi Validation: PydanticValidationError

Nguyên nhân: AI trả về dữ liệu không match với Pydantic model (sai format, thiếu required fields).

# ❌ Code gây lỗi - không handle validation error
result = client.chat.completions.create(
    model="gpt-4.1",
    messages=[...],
    response_model=ProductInfo,
)


✅ Fix - Wrap trong try-except và retry

from instructor.exceptions import InstructorError

def safe_extract_with_retry(text: str, max_retries: int = 3) -> Optional[ProductInfo]:
    for attempt in range(max_retries):
        try:
            return client.chat.completions.create(
                model="gpt-4.1",
                messages=[...],
                response_model=ProductInfo,
                max_retries=0,  # Disable instructor's internal retry
            )
        except InstructorError as e:
            if attempt == max_retries - 1:
                raise
            # Log và retry với model rẻ hơn
            time.sleep(0.5 * (attempt + 1))

2. Lỗi API Key: AuthenticationError

Nguyên nhân: API key

Tài nguyên liên quan

Bài viết liên quan