Đừng vật lộn với Regex nữa: Xây dựng hệ thống ETL thông minh với LLM và Pydantic

Thực trạng lộn xộn của dữ liệu phi cấu trúc

Các pipeline ETL (Trích xuất, Chuyển đổi, Tải) truyền thống hoạt động hoàn hảo khi nguồn dữ liệu là một tệp CSV sạch hoặc một API có cấu trúc. Nhưng ngay khi khách hàng gửi một thư mục đầy các email lộn xộn, báo cáo y tế dạng PDF hoặc log từ Slack, phương pháp cũ sẽ gặp bế tắc. Tôi đã từng dành cả cuối tuần vật lộn với một script Regular Expression (Regex) dài 200 dòng để phân tích phản hồi của khách hàng. Nó chỉ hoạt động đúng trong hai giờ—cho đến khi một người dùng quyết định sử dụng định dạng ngày tháng khác.

Mã nguồn thì mang tính xác định và cứng nhắc, trong khi ngôn ngữ tự nhiên lại linh hoạt và khó đoán. Chúng ta cần một cầu nối giữa giao tiếp của con người và các yêu cầu khắt khe của một cơ sở dữ liệu SQL. Bằng cách kết hợp các Mô hình Ngôn ngữ Lớn (LLM) với Pydantic, chúng ta có thể xây dựng các pipeline thực sự “hiểu” dữ liệu mà chúng đang xử lý.

Stack ETL hiện đại: LLM + Pydantic

Hãy coi sự kết hợp này như một bản hòa âm ba phần:

• Công cụ suy luận (LLM): Các mô hình như GPT-4o hoặc Claude 3.5 Sonnet không chỉ khớp các chuỗi ký tự. Chúng hiểu ngữ cảnh. Chúng biết rằng “bệnh nhân ở phòng số 4” và “Ông Henderson” cùng chỉ về một đối tượng.
• Trình xác thực (Pydantic): LLM dễ bị tình trạng “ảo giác” (hallucinations) hoặc thêm các từ đệm hội thoại không cần thiết. Pydantic đóng vai trò như một người gác cổng nghiêm ngặt. Nó bắt buộc LLM phải trả về dữ liệu khớp chính xác với schema của bạn. Nếu AI cố gắng đặt một cái tên vào trường tuổi, Pydantic sẽ từ chối nó.
• Lưu trữ vĩnh viễn (SQLAlchemy): Sau khi dữ liệu được làm sạch và xác thực, chúng ta đưa nó vào một môi trường có cấu trúc như PostgreSQL hoặc SQLite để lưu trữ lâu dài và phân tích BI.

Tôi đã triển khai mô hình này để xử lý hàng ngàn tài liệu. Trong một trường hợp, nó đã giảm bớt một tác vụ nhập liệu thủ công từ năm giờ làm việc của con người mỗi ngày xuống còn khoảng ba phút xử lý tự động.

Thiết lập Pipeline của bạn

Chúng ta sẽ sử dụng Python cho dự án này. Để kết nối khoảng cách giữa LLM và Pydantic, thư viện instructor là công cụ tốt nhất cho việc này. Nó “vá” (patch) client OpenAI tiêu chuẩn để trả về các đối tượng Pydantic thực thụ thay vì các chuỗi văn bản thô.

Cài đặt các thư viện phụ thuộc để bắt đầu:

pip install openai instructor pydantic sqlalchemy

Định nghĩa bản thiết kế dữ liệu

Schema luôn đi trước. Trước khi viết logic trích xuất, chúng ta phải định nghĩa một kết quả “thành công” trông như thế nào. Hãy tưởng tượng chúng ta đang phân tích các ghi chú chuyển tuyến y tế lộn xộn. Chúng ta cần tên bệnh nhân, tuổi, danh sách các triệu chứng và mức độ ưu tiên.

from pydantic import BaseModel, Field
from typing import List

class PatientExtraction(BaseModel):
    name: str = Field(..., description="Họ và tên đầy đủ của bệnh nhân")
    age: int = Field(..., description="Tuổi của bệnh nhân tính theo năm")
    symptoms: List[str] = Field(..., description="Danh sách các triệu chứng y tế cụ thể")
    priority: str = Field(..., description="Mức độ khẩn cấp: Low, Medium, hoặc High")

Các mô tả trong Field không chỉ để làm tài liệu. Thư viện instructor truyền các chuỗi này trực tiếp vào LLM dưới dạng hướng dẫn, giúp mô hình xác định đúng các điểm dữ liệu giữa những thông tin nhiễu.

Công cụ trích xuất

Logic trích xuất gọn nhẹ đến ngạc nhiên. Bằng cách sử dụng client đã được “vá”, mô hình bị buộc phải tuân theo schema Pydantic của chúng ta. Nếu mô hình không tạo ra được JSON hợp lệ, instructor có thể tự động thử lại yêu cầu, chỉ cho AI thấy chính xác nơi nó mắc lỗi xác thực.

import instructor
from openai import OpenAI

# Khởi tạo client đã được vá
client = instructor.from_openai(OpenAI(api_key="YOUR_OPENAI_API_KEY"))

messy_text = """
Nhận được cuộc gọi từ John Doe, ông ấy 45 tuổi. 
Phàn nàn về việc đau thắt lưng dữ dội và tê ở chân trái. 
Trường hợp này có vẻ khẩn cấp, hãy sắp xếp cho ông ấy khám ngay hôm nay.
"""

def extract_patient_data(text: str) -> PatientExtraction:
    return client.chat.completions.create(
        model="gpt-4o-mini",
        response_model=PatientExtraction,
        max_retries=3,
        messages=[{"role": "user", "content": text}],
    )

extracted = extract_patient_data(messy_text)
print(f"Đã trích xuất: {extracted.name} | Độ ưu tiên: {extracted.priority}")

Vòng lặp tự sửa lỗi này chính là “bí quyết”. Nó làm cho hệ thống linh hoạt hơn nhiều so với một lệnh gọi API thông thường có thể trả về văn bản sai định dạng.

Lưu trữ dữ liệu đã xác thực vào SQL

Bây giờ chúng ta đã có một đối tượng Python sạch, chúng ta có thể lưu nó vào cơ sở dữ liệu. Chúng ta sẽ sử dụng SQLAlchemy để ánh xạ dữ liệu Pydantic vào một bảng SQLite. Điều này giữ cho các kiểu dữ liệu nhất quán từ “quy trình tư duy” của AI cho đến khi lưu xuống đĩa.

from sqlalchemy import create_engine, Column, Integer, String, JSON
from sqlalchemy.orm import declarative_base, sessionmaker

Base = declarative_base()
engine = create_engine("sqlite:///patients.db")
Session = sessionmaker(bind=engine)

class PatientRecord(Base):
    __tablename__ = 'patients'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)
    symptoms = Column(JSON)
    priority = Column(String)

Base.metadata.create_all(engine)

def save_to_db(data: PatientExtraction):
    with Session() as session:
        new_record = PatientRecord(
            name=data.name,
            age=data.age,
            symptoms=data.symptoms,
            priority=data.priority
        )
        session.add(new_record)
        session.commit()

save_to_db(extracted)

Mở rộng cho môi trường Production

Việc chuyển từ một script chạy cục bộ sang một pipeline production đòi hỏi một vài điều chỉnh. Đầu tiên, hãy cân nhắc về chi phí. Xử lý 1.000 tài liệu với GPT-4o có thể tốn 30 USD, nhưng GPT-4o-mini có thể xử lý cùng một tác vụ đó với giá chưa đầy 1 USD với kết quả trích xuất cấu trúc gần như tương đương.

Thứ hai, hãy sử dụng field_validator của Pydantic để áp đặt logic nghiệp vụ. Ví dụ: nếu LLM trích xuất một mức độ ưu tiên không nằm trong danh sách cho phép, bạn có thể ép nó về một giá trị mặc định hoặc báo lỗi trước khi dữ liệu đi vào cơ sở dữ liệu.

from pydantic import field_validator

# Bên trong class PatientExtraction của bạn
@field_validator('priority')
@classmethod
def validate_priority(cls, v: str) -> str:
    allowed = ['Low', 'Medium', 'High']
    if v not in allowed:
        return 'Medium' # Giá trị dự phòng an toàn
    return v

Tổng kết

Việc kết hợp khả năng suy luận của LLM với tính xác thực nghiêm ngặt của Pydantic đã thay đổi cuộc chơi ETL. Bạn không còn phải lo sợ những văn bản phi cấu trúc lộn xộn nữa. Bằng cách định nghĩa một schema rõ ràng và để AI xử lý việc diễn giải, bạn có thể xây dựng các pipeline vừa linh hoạt vừa an toàn về kiểu dữ liệu.

Hãy bắt đầu nhỏ. Chọn một tác vụ nhập liệu thủ công, định nghĩa một model Pydantic cho nó và chạy thử vài lần. Một khi bạn thấy được độ tin cậy của mô hình này, bạn sẽ không bao giờ muốn viết một chuỗi Regex phức tạp nào nữa.