Xây dựng RAG pipeline production-ready với LangChain, pgvector và FastAPI

Vì sao tôi viết bài này

Trong vài dự án AI nội bộ ở VNG, câu hỏi lặp lại nhiều nhất không phải là “dùng model nào”, mà là: làm sao để RAG trả lời ổn định, có nguồn, có log và có thể debug? Prototype thường chỉ cần một notebook LangChain. Production thì cần pipeline rõ ràng hơn: ingestion, chunking, embedding, retrieval, generation, evaluation và monitoring.

Mục tiêu của bài này là dựng một baseline đủ tốt với LangChain, pgvector và FastAPI. Bạn có thể thay model, vector store hoặc framework sau này, nhưng cấu trúc tư duy nên giữ.

RAG production-ready không có nghĩa là “phức tạp nhất có thể”. Nó nghĩa là mỗi câu trả lời đều truy được nguồn, đo được chất lượng và biết thất bại ở bước nào.Minh Lê, AI Engineer @ VNG

Kiến trúc tổng thể

Một hệ thống RAG thực tế nên tách hai pipeline: offline ingestion và online query. Ingestion xử lý tài liệu theo batch hoặc event. Query chạy realtime, latency phải kiểm soát.

Luồng xử lý

• Ingestion: đọc PDF/HTML/Markdown, normalize text, chia chunk, sinh embedding, lưu vào PostgreSQL + pgvector.
• Retrieval: embed câu hỏi, tìm top-k chunks, rerank nếu cần, lọc theo quyền truy cập.
• Generation: đưa context vào prompt, yêu cầu model trả lời có citation.
• Evaluation: đo faithfulness, answer relevancy và context precision bằng bộ câu hỏi chuẩn.

Tech stack

• Python 3.11, Poetry để quản lý dependency.
• LangChain cho loader, splitter, retriever và prompt orchestration.
• PostgreSQL 15+ với extension vector.
• FastAPI cho endpoint query.
• Docker Compose cho local dev, Kubernetes cho production.

Bước 1: Setup project

Tạo project bằng Poetry, cài các package tối thiểu:

poetry new rag-production
cd rag-production
poetry add fastapi uvicorn langchain langchain-openai langchain-postgres psycopg[binary] pydantic-settings
poetry add --group dev pytest ruff

Với PostgreSQL, bật extension vector và tạo bảng lưu chunk. Ở production, tôi thường thêm các cột tenant_id, source_url, doc_version để debug và phân quyền.

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE rag_chunks (
  id bigserial PRIMARY KEY,
  tenant_id text NOT NULL,
  source_id text NOT NULL,
  source_url text,
  title text,
  chunk_index int NOT NULL,
  content text NOT NULL,
  metadata jsonb DEFAULT '{}'::jsonb,
  embedding vector(1536),
  created_at timestamptz DEFAULT now()
);

CREATE INDEX rag_chunks_embedding_idx
ON rag_chunks USING ivfflat (embedding vector_cosine_ops)
WITH (lists = 100);

Bước 2: Embedding và chunking

Chunking là nơi nhiều pipeline hỏng âm thầm. Chunk quá nhỏ thì thiếu ngữ cảnh; quá lớn thì retrieval nhiễu. Baseline tốt cho tài liệu tiếng Việt là chunk khoảng 700–1.000 token, overlap 100–150 token.

from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1200,
    chunk_overlap=180,
    separators=["\n\n", "\n", ". ", " ", ""],
)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

chunks = splitter.create_documents(
    texts=[raw_text],
    metadatas=[{"source_id": source_id, "title": title}],
)
vectors = embeddings.embed_documents([c.page_content for c in chunks])

Ở VNG, khi tài liệu có nhiều bảng và policy nội bộ, chúng tôi không chỉ split theo ký tự. Bảng thường cần normalize thành markdown table hoặc key-value bullets trước khi embedding, nếu không câu hỏi về “điều kiện áp dụng” rất dễ trả lời sai.

Bước 3: Retrieval

Baseline retrieval nên có metadata filter. Ví dụ nếu người dùng thuộc phòng Finance, retriever không được lấy tài liệu HR confidential. Đây là lý do nên lưu tenant_id, department hoặc acl trong metadata.

from langchain_postgres import PGVector

store = PGVector(
    connection="postgresql+psycopg://rag:rag@localhost:5432/rag",
    embeddings=embeddings,
    collection_name="company_docs",
    use_jsonb=True,
)

retriever = store.as_retriever(
    search_type="similarity_score_threshold",
    search_kwargs={"k": 8, "score_threshold": 0.72},
)

docs = retriever.invoke("Quy trình xin nghỉ phép khi làm remote là gì?")

Nếu câu hỏi phức tạp, tôi thường thêm reranking bằng cross-encoder hoặc LLM lightweight. Rerank giúp giảm hallucination vì context đưa vào prompt ít nhiễu hơn.

Bước 4: FastAPI endpoint

API nên trả về cả answer lẫn citations. Nếu chỉ trả text, team vận hành không thể kiểm tra câu trả lời đến từ chunk nào.

from fastapi import FastAPI
from pydantic import BaseModel
from langchain_openai import ChatOpenAI

app = FastAPI(title="RAG API")
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

class QueryRequest(BaseModel):
    question: str
    tenant_id: str

@app.post("/query")
def query(req: QueryRequest):
    docs = retriever.invoke(req.question)
    context = "\n\n".join(
        f"[source:{d.metadata.get('source_id')}] {d.page_content}"
        for d in docs
    )
    prompt = f"""
Bạn là trợ lý nội bộ. Chỉ trả lời dựa trên context.
Nếu thiếu dữ liệu, nói rõ là chưa đủ thông tin.

Context:
{context}

Câu hỏi: {req.question}
Trả lời bằng tiếng Việt, có citation [source:id].
"""
    answer = llm.invoke(prompt).content
    return {
        "answer": answer,
        "citations": [d.metadata for d in docs],
    }

Bước 5: Deploy

Khi deploy, hãy tách worker ingestion khỏi API realtime. Worker có thể chạy queue bằng Celery/RQ hoặc job Kubernetes. API chỉ phục vụ query, có timeout ngắn và circuit breaker cho model provider.

• Đặt timeout cho LLM call, ví dụ 20–30 giây.
• Cache embedding cho câu hỏi lặp lại nếu traffic cao.
• Log query_id, top-k chunk ids, prompt hash, model, latency và token cost.
• Không log tài liệu nhạy cảm nguyên văn nếu chưa có chính sách dữ liệu.

5 bài học từ triển khai thực tế

• Eval set quan trọng hơn prompt đẹp. Hãy có 50–100 câu hỏi thật từ user nội bộ.
• Chunking cần theo domain. Policy, FAQ, code docs và hợp đồng không nên split giống nhau.
• pgvector đủ tốt cho nhiều case. Trước khi mua vector DB riêng, hãy đo latency và scale thật.
• Citation là bắt buộc. Không có nguồn thì người dùng không tin, đội vận hành cũng khó debug.
• Cost phải monitor từ ngày đầu. Một prompt thừa context có thể nhân chi phí lên rất nhanh.

Kết luận

Với RAG, production-ready là một chuỗi quyết định nhỏ: schema đủ metadata, chunking có chủ đích, retrieval có filter, API trả citation, deploy tách ingestion và query, cuối cùng là eval liên tục. Nếu bạn đang bắt đầu, đừng cố xây platform hoàn hảo. Hãy dựng baseline trên, đo chất lượng, rồi cải tiến từng lớp.