Tôi là Minh, kiến trúc sư hệ thống AI tại một công ty thương mại điện tử Việt Nam quy mô 5 triệu người dùng. Hôm nay tôi sẽ chia sẻ cách chúng tôi giải quyết bài toán mà nhiều team gặp phải: làm sao để RAG hiểu được cả dữ liệu bảng (structured) lẫn văn bản thuần túy (unstructured)?
Bối cảnh thực chiến: Đỉnh dịch vụ khách hàng 11.11
Tháng 11 năm ngoái, hệ thống chatbot hỗ trợ khách hàng của chúng tôi sụp đổ vào ngày 11/11. Nguyên nhân không phải do traffic — mà là do prompt injection từ dữ liệu bảng. Khi khách hàng hỏi "Sản phẩm nào dưới 500k có đánh giá 5 sao?", hệ thống RAG cũ trả lời lạc chủ đề hoàn toàn vì nó không hiểu cấu trúc bảng.
Đó là lúc tôi bắt đầu nghiên cứu Table-aware RAG — giải pháp cho phép LLM đọc và truy vấn dữ liệu dạng bảng như CSV, Excel, database rows.
Tại sao bảng biểu là thách thức lớn cho RAG truyền thống?
- Chunking thông thường phá vỡ cấu trúc: Khi chia bảng 100 dòng thành từng đoạn 500 tokens, mỗi chunk mất context của các dòng liên quan
- Semantic search không hiểu relationships: "Tổng doanh thu Q3" không liên quan ngữ nghĩa đến "Doanh thu từng sản phẩm" nhưng chúng thực sự cùng bảng
- LLM không biết schema: Không có metadata, model không biết kiểu dữ liệu, constraints, hay foreign keys
Kiến trúc Table Data RAG hoàn chỉnh
"""
Table Data RAG - Hybrid Retrieval cho Structured + Unstructured Data
Tác giả: Minh, Kiến trúc sư AI @ E-commerce Việt Nam
"""
import json
import pandas as pd
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
import httpx
============ CẤU HÌNH HOLYSHEEP AI ============
Đăng ký và lấy API key tại: https://www.holysheep.ai/register
Tỷ giá ¥1 = $1 — tiết kiệm 85%+ so với OpenAI
HOLYSHEEP_CONFIG = {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY", # Thay bằng API key thật
"embedding_model": "text-embedding-3-large",
"chat_model": "gpt-4.1", # $8/MTok — rẻ hơn 60% GPT-4o
}
@dataclass
class TableMetadata:
"""Metadata cho mỗi bảng — giúp LLM hiểu schema"""
table_name: str
columns: List[Dict[str, str]] # {"name": "price", "type": "float", "description": "Giá VND"}
primary_key: str
description: str
@dataclass
class RetrievalResult:
content: str
source_type: str # "table" | "text"
relevance_score: float
table_metadata: Optional[TableMetadata] = None
class TableDataRAG:
"""
Hệ thống RAG lai ghép: truy vấn đồng thời dữ liệu bảng (structured)
và văn bản thường (unstructured) để trả lời câu hỏi phức tạp.
Ưu điểm thực chiến:
- < 50ms latency với HolySheep AI
- Hỗ trợ CSV, Excel, database tables
- Tự động sinh SQL-like queries từ câu hỏi tự nhiên
"""
def __init__(self, config: Dict):
self.config = config
self.client = httpx.Client(
base_url=config["base_url"],
headers={"Authorization": f"Bearer {config['api_key']}"},
timeout=30.0
)
self.tables: Dict[str, pd.DataFrame] = {}
self.table_metadata: Dict[str, TableMetadata] = {}
self.vector_store: Dict[str, List[float]] = {}
def register_table(
self,
df: pd.DataFrame,
metadata: TableMetadata,
id_column: str = "id"
):
"""Đăng ký một bảng vào hệ thống RAG"""
self.tables[metadata.table_name] = df
self.table_metadata[metadata.table_name] = metadata
# Embed toàn bộ bảng với schema description
schema_text = self._generate_schema_description(metadata)
embedding = self._get_embedding(schema_text)
self.vector_store[metadata.table_name] = embedding
print(f"✅ Đã đăng ký bảng '{metadata.table_name}' với {len(df)} dòng")
def _generate_schema_description(self, metadata: TableMetadata) -> str:
"""Sinh mô tả schema để embed — giúp retrieve đúng bảng"""
cols_desc = []
for col in metadata.columns:
cols_desc.append(
f"- {col['name']} ({col['type']}): {col['description']}"
)
return f"""
Bảng: {metadata.table_name}
Mô tả: {metadata.description}
Cột:
{chr(10).join(cols_desc)}
Primary Key: {metadata.primary_key}
"""
def _get_embedding(self, text: str) -> List[float]:
"""Gọi HolySheep embedding API — latency < 50ms"""
response = self.client.post(
"/embeddings",
json={
"model": self.config["embedding_model"],
"input": text
}
)
response.raise_for_status()
return response.json()["data"][0]["embedding"]
def _semantic_search_tables(self, query: str, top_k: int = 3) -> List[str]:
"""Tìm bảng liên quan bằng semantic search"""
query_embedding = self._get_embedding(query)
# Tính similarity với mỗi bảng đã đăng ký
similarities = []
for table_name, table_embedding in self.vector_store.items():
sim = self._cosine_similarity(query_embedding, table_embedding)
similarities.append((table_name, sim))
similarities.sort(key=lambda x: x[1], reverse=True)
return [name for name, _ in similarities[:top_k]]
def _cosine_similarity(self, a: List[float], b: List[float]) -> float:
"""Tính cosine similarity đơn giản"""
dot = sum(x * y for x, y in zip(a, b))
norm_a = sum(x * x for x in a) ** 0.5
norm_b = sum(x * x for x in b) ** 0.5
return dot / (norm_a * norm_b + 1e-8)
def retrieve(self, query: str, top_k: int = 5) -> List[RetrievalResult]:
"""
Retrieval lai ghép: tìm cả bảng liên quan VÀ text chunks
Chiến lược:
1. Semantic search để tìm bảng phù hợp
2. Nếu query hỏi về dữ liệu cụ thể → sinh table snapshot
3. Nếu query cần giải thích → tìm text chunks
"""
results = []
# Bước 1: Tìm bảng liên quan
relevant_tables = self._semantic_search_tables(query)
# Bước 2: Phân tích query để quyết định retrieval strategy
query_lower = query.lower()
is_structured_query = any(
keyword in query_lower
for keyword in ["bao nhiêu", "tổng", "trung bình", "so sánh",
"nhiều nhất", "ít nhất", "dưới", "trên", "count", "sum", "avg"]
)
# Bước 3: Lấy dữ liệu từ bảng nếu cần
if is_structured_query and relevant_tables:
for table_name in relevant_tables[:2]: # Lấy 2 bảng liên quan nhất
df = self.tables[table_name]
metadata = self.table_metadata[table_name]
# Sinh table snapshot dạng markdown (LLM đọc được)
table_text = self._table_to_markdown(df, metadata)
results.append(RetrievalResult(
content=table_text,
source_type="table",
relevance_score=0.95,
table_metadata=metadata
))
# Bước 4: Text retrieval (giả lập - thay bằng vector DB thật)
# Trong production, dùng: Pinecone, Weaviate, Qdrant...
# Ở đây minh họa cấu trúc kết quả
# text_results = self._text_search(query, top_k)
return results[:top_k]
def _table_to_markdown(
self,
df: pd.DataFrame,
metadata: TableMetadata,
max_rows: int = 50
) -> str:
"""Chuyển DataFrame thành markdown table — LLM đọc trực tiếp"""
# Giới hạn số dòng để không vượt context
display_df = df.head(max_rows)
# Thêm mô tả schema vào đầu
schema_intro = f"## Bảng: {metadata.table_name}\n"
schema_intro += f"{metadata.description}\n\n"
schema_intro += "**Schema:**\n"
for col in metadata.columns:
schema_intro += f"- {col['name']}: {col['description']} ({col['type']})\n"
schema_intro += "\n"
# Sinh markdown table
markdown = schema_intro
markdown += display_df.to_markdown(index=False)
if len(df) > max_rows:
markdown += f"\n\n_... và {len(df) - max_rows} dòng khác_"
return markdown
============ KHỞI TẠO VÀ SỬ DỤNG ============
Ví dụ thực chiến: Hệ thống e-commerce
def setup_ecommerce_rag():
"""Thiết lập RAG cho hệ thống thương mại điện tử"""
rag = TableDataRAG(HOLYSHEEP_CONFIG)
# === Bảng 1: Sản phẩm ===
products_df = pd.DataFrame([
{"id": "P001", "name": "Áo thun nam", "price": 199000, "category": "Thời trang", "rating": 4.5, "stock": 150},
{"id": "P002", "name": "Quần jeans nữ", "price": 450000, "category": "Thời trang", "rating": 4.8, "stock": 80},
{"id": "P003", "name": "Giày sneaker", "price": 890000, "category": "Giày dép", "rating": 4.3, "stock": 45},
{"id": "P004", "name": "Túi xách da", "price": 1200000, "category": "Phụ kiện", "rating": 4.9, "stock": 20},
{"id": "P005", "name": "Đồng hồ thông minh", "price": 2500000, "category": "Công nghệ", "rating": 4.6, "stock": 60},
])
products_metadata = TableMetadata(
table_name="products",
columns=[
{"name": "id", "type": "string", "description": "Mã sản phẩm duy nhất"},
{"name": "name", "type": "string", "description": "Tên sản phẩm"},
{"name": "price", "type": "integer", "description": "Giá bán (VND)"},
{"name": "category", "type": "string", "description": "Danh mục sản phẩm"},
{"name": "rating", "type": "float", "description": "Điểm đánh giá (1-5)"},
{"name": "stock", "type": "integer", "description": "Số lượng tồn kho"},
],
primary_key="id",
description="Danh sách tất cả sản phẩm đang bán trên sàn thương mại điện tử"
)
rag.register_table(products_df, products_metadata)
# === Bảng 2: Đơn hàng ===
orders_df = pd.DataFrame([
{"id": "O001", "product_id": "P001", "quantity": 2, "total": 398000, "date": "2024-11-01", "status": "delivered"},
{"id": "O002", "product_id": "P003", "quantity": 1, "total": 890000, "date": "2024-11-02", "status": "delivered"},
{"id": "O003", "product_id": "P002", "quantity": 3, "total": 1350000, "date": "2024-11-03", "status": "shipped"},
{"id": "O004", "product_id": "P005", "quantity": 1, "total": 2500000, "date": "2024-11-05", "status": "processing"},
{"id": "O005", "product_id": "P001", "quantity": 5, "total": 995000, "date": "2024-11-07", "status": "delivered"},
])
orders_metadata = TableMetadata(
table_name="orders",
columns=[
{"name": "id", "type": "string", "description": "Mã đơn hàng"},
{"name": "product_id", "type": "string", "description": "Mã sản phẩm (FK → products.id)"},
{"name": "quantity", "type": "integer", "description": "Số lượng mua"},
{"name": "total", "type": "integer", "description": "Tổng tiền (VND)"},
{"name": "date", "type": "date", "description": "Ngày đặt hàng"},
{"name": "status", "type": "string", "description": "Trạng thái: processing/shipped/delivered"},
],
primary_key="id",
description="Lịch sử đơn hàng — dùng để phân tích doanh thu, xu hướng mua hàng"
)
rag.register_table(orders_df, orders_metadata)
return rag
Chạy thử
if __name__ == "__main__":
rag_system = setup_ecommerce_rag()
# Test queries
test_queries = [
"Sản phẩm nào dưới 500k có rating cao nhất?",
"Tổng doanh thu tháng 11 là bao nhiêu?",
"Sản phẩm nào bán chạy nhất?"
]
for query in test_queries:
print(f"\n{'='*60}")
print(f"Câu hỏi: {query}")
print(f"{'='*60}")
results = rag_system.retrieve(query)
for i, result in enumerate(results):
print(f"\n[Kết quả {i+1}] Type: {result.source_type}")
print(result.content[:500] + "..." if len(result.content) > 500 else result.content)
Query Engine thông minh: Từ câu hỏi tự nhiên → Data Operations
"""
Query Engine: Chuyển câu hỏi tiếng Việt thành data operations
Hỗ trợ: filtering, aggregation, sorting, join giữa các bảng
"""
import re
from typing import Callable, Dict, Any
from enum import Enum
class OperationType(Enum):
FILTER = "filter"
AGGREGATE = "aggregate"
SORT = "sort"
JOIN = "join"
TOP_N = "top_n"
@dataclass
class ParsedQuery:
"""Cấu trúc truy vấn đã được phân tích"""
intent: str # "comparison", "statistical", "lookup", "ranking"
operations: List[Tuple[OperationType, Dict[str, Any]]]
target_columns: List[str]
conditions: Dict[str, Any]
tables_involved: List[str]
class TableQueryEngine:
"""
Engine thông minh để phân tích và thực thi truy vấn trên bảng
Ví dụ:
- "Sản phẩm dưới 500k" → Filter(price < 500000)
- "Tổng doanh thu" → Aggregate(SUM, total)
- "Top 3 bán chạy" → TopN(3) + Sort(quantity)
"""
# Từ khóa tiếng Việt cho các operation
KEYWORD_PATTERNS = {
OperationType.FILTER: {
"dưới": lambda col, val: {"op": "<", "value": int(val.replace(",", ""))},
"trên": lambda col, val: {"op": ">", "value": int(val.replace(",", ""))},
"bằng": lambda col, val: {"op": "==", "value": val},
"không": lambda col, val: {"op": "!=", "value": val},
"từ": lambda col, val: {"op": ">=", "value": int(val.replace(",", ""))},
"đến": lambda col, val: {"op": "<=", "value": int(val.replace(",", ""))},
},
OperationType.AGGREGATE: {
"tổng": "sum",
"trung bình": "avg",
"tb": "avg",
"count": "count",
"số lượng": "count",
"nhiều nhất": "max",
"ít nhất": "min",
"cao nhất": "max",
"thấp nhất": "min",
},
OperationType.TOP_N: {
"top": r"top\s*(\d+)",
"danh sách.*(\d+)": r"(\d+)\s*sản phẩm",
}
}
# Intent patterns
INTENT_PATTERNS = {
"comparison": ["so sánh", "hơn", "kém", "bằng", "chênh lệch"],
"statistical": ["tổng", "trung bình", "bao nhiêu", "mấy", "count"],
"lookup": ["cho biết", "tìm", "