LangChain Output Parsing: Hướng Dẫn Toàn Diện Để Lấy Dữ Liệu Có Cấu Trúc Từ LLM

Khi tôi triển khai hệ thống RAG cho doanh nghiệp thương mại điện tử với 50,000 sản phẩm và 2 triệu đánh giá khách hàng, vấn đề lớn nhất không phải là tìm kiếm — mà là parse output từ LLM sao cho ứng dụng có thể xử lý tự động. Đêm trước ngày ra mắt, đội ngũ backend nhận ra: 30% responses từ model trả về không đúng format mong đợi, khiến pipeline xử lý đơn hàng bị deadlock. Đó là lý do tôi viết bài hướng dẫn này — để bạn không phải đối mặt với cùng một vấn đề.

Output Parsing Là Gì và Tại Sao Nó Quan Trọng?

LangChain Output Parsing là cơ chế cho phép bạn chuyển đổi raw text response từ LLM thành structured data (Python objects, dictionaries, lists) một cách đáng tin cậy. Thay vì phải regex parsing thủ công và xử lý edge cases, Output Parsers cung cấp:

• Type validation tự động
• Error handling khi LLM trả về format không hợp lệ
• Retry mechanism thông minh
• Schema enforcement nhất quán

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

# Cài đặt các thư viện cần thiết
pip install langchain langchain-core langchain-holysheep pydantic

Kiểm tra version
python -c "import langchain; print(langchain.__version__)"

Với HolySheep AI, bạn nhận được tín dụng miễn phí khi đăng ký và tỷ giá chỉ từ $0.42/MTok cho DeepSeek V3.2 — tiết kiệm đến 85% so với OpenAI. Độ trễ trung bình dưới 50ms giúp việc retry parse không ảnh hưởng đáng kể đến UX.

Pydantic Output Parser — King Of The Hill

Đây là output parser mạnh mẽ nhất trong LangChain. Bạn định nghĩa schema bằng Pydantic models và LangChain sẽ đảm bảo LLM trả về đúng structure.

import os
from langchain_holysheep import ChatHolySheep
from langchain_core.output_parsers import PydanticOutputParser
from langchain_core.prompts import ChatPromptTemplate
from pydantic import BaseModel, Field
from typing import List, Optional

Cấu hình API - base_url BẮT BUỘC là https://api.holysheep.ai/v1
os.environ["HOLYSHEEP_API_KEY"] = "YOUR_HOLYSHEEP_API_KEY"

llm = ChatHolySheep(
    base_url="https://api.holysheep.ai/v1",
    api_key=os.environ["HOLYSHEEP_API_KEY"],
    model="gpt-4.1"  # Hoặc deepseek-v3.2, claude-sonnet-4.5
)

Định nghĩa schema cho phản hồi sản phẩm
class ProductReview(BaseModel):
    product_id: str = Field(description="Mã sản phẩm")
    rating: float = Field(description="Điểm đánh giá từ 1-5", ge=1, le=5)
    pros: List[str] = Field(description="Những điểm tích cực")
    cons: List[str] = Field(description="Những điểm cần cải thiện")
    recommended: bool = Field(description="Có nên mua không")
    sentiment_score: Optional[float] = Field(
        default=None, 
        description="Điểm cảm xúc từ -1.0 đến 1.0"
    )

Khởi tạo parser
parser = PydanticOutputParser(pydantic_schema=ProductReview)

Tạo prompt với format instructions tự động
prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là chuyên gia phân tích đánh giá sản phẩm. "
               "Trả về thông tin theo đúng format được yêu cầu."),
    ("human", "Phân tích đánh giá sau và trả về structured data:\n\n{review}")
])

Chain everything together
chain = prompt | llm | parser

Thực thi
review_text = """
Sản phẩm tuyệt vời! Giao hàng nhanh, đóng gói cẩn thận. 
Camera chụp đẹp nhưng pin hơi yếu, dùng được 4 tiếng.
Giá cả hợp lý, tôi sẽ giới thiệu cho bạn bè. Mua lại lần 2 rồi!
"""

result = chain.invoke({"review": review_text})

print(f"Product ID: {result.product_id}")
print(f"Rating: {result.rating}/5")
print(f"Recommended: {'Có' if result.recommended else 'Không'}")
print(f"Pros: {', '.join(result.pros)}")
print(f"Cons: {', '.join(result.cons)}")

JsonOutputParser — Khi Bạn Cần Linh Hoạt Hơn

Đối với những trường hợp schema có thể thay đổi động hoặc bạn muốn kiểm soát format JSON output một cách chi tiết, JsonOutputParser là lựa chọn tối ưu.

from langchain_core.output_parsers import JsonOutputParser
from langchain_core.runnables import RunnableLambda
from typing import Dict, List, Any

Định nghĩa schema động bằng dict
schema = {
    "type": "object",
    "properties": {
        "analysis": {
            "type": "object",
            "properties": {
                "category": {"type": "string", "enum": ["positive", "neutral", "negative"]},
                "urgency": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
                "topics": {"type": "array", "items": {"type": "string"}}
            },
            "required": ["category", "urgency"]
        },
        "action_items": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {
                    "task": {"type": "string"},
                    "assignee": {"type": "string"},
                    "deadline": {"type": "string"}
                }
            }
        },
        "summary": {"type": "string", "minLength": 50}
    },
    "required": ["analysis", "summary"]
}

parser = JsonOutputParser(schema=schema)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Bạn là trợ lý AI phân tích ticket hỗ trợ khách hàng.
Hãy trả về JSON hợp lệ theo schema được cung cấp, không thêm text giải thích."""),
    ("human", "Phân tích ticket:\n\n{ticket}")
])

Xử lý thêm sau khi parse
def enrich_with_timestamp(data: Dict) -> Dict:
    from datetime import datetime
    data["processed_at"] = datetime.now().isoformat()
    return data

chain = (
    prompt 
    | llm 
    | parser 
    | RunnableLambda(enrich_with_timestamp)
)

ticket = """
[TICKET #89234] - Khách hàng: Minh Tran - 15/03/2024 14:32
Subject: Không nhận được hàng đã đặt

Tôi đã đặt hàng từ 5 ngày trước (order #ORD-55123) nhưng vẫn chưa 
nhận được. Đơn hàng status vẫn là "shipping". Tôi cần hàng gấp 
vì làm quà sinh nhật cho mẹ vào cuối tuần này. Đã liên hệ shipper 
2 lần nhưng không ai nghe máy. Very disappointed!
"""

result = chain.invoke({"ticket": ticket})

print("Category:", result["analysis"]["category"])
print("Urgency:", result["analysis"]["urgency"])
print("Topics:", result["analysis"]["topics"])
print("Processed at:", result["processed_at"])

CSV Parser — Trích Xuất Dữ Liệu Bảng

Khi bạn cần extract structured data dạng bảng từ LLM response (ví dụ: so sánh sản phẩm, báo cáo tài chính), CSVOutputParser rất hữu ích.

from langchain_core.output_parsers import CSVOutputParser
from typing import List

Định nghĩa columns
columns = ["Tên sản phẩm", "Giá (VND)", "Điểm đánh giá", "Tồn kho", "Khuyến nghị"]

parser = CSVOutputParser(columns=columns)

prompt = ChatPromptTemplate.from_messages([
    ("system", """Bạn là chuyên gia tư vấn mua sắm. So sánh các sản phẩm 
và trả về kết quả dạng CSV với header. Không thêm text giải thích nào."""),
    ("human", "So sánh 5 laptop tốt nhất dưới 20 triệu:\n\n{request}")
])

chain = prompt | llm | parser

result = chain.invoke({
    "request": "Tìm 5 laptop gaming dưới 20 triệu VNĐ, ưu tiên RTX 3050 trở lên"
})

Chuyển đổi thành list of dicts
import csv
from io import StringIO

reader = csv.DictReader(StringIO(result), fieldnames=columns)
products = list(reader)

for i, product in enumerate(products, 1):
    print(f"\n--- Sản phẩm {i} ---")
    print(f"Tên: {product['Tên sản phẩm']}")
    print(f"Giá: {product['Giá (VND)']}")
    print(f"Điểm: {product['Điểm đánh giá']}")
    print(f"Tồn kho: {product['Tồn kho']}")
    print(f"Khuyến nghị: {product['Khuyến nghị']}")

Custom Output Parser — Khi Logic Đặc Biệt

Đôi khi bạn cần logic parse phức tạp mà không có sẵn parser. Lúc đó, subclass BaseOutputParser là giải pháp.

from langchain_core.output_parsers import BaseOutputParser
from typing import Type
from pydantic import BaseModel

class NaturalLanguagePlan(BaseModel):
    steps: List[str] = Field(description="Danh sách các bước cần thực hiện")
    estimated_time: str = Field(description="Thời gian ước tính")
    priority: str = Field(description="Mức độ ưu tiên")

class PlanOutputParser(BaseOutputParser):
    """Custom parser cho kế hoạch dạng text có cấu trúc"""
    
    def parse(self, text: str) -> Type[NaturalLanguagePlan]:
        import json
        import re
        
        # Clean response
        text = text.strip()
        
        # Tìm JSON block nếu có
        json_match = re.search(r'\{.*\}', text, re.DOTALL)
        if json_match:
            data = json.loads(json_match.group())
        else:
            # Parse text-based format
            lines = text.split('\n')
            data = {
                "steps": [],
                "estimated_time": "Không xác định",
                "priority": "medium"
            }
            
            for line in lines:
                line = line.strip()
                if line.startswith(('1.', '2.', '3.', '4.', '5.')):
                    step = re.sub(r'^\d+\.\s*', '', line)
                    data["steps"].append(step)
                elif 'phút' in line or 'giờ' in line:
                    time_match = re.search(r'(\d+\s*(phút|giờ))', line)
                    if time_match:
                        data["estimated_time"] = time_match.group(1)
        
        return NaturalLanguagePlan(**data)
    
    def get_format_instructions(self) -> str:
        return """Trả về kế hoạch với format:
{
    "steps": ["Bước 1", "Bước 2", ...],
    "estimated_time": "X phút/Y giờ",
    "priority": "low/medium/high"
}"""

Sử dụng custom parser
custom_parser = PlanOutputParser()

prompt = ChatPromptTemplate.from_messages([
    ("system", "Bạn là chuyên gia lập kế hoạch. Tạo kế hoạch chi tiết theo format."),
    ("human", "Lập kế hoạch deploy ứng dụng React lên production với CI/CD pipeline")
])

chain = prompt | llm | custom_parser

plan = chain.invoke({})

print(f"Priority: {plan.priority}")
print(f"Estimated Time: {plan.estimated_time}")
print("Steps:")
for i, step in enumerate(plan.steps, 1):
    print(f"  {i}. {step}")

Retry Mechanism — Đảm Bảo Reliability

Trong production, bạn cần handle trường hợp parse fail. LangChain cung cấp OutputFixingParser để tự động retry khi parse thất bại.

from langchain_core.output_parsers import PydanticOutputParser
from langchain.output_parsers.error_handling import OutputParserException
from langchain.output_parsers import RetryWithErrorOutputParser
from pydantic import BaseModel, Field
from typing import List

class FAQAnswer(BaseModel):
    question: str = Field(description="Câu hỏi FAQ")
    answer: str = Field(description="Câu trả lời ngắn gọn", min_length=20)
    related_topics: List[str] = Field(description="Các chủ đề liên quan")
    confidence: float = Field(description="Độ tự tin từ 0-1", ge=0, le=1)

Base parser
base_parser = PydanticOutputParser(pydantic_schema=FAQAnswer)

Retry parser với max retries
retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=base_parser,
    llm=llm,
    max_retries=3
)

Test với input khó
test_input = """
Tạo FAQ cho phần giới thiệu sản phẩm skincare:
- Công dụng của serum vitamin C
- Cách sử dụng kem chống nắng
- Thời gian shipping
"""

Invoke với retry tự động
try:
    result = (prompt | llm | retry_parser).invoke({"input": test_input})
    print(f"Q: {result.question}")
    print(f"A: {result.answer}")
    print(f"Confidence: {result.confidence}")
except OutputParserException as e:
    print(f"Parse failed after all retries: {e}")

Bảng So Sánh Các Output Parser Types

Parser Type	Use Case	Pros	Cons
PydanticOutputParser	Schema phức tạp, validation chặt chẽ	Type safety, validation tự động	Cần định nghĩa Pydantic model
JsonOutputParser	Schema động, JSON thuần	Linh hoạt, không cần class	Validation thủ công
CSVOutputParser	Data dạng bảng	Đơn giản, dễ export	Hạn chế nested data
Custom Parser	Logic đặc biệt	Tùy biến hoàn toàn	Cần code nhiều

Bảng Giá HolySheheep AI — Cập Nhật 2026

Model	Giá/MTok	Use Case Tối Ưu
DeepSeek V3.2	$0.42	Parse có khối lượng lớn, tiết kiệm 85%
Gemini 2.5 Flash	$2.50	Parse real-time, low latency
GPT-4.1	$8	Parse phức tạp, accuracy cao
Claude Sonnet 4.5	$15	Structured output, reasoning

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

1. Lỗi: "Expected dict but got str" khi parse

Nguyên nhân: Output parser nhận raw string thay vì AIMessage object.

# ❌ SAI - parse trực tiếp string
raw_response = llm.invoke("Extract product info")
result = parser.parse(raw_response)  # Lỗi!

✅ ĐÚNG - parse từ AIMessage content
raw_response = llm.invoke("Extract product info")
result = parser.parse(raw_response.content)  # Hoặc dùng chain

✅ TỐT NHẤT - dùng LCEL chain
chain = prompt | llm | parser
result = chain.invoke({"input": "Extract product info"})

2. Lỗi: "ValidationError" với Pydantic Field

Nguyên nhân: LLM trả về giá trị ngoài range/format định nghĩa.

from pydantic import BaseModel, Field, validator

class PriceRange(BaseModel):
    min_price: int = Field(ge=0, description="Giá tối thiểu")
    max_price: int = Field(le=100000000, description="Giá tối đa")
    
    @validator('max_price')
    def max_must_be_greater(cls, v, values):
        if 'min_price' in values and v < values['min_price']:
            raise ValueError('max_price phải >= min_price')
        return v

Nếu LLM trả về format không đúng, dùng retry parser
retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=PydanticOutputParser(pydantic_schema=PriceRange),
    llm=llm
)

Hoặc relax constraints nếu data không quan trọng
class FlexiblePriceRange(BaseModel):
    min_price: float = Field(default=0)  # Không strict validation
    max_price: float = Field(default=0)
    currency: str = "VND"  # Default value

3. Lỗi: "Could not parse LLM output" - JSON Invalid

Nguyên nhân: LLM thêm markdown code blocks hoặc text ngoài JSON.

import json
import re

class RobustJsonParser(BaseOutputParser):
    """Parser xử lý các format JSON không chuẩn"""
    
    def parse(self, text: str):
        text = text.strip()
        
        # Loại bỏ markdown code blocks
        text = re.sub(r'^```json\s*', '', text, flags=re.MULTILINE)
        text = re.sub(r'^```\s*$', '', text, flags=re.MULTILINE)
        
        # Loại bỏ text trước/sau JSON
        json_match = re.search(r'\{[\s\S]*\}', text)
        if json_match:
            try:
                return json.loads(json_match.group())
            except json.JSONDecodeError:
                pass
        
        # Thử parse trực tiếp
        return json.loads(text)

Cách khác: dùng prompt engineering
prompt_with_constraint = ChatPromptTemplate.from_messages([
    ("system", """Trả về CHỈ JSON thuần, không có markdown, không có text giải thích.
Không dùng ```json code blocks. Bắt đầu trực tiếp bằng {{."""),
    ("human", "{question}")
])

chain = prompt_with_constraint | llm | RobustJsonParser()

4. Lỗi: "Rate limit exceeded" khi retry nhiều lần

Nguyên nhân: RetryWithErrorOutputParser gọi LLM nhiều lần, có thể trigger rate limit.

import time
from functools import wraps

def rate_limit_handler(max_retries=3, backoff_factor=2):
    """Handler rate limit với exponential backoff"""
    def decorator(chain):
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return chain.invoke(*args, **kwargs)
                except RateLimitError as e:
                    if attempt == max_retries - 1:
                        raise
                    wait_time = backoff_factor ** attempt
                    print(f"Rate limited. Waiting {wait_time}s...")
                    time.sleep(wait_time)
        return wrapper
    return decorator

Giải pháp: Dùng model rẻ hơn cho retry
from langchain_holysheep import ChatHolySheep

Model chính cho parse
main_llm = ChatHolySheep(
    base_url="https://api.holysheep.ai/v1",
    model="gpt-4.1"
)

Model rẻ hơn cho retry (tiết kiệm 95% chi phí)
retry_llm = ChatHolySheep(
    base_url="https://api.holysheep.ai/v1",
    model="deepseek-v3.2"  # Chỉ $0.42/MTok!
)

retry_parser = RetryWithErrorOutputParser.from_llm(
    parser=base_parser,
    llm=retry_llm  # Model rẻ cho retry
)

Kết Luận

Output Parsing là một trong những kỹ năng quan trọng nhất khi làm việc với LangChain và LLM trong production. Từ kinh nghiệm triển khai hệ thống RAG cho doanh nghiệp thương mại điện tử, tôi đã học được:

• Luôn dùng PydanticOutputParser khi schema cố định — nó giúp catch lỗi sớm
• Implement retry mechanism với model rẻ hơn (DeepSeek V3.2) để tiết kiệm chi phí
• Test với edge cases — LLM luôn trả về những format không mong đợi
• Monitoring parse success rate — nếu dưới 95%, cần điều chỉnh prompt

Với HolySheheep AI, bạn có thể deploy các giải pháp parse quy mô lớn với chi phí chỉ từ $0.42/MTok, hỗ trợ WeChat/Alipay thanh toán, và độ trễ dưới 50ms. Đăng ký hôm nay để nhận tín dụng miễn phí và bắt đầu xây dựng ứng dụng AI production-ready!

👉 Đăng ký HolySheheep AI — nhận tín dụng miễn phí khi đăng ký