Kiến trúc RAG cho xử lý tài liệu doanh nghiệp

RAG (Retrieval-Augmented Generation) là kiến trúc AI cho phép LLM trả lời câu hỏi dựa trên dữ liệu nội bộ doanh nghiệp mà không cần fine-tune model tốn kém. Ventra Rocket đã triển khai nhiều hệ thống RAG giúp doanh nghiệp khai thác hàng nghìn tài liệu PDF, Word, và database một cách hiệu quả.

1. Tổng quan kiến trúc RAG

Hệ thống RAG gồm hai pipeline chính:

Indexing Pipeline — Chạy offline, xử lý và lưu trữ tài liệu vào vector database.

Query Pipeline — Chạy real-time khi có câu hỏi từ người dùng.

Indexing:
Documents → Chunking → Embedding → Vector Store

Query:
User Question → Embedding → Vector Search → Context Assembly → LLM → Answer

2. Document Ingestion và Chunking

Chiến lược chunking ảnh hưởng trực tiếp đến chất lượng retrieval. Chunk quá lớn làm loãng context, chunk quá nhỏ mất đi ngữ cảnh.

from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.document_loaders import PyPDFLoader, Docx2txtLoader
import os

def load_documents(file_path: str):
    ext = os.path.splitext(file_path)[1].lower()
    if ext == '.pdf':
        loader = PyPDFLoader(file_path)
    elif ext in ('.docx', '.doc'):
        loader = Docx2txtLoader(file_path)
    else:
        raise ValueError(f"Unsupported file type: {ext}")
    return loader.load()

def chunk_documents(documents, chunk_size=800, chunk_overlap=100):
    """
    chunk_size=800 tokens là điểm cân bằng tốt cho tài liệu kỹ thuật.
    chunk_overlap=100 đảm bảo không mất thông tin ở ranh giới chunk.
    """
    splitter = RecursiveCharacterTextSplitter(
        chunk_size=chunk_size,
        chunk_overlap=chunk_overlap,
        separators=["\n\n", "\n", ".", "!", "?", ",", " "],
        length_function=len,
    )
    return splitter.split_documents(documents)

3. Embedding và Vector Storage với Qdrant

from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
from openai import OpenAI
import uuid

client = OpenAI()
qdrant = QdrantClient(url="http://localhost:6333")

COLLECTION_NAME = "enterprise_docs"
EMBEDDING_DIM = 1536  # text-embedding-3-small

def create_collection():
    qdrant.recreate_collection(
        collection_name=COLLECTION_NAME,
        vectors_config=VectorParams(
            size=EMBEDDING_DIM,
            distance=Distance.COSINE,
        ),
    )

def embed_and_store(chunks: list, metadata: dict):
    texts = [chunk.page_content for chunk in chunks]

    # Batch embedding để giảm API calls
    response = client.embeddings.create(
        input=texts,
        model="text-embedding-3-small",
    )
    embeddings = [item.embedding for item in response.data]

    points = [
        PointStruct(
            id=str(uuid.uuid4()),
            vector=embedding,
            payload={
                "text": text,
                "source": metadata.get("source", ""),
                "department": metadata.get("department", ""),
                "doc_type": metadata.get("doc_type", ""),
                "created_at": metadata.get("created_at", ""),
            },
        )
        for text, embedding in zip(texts, embeddings)
    ]

    qdrant.upsert(collection_name=COLLECTION_NAME, points=points)

4. Retrieval với Hybrid Search

Kết hợp semantic search (vector) và keyword search (BM25) cho kết quả tốt nhất.

from qdrant_client.models import Filter, FieldCondition, MatchValue

def retrieve_context(
    query: str,
    department_filter: str | None = None,
    top_k: int = 5,
) -> list[dict]:
    # Embed câu hỏi
    query_embedding = client.embeddings.create(
        input=query,
        model="text-embedding-3-small",
    ).data[0].embedding

    # Build filter nếu có
    search_filter = None
    if department_filter:
        search_filter = Filter(
            must=[
                FieldCondition(
                    key="department",
                    match=MatchValue(value=department_filter),
                )
            ]
        )

    results = qdrant.search(
        collection_name=COLLECTION_NAME,
        query_vector=query_embedding,
        query_filter=search_filter,
        limit=top_k,
        score_threshold=0.72,  # Lọc kết quả không liên quan
        with_payload=True,
    )

    return [
        {
            "text": hit.payload["text"],
            "source": hit.payload["source"],
            "score": hit.score,
        }
        for hit in results
    ]

5. Generation với Context Assembly

def build_prompt(query: str, contexts: list[dict]) -> str:
    context_text = "\n\n---\n\n".join(
        f"[Nguồn: {ctx['source']}]\n{ctx['text']}"
        for ctx in contexts
    )

    return f"""Bạn là trợ lý AI nội bộ của doanh nghiệp. Trả lời câu hỏi dựa trên tài liệu được cung cấp.
Nếu thông tin không có trong tài liệu, hãy nói rõ điều đó thay vì đoán mò.
Luôn trích dẫn nguồn tài liệu trong câu trả lời.

TÀI LIỆU THAM KHẢO:
{context_text}

CÂU HỎI: {query}

TRẢ LỜI:"""

def answer_question(query: str, department: str | None = None) -> dict:
    contexts = retrieve_context(query, department_filter=department)

    if not contexts:
        return {
            "answer": "Không tìm thấy thông tin liên quan trong tài liệu nội bộ.",
            "sources": [],
        }

    prompt = build_prompt(query, contexts)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1,  # Thấp để đảm bảo tính chính xác
        max_tokens=1000,
    )

    return {
        "answer": response.choices[0].message.content,
        "sources": list({ctx["source"] for ctx in contexts}),
    }

6. Đánh giá chất lượng RAG

Đo lường bằng ba chỉ số RAGAS:

• Faithfulness — Câu trả lời có trung thực với context không?
• Answer Relevancy — Câu trả lời có liên quan đến câu hỏi không?
• Context Recall — Context retrieved có đủ thông tin để trả lời không?

from ragas import evaluate
from ragas.metrics import faithfulness, answer_relevancy, context_recall

results = evaluate(
    dataset=test_dataset,
    metrics=[faithfulness, answer_relevancy, context_recall],
)
print(results)
# {'faithfulness': 0.91, 'answer_relevancy': 0.87, 'context_recall': 0.84}

Kết luận

RAG là giải pháp thực tế nhất để đưa AI vào quy trình nội bộ doanh nghiệp. Kiến trúc này không đòi hỏi fine-tune model, dễ cập nhật dữ liệu, và có thể kiểm soát phạm vi trả lời. Ventra Rocket đã triển khai thành công cho nhiều doanh nghiệp với độ chính xác trên 90%.