Capítulo 3: Implementaciones personalizadas de RAG

Cuándo construir personalizado

✅ Construya personalizado cuando:

• Necesite control detallado
• Requisitos específicos de rendimiento
• Lógica de recuperación única
• Dependencias mínimas

❌ Use frameworks cuando:

• Prototipado rápido
• Casos de uso estándar
• Importa la familiaridad del equipo

Implementación mínima de RAG

import openai
from chromadb import Client
from chromadb.config import Settings

class SimpleRAG:
    def __init__(self, collection_name="docs"):
        self.client = Client(Settings(anonymized_telemetry=False))
        self.collection = self.client.create_collection(collection_name)
        self.openai = openai.OpenAI()
    
    def add_documents(self, texts, ids=None):
        """Add documents to the knowledge base"""
        if ids is None:
            ids = [f"doc_{i}" for i in range(len(texts))]
        
        embeddings = self._embed(texts)
        self.collection.add(
            embeddings=embeddings,
            documents=texts,
            ids=ids
        )
    
    def _embed(self, texts):
        """Generate embeddings"""
        response = self.openai.embeddings.create(
            input=texts,
            model="text-embedding-3-small"
        )
        return [e.embedding for e in response.data]
    
    def query(self, question, top_k=3):
        """Query the RAG system"""
        # Retrieve
        query_embedding = self._embed([question])[0]
        results = self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k
        )
        
        # Generate
        context = "\n".join(results["documents"][0])
        prompt = f"""Context: {context}

Question: {question}

Answer based on the context:"""
        
        response = self.openai.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        
        return response.choices[0].message.content

# Usage
rag = SimpleRAG()
rag.add_documents(["Doc 1 content...", "Doc 2 content..."])
answer = rag.query("What is discussed?")

Agregando funcionalidades

Streaming

def query_stream(self, question, top_k=3):
    context = self._retrieve(question, top_k)
    prompt = self._build_prompt(context, question)
    
    stream = self.openai.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": prompt}],
        stream=True
    )
    
    for chunk in stream:
        if chunk.choices[0].delta.content:
            yield chunk.choices[0].delta.content

Caché

from functools import lru_cache

class CachedRAG(SimpleRAG):
    @lru_cache(maxsize=1000)
    def _embed_cached(self, text):
        return tuple(self._embed([text])[0])

Resumen comparativo

Enfoque	Ventajas	Desventajas
LangChain	Rico en funcionalidades, componible	Complejo, grande
LlamaIndex	Enfocado en datos, sencillo	Menos flexible
Personalizado	Control total, mínimo	Más trabajo

---

Próximos pasos

¡Ponga a prueba sus conocimientos con el Cuestionario del Módulo 6!