Módulo 4: ChromaDB Setup y Configuración

Cápsula 08: Mini-Proyecto - Document Search System

Siguiente →

🎯 Objetivo del Mini-Proyecto

Construir sistema de búsqueda de documentos completo con ChromaDB: Ingest 10K docs, metadata filtering, query optimization, y benchmark performance.

Tiempo estimado: 25-30 minutos

📋 Especificaciones del Proyecto

Requisitos

  1. Indexar 10K documentos de Wikipedia (simulados)
  2. Metadata: category, language, timestamp
  3. Batch ingestion eficiente (<10 segundos)
  4. Queries con metadata filtering
  5. Benchmark latency (p95 <20ms)

Success Criteria

  • ✅ 10K docs ingested en <10s
  • ✅ Query latency p95 <20ms
  • ✅ Filtering reduce latency 3x+
  • ✅ Accuracy: Top-10 recall >90%

💻 Implementación Completa

import chromadb
import time
import numpy as np
from tqdm import tqdm
from datetime import datetime, timedelta

# ============================================
# STEP 1: Setup ChromaDB
# ============================================

print("Step 1: Setup ChromaDB")
client = chromadb.PersistentClient(path="./document_search_db")

# Collection con HNSW optimizado
collection = client.get_or_create_collection(
    name="wikipedia_docs",
    metadata={
        "hnsw:space": "cosine",
        "hnsw:M": 32,  # High accuracy
        "hnsw:construction_ef": 200
    }
)

print(f"✅ Collection created: {collection.name}")

# ============================================
# STEP 2: Generate Mock Data
# ============================================

print("\nStep 2: Generate mock Wikipedia data (10K docs)")

categories = ["science", "history", "technology", "arts", "sports"]
languages = ["en", "es", "fr"]

def generate_mock_docs(num_docs=10000):
    """Generate mock Wikipedia-like documents."""
    docs = []
    metadatas = []
    ids = []
    
    base_date = datetime.now() - timedelta(days=365)
    
    for i in range(num_docs):
        # Document content
        category = categories[i % len(categories)]
        doc = f"Article about {category} topic number {i}. " \
              f"This article contains detailed information about {category}."
        
        # Metadata
        metadata = {
            "category": category,
            "language": languages[i % len(languages)],
            "timestamp": int((base_date + timedelta(days=i % 365)).timestamp()),
            "word_count": len(doc.split())
        }
        
        docs.append(doc)
        metadatas.append(metadata)
        ids.append(f"doc_{i}")
    
    return docs, metadatas, ids

docs, metadatas, ids = generate_mock_docs(10000)
print(f"✅ Generated {len(docs)} documents")

# ============================================
# STEP 3: Batch Ingestion
# ============================================

print("\nStep 3: Batch ingestion (target: <10 seconds)")

batch_size = 1000
start_time = time.time()

with tqdm(total=len(docs), desc="Ingesting") as pbar:
    for i in range(0, len(docs), batch_size):
        batch_docs = docs[i:i+batch_size]
        batch_meta = metadatas[i:i+batch_size]
        batch_ids = ids[i:i+batch_size]
        
        collection.add(
            documents=batch_docs,
            metadatas=batch_meta,
            ids=batch_ids
        )
        
        pbar.update(len(batch_docs))

ingestion_time = time.time() - start_time
throughput = len(docs) / ingestion_time

print(f"✅ Ingestion complete:")
print(f"   Time: {ingestion_time:.2f}s")
print(f"   Throughput: {throughput:.0f} docs/sec")
print(f"   Target met: {'✅' if ingestion_time < 10 else '❌'}")

# ============================================
# STEP 4: Query Benchmarks
# ============================================

print("\nStep 4: Query performance benchmarks")

def benchmark_queries(collection, query_texts, where=None, n_results=10, iterations=50):
    """Benchmark query latency."""
    latencies = []
    
    for _ in range(iterations):
        start = time.time()
        collection.query(
            query_texts=query_texts,
            where=where,
            n_results=n_results
        )
        latencies.append((time.time() - start) * 1000)
    
    return {
        "p50": np.percentile(latencies, 50),
        "p95": np.percentile(latencies, 95),
        "p99": np.percentile(latencies, 99),
        "mean": np.mean(latencies)
    }

# Benchmark A: Sin filtering
print("\nBenchmark A: Without filtering")
results_no_filter = benchmark_queries(
    collection,
    query_texts=["article about science"],
    where=None
)

print(f"p50: {results_no_filter['p50']:.1f}ms")
print(f"p95: {results_no_filter['p95']:.1f}ms")
print(f"p99: {results_no_filter['p99']:.1f}ms")
print(f"Target (p95 <20ms): {'✅' if results_no_filter['p95'] < 20 else '❌'}")

# Benchmark B: Con filtering
print("\nBenchmark B: With filtering (category='science')")
results_with_filter = benchmark_queries(
    collection,
    query_texts=["article about science"],
    where={"category": "science"}
)

print(f"p50: {results_with_filter['p50']:.1f}ms")
print(f"p95: {results_with_filter['p95']:.1f}ms")
print(f"p99: {results_with_filter['p99']:.1f}ms")

speedup = results_no_filter['mean'] / results_with_filter['mean']
print(f"Speedup with filtering: {speedup:.1f}x")
print(f"Target (3x+ speedup): {'✅' if speedup >= 3 else '❌'}")

# ============================================
# STEP 5: Accuracy Test (Recall@10)
# ============================================

print("\nStep 5: Accuracy test (Recall@10)")

def calculate_recall_at_k(collection, test_queries, k=10):
    """Calculate recall@k for test queries."""
    
    recalls = []
    
    for query_text, expected_category in test_queries:
        # Query
        results = collection.query(
            query_texts=[query_text],
            n_results=k
        )
        
        # Check how many results match expected category
        retrieved_categories = [
            meta['category'] for meta in results['metadatas'][0]
        ]
        
        relevant_count = sum(1 for cat in retrieved_categories if cat == expected_category)
        recall = relevant_count / k
        recalls.append(recall)
    
    return np.mean(recalls)

# Test queries
test_queries = [
    ("article about science topic", "science"),
    ("information about history", "history"),
    ("technology article content", "technology"),
    ("arts and culture topic", "arts"),
    ("sports related article", "sports"),
]

recall = calculate_recall_at_k(collection, test_queries, k=10)
print(f"Average Recall@10: {recall*100:.1f}%")
print(f"Target (>90%): {'✅' if recall > 0.9 else '❌'}")

# ============================================
# STEP 6: Final Summary
# ============================================

print("\n" + "="*60)
print("MINI-PROYECTO SUMMARY")
print("="*60)

print(f"\n✅ Ingestion:")
print(f"   - Docs: {len(docs)}")
print(f"   - Time: {ingestion_time:.2f}s")
print(f"   - Throughput: {throughput:.0f} docs/sec")
print(f"   - Target: <10s {'✅' if ingestion_time < 10 else '❌'}")

print(f"\n✅ Query Performance:")
print(f"   - p95 latency (no filter): {results_no_filter['p95']:.1f}ms")
print(f"   - p95 latency (with filter): {results_with_filter['p95']:.1f}ms")
print(f"   - Speedup: {speedup:.1f}x")
print(f"   - Target p95 <20ms: {'✅' if results_no_filter['p95'] < 20 else '❌'}")

print(f"\n✅ Accuracy:")
print(f"   - Recall@10: {recall*100:.1f}%")
print(f"   - Target >90%: {'✅' if recall > 0.9 else '❌'}")

all_passed = (
    ingestion_time < 10 and
    results_no_filter['p95'] < 20 and
    speedup >= 3 and
    recall > 0.9
)

print(f"\n{'='*60}")
print(f"ALL TESTS: {'✅ PASSED' if all_passed else '❌ FAILED'}")
print(f"{'='*60}")

# ============================================
# CLEANUP (Optional)
# ============================================

# import shutil
# shutil.rmtree("./document_search_db")
# print("\n✅ Cleaned up test database")

📊 Expected Output

Step 1: Setup ChromaDB
✅ Collection created: wikipedia_docs

Step 2: Generate mock Wikipedia data (10K docs)
✅ Generated 10000 documents

Step 3: Batch ingestion (target: <10 seconds)
Ingesting: 100%|██████████| 10000/10000 [00:08<00:00, 1250 docs/s]
✅ Ingestion complete:
   Time: 8.0s
   Throughput: 1250 docs/sec
   Target met: ✅

Step 4: Query performance benchmarks

Benchmark A: Without filtering
p50: 3.2ms
p95: 8.5ms
p99: 12.3ms
Target (p95 <20ms): ✅

Benchmark B: With filtering (category='science')
p50: 0.9ms
p95: 2.1ms
p99: 3.8ms
Speedup with filtering: 3.6x
Target (3x+ speedup): ✅

Step 5: Accuracy test (Recall@10)
Average Recall@10: 94.0%
Target (>90%): ✅

============================================================
MINI-PROYECTO SUMMARY
============================================================

✅ Ingestion:
   - Docs: 10000
   - Time: 8.00s
   - Throughput: 1250 docs/sec
   - Target: <10s ✅

✅ Query Performance:
   - p95 latency (no filter): 8.5ms
   - p95 latency (with filter): 2.1ms
   - Speedup: 3.6x
   - Target p95 <20ms: ✅

✅ Accuracy:
   - Recall@10: 94.0%
   - Target >90%: ✅

============================================================
ALL TESTS: ✅ PASSED
============================================================

✅ Learning Outcomes

Al completar este mini-proyecto, consolidaste:

  1. ✅ Setup ChromaDB con HNSW optimizado
  2. ✅ Batch ingestion eficiente (1250 docs/sec)
  3. ✅ Metadata filtering con 3.6x speedup
  4. ✅ Query optimization (p95 <20ms)
  5. ✅ Benchmarking completo (latency, throughput, accuracy)

🚀 Próximo paso: del Document Search al RAG completo

Acabas de cerrar el ciclo de "ChromaDB como vector database básica". Tu sistema funciona, los benchmarks confirman p95 <20ms con 10K documentos, y dominaste batch ingestion, metadata filtering y query optimization. Pero hay tres preguntas que tu sistema actual no resuelve para RAG real:

  1. ¿Qué embeddings está usando ChromaDB? Hasta ahora aceptaste el default sin cuestionarlo. ¿Es suficiente para producción o necesitas algo mejor? ¿Cuánto cuesta cambiar?
  2. ¿Qué pasa con un documento largo de 50 páginas? El default de ChromaDB lo trunca silenciosamente. ¿Cómo conservas la información sin perder resolución de búsqueda?
  3. ¿Cómo se conecta esto con un LLM? Tu Document Search devuelve chunks similares, pero un sistema RAG necesita ir un paso más: usar esos chunks como contexto para que un LLM redacte una respuesta.

Las cápsulas 09, 10 y 11 cierran exactamente esos tres frentes. Vas a aprender a tomar decisiones conscientes sobre embeddings (cuándo cambiar del default a OpenAI), a chunkear documentos largos sin perder calidad de retrieval, y a construir un pipeline RAG end-to-end mínimo viable que es la base del proyecto integrador del Módulo 8.

Qué te falta del Módulo 4:

  • Cápsula 09: Embeddings con OpenAI — cuándo cambiar del default y por qué
  • Cápsula 10: Chunking de documentos — RecursiveCharacterTextSplitter y decisiones de tamaño/overlap
  • Cápsula 11: Pipeline RAG end-to-end — chunking + embeddings + ChromaDB + GPT con citas de fuentes

Después del Módulo 4 completo, el Módulo 5 abre el panorama de proveedores (Pinecone, Weaviate, Qdrant, Milvus) para que decidas si ChromaDB es la elección correcta para tu próximo proyecto.

Tiempo: 25-30 minutos Siguiente: 09-embeddings-con-openai.md

← AnteriorCápsula 07: Persistence y Durability — qué pasa cuando algo se rompeSiguiente →Cápsula 09: Embeddings con OpenAI — Cuándo cambiar del default