Bài 25 của series LLM từ zero dạy RAG theo kiểu pipeline: embed câu hỏi, tìm top-K chunk, nhét vào prompt, gọi LLM, lấy câu trả lời. Một lần. Tuyến tính. Cố định.

Pipeline đó hoạt động tốt cho chatbot QA đơn giản. Nhưng khi bạn build agent, nó bắt đầu bộc lộ giới hạn ngay câu hỏi thứ hai.

Tôi nhớ lần đầu gắn RAG vào một agent phân tích codebase. Agent cần tìm “tất cả các endpoint có dùng JWT auth”. Câu trả lời nằm rải rác: một phần ở file middleware, một phần ở route definition, một phần ở config. Pipeline RAG cổ điển query một lần với câu hỏi gốc, trả về 5 chunk đầu tiên khớp về mặt ngữ nghĩa với “JWT”. Toàn chunk về cách JWT hoạt động, không phải danh sách endpoint. Agent trả lời sai.

Vấn đề không phải ở vector DB hay embedding model. Vấn đề là retrieval được gọi một lần, trước khi agent bắt đầu suy nghĩ.

Pipeline RAG vs Agentic RAG

Sự khác biệt cốt lõi nằm ở chỗ ai quyết định query gì và khi nào.

Pipeline RAG (QA)Agentic RAG
Ai querySystem tự độngLLM quyết định
Query là gìCâu hỏi gốc của userReformulated query, có thể nhiều lần
Khi nào queryTrước khi gọi LLMBất cứ khi nào LLM thấy cần
Số lần retrieval1 lần cố địnhNhiều lần, phụ thuộc task
Xử lý kết quảNhét thẳng vào promptLLM đọc, quyết định query tiếp hay không
Phù hợp vớiFAQ, document QAResearch, phân tích, multi-hop lookup

Trong agentic RAG, retrieval là một tool. LLM gọi nó như gọi bất kỳ tool nào khác: khi cần thông tin, gọi search_docs. Khi thấy kết quả chưa đủ, query lại với keyword khác. Khi task xong, dừng.

Retrieval như là tool

Cách đơn giản nhất để biến RAG thành agentic là expose retrieval như một function.

import anthropic
from qdrant_client import QdrantClient
from qdrant_client.models import VectorParams, Distance
from sentence_transformers import SentenceTransformer

client = anthropic.Anthropic()
qdrant = QdrantClient(":memory:")
embed_model = SentenceTransformer("BAAI/bge-m3")

# Setup collection (giả sử đã có documents indexed)
COLLECTION = "codebase"

def search_docs(query: str, top_k: int = 5) -> list[dict]:
    """Tìm chunks liên quan từ vector DB."""
    query_vector = embed_model.encode(query).tolist()
    results = qdrant.search(
        collection_name=COLLECTION,
        query_vector=query_vector,
        limit=top_k,
    )
    return [
        {"score": hit.score, "content": hit.payload["content"], "source": hit.payload.get("source", "")}
        for hit in results
    ]

TOOLS = [
    {
        "name": "search_docs",
        "description": (
            "Tìm kiếm trong codebase / tài liệu nội bộ. "
            "Dùng khi cần tìm code, API, convention, hoặc thông tin kỹ thuật. "
            "Có thể gọi nhiều lần với query khác nhau nếu kết quả chưa đủ."
        ),
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "Query tìm kiếm. Ngắn gọn, tập trung vào một khía cạnh."
                },
                "top_k": {
                    "type": "integer",
                    "description": "Số chunk trả về. Mặc định 5.",
                    "default": 5
                }
            },
            "required": ["query"]
        }
    }
]

def execute_tool(name: str, args: dict) -> str:
    if name == "search_docs":
        results = search_docs(args["query"], args.get("top_k", 5))
        if not results:
            return "Không tìm thấy kết quả."
        lines = []
        for r in results:
            lines.append(f"[score={r['score']:.3f}] {r['source']}\n{r['content']}")
        return "\n\n---\n\n".join(lines)
    return f"Unknown tool: {name}"

Tool description quan trọng không kém schema. Câu "Có thể gọi nhiều lần với query khác nhau" là hint cho LLM biết rằng nó không bị giới hạn ở một lần search.

Multi-hop retrieval

Đây là nơi agentic RAG thật sự tỏa sáng. Câu hỏi phức tạp thường cần nhiều bước retrieval, mỗi bước phụ thuộc vào kết quả bước trước.

Ví dụ câu hỏi: “Service order-processor có rate limit không, và nếu có thì config ở đâu?”

Pipeline RAG cổ điển: query một lần với “order-processor rate limit”, nhận về một vài chunk, trả lời.

Agent với agentic RAG:

  1. Query search_docs("order-processor service") để tìm file liên quan
  2. Đọc kết quả, thấy service dùng middleware tên throttle-middleware
  3. Query search_docs("throttle-middleware config") để tìm config
  4. Thấy config reference tới rate-limit.yaml
  5. Query search_docs("rate-limit.yaml order-processor") để tìm file config thật
  6. Có đủ thông tin, tổng hợp câu trả lời

Mỗi bước, agent dùng thông tin từ bước trước để hình thành query tiếp theo. Đây là multi-hop: nhảy từ node này sang node khác trong knowledge graph ngầm của codebase.

def agent_with_rag(user_question: str, max_iterations: int = 15) -> str:
    messages = [{"role": "user", "content": user_question}]

    for _ in range(max_iterations):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=2048,
            tools=TOOLS,
            messages=messages,
        )
        messages.append({"role": "assistant", "content": response.content})

        if response.stop_reason == "end_turn":
            for block in response.content:
                if hasattr(block, "text"):
                    return block.text
            return ""

        if response.stop_reason == "tool_use":
            tool_results = []
            for block in response.content:
                if block.type == "tool_use":
                    result = execute_tool(block.name, block.input)
                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": result,
                    })
            messages.append({"role": "user", "content": tool_results})

    return "Max iterations reached without answer."

Không có gì đặc biệt trong control loop này so với agent thông thường ở bài 1. Multi-hop xảy ra tự nhiên vì agent có quyền gọi search_docs nhiều lần.

Query rewriting

Một vấn đề tinh vi của agentic RAG: câu hỏi gốc của user thường là ngôn ngữ tự nhiên, không phải keyword tốt cho vector search.

“Tại sao login bị chậm vào buổi sáng?” không match tốt với các chunk về “authentication latency spike”, “database connection pool exhausted”, hay “Redis cache miss rate”.

Có hai cách xử lý:

Cách 1: Để LLM tự rewrite. Vì LLM đã biết task, nó tự nhiên sẽ dùng keyword kỹ thuật khi gọi search_docs. Câu hỏi “login bị chậm” sẽ được translate thành query "authentication latency", rồi "connection pool", rồi "cache performance".

Cách này đơn giản và hoạt động tốt nếu system prompt rõ ràng.

Cách 2: Tool layer riêng cho query expansion. Thêm một tool expand_query hoặc generate_search_queries để LLM dùng trước khi search.

{
    "name": "generate_search_queries",
    "description": (
        "Sinh ra 3-5 search queries từ một câu hỏi người dùng. "
        "Dùng khi câu hỏi gốc quá tự nhiên, cần rewrite thành keyword kỹ thuật trước khi search."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "question": {"type": "string", "description": "Câu hỏi cần expand"},
            "domain": {"type": "string", "description": "Domain của codebase: backend, frontend, infra, v.v."}
        },
        "required": ["question"]
    }
}

Tool này không cần vector DB, chỉ cần một LLM call nhỏ (hoặc thậm chí gọi lại chính model đang chạy). Nhưng nó thêm latency và token cost. Dùng khi user input đặc biệt “noisy” về mặt ngôn ngữ, ví dụ Slack message dump hay bug report tự do.

Vector DB: pgvector, Qdrant, Pinecone

Ba lựa chọn phổ biến, mỗi cái phù hợp với một context:

pgvector: Extension của PostgreSQL. Nếu stack đã có Postgres, đây là lựa chọn ít overhead nhất. Không cần dịch vụ mới, backup cùng với data, query kết hợp được với metadata filter bằng SQL thuần.

# pgvector với psycopg2
import psycopg2
import json

conn = psycopg2.connect("postgresql://localhost/mydb")
cur = conn.cursor()

# Tạo table
cur.execute("""
    CREATE TABLE IF NOT EXISTS docs (
        id SERIAL PRIMARY KEY,
        content TEXT,
        source TEXT,
        embedding vector(768)
    )
""")

# Insert
embedding = embed_model.encode("nội dung chunk").tolist()
cur.execute(
    "INSERT INTO docs (content, source, embedding) VALUES (%s, %s, %s)",
    ("nội dung chunk", "src/api/auth.py", json.dumps(embedding))
)

# Search top-5
query_emb = embed_model.encode("authentication endpoint").tolist()
cur.execute("""
    SELECT content, source, 1 - (embedding <=> %s::vector) AS score
    FROM docs
    ORDER BY embedding <=> %s::vector
    LIMIT 5
""", (json.dumps(query_emb), json.dumps(query_emb)))
rows = cur.fetchall()

Qdrant: Vector DB chuyên dụng, self-hosted hoặc cloud. Filter metadata mạnh hơn pgvector. Hỗ trợ named vectors (dùng nhiều embedding model cùng lúc). Tốt khi collection lớn (triệu+ vectors) hoặc cần advanced filtering.

Pinecone: Serverless, zero ops. Tốt cho prototype nhanh hoặc khi không muốn quản lý infrastructure. Cost cao hơn ở scale lớn.

Cho agent dùng nội bộ (codebase, docs): pgvector thường là lựa chọn thực tế nhất. Cho production với nhiều tenant hoặc collection lớn: Qdrant.

Hybrid search: BM25 và embedding

Embedding search tốt cho semantic match. Nhưng khi user tìm tên function cụ thể, class name, hoặc error message chính xác, BM25 (full-text search) thường tốt hơn.

User: "Tìm hàm `validate_jwt_expiry`"
Embedding search: trả về chunk về JWT validation nói chung
BM25: trả về đúng file chứa hàm đó

Hybrid search kết hợp cả hai, dùng RRF (Reciprocal Rank Fusion) để merge kết quả:

def hybrid_search(query: str, top_k: int = 5) -> list[dict]:
    # Dense search (embedding)
    query_vector = embed_model.encode(query).tolist()
    dense_results = qdrant.search(
        collection_name=COLLECTION,
        query_vector=query_vector,
        limit=top_k * 2,  # lấy nhiều hơn để merge
    )

    # Sparse search (BM25) - Qdrant hỗ trợ sparse vectors
    # Hoặc dùng Elasticsearch / PostgreSQL full-text song song
    sparse_results = bm25_search(query, top_k=top_k * 2)

    # RRF merge
    scores: dict[str, float] = {}
    k = 60  # constant trong RRF

    for rank, hit in enumerate(dense_results):
        doc_id = hit.id
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    for rank, hit in enumerate(sparse_results):
        doc_id = hit["id"]
        scores[doc_id] = scores.get(doc_id, 0) + 1 / (k + rank + 1)

    # Sort by merged score
    sorted_ids = sorted(scores, key=lambda x: scores[x], reverse=True)
    return [get_doc_by_id(doc_id) for doc_id in sorted_ids[:top_k]]

Với agent tool interface, hybrid search là implementation detail. LLM không biết và không cần biết. Nó vẫn gọi search_docs(query), kết quả trả về tốt hơn.

Đây là bug tôi gặp sau khoảng một tuần chạy agent RAG trong production.

Agent nhận câu hỏi về codebase. Task đơn giản. Nhưng trace log cho thấy search_docs được gọi 23 lần cho một câu hỏi. Mỗi lần thêm ~1500 token vào context. Context window phồng lên, latency tăng, cost nhân đôi so với dự kiến.

Vấn đề: không phải agent “ngốc”. Agent đang làm quá cẩn thận. Nó tìm một chunk, thấy thông tin chưa đủ chắc, tìm thêm. Tìm lại từ góc khác. Confirm bằng một query khác. Đây là over-search.

Một số heuristic để fix:

Giới hạn số lần gọi tool cùng tên. Track trong execute_tool, raise lỗi sau N lần:

tool_call_counts: dict[str, int] = {}
MAX_SEARCH_CALLS = 8

def execute_tool(name: str, args: dict) -> str:
    tool_call_counts[name] = tool_call_counts.get(name, 0) + 1
    if name == "search_docs" and tool_call_counts[name] > MAX_SEARCH_CALLS:
        return (
            "Đã tìm kiếm nhiều lần. Hãy tổng hợp từ thông tin đã có, "
            "hoặc báo cáo rằng không đủ thông tin để trả lời chính xác."
        )
    # ... execute normally

System prompt rõ ràng về kỳ vọng. Thêm vào system prompt:

Khi search_docs, ưu tiên đọc kỹ kết quả trước khi quyết định search tiếp.
Nếu đã có 3-4 chunk liên quan, đủ để tổng hợp thì không cần search thêm.
Chỉ search thêm khi thiếu thông tin cụ thể, không phải để confirm thứ đã biết.

Token budget tracking. Nếu context đã dùng quá 60% max_tokens, thêm cảnh báo vào tool result:

def execute_tool(name: str, args: dict, current_tokens: int, max_tokens: int) -> str:
    result = _do_execute(name, args)
    if current_tokens / max_tokens > 0.6:
        result += "\n\n[Ghi chú: Context window gần đầy. Ưu tiên tổng hợp câu trả lời từ thông tin hiện có.]"
    return result

Over-search tốn tiền và tốn thời gian, nhưng không gây sai kết quả. Nguy hiểm hơn là under-search: agent search một lần, nhận kết quả mediocre, tự tin trả lời sai. Monitor cả hai hướng khi trace production.

Cheatsheet: pipeline RAG vs agentic RAG

Câu hỏi thiết kếPipeline RAGAgentic RAG
Query do ai tạo?System (câu hỏi gốc)LLM (tự rewrite)
Query cố định?Có, 1 lầnKhông, nhiều lần
Có thể multi-hop?KhôngCó
Kết quả dùng ngay?Nhét vào promptLLM đọc rồi quyết định
Dễ implement?Rất dễTrung bình
Phù hợp vớiFAQ, knowledge base QAAgent phân tích, research, debug
LatencyThấp (1 round-trip)Cao hơn (nhiều round-trip)
CostThấpCao hơn, cần giới hạn
Pitfall chínhRetrieval missOver-search, context bloat

Không phải agentic RAG lúc nào cũng tốt hơn. Khi task là “trả lời câu hỏi từ tài liệu cố định”, pipeline RAG đơn giản hơn, rẻ hơn, dễ debug hơn. Chỉ nâng lên agentic khi câu trả lời đòi hỏi nhiều bước lookup, hoặc khi agent cần kết hợp thông tin từ nhiều nguồn khác nhau.

Lời kết

RAG trong agent không phải RAG được nhét vào agent. Đó là sự thay đổi tư duy: từ “retrieve rồi mới gọi LLM” sang “LLM quyết định có cần retrieve không, và query gì”.

Memory của agent ở bài 4 lưu thông tin trong context window. RAG mở rộng khả năng đó ra ngoài context window, vào một knowledge base lớn tùy ý. Hai cơ chế bổ sung cho nhau, không thay thế nhau.

Bài tiếp theo, bài 15, là MCP: Model Context Protocol. MCP chuẩn hóa cách agent kết nối với tools và data sources, bao gồm cả các RAG server. Thay vì mỗi agent tự viết search_docs, MCP cho phép một server expose tool đó cho nhiều agent và nhiều model khác nhau dùng chung.