RAG sederhana: Claude + pgvector + chunks

LangChain bikin RAG kelihatan ribet padahal core-nya cuma 3 langkah: chunk → embed → retrieve → prompt. Snippet ini bangun RAG buat tanya jawab kebijakan BPJS Kesehatan, pakai pgvector + Claude. Tanpa abstraksi yang bikin pusing.

Kode

import os
import re
from dataclasses import dataclass

import anthropic
import psycopg
from openai import OpenAI

openai_client = OpenAI()
claude_client = anthropic.Anthropic()
DATABASE_URL = os.environ["DATABASE_URL"]


@dataclass
class Chunk:
    id: int
    sumber: str
    teks: str
    distance: float = 0.0


def chunk_text(text: str, max_chars: int = 800, overlap: int = 150) -> list[str]:
    """Pecah teks per paragraf, gabung sampai max_chars, dengan overlap."""
    paragraphs = re.split(r"\n\s*\n", text.strip())
    chunks: list[str] = []
    buffer = ""

    for para in paragraphs:
        if len(buffer) + len(para) <= max_chars:
            buffer = (buffer + "\n\n" + para).strip()
        else:
            if buffer:
                chunks.append(buffer)
            # Mulai chunk baru dengan tail overlap chunk sebelumnya
            tail = buffer[-overlap:] if buffer else ""
            buffer = (tail + "\n\n" + para).strip()

    if buffer:
        chunks.append(buffer)
    return chunks


def embed(text: str) -> list[float]:
    resp = openai_client.embeddings.create(
        model="text-embedding-3-small", input=text
    )
    return resp.data[0].embedding


def setup(conn: psycopg.Connection) -> None:
    with conn.cursor() as cur:
        cur.execute("CREATE EXTENSION IF NOT EXISTS vector")
        cur.execute(
            """
            CREATE TABLE IF NOT EXISTS dokumen_chunk (
                id BIGSERIAL PRIMARY KEY,
                sumber TEXT NOT NULL,
                teks TEXT NOT NULL,
                embedding vector(1536) NOT NULL
            )
            """
        )
        cur.execute(
            """
            CREATE INDEX IF NOT EXISTS dokumen_chunk_hnsw
            ON dokumen_chunk USING hnsw (embedding vector_cosine_ops)
            """
        )
    conn.commit()


def index_dokumen(conn: psycopg.Connection, sumber: str, isi: str) -> int:
    """Chunk + embed + simpan. Return jumlah chunk."""
    chunks = chunk_text(isi)
    with conn.cursor() as cur:
        for chunk in chunks:
            vec = embed(chunk)
            cur.execute(
                "INSERT INTO dokumen_chunk (sumber, teks, embedding) "
                "VALUES (%s, %s, %s)",
                (sumber, chunk, vec),
            )
    conn.commit()
    return len(chunks)


def retrieve(conn: psycopg.Connection, query: str, k: int = 4) -> list[Chunk]:
    vec = embed(query)
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT id, sumber, teks, embedding <=> %s::vector AS dist
            FROM dokumen_chunk
            ORDER BY embedding <=> %s::vector
            LIMIT %s
            """,
            (vec, vec, k),
        )
        rows = cur.fetchall()
    return [Chunk(id=r[0], sumber=r[1], teks=r[2], distance=float(r[3])) for r in rows]


def ask(conn: psycopg.Connection, pertanyaan: str) -> str:
    """RAG: retrieve context + tanya Claude dengan instruksi grounding."""
    chunks = retrieve(conn, pertanyaan, k=4)
    context = "\n\n---\n\n".join(
        f"[Sumber: {c.sumber}]\n{c.teks}" for c in chunks
    )

    response = claude_client.messages.create(
        model="claude-opus-4-5",
        max_tokens=800,
        system=(
            "Jawab hanya berdasarkan konteks yang diberikan. "
            "Kalau jawaban tidak ada di konteks, katakan 'Saya tidak menemukan info ini'. "
            "Selalu cantumkan [Sumber: ...] di akhir jawaban."
        ),
        messages=[
            {
                "role": "user",
                "content": f"Konteks:\n{context}\n\nPertanyaan: {pertanyaan}",
            }
        ],
    )

    text_block = next((b for b in response.content if b.type == "text"), None)
    return text_block.text if text_block else ""

Pemakaian

# Indexing dokumen BPJS
dokumen_bpjs = """
Iuran BPJS Kesehatan Kelas I sebesar Rp150.000 per orang per bulan.
Iuran Kelas II sebesar Rp100.000 per orang per bulan.
Iuran Kelas III sebesar Rp42.000 per orang per bulan, dengan subsidi pemerintah Rp7.000.

Peserta yang terlambat membayar iuran dikenakan denda pelayanan...
(lanjut dokumen panjang)
"""

with psycopg.connect(DATABASE_URL) as conn:
    setup(conn)
    n = index_dokumen(conn, sumber="Peraturan BPJS 2026", isi=dokumen_bpjs)
    print(f"Index {n} chunks")

# Tanya jawab
with psycopg.connect(DATABASE_URL) as conn:
    print(ask(conn, "Iuran BPJS kelas III berapa?"))
    # → "Iuran Kelas III sebesar Rp42.000 per orang per bulan...
    #     [Sumber: Peraturan BPJS 2026]"

    print(ask(conn, "Apa hukum jadi peserta BPJS?"))
    # → "Saya tidak menemukan info ini di konteks yang tersedia."

Kapan dipakai

• Internal knowledge base — tanya jawab kebijakan HR, SOP, dokumentasi teknis.
• Chatbot CS dengan FAQ panjang yang sering update.
• Legal/regulatory Q&A — penting cite source biar accountable.
• Documentation search engine yang lebih pintar dari keyword.

Catatan

• Chunk overlap penting — kalau jawaban ada di boundary antara 2 chunk, tanpa overlap bisa miss. 150-200 char overlap reasonable.
• Reranking — top-K embedding retrieval kadang miss. Tambahkan reranker (Cohere rerank, BGE-reranker) untuk improve accuracy.
• Grounding instruction — system prompt harus eksplisit “jangan jawab di luar konteks”, kalau tidak Claude bakal halusinasi.
• Hybrid search — combine vector similarity dengan keyword BM25 (Postgres ts_rank) untuk tangkap exact term yang vector miss.
• Chunk metadata — simpan halaman, section, timestamp publikasi. Berguna buat filter “cari di dokumen 2026 aja”.

Hindari chunk yang terlalu kecil (100 token). Vector embedding text pendek noisy dan retrieval-nya jadi ngaco. Sweet spot 500-1000 token.