Bài 14: ChromaDB — local vector DB cho prototype

Sau bài này bạn sẽ:

• ✅ Hiểu ChromaDB hoạt động theo cách nào và phù hợp với bài toán nào
• ✅ Biết chọn đúng chế độ chạy (ephemeral / persistent / server) cho từng tình huống
• ✅ Thực hiện được CRUD trên collection: add, query, update, delete với metadata
• ✅ Dùng metadata filter để thu hẹp kết quả search
• ✅ Cài embedding function tùy chỉnh (OpenAI, HuggingFace, hoặc custom)
• ✅ Chạy được ví dụ RAG mini end-to-end

ChromaDB (github.com/chroma-core/chroma) là vector database open source, license Apache 2.0. Core viết bằng Python + Rust. Backend lưu trữ mặc định dùng DuckDB + Parquet (persistent mode) hoặc in-memory (ephemeral mode).

Điểm khác biệt so với các vector DB khác

• Zero-config để bắt đầu: pip install chromadb rồi dùng luôn, không cần server riêng.
• Embedding function tích hợp: Gọi add(documents=[...]) — Chroma tự embed text bằng model mặc định (all-MiniLM-L6-v2). Bạn không cần gọi model embedding thủ công.
• API thuần Python: Không cần học query language riêng.

Khi nào nên dùng ChromaDB

• Dataset dưới vài triệu vectors, chạy trên một máy (single machine).
• Prototype RAG, internal tool, demo cá nhân.
• Không muốn quản lý infrastructure cho vector DB.

Khi nào không nên dùng

• Cần scale horizontal qua nhiều node — ChromaDB không hỗ trợ distributed mode trong phiên bản hiện tại.
• Cần SLA, replication, managed backup — xem bài 15 (Pinecone) hoặc bài 16 (Qdrant, Weaviate).
• Multi-process write đồng thời vào PersistentClient — cần chuyển sang server mode.

Phiên bản hiện tại là 0.5.x (tháng 5/2026).

# Cài đầy đủ (server + client + embedding functions)
pip install chromadb

# Nếu chỉ cần connect HTTP server (môi trường production, không cần local storage)
pip install chromadb-client

Gói chromadb kéo theo sentence-transformers (cho default embedding function). Lần đầu chạy, model all-MiniLM-L6-v2 (~80 MB) sẽ được tải về ~/.cache/chroma/onnx_models/.

Kiểm tra version sau cài:

import chromadb
print(chromadb.__version__)  # 0.5.x

Ephemeral (in-memory)

Data tồn tại trong RAM, mất khi process exit. Phù hợp test, unit test, notebook demo.

import chromadb

client = chromadb.Client()
# Tương đương chromadb.EphemeralClient()

Persistent (local file)

Data lưu xuống disk tại đường dẫn chỉ định. Phù hợp prototype chạy local.

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")
# Thư mục ./chroma_db được tạo tự động nếu chưa có
# Lần chạy sau data vẫn còn

Lưu ý: PersistentClient dùng file lock — chỉ một process được write cùng lúc. Nếu cần nhiều process đồng thời, chuyển sang server mode.

Server (HTTP)

Chạy Chroma như một service riêng, client connect qua REST. Phù hợp khi nhiều process hoặc nhiều service cần dùng chung một vector DB.

# Khởi động server (terminal 1)
chroma run --path ./chroma_db --host 0.0.0.0 --port 8000

# Connect từ application (terminal 2 hoặc service khác)
import chromadb

client = chromadb.HttpClient(host="localhost", port=8000)
# hoặc host="192.168.x.x" nếu server ở máy khác

API HttpClient hoàn toàn giống EphemeralClient và PersistentClient — cùng method, cùng return type. Đổi client không cần sửa code business logic.

Tóm tắt lựa chọn

Collection trong ChromaDB tương đương table trong SQL. Mỗi collection có một index HNSW riêng, một embedding function riêng, và một distance metric riêng.

Tạo collection

import chromadb

client = chromadb.PersistentClient(path="./chroma_db")

# Tạo mới — báo lỗi nếu đã tồn tại
collection = client.create_collection(name="docs")

# Lấy nếu có, tạo nếu chưa — an toàn hơn trong production
collection = client.get_or_create_collection(name="docs")

Tham số quan trọng khi tạo collection

from chromadb.utils import embedding_functions

# 1. Chỉ định distance metric (mặc định là "l2")
collection = client.get_or_create_collection(
    name="docs",
    metadata={"hnsw:space": "cosine"}  # hoặc "l2", "ip"
)

# 2. Chỉ định embedding function
ef = embedding_functions.SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-small-en-v1.5"
)
collection = client.get_or_create_collection(
    name="docs_bge",
    embedding_function=ef,
    metadata={"hnsw:space": "cosine"}
)

Distance metric:

• l2 — Euclidean distance (default). Khoảng cách nhỏ hơn = gần hơn.
• cosine — cosine distance (= 1 - cosine similarity). Phổ biến nhất cho text search.
• ip — inner product (dot product âm). Dùng khi vector đã normalize.

Lưu ý: Nếu collection đã tạo với metric X rồi, get_or_create_collection gọi lại với metadata khác không đổi được metric. Phải xóa collection và tạo lại.

Các method quản lý collection

# Liệt kê tất cả collection
client.list_collections()

# Lấy collection đã có (lỗi nếu không tồn tại)
col = client.get_collection(name="docs")

# Xóa collection
client.delete_collection(name="docs")

# Xem số lượng document trong collection
collection.count()

Mỗi document trong Chroma có 4 thành phần: id (bắt buộc, unique), document (text gốc), embedding (vector), metadata (dict tùy ý).

Cách 1: Documents only (Chroma tự embed)

collection.add(
    ids=["doc1", "doc2", "doc3"],
    documents=[
        "FastAPI supports async endpoints natively.",
        "Use Pydantic models to validate request bodies.",
        "Dependency injection in FastAPI uses Depends().",
    ]
)
# Chroma gọi embedding function tự động
# Mặc định dùng all-MiniLM-L6-v2 (384-dim)

Cách 2: Custom embeddings (đã có vector)

import numpy as np

# Giả sử đã có vector từ model khác
my_embeddings = [
    np.random.rand(384).tolist(),  # vector cho doc1
    np.random.rand(384).tolist(),  # vector cho doc2
]

collection.add(
    ids=["doc1", "doc2"],
    embeddings=my_embeddings,
    documents=[
        "FastAPI supports async endpoints natively.",
        "Use Pydantic models to validate request bodies.",
    ]
)

Cách 3: Với metadata

collection.add(
    ids=["doc1", "doc2", "doc3"],
    documents=[
        "FastAPI supports async endpoints natively.",
        "Use Pydantic models to validate request bodies.",
        "Upload files with UploadFile and File parameters.",
    ],
    metadatas=[
        {"source": "fastapi-docs.pdf", "page": 1, "chapter": "async"},
        {"source": "fastapi-docs.pdf", "page": 5, "chapter": "pydantic"},
        {"source": "fastapi-docs.pdf", "page": 12, "chapter": "files"},
    ]
)

Lưu ý về metadata: Chỉ hỗ trợ kiểu str, int, float, bool. Không dùng nested dict hay list trong metadata.

Batch add — tốt hơn nhiều vòng lặp

docs = [...]       # list 1000 đoạn text
ids = [f"id_{i}" for i in range(len(docs))]
metas = [{"source": "corpus.txt", "idx": i} for i in range(len(docs))]

# Add một lần — Chroma xây HNSW index theo batch, nhanh hơn add từng cái
collection.add(ids=ids, documents=docs, metadatas=metas)

Query bằng text (Chroma tự embed query)

results = collection.query(
    query_texts=["How to handle file upload?"],
    n_results=3
)

Query bằng vector

query_vec = my_embed_function("How to handle file upload?")

results = collection.query(
    query_embeddings=[query_vec],  # list-of-list (batch support)
    n_results=3
)

Cấu trúc kết quả trả về

collection.query() trả về dict, mỗi key là list-of-list (outer list = batch size, inner list = n_results):

# results có dạng:
{
    "ids": [["doc3", "doc1", "doc2"]],          # outer list vì batch=1
    "distances": [[0.12, 0.34, 0.56]],           # khoảng cách tương ứng
    "documents": [["Upload files...", "FastAPI...", "Use Pydantic..."]],
    "metadatas": [[
        {"source": "fastapi-docs.pdf", "page": 12, "chapter": "files"},
        {"source": "fastapi-docs.pdf", "page": 1,  "chapter": "async"},
        {"source": "fastapi-docs.pdf", "page": 5,  "chapter": "pydantic"},
    ]],
    "embeddings": None  # None nếu không request (mặc định)
}

# Lấy top-1 document
top_doc = results["documents"][0][0]
top_distance = results["distances"][0][0]

Include thêm trường trong kết quả

results = collection.query(
    query_texts=["file upload"],
    n_results=3,
    include=["documents", "metadatas", "distances", "embeddings"]
    # Mặc định include ["documents", "metadatas", "distances"]
)

Batch query

results = collection.query(
    query_texts=["file upload", "async endpoint", "authentication"],
    n_results=2
)
# results["ids"] là list-of-list, shape: [3, 2]
# results["ids"][0] = top-2 cho query đầu tiên
# results["ids"][1] = top-2 cho query thứ hai

Metadata filter giới hạn không gian search trước khi tính ANN (approximate nearest neighbor). Quan trọng khi collection chứa nhiều loại document từ nhiều nguồn.

Equality filter

# Chỉ search trong document từ "fastapi-docs.pdf"
results = collection.query(
    query_texts=["file upload"],
    n_results=3,
    where={"source": "fastapi-docs.pdf"}
)

Comparison operators

# Tìm trong trang > 10
results = collection.query(
    query_texts=["middleware"],
    n_results=5,
    where={"page": {"$gt": 10}}
)

# Các operator: $eq, $ne, $gt, $gte, $lt, $lte
# Ví dụ $in (thuộc list)
results = collection.query(
    query_texts=["authentication"],
    n_results=5,
    where={"chapter": {"$in": ["security", "oauth", "jwt"]}}
)

# $nin (không thuộc list)
where={"chapter": {"$nin": ["intro", "installation"]}}

Composite filter ($and, $or)

# source = "fastapi-docs.pdf" AND page < 20
results = collection.query(
    query_texts=["request body"],
    n_results=3,
    where={
        "$and": [
            {"source": "fastapi-docs.pdf"},
            {"page": {"$lt": 20}}
        ]
    }
)

# source = "fastapi-docs.pdf" OR source = "starlette-docs.pdf"
where={
    "$or": [
        {"source": "fastapi-docs.pdf"},
        {"source": "starlette-docs.pdf"}
    ]
}

Full-text filter trong document

# Kết hợp: vector search + filter document chứa "fastapi"
results = collection.query(
    query_texts=["dependency injection"],
    n_results=5,
    where_document={"$contains": "fastapi"}
    # where_document không dùng index — scan document text
)

where và where_document có thể dùng đồng thời. where lọc theo metadata (nhanh, dùng index), where_document lọc theo nội dung text (chậm hơn, scan).

Update — chỉ cập nhật document đã tồn tại

# Cập nhật document và metadata của "doc1"
collection.update(
    ids=["doc1"],
    documents=["FastAPI supports async/await natively using asyncio."],
    metadatas=[{"source": "fastapi-docs.pdf", "page": 1, "chapter": "async", "updated": True}]
)
# Nếu id không tồn tại → lỗi. Dùng upsert thay thế nếu không chắc.

Upsert — insert hoặc update

# Insert nếu id chưa có, update nếu đã có
collection.upsert(
    ids=["doc1", "doc_new"],
    documents=[
        "FastAPI supports async/await natively.",
        "Background tasks in FastAPI use BackgroundTasks class.",
    ],
    metadatas=[
        {"source": "fastapi-docs.pdf", "page": 1, "chapter": "async"},
        {"source": "fastapi-docs.pdf", "page": 25, "chapter": "background"},
    ]
)

Delete

# Xóa theo ids
collection.delete(ids=["doc1", "doc2"])

# Xóa theo where filter
collection.delete(
    where={"source": "old-docs.pdf"}
)

# Xóa tất cả document trong collection (giữ collection)
# Chú ý: không có method clear() — dùng where={} với get() rồi delete theo ids
all_ids = collection.get()["ids"]
if all_ids:
    collection.delete(ids=all_ids)

Default embedding function

Khi không chỉ định, Chroma dùng DefaultEmbeddingFunction — chạy sentence-transformers/all-MiniLM-L6-v2 locally qua ONNX runtime (384 dimensions, không cần API key, miễn phí).

OpenAI embedding

from chromadb.utils.embedding_functions import OpenAIEmbeddingFunction

ef_openai = OpenAIEmbeddingFunction(
    api_key="sk-...",
    model_name="text-embedding-3-small"  # 1536-dim, hoặc text-embedding-3-large
)

collection = client.get_or_create_collection(
    name="docs_openai",
    embedding_function=ef_openai,
    metadata={"hnsw:space": "cosine"}
)

HuggingFace embedding (qua API)

from chromadb.utils.embedding_functions import HuggingFaceEmbeddingFunction

ef_hf = HuggingFaceEmbeddingFunction(
    api_key="hf_...",
    model_name="BAAI/bge-large-en-v1.5"  # 1024-dim
)

SentenceTransformer local

from chromadb.utils.embedding_functions import SentenceTransformerEmbeddingFunction

ef_st = SentenceTransformerEmbeddingFunction(
    model_name="BAAI/bge-small-en-v1.5"  # 384-dim, nhanh hơn bge-large
)

Custom embedding function

from chromadb import Documents, EmbeddingFunction, Embeddings

class MyEmbeddingFunction(EmbeddingFunction):
    def __init__(self, model):
        self.model = model

    def __call__(self, input: Documents) -> Embeddings:
        # input là list[str]
        # trả về list[list[float]]
        return self.model.encode(input).tolist()

ef_custom = MyEmbeddingFunction(my_model)
collection = client.get_or_create_collection(
    name="docs_custom",
    embedding_function=ef_custom
)

Quan trọng: Embedding function phải nhất quán — collection đã tạo với model A thì query cũng phải dùng model A. Nếu trộn model, distance không có nghĩa so sánh giữa các vector.

Ví dụ dưới đây dựng pipeline RAG đơn giản trên 10 đoạn text về FastAPI. Chạy được offline, không cần API key.

"""
rag_mini.py — RAG mini với ChromaDB 0.5.x
Không cần API key — dùng default embedding (all-MiniLM-L6-v2)
"""
import chromadb

# ── 1. Data: 10 đoạn text về FastAPI ──────────────────────────────────────────
CHUNKS = [
    {
        "id": "c0",
        "text": "FastAPI is a modern web framework for building APIs with Python 3.8+ based on standard Python type hints.",
        "meta": {"source": "fastapi-docs", "page": 1, "topic": "overview"}
    },
    {
        "id": "c1",
        "text": "FastAPI supports async and await keywords natively, enabling non-blocking I/O for high concurrency.",
        "meta": {"source": "fastapi-docs", "page": 3, "topic": "async"}
    },
    {
        "id": "c2",
        "text": "Pydantic models are used in FastAPI to declare request bodies. FastAPI validates incoming JSON automatically.",
        "meta": {"source": "fastapi-docs", "page": 5, "topic": "pydantic"}
    },
    {
        "id": "c3",
        "text": "To upload files in FastAPI, use UploadFile type parameter. Access file content via await file.read().",
        "meta": {"source": "fastapi-docs", "page": 12, "topic": "files"}
    },
    {
        "id": "c4",
        "text": "Background tasks let you run code after returning a response. Use BackgroundTasks and add_task().",
        "meta": {"source": "fastapi-docs", "page": 18, "topic": "background"}
    },
    {
        "id": "c5",
        "text": "Dependency injection in FastAPI uses Depends(). Dependencies can be functions or classes.",
        "meta": {"source": "fastapi-docs", "page": 22, "topic": "dependency"}
    },
    {
        "id": "c6",
        "text": "FastAPI generates OpenAPI documentation automatically at /docs (Swagger UI) and /redoc.",
        "meta": {"source": "fastapi-docs", "page": 7, "topic": "docs"}
    },
    {
        "id": "c7",
        "text": "OAuth2 with Password flow in FastAPI uses OAuth2PasswordBearer and OAuth2PasswordRequestForm.",
        "meta": {"source": "fastapi-docs", "page": 35, "topic": "security"}
    },
    {
        "id": "c8",
        "text": "StreamingResponse allows returning large files or LLM tokens incrementally to the client.",
        "meta": {"source": "fastapi-docs", "page": 40, "topic": "streaming"}
    },
    {
        "id": "c9",
        "text": "Middleware in FastAPI wraps every request and response. Use app.add_middleware() to register.",
        "meta": {"source": "fastapi-docs", "page": 28, "topic": "middleware"}
    },
]

# ── 2. Khởi tạo client và collection ──────────────────────────────────────────
client = chromadb.PersistentClient(path="./rag_demo_db")

collection = client.get_or_create_collection(
    name="fastapi_docs",
    metadata={"hnsw:space": "cosine"}
    # embedding_function không chỉ định → dùng all-MiniLM-L6-v2
)

# Chỉ add nếu collection chưa có data (tránh duplicate khi chạy lại)
if collection.count() == 0:
    collection.add(
        ids=[c["id"] for c in CHUNKS],
        documents=[c["text"] for c in CHUNKS],
        metadatas=[c["meta"] for c in CHUNKS]
    )
    print(f"Đã add {collection.count()} chunks vào collection.")
else:
    print(f"Collection đã có {collection.count()} chunks, bỏ qua add.")

# ── 3. Query ───────────────────────────────────────────────────────────────────
query = "How to handle file upload?"
results = collection.query(
    query_texts=[query],
    n_results=3,
    include=["documents", "metadatas", "distances"]
)

print(f"\nQuery: {query}")
print("─" * 60)

docs = results["documents"][0]
metas = results["metadatas"][0]
dists = results["distances"][0]

for rank, (doc, meta, dist) in enumerate(zip(docs, metas, dists), start=1):
    print(f"#{rank}  distance={dist:.4f}  page={meta['page']}  topic={meta['topic']}")
    print(f"     {doc}")
    print()

Output mẫu:

Đã add 10 chunks vào collection.

Query: How to handle file upload?
────────────────────────────────────────────────────────────
#1  distance=0.1823  page=12  topic=files
     To upload files in FastAPI, use UploadFile type parameter. Access file content via await file.read().

#2  distance=0.5241  page=3   topic=async
     FastAPI supports async and await keywords natively, enabling non-blocking I/O for high concurrency.

#3  distance=0.5879  page=40  topic=streaming
     StreamingResponse allows returning large files or LLM tokens incrementally to the client.

Chunk #1 (topic=files) có distance gần nhất (0.18), đúng nội dung. Chunk #2 và #3 ở distance xa hơn (~0.52–0.59) — trong RAG thực tế bạn nên dùng threshold để loại kết quả quá xa:

DISTANCE_THRESHOLD = 0.5  # cosine distance, điều chỉnh tùy model

relevant = [
    (doc, meta, dist)
    for doc, meta, dist in zip(docs, metas, dists)
    if dist < DISTANCE_THRESHOLD
]

HNSW index

ChromaDB dùng HNSW (Hierarchical Navigable Small World) cho approximate nearest neighbor search. Query time O(log N). Tham số HNSW có thể điều chỉnh qua metadata collection:

collection = client.get_or_create_collection(
    name="docs",
    metadata={
        "hnsw:space": "cosine",
        "hnsw:construction_ef": 200,   # độ chính xác khi build index (default 100)
        "hnsw:search_ef": 100,         # độ chính xác khi query (default 10)
        "hnsw:M": 16,                  # số kết nối mỗi node (default 16)
    }
)
# construction_ef cao → index chính xác hơn nhưng build chậm hơn
# search_ef cao → query chính xác hơn nhưng query chậm hơn

Capacity theo môi trường

• In-memory (EphemeralClient): ~vài trăm nghìn vectors trên laptop 16 GB RAM với 384-dim vector. Vector 1536-dim (OpenAI) giảm xuống còn ~100k vectors trong cùng RAM.
• PersistentClient: Giới hạn bởi disk và RAM cho index. Thực tế tới ~5–10 triệu vectors vẫn dùng được nếu máy đủ RAM.
• Server mode: Tương tự PersistentClient về capacity, nhưng hỗ trợ multi-process.

Insert vs query speed

Insert chậm hơn query vì phải cập nhật HNSW index online. Nếu cần insert số lượng lớn, nên dùng batch add thay vì loop add từng record.

Concurrent access

• EphemeralClient: single-threaded safe, multi-threaded cần GIL Python.
• PersistentClient: chỉ an toàn trong single process (SQLite file lock).
• HttpClient: server handle concurrent request, an toàn cho multi-process.

1. Trộn embedding function giữa các collection

Nếu collection A dùng all-MiniLM-L6-v2 (384-dim) và collection B dùng OpenAI text-embedding-3-small (1536-dim), kết quả distance của 2 collection không so sánh được. Mỗi collection phải nhất quán từ đầu đến cuối.

# Sai: query bằng embedding function khác với lúc add
col = client.get_collection("docs")          # đã add bằng ef_A
results = col.query(query_embeddings=[ef_B("some text")])  # BAD
# vector dim khác → Chroma báo lỗi hoặc trả kết quả vô nghĩa

# Đúng: dùng cùng embedding function
results = col.query(query_embeddings=[ef_A("some text")])
# hoặc dùng query_texts để Chroma tự embed bằng ef đã đăng ký
results = col.query(query_texts=["some text"])

2. Trùng id khi add

Chroma yêu cầu id unique. Nếu gọi add() với id đã tồn tại, phiên bản 0.5.x báo lỗi (không silent override). Dùng upsert() nếu không chắc id đã có hay chưa.

# Thay vì:
collection.add(ids=["doc1"], documents=["updated text"])  # lỗi nếu doc1 đã có

# Dùng:
collection.upsert(ids=["doc1"], documents=["updated text"])

3. Add document dài mà không chunk

Một document 10 trang PDF đưa thẳng vào add(documents=[full_pdf_text]) sẽ bị embed thành một vector duy nhất. Model embedding có max token limit (~512 tokens với MiniLM) — text dài bị truncate, mất thông tin cuối. Recall thấp vì một vector đại diện quá nhiều nội dung. Nên chunk trước khi add (xem bài 22: Text Splitters).

4. HNSW không deterministic

Kết quả query ANN có thể thay đổi nhẹ giữa các lần build index khác nhau (do randomness trong HNSW construction). Kết quả top-K có thể đổi thứ tự hoặc xuất hiện / mất một vài document ở biên. Đây là đặc tính của approximate search, không phải bug.

5. Không đặt distance threshold

collection.query(n_results=5) luôn trả về đúng 5 kết quả, kể cả khi tất cả đều xa query. Trong RAG, kết quả có distance cao không có giá trị, thậm chí gây hallucination nếu đưa vào context LLM. Luôn filter theo threshold sau khi query.

• ChromaDB là vector DB local-first, zero-config, phù hợp prototype và dataset < vài triệu vectors.
• Ba chế độ: EphemeralClient (RAM), PersistentClient (disk, single-process), HttpClient (server, multi-process).
• Collection = đơn vị storage. Mỗi collection có embedding function và distance metric riêng — phải nhất quán.
• add(documents=[...]) tự embed; add(embeddings=[...]) nếu đã có vector.
• Metadata filter dùng where={} với operators $eq/$ne/$gt/$gte/$lt/$lte/$in/$nin và composite $and/$or.
• Dùng upsert() thay vì add() khi không chắc id đã tồn tại.
• Luôn đặt distance threshold sau query để loại kết quả không liên quan.

Bài 15: Pinecone — managed vector DB cho production — khi nào cần rời khỏi ChromaDB, cách dùng Pinecone serverless, so sánh chi phí và trade-off.