Fase 7 · Capstone: RAG Chatbot

Chroma — Introduction

ChromaDB: menyimpan embedding dan mencari yang paling mirip — komponen retrieval dari capstone-mu.

Sumber asli docs.trychroma.com Resmi Rangkuman ~6 menit baca

Intisari

Bisa jalan in-process tanpa server terpisah — sempurna untuk belajar dan prototipe.
Chroma bisa membuat embedding sendiri, tapi untuk kendali penuh sebaiknya kamu sediakan sendiri.
Tiga operasi inti: add, query, delete. Itu saja.
Metadata memungkinkan filter — penting untuk membatasi pencarian per pengguna atau per sumber.
Nanti di AWS ini digantikan Bedrock Knowledge Bases, OpenSearch, atau pgvector.

Apa yang dilakukan vector store

Teks  ──[model embedding]──>  Vektor 768 dimensi
                                     │
                          disimpan di vector store
                                     │
Pertanyaan ──[embedding]──> Vektor ──┘
                                     │
                          cari vektor paling mirip
                                     │
                          kembalikan teks aslinya

Itulah seluruh isi "R" dalam RAG. Vector store adalah indeks yang membuat pencarian kemiripan tetap cepat meski ada jutaan vektor.

Pemasangan

uv add chromadb

import chromadb

client = chromadb.Client()                              # di memori, hilang saat program selesai
client = chromadb.PersistentClient(path="./chroma")     # tersimpan ke disk
client = chromadb.HttpClient(host="chroma", port=8000)  # server terpisah (Docker)

Mulai dengan PersistentClient. Tidak ada server yang perlu dijalankan, datanya bertahan antar percobaan, dan kamu bisa fokus ke logika RAG-nya. Pindah ke HttpClient saat sudah masuk Docker Compose.

Collection

collection = client.get_or_create_collection(
    name="dokumen",
    metadata={"hnsw:space": "cosine"},     # cosine similarity — sesuai untuk teks
)

client.list_collections()
client.delete_collection("dokumen")

Setel hnsw:space saat membuat collection. Default-nya jarak L2 (Euclidean), sedangkan untuk embedding teks yang benar adalah cosine. Nilai ini tidak bisa diubah setelah collection dibuat — kamu harus membuat ulang.

Menambahkan dokumen

Cara cepat — Chroma yang membuat embedding

collection.add(
    ids=["1", "2"],
    documents=["Python adalah bahasa pemrograman", "RAG menggabungkan retrieval dan generation"],
    metadatas=[{"sumber": "docs.md"}, {"sumber": "rag.md"}],
)

Chroma mengunduh model embedding kecil dan menjalankannya secara lokal. Praktis untuk mencoba.

Cara yang kamu pakai di capstone — embedding disediakan sendiri

collection.add(
    ids=[c.id for c in chunks],
    embeddings=[c.embedding for c in chunks],      # list[list[float]]
    documents=[c.teks for c in chunks],
    metadatas=[{"sumber": c.sumber, "halaman": c.halaman} for c in chunks],
)

Kenapa menyediakan embedding sendiri: kamu memilih model embedding-nya, tahu persis berapa dimensinya, bisa membuatnya secara batch dan paralel, dan bisa mengganti model tanpa bergantung pada apa yang kebetulan didukung Chroma. Untuk deployment AWS nanti, ini juga yang memungkinkan kamu memakai model embedding dari Bedrock.

Mencari

hasil = collection.query(
    query_embeddings=[embedding_pertanyaan],
    n_results=5,
)

hasil["ids"]         # [["3", "7", "1", ...]]
hasil["documents"]   # [["teks chunk 3", ...]]
hasil["metadatas"]   # [[{"sumber": "..."}, ...]]
hasil["distances"]   # [[0.12, 0.31, ...]]  ← JARAK, bukan skor kemiripan

distances adalah jarak, bukan kemiripan. Untuk ruang cosine, kemiripan = 1 - jarak. Semakin kecil jaraknya, semakin mirip. Salah menafsirkan ini akan membuat filter ambang batasmu terbalik dan mengembalikan chunk yang paling tidak relevan.

for id_, teks, meta, jarak in zip(
    hasil["ids"][0], hasil["documents"][0],
    hasil["metadatas"][0], hasil["distances"][0],
):
    kemiripan = 1 - jarak
    if kemiripan < 0.6:
        continue
    print(f"{kemiripan:.3f}  {meta['sumber']}  {teks[:80]}")

Filter metadata

collection.query(
    query_embeddings=[emb],
    n_results=5,
    where={"sumber": "manual.pdf"},
    where_document={"$contains": "instalasi"},      # filter isi teks
)

# Operator
where={"tahun": {"$gte": 2024}}
where={"kategori": {"$in": ["teknis", "panduan"]}}
where={"$and": [{"sumber": "a.md"}, {"halaman": {"$lt": 10}}]}

Filter metadata sering kali yang membuat RAG berguna di dunia nyata. Aplikasi multi-tenant harus memfilter berdasarkan user_id atau org_id — tanpa itu, pengguna A bisa mendapat jawaban yang bersumber dari dokumen pengguna B. Rancang skema metadata-mu sejak awal.

Operasi lain

collection.count()
collection.get(ids=["1", "2"])
collection.get(where={"sumber": "lama.md"})

collection.update(ids=["1"], documents=["teks baru"], embeddings=[emb_baru])
collection.upsert(ids=["1"], documents=["..."], embeddings=[...])   # tambah atau perbarui

collection.delete(ids=["1"])
collection.delete(where={"sumber": "usang.md"})

Pola ingest

import asyncio
from pathlib import Path

async def ingest(folder: Path, collection) -> int:
    chunks = []
    for path in folder.rglob("*.md"):
        teks = path.read_text(encoding="utf-8")
        for i, potongan in enumerate(potong(teks, ukuran=1000, tumpang=200)):
            chunks.append({
                "id": f"{path.stem}-{i}",
                "teks": potongan,
                "sumber": str(path.relative_to(folder)),
                "indeks": i,
            })

    # Embed secara batch — jauh lebih cepat daripada satu per satu
    for batch in batched(chunks, 100):
        emb = await buat_embedding([c["teks"] for c in batch])
        collection.upsert(
            ids=[c["id"] for c in batch],
            embeddings=emb,
            documents=[c["teks"] for c in batch],
            metadatas=[{"sumber": c["sumber"], "indeks": c["indeks"]} for c in batch],
        )
    return len(chunks)

ID yang deterministik ({nama_file}-{indeks}) plus upsert membuat ingest bisa dijalankan ulang tanpa menduplikasi data. Kalau ID-nya acak, menjalankan ingest dua kali menggandakan seluruh isi collection.

Menjalankan sebagai server

services:
  chroma:
    image: chromadb/chroma:latest
    ports:
      - "8000:8000"
    volumes:
      - chroma-data:/chroma/chroma
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/api/v2/heartbeat"]
      interval: 10s

volumes:
  chroma-data:

client = chromadb.HttpClient(host="chroma", port=8000)

Batasan yang perlu kamu sadari

Aspek	Chroma	Alternatif production
Skala	Ratusan ribu vektor	OpenSearch, pgvector, Pinecone
Hybrid search (vektor + kata kunci)	Terbatas	OpenSearch
High availability	Tidak ada	Layanan terkelola
Kemudahan belajar	Terbaik	Perlu setup

Chroma dipilih untuk roadmap ini justru karena batasannya tidak menghalangi belajar. Tidak ada server, tidak ada konfigurasi cluster, tidak ada biaya. Setelah kamu paham konsep-nya — chunking, embedding, similarity search, filter metadata — memindahkan ke OpenSearch atau Bedrock Knowledge Bases hanya soal mengganti implementasi di balik interface yang sama.

Bungkus di balik interface

from typing import Protocol

class VectorStore(Protocol):
    async def simpan(self, chunks: list[Chunk]) -> None: ...
    async def cari(self, embedding: list[float], k: int,
                   filter: dict | None = None) -> list[Chunk]: ...

Dengan Protocol dari Fase 2, kamu bisa mengganti ChromaStore dengan OpenSearchStore tanpa menyentuh logika RAG-mu — dan memakai StorePalsu di unit test.

Rangkuman ini sengaja dipangkas ke bagian yang dipakai di roadmap. Buka sumber aslinya saat kamu butuh detail lengkap atau referensi parameter.