การสร้าง Production RAG System: ก้าวข้ามขั้น Demo

Retrieval-Augmented Generation ดูเหมือนเรื่องง่ายใน notebook แต่โหดร้ายใน production ช่องว่างระหว่างสองสิ่งนี้เต็มไปด้วย PDF ที่ยุ่งเหยิง คำถามที่กำกวม hallucination ที่เงียบงัน และค่าใช้จ่ายที่พุ่งสูงโดยไม่มีใครคาดการณ์ไว้ บทความนี้คือ field guide ที่ผมอยากจะมีตั้งแต่ครั้งแรกที่ RAG demo ได้พบกับ corpus จริง

TL;DR

• คุณภาพของ retrieval คือเพดานของคุณภาพ RAG — model จะตอบได้เฉพาะจาก context ที่มันได้รับจริงเท่านั้น
• แยกระบบออกเป็น offline indexing pipeline และ online query pipeline อย่ายุบรวมเข้าด้วยกัน
• ทุ่มเวลาให้กับ ingestion อย่างไม่สมส่วน: HTML chrome, PDF OCR, ตาราง และ near-duplicates ล้วนทำลาย retrieval
• ใช้ chunking แบบ document-structure-aware เมื่อมีโครงสร้าง และ fallback ไปที่ recursive splitting พร้อม overlap
• ใช้ hybrid search เป็น default (BM25 + vector รวมกันด้วย RRF) แล้วจึงเพิ่ม cross-encoder reranker เมื่อ latency เอื้ออำนวย
• บังคับให้มี citations ใน prompt และปฏิเสธคำตอบที่มีข้อกล่าวอ้างโดยไม่มี source — สิ่งนี้กำจัด hallucination ทั้งคลาส
• สร้าง gold eval set ขนาดเล็กตั้งแต่เนิ่น ๆ วัด context recall / answer relevance / faithfulness และเฝ้าดู production traffic

ทำไม RAG Demos ส่วนใหญ่ไม่รอดเมื่อเจอข้อมูลจริง

ทุก LLM demo เดินตามเส้นทางเดียวกัน คุณ load โฟลเดอร์ของ markdown สะอาด ๆ แบ่งทุก ๆ 1,000 ตัวอักษร embed ด้วย text-embedding-3-small ยัด top-5 chunks เข้า prompt และ model ตอบได้สมบูรณ์แบบกับสามคำถามที่คุณซ้อมไว้

จากนั้นคุณ ship มันออกไป corpus จริงปรากฏตัว: scanned PDFs ที่มีหน้าหมุน layouts หลายคอลัมน์ ตารางที่มีความสำคัญ code blocks ที่ห้ามแตก HTML dumps ที่มี navigation chrome เอกสารซ้ำกันจาก CMS exports สามแห่ง และคำถามอย่าง “เราตัดสินใจอะไรเกี่ยวกับ vendor X ใน Q3?” ที่ไม่มี cosine similarity ตัวไหนจะค้นพบได้

บทความนี้พูดถึงสิ่งที่อยู่ระหว่าง demo และระบบ RAG ที่อยู่รอดใน production ผมสมมติว่าคุณสร้าง demo มาแล้ว ถ้ายัง — สร้างมันก่อน เพราะ failure modes ด้านล่างจะเข้าใจได้ก็ต่อเมื่อคุณรู้สึกถึงแรงดึงดูดของ happy path แบบไร้เดียงสาแล้ว

ธีมตลอดทั้งบทความ: คุณภาพของ retrieval คือเพดานของคุณภาพ RAG model ที่ฉลาดที่สุดในโลกก็ตอบจาก context ที่มันไม่เคยได้รับไม่ได้ ทุกหัวข้อด้านล่างสุดท้ายแล้วล้วนเกี่ยวกับการป้อน context ที่ดีกว่าให้กับ model

ภาพรวมของ Pipeline

ระบบ RAG ระดับ production ไม่ใช่ pipeline เดียว แต่เป็นสอง: offline indexing pipeline ที่ประมวลผลเอกสารต้นฉบับ และ online query pipeline ที่รับ requests ทั้งสองใช้ index ร่วมกัน และไม่มีอะไรร่วมกันอีก

flowchart LR
    subgraph Offline[Indexing pipeline]
      A[Source docs] --> B[Parse / OCR]
      B --> C[Chunk]
      C --> D[Embed]
      D --> E[(Vector + BM25 index)]
    end
    subgraph Online[Query pipeline]
      Q[User query] --> R[Retrieve<br/>BM25 + vector]
      R --> F[Fuse RRF]
      F --> RR[Rerank<br/>cross-encoder]
      RR --> P[Prompt with citations]
      P --> L[LLM]
      L --> ANS[Answer + sources]
      ANS --> EV[Eval / telemetry]
    end
    E --> R

ทีมส่วนใหญ่ยุบทั้งสองให้เป็น script เดียวแล้วเสียใจในภายหลัง แยกพวกมันออกจากกัน indexing pipeline เป็น batch job ที่อาจใช้เวลาหลายชั่วโมง ส่วน query pipeline ต้องตอบภายในไม่กี่ร้อยมิลลิวินาที

Ingestion: 40% ที่ไม่หรูหราแต่จำเป็น

คุณจะใช้เวลากับการ parse มากกว่าใช้กับ prompts, embeddings และ vector DBs รวมกัน นี่ไม่ใช่ความล้มเหลวของ architecture — มันคือธรรมชาติของเอกสารจริง

HTML

ความผิดพลาดแรกคือใช้ BeautifulSoup.get_text() แล้ว ship เลย HTML จริงประกอบด้วย navigation, cookie banners, footers, “related posts,” ad slots และ <script> tags ที่รอดพ้นจากการ extract แบบไร้เดียงสา พวกมันสร้างมลพิษให้ index ของคุณด้วย boilerplate ที่ใกล้เคียงกันซึ่งทำลาย retrieval precision — “privacy policy” จะ outrank คอนเทนต์จริงของคุณ

ใช้ reader-mode extractor (trafilatura, readability-lxml หรือ readability ของ Mozilla) ก่อนทำอะไรอื่น เครื่องมือเหล่านี้แยก main content block ลอก chrome ออก และรักษาโครงสร้างไว้

import trafilatura

def extract_html(url_or_html: str) -> str | None:
    return trafilatura.extract(
        url_or_html,
        include_tables=True,
        include_links=False,
        include_comments=False,
        favor_precision=True,
    )

PDF

PDF เป็น format ที่ยากที่สุดใน corpus มันมาในสามรูปแบบ และคุณต้องตรวจสอบให้ได้ว่าคุณมีแบบไหน:

1. Born-digital text PDFs — text layer สมบูรณ์ pypdf หรือ pdfplumber ใช้ได้ แต่ลำดับคอลัมน์มักผิด
2. Scanned PDFs — “text” คือ pixels คุณต้องใช้ OCR (tesseract, paddleocr หรือ cloud vision API)
3. Hybrid — มี text layer แต่สร้างจาก OCR ห่วย ๆ ตอนสร้าง คุณจึงได้ rn แทน m และความเสียหายในลักษณะเดียวกัน

heuristic ที่มีประโยชน์: ถ้า len(extracted_text) / num_pages < 100 ให้สมมติว่ามันคือ scanned หรือเสียหาย แล้ว fallback ไปที่ OCR สำหรับการ extract แบบ layout-aware (ตาราง หลายคอลัมน์ figures) docling, unstructured หรือ marker ให้ผลลัพธ์ที่ดีกว่า PDF libraries ดิบ ๆ อย่างน่าทึ่ง — แลกกับ compute ต่อหน้าที่สูงขึ้นมาก

from pathlib import Path
import pypdf
from unstructured.partition.pdf import partition_pdf

def parse_pdf(path: Path) -> list[dict]:
    # Fast path: try text extraction first
    reader = pypdf.PdfReader(path)
    raw = "\n".join(p.extract_text() or "" for p in reader.pages)

    if len(raw) / max(len(reader.pages), 1) < 100:
        # Likely scanned — fall back to layout-aware parser with OCR
        elements = partition_pdf(
            filename=str(path),
            strategy="hi_res",
            infer_table_structure=True,
            languages=["eng"],
        )
        return [{"type": e.category, "text": e.text, "page": e.metadata.page_number}
                for e in elements if e.text]

    return [{"type": "NarrativeText", "text": raw, "page": None}]

ตาราง

ตารางเป็นจุดที่ระบบ RAG ส่วนใหญ่สูญเสียความแม่นยำอย่างเงียบ ๆ โครงสร้าง 2 มิติที่ถูก serialize เป็น text คั่นด้วย whitespace แทบจะไร้ประโยชน์สำหรับ embedding model สองตัวเลือก:

• Serialize เป็น markdown tables ซึ่ง LLMs อ่านได้แบบ native ดีสำหรับตารางขนาดเล็ก
• Serialize แต่ละแถวเป็นประโยค (“In 2024, revenue was $12M and headcount was 82.”) แล้ว embed แต่ละแถวแยกกัน ดีกว่าสำหรับตารางขนาดใหญ่ที่คุณต้องการ retrieve

Code blocks

ถ้า corpus ของคุณมี code — docs, runbooks, internal wikis — ให้รักษา fence markers และ language tags ผ่านทุกการแปลง code ที่แตกข้าม chunks แย่กว่าไร้ประโยชน์ model จะ hallucinate ครึ่งที่หายไป

Duplicates

การตรวจ near-duplicate มีความสำคัญมากกว่าที่คนคิด การ migrate CMS แบบ multi-export มักผลิตบทความเดียวกันสามครั้งโดยมี whitespace ต่างกัน MinHash (ผ่าน datasketch) บน text ที่ normalize แล้วจับได้ 95% อย่างประหยัด pass ที่สองด้วย exact-hash บน normalized content จับที่เหลือ

Chunking: ไม่มีของฟรี

Chunking เป็น optimization problem ร่วมระหว่าง embedding quality, retrieval recall และ prompt budget ทุก strategy แลกอย่างหนึ่งกับอีกอย่าง

Fixed-size token chunks

ง่ายที่สุด และมักจะผิด ตัดประโยค ตาราง และ code

Recursive character splitting

de facto default แบ่งที่ paragraph breaks ก่อน แล้ว sentence breaks แล้ว words — ตกลงไปถึง character splits แข็ง ๆ เฉพาะเมื่อไม่มีอะไรอื่นใช้ได้ ใช้ overlap 10–20% เพื่อรักษา context ข้ามขอบเขต

def recursive_split(
    text: str,
    chunk_size: int = 800,
    overlap: int = 120,
    separators: tuple[str, ...] = ("\n\n", "\n", ". ", " ", ""),
) -> list[str]:
    def split(text: str, seps: tuple[str, ...]) -> list[str]:
        if len(text) <= chunk_size or not seps:
            return [text]
        sep, rest = seps[0], seps[1:]
        parts = text.split(sep) if sep else list(text)
        chunks, buf = [], ""
        for p in parts:
            candidate = buf + (sep if buf else "") + p
            if len(candidate) <= chunk_size:
                buf = candidate
            else:
                if buf:
                    chunks.append(buf)
                buf = p if len(p) <= chunk_size else ""
                if len(p) > chunk_size:
                    chunks.extend(split(p, rest))
        if buf:
            chunks.append(buf)
        return chunks

    raw = split(text, separators)
    # Apply overlap by prepending the tail of the previous chunk
    out: list[str] = []
    prev_tail = ""
    for c in raw:
        out.append((prev_tail + c)[-chunk_size - overlap:])
        prev_tail = c[-overlap:] if overlap else ""
    return out

Semantic chunking

Embed แต่ละประโยค วัด cosine distance ระหว่างเพื่อนบ้าน แบ่งตรงขอบเขตที่ระยะห่างสูง ผลิต chunks ที่เคารพการเปลี่ยนหัวข้อแทนที่จะเป็น paragraph breaks คุ้มค่าสำหรับ long-form content (papers, book chapters) ที่ paragraphs ดูประดิษฐ์

ค่าใช้จ่าย: คุณ embed ทั้ง corpus สองครั้ง (ครั้งหนึ่งสำหรับ chunking ครั้งหนึ่งสำหรับ indexing)

Document-structure-aware chunking

สำหรับสิ่งที่มีหัวข้อ — docs, runbooks, legal — แบ่งตามลำดับชั้นของหัวข้อและเก็บ heading trail ไว้เป็น metadata ทุก chunk รู้ว่ามันอยู่ในส่วนไหน ซึ่งหมายความว่าคุณกรองได้ (“เฉพาะ chunks จากส่วน Security”) และคุณแสดง breadcrumbs จริงให้ users เห็นได้

from dataclasses import dataclass, field

@dataclass
class Chunk:
    text: str
    doc_id: str
    heading_trail: list[str] = field(default_factory=list)
    page: int | None = None
    token_count: int = 0

def chunk_by_headings(doc_id: str, markdown: str, max_tokens: int = 400) -> list[Chunk]:
    chunks: list[Chunk] = []
    trail: list[str] = []
    buf: list[str] = []

    def flush():
        if not buf:
            return
        text = "\n".join(buf).strip()
        if text:
            chunks.append(Chunk(text=text, doc_id=doc_id, heading_trail=list(trail)))
        buf.clear()

    for line in markdown.splitlines():
        if line.startswith("#"):
            flush()
            level = len(line) - len(line.lstrip("#"))
            title = line.lstrip("# ").strip()
            trail = trail[: level - 1] + [title]
        else:
            buf.append(line)
            if sum(len(x) for x in buf) > max_tokens * 4:  # rough char→token
                flush()
    flush()
    return chunks

Rule of thumb: เริ่มด้วย document-structure-aware chunking เมื่อมีโครงสร้าง recursive splitting เป็น fallback และ semantic chunking เฉพาะเมื่อคุณวัดแล้วว่าสองอันแรกไม่พอ

Embeddings: เลือกสองในสาม

คุณกำลังแลกบนสามแกน: คุณภาพ, dimensionality (storage + retrieval cost) และ price/latency ไม่มี model ที่ชนะทั้งสาม และ rankings เปลี่ยนทุกไตรมาส ดังนั้นจงมองส่วนนี้เป็น framework มากกว่าคำตัดสิน

shortlist ที่สมจริงในปัจจุบัน:

• OpenAI text-embedding-3-large / -small — ภาษาอังกฤษทั่วไปแข็งแรง API ง่าย และ -small ราคาถูกมาก รองรับ dimension truncation (Matryoshka) ดังนั้นคุณแลก recall กับ storage ได้โดยไม่ต้อง re-embed
• Cohere embed-v3 — คุณภาพแข่งขันได้ มี multilingual variants ที่ใช้ได้จริงกับ corpus ที่ไม่ใช่ภาษาอังกฤษ
• bge-* (BAAI general embedding) — open-weight, self-hostable แข็งแรงบน English benchmarks ดีเมื่อข้อมูลออกจาก network ของคุณไม่ได้
• gte-* (Alibaba) — niche คล้ายกับ bge บางครั้งแข็งแรงกว่าใน domain เฉพาะ

ไม่ว่าจะเลือกอันไหน ประเมินบนข้อมูลของคุณเอง — published benchmark rankings ไม่เคยตรงกับ corpus เฉพาะ สร้าง gold set ขนาดเล็ก (หัวข้อถัดไป) แล้ว run แต่ละตัวเลือกกับมัน

Dimensions vs accuracy

embeddings ที่มี dimension สูงพกพา signal มากกว่าและ retrieve ได้แม่นยำกว่า แต่มีต้นทุนในการเก็บ index และเปรียบเทียบสูงกว่า สำหรับ apps ส่วนใหญ่ 768–1024 dimensions เป็น sweet spot Models ที่รองรับ Matryoshka representations (text-embedding-3-*, บาง bge variants) ให้คุณเก็บ vector เต็มและ truncate ตอน query

เมื่อไรควร fine-tune

แทบไม่เคย และช้ากว่าที่คุณคิด เรียงตามความคุ้มค่า:

1. Chunking และ metadata ที่ดีกว่า — แก้ปัญหา “retrieval แย่” ส่วนใหญ่ได้
2. Hybrid search + reranker — แก้ที่เหลือส่วนใหญ่
3. Instruction-tuned prompts ตอน embedding — บาง models (bge, e5) รับ task prefix ที่ช่วย boost retrieval ของฟรีถ้า model คุณรองรับ
4. Fine-tune embedding model — เฉพาะเมื่อคุณมี domain query/document pairs หลายพันคู่และทำข้างต้นจนหมดแล้ว

Batching และ backoff

Embedding APIs throttle อย่างก้าวร้าวและล้มเหลวแบบชั่วคราว Batch ให้ถึง max ของ provider, retry ด้วย exponential backoff และ persist embeddings โดย key ด้วย content hash เพื่อไม่ให้ re-runs จ่ายบิลซ้ำ

import hashlib, time
from openai import OpenAI, RateLimitError

client = OpenAI()

def content_hash(text: str, model: str) -> str:
    return hashlib.sha256(f"{model}\x00{text}".encode()).hexdigest()

def embed_batch(
    texts: list[str],
    model: str = "text-embedding-3-small",
    batch_size: int = 128,
    max_retries: int = 6,
) -> list[list[float]]:
    out: list[list[float]] = []
    for i in range(0, len(texts), batch_size):
        batch = texts[i : i + batch_size]
        for attempt in range(max_retries):
            try:
                resp = client.embeddings.create(model=model, input=batch)
                out.extend(d.embedding for d in resp.data)
                break
            except RateLimitError:
                time.sleep(min(60, 2 ** attempt))
        else:
            raise RuntimeError(f"embedding failed after {max_retries} retries")
    return out

Vector DB: Trade-offs ที่ตรงไปตรงมา

ไม่มี vector database ที่ดีที่สุดในทุกกรณี เลือกตามความเหมาะสมในการ operate ไม่ใช่ตาม benchmark charts

• pgvector — ถ้าคุณ run Postgres อยู่แล้ว เริ่มที่นี่ ระบบเดียวที่ต้อง operate transactional กับ relational data ของคุณ filters เป็น SQL จริง index HNSW พร้อม production จุดอ่อน: multi-tenant ที่ scale ใหญ่ต้องระวัง ไม่มี BM25 native (ใช้ tsvector แยก)
• Qdrant — filter performance ดีเยี่ยม native hybrid search throughput ระดับ Rust ดี self-host ตรงไปตรงมา จุดอ่อนในอดีตคือ multi-tenant loads ที่ metadata เยอะ versions ล่าสุดแก้ส่วนใหญ่แล้ว
• Weaviate — schema-first มี modules ในตัวสำหรับ hybrid และ multi-modal opinionated มากกว่า มี features ให้เรียนมากกว่า
• Pinecone — serverless, zero ops, เริ่มต้นเร็วมาก ค่าใช้จ่ายอาจน่าตกใจที่ scale ใหญ่ self-host ไม่ได้
• Chroma — เยี่ยมสำหรับ prototypes และ local dev ผมจะไม่ใช้มันเป็น durable store ของระบบ production
• Elasticsearch / OpenSearch — ถูกประเมินค่าต่ำเกินไปสำหรับ hybrid คุณได้ BM25 ที่สมบูรณ์ filters, aggregations และ kNN ในระบบเดียว หนักในการ operate ถ้ายังไม่ได้ run อยู่

Stack default ของผม ถ้าไม่มีอะไรอื่นจำกัดทางเลือก: Postgres กับ pgvector สำหรับ vectors และ companion tsvector full-text index สำหรับ BM25 ระบบเดียว transactional updates SQL filters จริง คุณ outgrow มันได้ แต่คุณจะรู้ตอนนั้น

CREATE EXTENSION IF NOT EXISTS vector;

CREATE TABLE chunks (
  id           BIGSERIAL PRIMARY KEY,
  doc_id       TEXT    NOT NULL,
  text         TEXT    NOT NULL,
  heading_trail TEXT[] NOT NULL DEFAULT '{}',
  metadata     JSONB   NOT NULL DEFAULT '{}',
  content_hash TEXT    NOT NULL,
  embedding    VECTOR(1536),
  tsv          TSVECTOR
    GENERATED ALWAYS AS (to_tsvector('english', text)) STORED,
  created_at   TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE INDEX chunks_embedding_hnsw
  ON chunks USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 64);

CREATE INDEX chunks_tsv_gin ON chunks USING GIN (tsv);
CREATE INDEX chunks_doc_id ON chunks (doc_id);
CREATE INDEX chunks_metadata_gin ON chunks USING GIN (metadata);

Hybrid Search: ทำไม Pure Vector ถึงแพ้

Vector search จับ semantic similarity นั่นคือสิ่งที่คุณต้องการสำหรับ “ฉันยกเลิก subscription ยังไง” matching กับ doc ที่ชื่อว่า “Ending your plan.” มันเป็นสิ่งที่คุณ ไม่ต้องการ เมื่อผู้ใช้แปะ error code, SKU, ชื่อคน หรือ token อื่นใดที่การ match ตรงตัวคือ signal

Vector embeddings ทำให้ rare tokens เรียบเนียน เพราะนั่นคือสิ่งที่มันถูกเทรนให้ทำ ผู้ใช้ที่ค้นหา ERR_CERT_AUTHORITY_INVALID ต้องการเอกสารที่มี string นั้นเป๊ะ ๆ ไม่ใช่เอกสารเกี่ยวกับ “ปัญหา certificate validity โดยทั่วไป”

ทางแก้คือ hybrid search: run BM25 และ vector retrieval ขนานกัน แล้ว fuse ranked lists สองอัน Reciprocal Rank Fusion (RRF) คือ default ที่ถูกที่สุด-และ-ดีพอ — มันไม่ต้องการ score calibration ระหว่าง retrievers สองตัว ซึ่งเป็นเหตุผลหลักที่ naive score-weighted fusion ล้มเหลว

from collections import defaultdict

def rrf_fuse(
    rankings: list[list[str]],   # each inner list is chunk_ids in rank order
    k: int = 60,
) -> list[tuple[str, float]]:
    scores: dict[str, float] = defaultdict(float)
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking):
            scores[doc_id] += 1.0 / (k + rank + 1)
    return sorted(scores.items(), key=lambda x: x[1], reverse=True)


def hybrid_search(conn, query: str, query_vec: list[float], top_k: int = 50):
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT id::text FROM chunks
            WHERE tsv @@ plainto_tsquery('english', %s)
            ORDER BY ts_rank(tsv, plainto_tsquery('english', %s)) DESC
            LIMIT %s
            """,
            (query, query, top_k),
        )
        bm25_ids = [r[0] for r in cur.fetchall()]

        cur.execute(
            "SELECT id::text FROM chunks ORDER BY embedding <=> %s::vector LIMIT %s",
            (query_vec, top_k),
        )
        vec_ids = [r[0] for r in cur.fetchall()]

    return rrf_fuse([bm25_ids, vec_ids])

Over-retrieve ในขั้นนี้ ดึง 50–100 candidates ต่อ retriever แม้ว่าคุณวางแผนจะส่งแค่ 5 chunks ไป LLM reranker ในขั้นถัดไปขึ้นกับการมี candidate pool ที่หลากหลาย

Reranking: จ่ายเพื่อ Precision ตรงที่มันสำคัญ

Bi-encoder embeddings (สิ่งที่ vector DB ใช้) เร็วเพราะ query และ document ถูก embed อย่างอิสระ ความอิสระนั้นเองคือเหตุผลที่มันพลาดความละเอียดอ่อน — มันไม่สามารถ “อ่าน” query และ document ด้วยกัน

Cross-encoder reranker อ่าน query และแต่ละ candidate ร่วมกันและผลิต relevance score ช้ากว่า 10–100 เท่าต่อคู่เมื่อเทียบกับ bi-encoder comparison ซึ่งเป็นเหตุผลที่คุณ run มันเฉพาะกับ top-N fused candidates ไม่ใช่ทั้ง corpus

สองตัวเลือกที่สมจริง:

• Cohere rerank-v3 — hosted API คุณภาพแข็งแรงมาก คิดราคาต่อ 1K documents
• bge-reranker-large / bge-reranker-v2-m3 — open-weight, self-hostable บน GPU เดียว

from sentence_transformers import CrossEncoder

reranker = CrossEncoder("BAAI/bge-reranker-large", max_length=512)

def rerank(query: str, candidates: list[tuple[str, str]], top_n: int = 5):
    """candidates: list of (chunk_id, chunk_text)."""
    pairs = [(query, text) for _, text in candidates]
    scores = reranker.predict(pairs)
    ranked = sorted(zip(candidates, scores), key=lambda x: x[1], reverse=True)
    return [(cid, text, float(score)) for (cid, text), score in ranked[:top_n]]

เมื่อไร rerankers คุ้มค่า

• Queries สั้นและกำกวม Bi-encoders มีปัญหามากที่สุดตรงนี้ rerankers ช่วยมากที่สุด
• Context budget แน่น เมื่อคุณส่งได้แค่ 3–5 chunks ใน prompt precision บน 3–5 นั้นสำคัญกว่า broad recall
• คุณภาพคำตอบสัมพันธ์กับคุณภาพ citation Grounded answers ต้องการ retrieval precision สูง

เมื่อไรไม่คุ้ม

• Latency budget แน่น การเพิ่ม reranker เพิ่ม 100–500ms แม้บน GPU ถ้าคุณ end-to-end ต่ำกว่า 300ms ข้ามไปหรือย้ายไปหลัง cache
• Queries ยาวมาก เมื่อ queries เองเป็น paragraphs bi-encoders ปิดช่องว่างส่วนใหญ่ได้

วัด อย่าสมมติ Reranker ที่ใช้ 200ms และเลื่อน answer faithfulness จาก 0.72 เป็น 0.78 บน eval set ของคุณคือ trade ที่ดี Reranker เดียวกันที่เลื่อนจาก 0.88 เป็น 0.89 ไม่ใช่

Prompt Design สำหรับ Citations และ Grounding

RAG answer ที่ไม่มี citations แยกไม่ออกจาก hallucination ผู้ใช้ verify ไม่ได้ operators debug ไม่ได้ และ legal ปกป้องไม่ได้ สร้าง citations เข้าไปตั้งแต่วันแรก

Prompt pattern ที่ใช้ได้:

SYSTEM = """You are a helpful assistant that answers questions using only the
provided sources. Every factual claim MUST be followed by a citation in the
form [S#], where # is the source index. If the sources do not contain the
answer, say so explicitly. Do not use prior knowledge. Do not invent sources."""

def build_prompt(query: str, chunks: list[dict]) -> list[dict]:
    sources = "\n\n".join(
        f"[S{i+1}] ({c['doc_id']} — {' > '.join(c['heading_trail'])})\n{c['text']}"
        for i, c in enumerate(chunks)
    )
    user = f"Question: {query}\n\nSources:\n{sources}\n\nAnswer with citations:"
    return [{"role": "system", "content": SYSTEM},
            {"role": "user", "content": user}]

สี่ design choices ที่สำคัญ:

1. เลขลำดับ sources ใน prompt [S1], [S2] Model เขียนพวกนี้กลับเข้าไปในคำตอบ Post-processing แปลงพวกมันเป็น doc URLs
2. ใส่ heading trail และ doc_id กับแต่ละ chunk Model ใช้พวกมันเพื่อแยกแยะ chunks ที่คล้ายกันและเขียน citations ที่ดีกว่า
3. clause “say so” ชัดเจนสำหรับคำตอบที่ขาดหาย ถ้าไม่มี models เดา ถ้ามี พวกมันก็ยังเดาเป็นครั้งคราว แต่น้อยลงมาก
4. เรียง chunks ตาม rerank score ดีที่สุดก่อน Models สนใจจุดเริ่มของ context window มากกว่ากลาง — วาง evidence ที่แข็งแรงสุดไว้ทั้งสองปลายถ้ามี budget

หลังจาก LLM ตอบ parse [S#] references และแนบ source metadata จริง ปฏิเสธคำตอบที่มีข้อกล่าวอ้างโดยไม่มี citation แนบ — แค่นี้ก็ป้องกัน hallucination เงียบทั้งคลาสได้แล้ว

Evaluation: ส่วนที่ทุกคนข้าม

ถ้าไม่มี eval ทุก “improvement” ที่คุณ ship คือความรู้สึก RAG มีข้อดีที่คำตอบ grounded ใน text ที่ retrieve มา ซึ่งหมายความว่าคุณ check แบบ mechanical ได้เยอะ — มากกว่า open-ended generation

วัดอะไร

อย่างน้อย สามตัวเลข:

• Context recall — จาก facts ที่จำเป็นต้องตอบคำถาม fraction เท่าไรปรากฏใน chunks ที่ retrieve มา? วัด retrieval
• Answer relevance — คำตอบตอบคำถามจริงไหม? จับเรื่องนอกประเด็น
• Faithfulness (groundedness) — ทุก claim ในคำตอบมี retrieved context รองรับไหม? จับ hallucination

Ragas และ TruLens implement ทั้งสามตัวพร้อมใช้โดยใช้ LLM-as-judge คุณเขียน rubric ของตัวเองได้ — LLM-as-judge มักโอเคสำหรับสองตัวแรก แต่ faithfulness ได้ประโยชน์จาก verifier ที่เข้มงวดกว่า claim-by-claim

# Minimal faithfulness check: split the answer into claims and ask
# the judge whether each is supported by the retrieved context.
from openai import OpenAI
client = OpenAI()

CLAIM_PROMPT = """Split the answer into atomic factual claims.
Return a JSON array of strings. Ignore hedges and meta-statements."""

VERIFY_PROMPT = """Given CONTEXT and CLAIM, answer SUPPORTED or NOT_SUPPORTED.
A claim is SUPPORTED only if a careful reader could justify it from CONTEXT alone."""

def faithfulness(answer: str, context: str) -> float:
    claims = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": CLAIM_PROMPT},
                  {"role": "user",   "content": answer}],
        response_format={"type": "json_object"},
    )
    import json
    parsed = json.loads(claims.choices[0].message.content)
    claim_list: list[str] = parsed.get("claims", [])
    if not claim_list:
        return 1.0
    supported = 0
    for c in claim_list:
        v = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "system", "content": VERIFY_PROMPT},
                      {"role": "user",   "content": f"CONTEXT:\n{context}\n\nCLAIM: {c}"}],
        )
        if "SUPPORTED" in v.choices[0].message.content.upper().split("NOT")[0]:
            supported += 1
    return supported / len(claim_list)

สร้าง gold set ตั้งแต่เนิ่น ๆ

คู่ question/answer แบบ hand-labeled สัก 20 คู่พร้อมเอกสาร source ที่คาดหวัง มีค่ามากกว่า benchmark ที่ตีพิมพ์ใด ๆ ทุกการเปลี่ยน feature run ใหม่กับชุดนี้ ขยายไป 100–300 ตัวอย่างครอบคลุม edge cases พอที่จะทำให้ตัวเลขเสถียร

แบ่งชุด: ส่วน frozen ที่คุณแทบไม่ตรวจ (เพื่อจับ silent regressions) และส่วน working ที่คุณใช้ทุกวันสำหรับ tuning กับดักเดียวกับ ML โดยทั่วไป — ถ้าคุณดู holdout บ่อยเกินไป คุณก็ train บนมันไปแล้ว

Offline ไม่พอ

Log production queries พร้อม retrieval scores, rerank scores, chunks ที่คุณส่ง และคำตอบสุดท้าย Sample เปอร์เซ็นต์หนึ่งเพื่อให้คนรีวิวหรือ judge รีวิว Production distribution drift จาก gold set ของคุณในแบบที่คุณทำนายไม่ได้ การป้องกันเดียวคือเฝ้าดู traffic จริง

ข้อกังวลใน Production

Caching

สาม tier แต่ละอันมี invalidation semantics ต่างกัน:

• Embedding cache ตาม content hash การ re-index เอกสารที่เนื้อหาไม่เปลี่ยนใช้ API calls ศูนย์
• Retrieval cache ตาม (normalized query, filter set) TTL สั้น (นาที) กำจัด searches ซ้ำซ้อนสำหรับคำถามเดียวกันที่ทีมถามซ้ำ
• Answer cache ตาม (query hash, retrieved chunk IDs, prompt version) ยุ่งยาก — ต้อง invalidate เมื่อ source เปลี่ยน ใช้ retrieved chunk IDs เป็นส่วนของ key เพื่อให้ index ที่ update แล้ว bypass คำตอบเก่าโดยธรรมชาติ

Updates กับเอกสาร source

วิธีผิด: ลบเอกสาร re-embed จากศูนย์ insert คุณเสีย URL stability ทำลาย referential integrity กับ queries ที่ log ไว้ และจ่ายค่า embedding เต็ม

วิธีถูก: chunk version ใหม่ คำนวณ content hashes diff กับ chunks ที่มีอยู่ด้วย hash และ embed เฉพาะอันใหม่หรืออันที่เปลี่ยน ลบ chunks ที่ hashes ไม่ปรากฏแล้ว document updates ส่วนใหญ่แตะ chunks เพียงเศษเสี้ยว และบิล embedding ลดลงตามสัดส่วน

def reindex_document(conn, doc_id: str, new_chunks: list[Chunk]):
    new_hashes = {content_hash(c.text, "text-embedding-3-small"): c for c in new_chunks}

    with conn.cursor() as cur:
        cur.execute("SELECT content_hash FROM chunks WHERE doc_id = %s", (doc_id,))
        existing = {r[0] for r in cur.fetchall()}

        to_delete = existing - new_hashes.keys()
        to_insert = [c for h, c in new_hashes.items() if h not in existing]

        if to_delete:
            cur.execute(
                "DELETE FROM chunks WHERE doc_id = %s AND content_hash = ANY(%s)",
                (doc_id, list(to_delete)),
            )

        if to_insert:
            vectors = embed_batch([c.text for c in to_insert])
            for c, v in zip(to_insert, vectors):
                cur.execute(
                    """
                    INSERT INTO chunks (doc_id, text, heading_trail, content_hash, embedding)
                    VALUES (%s, %s, %s, %s, %s)
                    """,
                    (c.doc_id, c.text, c.heading_trail,
                     content_hash(c.text, "text-embedding-3-small"), v),
                )
    conn.commit()

Freshness

ถ้า corpus ของคุณเปลี่ยนระหว่างวัน retrieval อาจ serve chunks เก่าได้หลายนาทีหลัง write ตัวเลือกตามค่าใช้จ่าย:

• ยอมรับมัน knowledge bases ส่วนใหญ่ทน staleness 5 นาทีได้
• Dual-read: query index และ recent write-ahead log แล้ว merge
• Real-time indexing pipeline พร้อม per-document versioning แพง คุ้มเฉพาะ domain ที่ tolerance ต่ำจริง ๆ (ops, incident response)

PII และ redaction

ถ้า docs ของคุณมี PII หรือ secrets redact ก่อน embedding Embeddings รั่วข้อมูล vector ของ SSN ของใครคนหนึ่งไม่ใช่ format storage ที่ปลอดภัย Run redactor (presidio, regex pass สำหรับ patterns ชัดเจน หรือ classifier ขนาดเล็ก) ใน ingestion pipeline และเก็บ text ที่ redact แล้วพร้อม pointer ไปยังต้นฉบับสำหรับ users ที่ได้รับอนุญาต

Cost

ตัวฆ่า cost เงียบ ๆ เรียงตามความถี่ที่ผมเห็น:

1. Re-embed ทุกอย่างทุกครั้งที่ rebuild index Cache ตาม content hash
2. Embed ที่ dimensions large ในเมื่อ small ก็พอ ประเมิน อย่าสมมติว่าใหญ่กว่าชนะ
3. Rerank ทุกอย่าง รวมถึง top-1 hits ที่ชัดเจน ข้าม reranker เมื่อ top bi-encoder hit มี score สูงมาก
4. ส่ง 20 chunks ตอนที่ 5 ก็ตอบคำถามได้ Prompts ยาวกว่ามีต้นทุนสูงกว่าและมักได้คำตอบที่แย่กว่าเพราะ middle-of-context attention drop

Architecture สำหรับ Production RAG Service

flowchart TB
    subgraph Ingest[Ingestion]
      S1[Source connectors<br/>S3 / GDrive / CMS] --> S2[Parse + OCR<br/>worker pool]
      S2 --> S3[Chunker]
      S3 --> S4[Redactor]
      S4 --> S5[Embedding<br/>batch worker]
      S5 --> DB[(Postgres<br/>pgvector + tsvector)]
    end

    subgraph API[Query service]
      U[Client] --> GW[API gateway]
      GW --> QS[Query service]
      QS --> QR[Query rewriter<br/>optional]
      QR --> HR[Hybrid retriever<br/>BM25 + vector]
      HR --> DB
      HR --> RR[Reranker]
      RR --> LL[LLM client]
      LL --> CV[Citation validator]
      CV --> GW
    end

    subgraph Obs[Observability]
      QS -.logs.-> TEL[OTLP collector]
      TEL --> TR[Traces + metrics]
      CV -.samples.-> EV[Eval harness]
      EV --> DASH[Quality dashboard]
    end

การแยก concerns ที่สำคัญในทางปฏิบัติ:

• Query service เป็น stateless Scale แนวนอนได้ DB เป็น shared state ตัวเดียว
• Reranker เป็น service ของตัวเองหรือ sidecar ความต้องการ GPU ต่างกัน scaling curve ต่างกัน บางครั้ง vendor ต่างกัน
• Embedding worker แยกจาก query service คุณไม่ต้องการให้ ingestion spike แย่ง CPU จาก user queries
• Citation validator run บน critical path คำตอบที่ยังไม่ validate ไม่ ship ราคาถูก และจับ hallucination คลาสเฉพาะที่ eval harnesses บางครั้งพลาด

Node.js API endpoint

ทีมส่วนใหญ่ที่ผมร่วมงาน run retrieval stack ใน Python และ expose ให้ส่วนอื่นของ product ผ่าน typed API TypeScript wrapper บางหน้า Python service ทำให้ codebase ของ product เรียบง่ายและให้ frontend engineers อยู่ในภาษาที่พวกเขารู้

// apps/api/src/routes/rag.ts
import type { Request, Response } from "express";
import { z } from "zod";

const RagRequest = z.object({
  query: z.string().min(1).max(2000),
  filters: z.record(z.string(), z.unknown()).optional(),
  top_k: z.number().int().min(1).max(20).default(5),
});

interface RagClient {
  answer(input: z.infer<typeof RagRequest>): Promise<{
    answer: string;
    citations: { id: string; doc_id: string; url: string; snippet: string }[];
    retrieval_ms: number;
    rerank_ms: number;
    llm_ms: number;
  }>;
}

export function createRagHandler(client: RagClient) {
  return async (req: Request, res: Response) => {
    const parsed = RagRequest.safeParse(req.body);
    if (!parsed.success) {
      return res.status(400).json({ error: parsed.error.flatten() });
    }

    const started = Date.now();
    try {
      const result = await client.answer(parsed.data);
      res.setHeader("x-rag-retrieval-ms", result.retrieval_ms);
      res.setHeader("x-rag-rerank-ms", result.rerank_ms);
      res.setHeader("x-rag-llm-ms", result.llm_ms);
      res.setHeader("x-rag-total-ms", Date.now() - started);
      res.json({
        answer: result.answer,
        citations: result.citations,
      });
    } catch (err) {
      req.log.error({ err, query: parsed.data.query }, "rag.answer failed");
      res.status(502).json({ error: "retrieval_service_unavailable" });
    }
  };
}

Endpoint นี้น่าเบื่อโดยเจตนา code ฉลาด ๆ อยู่ใน Python service ฝั่ง TypeScript validate input, pass through, surface timing headers สำหรับ tracing และคืน response shape ที่เสถียรให้ clients

สรุปเช็คลิสต์

ก่อน ship ระบบ RAG ให้ users จริง เดินผ่าน list นี้ตั้งแต่ต้นจนจบ ทุก item map กับ failure ที่ผมเฝ้าดูทีม (รวมถึงทีมของผมเอง) เจอมา

• Ingestion จัดการ PDFs ด้วย OCR fallback ไม่ใช่แค่ text extraction
• ตารางและ code blocks รอด chunking โดยสมบูรณ์
• Near-duplicates ถูก deduplicate ก่อน indexing
• Chunking เป็น document-structure-aware เมื่อมีโครงสร้าง
• Embeddings ถูก cache ตาม content hash การ re-index เป็น incremental
• Hybrid retrieval (BM25 + vector รวมด้วย RRF) เป็น default ไม่ใช่ “future improvement”
• Reranker อยู่ในที่ พร้อม latency budget ที่วัดและให้เหตุผลแล้ว
• Prompts บังคับ citations validator ปฏิเสธ claims ที่ไม่มี citation
• Gold eval set อย่างน้อย 50 ตัวอย่าง CI run มันบนทุกการเปลี่ยน prompt หรือ retrieval
• Production queries ถูก log พร้อม retrieval metadata sample ถูก judge รายสัปดาห์
• PII ถูก redact ก่อน embedding ไม่ใช่หลัง
• การ update เอกสาร source ใช้ hash-diffing ไม่ใช่ full re-index
• Dashboards แสดง retrieval latency, rerank latency, LLM latency, faithfulness score และ cost ต่อ query

อ่านเพิ่มเติม

• Dense Passage Retrieval — Karpukhin et al., 2020 paper รากฐานสำหรับ bi-encoder retrieval
• ColBERT — Khattab and Zaharia, 2020 Late-interaction retrieval ทางสายกลางระหว่าง bi-encoder และ cross-encoder
• Lost in the Middle — Liu et al., 2023 ทำไม long contexts ทำได้ไม่ดีและจะทำอะไรกับมัน
• Docs ของโปรเจกต์ Ragas และ TruLens — entry points ที่ใช้งานได้จริงที่สุดสำหรับ RAG evaluation

ระบบ RAG ที่ทำงานใน production คือการฝึกทำสิ่งที่น่าเบื่อหลายอย่างให้ดี Parse อย่างระมัดระวัง chunk อย่างไตร่ตรอง retrieve ด้วยทั้ง keywords และ semantics rerank เมื่อมันคุ้มกับ latency cite ทุกอย่าง และวัดอย่างต่อเนื่อง ไม่มีขั้นตอนไหนหรูหรา และทุกขั้นตอนสำคัญกว่าโมเดลตัวไหนก็ตามที่อยู่บน leaderboard ในไตรมาสนี้