Construye un Chatbot de Documentación con Claude y RAG

La documentación suele estar dispersa en Notion, Confluence, Google Docs y múltiples wikis. Los desarrolladores pasan horas buscando respuestas que deberían tomar segundos. ¿Te suena familiar?

Los estudios muestran que los equipos de desarrollo desperdician entre un 20-40% de su tiempo buscando información. Preguntas como "¿Cómo me autentico con nuestra API de pagos?" o "¿Cuál es el patrón de manejo de errores para tareas en segundo plano?" no deberían requerir revisar docenas de documentos.

¿La solución? Un chatbot de documentación potenciado con RAG usando Claude AI. Este tutorial te mostrará cómo construir un prototipo funcional en dos semanas o menos. Una vez desplegados, estos sistemas típicamente manejan más de 100 consultas diarias y pueden reducir el tiempo de búsqueda de documentación en un 70-80%.

Lo que aprenderás:

• Qué es RAG y por qué es perfecto para documentación
• Cómo construir un chatbot de documentación desde cero
• Configurar bases de datos vectoriales y embeddings
• Integrar la API de Claude para respuestas inteligentes
• Despliegue en producción y optimización de costos
• Errores comunes y cómo evitarlos

Tiempo de lectura: 15 minutos Nivel de habilidad: Intermedio Requisitos previos: Conocimientos básicos de Python, experiencia con APIs, familiaridad con bases de datos

¿Qué es RAG y Por Qué Usarlo?

RAG (Retrieval Augmented Generation o Generación Aumentada por Recuperación) es una técnica que combina búsqueda de documentos con generación de IA. En lugar de depender únicamente de los datos de entrenamiento de un modelo de IA, RAG:

1. Recupera documentos relevantes de tu base de conocimiento
2. Aumenta el prompt de IA con ese contexto
3. Genera una respuesta basada en tu documentación real

Piénsalo como darle a Claude una memoria fotográfica de tus documentos más la inteligencia para sintetizar respuestas.

Por Qué RAG Supera a las Alternativas

¿Fine-tuning? Costoso, consume tiempo y queda obsoleto en cuanto los documentos cambian.

¿Prompt stuffing? Limitado por las ventanas de contexto y los costos.

¿RAG? Dinámico, rentable y siempre actualizado.

La Arquitectura Que Construiremos

Pregunta del Usuario
    ↓
Consulta → Base de Datos Vectorial (encontrar docs relevantes)
    ↓
Mejores Resultados + Pregunta → API de Claude
    ↓
Respuesta Inteligente (con fuentes)

Simple, efectivo, listo para producción.

Requisitos Previos y Configuración

Antes de sumergirnos, preparemos nuestro entorno.

Lo Que Necesitarás

Requerido:

• Python 3.9+ instalado
• Clave API de Anthropic (obtén una aquí)
• Cuenta de Pinecone (el nivel gratuito funciona) o base de datos vectorial alternativa
• 30 minutos de tiempo concentrado

Opcional pero Recomendado:

• Docker (para desarrollo local)
• Cuenta de Vercel/Railway (para despliegue)
• Repositorio GitHub (para control de versiones)

Configuración del Proyecto

# Crear directorio del proyecto
mkdir doc-chatbot
cd doc-chatbot

# Crear entorno virtual
python -m venv venv
source venv/bin/activate  # En Windows: venv\Scripts\activate

# Crear estructura del proyecto
mkdir -p src/{data,embeddings,retrieval,generation}
touch src/__init__.py
touch src/main.py
touch requirements.txt
touch .env

# Inicializar git
git init
echo "venv/" >> .gitignore
echo ".env" >> .gitignore
echo "__pycache__/" >> .gitignore

Instalar Dependencias

# requirements.txt
anthropic==0.18.1
pinecone-client==3.0.0
python-dotenv==1.0.0
openai==1.12.0  # Para embeddings
beautifulsoup4==4.12.0  # Para parsear docs
markdown==3.5.0
tiktoken==0.5.2
fastapi==0.109.0  # Para servidor API
uvicorn==0.27.0

Instalar todo:

pip install -r requirements.txt

Variables de Entorno

Crea tu archivo .env:

# .env
ANTHROPIC_API_KEY=sk-ant-xxxxx
OPENAI_API_KEY=sk-xxxxx  # Para embeddings
PINECONE_API_KEY=xxxxx
PINECONE_ENVIRONMENT=us-west1-gcp
PINECONE_INDEX_NAME=docs-chatbot

Nota de Seguridad: Nunca hagas commit de .env al control de versiones. Usa gestión de secretos en producción (AWS Secrets Manager, variables de entorno de Vercel, etc.)

Paso 1: Preparar Tu Documentación

RAG es tan bueno como los documentos que le alimentas. Preparemos fragmentos de documentación de alta calidad.

Cargar Documentos

# src/data/loader.py
import os
from pathlib import Path
from typing import List, Dict
import markdown
from bs4 import BeautifulSoup

class DocumentLoader:
    """Cargar y parsear documentación de varias fuentes."""

    def __init__(self, docs_dir: str):
        self.docs_dir = Path(docs_dir)

    def load_markdown_files(self) -> List[Dict[str, str]]:
        """Cargar todos los archivos markdown del directorio."""
        documents = []

        for md_file in self.docs_dir.rglob("*.md"):
            with open(md_file, 'r', encoding='utf-8') as f:
                content = f.read()

                # Convertir markdown a texto
                html = markdown.markdown(content)
                text = BeautifulSoup(html, 'html.parser').get_text()

                documents.append({
                    'content': text,
                    'source': str(md_file.relative_to(self.docs_dir)),
                    'filename': md_file.name
                })

        return documents

    def load_from_url(self, url: str) -> Dict[str, str]:
        """Cargar documentación desde una URL (para web scraping)."""
        # Implementación para web scraping
        # Agregar requests, parsing con beautifulsoup
        pass

# Uso
loader = DocumentLoader("./docs")
documents = loader.load_markdown_files()
print(f"Cargados {len(documents)} documentos")

Estrategia de Fragmentación

Los documentos necesitan dividirse en fragmentos que quepan en embeddings y proporcionen buen contexto.

# src/data/chunker.py
from typing import List, Dict
import tiktoken

class DocumentChunker:
    """Dividir documentos en fragmentos óptimos para RAG."""

    def __init__(self,
                 chunk_size: int = 512,
                 chunk_overlap: int = 50,
                 encoding_name: str = "cl100k_base"):
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.encoding = tiktoken.get_encoding(encoding_name)

    def chunk_document(self, document: Dict[str, str]) -> List[Dict[str, str]]:
        """Dividir un documento en fragmentos con superposición."""
        text = document['content']
        tokens = self.encoding.encode(text)

        chunks = []
        start = 0

        while start < len(tokens):
            # Obtener fragmento
            end = start + self.chunk_size
            chunk_tokens = tokens[start:end]
            chunk_text = self.encoding.decode(chunk_tokens)

            chunks.append({
                'content': chunk_text,
                'source': document['source'],
                'filename': document['filename'],
                'chunk_index': len(chunks)
            })

            # Mover al siguiente fragmento con superposición
            start += self.chunk_size - self.chunk_overlap

        return chunks

    def chunk_documents(self, documents: List[Dict]) -> List[Dict]:
        """Fragmentar todos los documentos."""
        all_chunks = []

        for doc in documents:
            chunks = self.chunk_document(doc)
            all_chunks.extend(chunks)

        return all_chunks

# Uso
chunker = DocumentChunker(chunk_size=512, chunk_overlap=50)
chunks = chunker.chunk_documents(documents)
print(f"Creados {len(chunks)} fragmentos de {len(documents)} documentos")

¿Por Qué Estos Parámetros?

• 512 tokens por fragmento: Balancea contexto vs. especificidad
• 50 tokens de superposición: Previene pérdida de contexto en los límites
• Preservación de metadatos: Seguimiento de fuentes para citas

Consejo Pro: Ajusta el tamaño del fragmento según tus documentos. ¿Referencias de API? Fragmentos más pequeños (256). ¿Guías conceptuales? Fragmentos más grandes (1024).

Paso 2: Crear Embeddings

Los embeddings convierten texto en vectores que capturan significado semántico. Conceptos similares tienen vectores similares.

Configurar Embeddings de OpenAI

Usaremos el modelo text-embedding-3-small de OpenAI—es rápido, económico y excelente para RAG.

# src/embeddings/embedder.py
from typing import List, Dict
from openai import OpenAI
import os
from dotenv import load_dotenv

load_dotenv()

class DocumentEmbedder:
    """Generar embeddings para fragmentos de documentos."""

    def __init__(self, model: str = "text-embedding-3-small"):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
        self.model = model
        self.dimension = 1536  # Dimensión para text-embedding-3-small

    def embed_text(self, text: str) -> List[float]:
        """Generar embedding para un texto único."""
        response = self.client.embeddings.create(
            model=self.model,
            input=text
        )
        return response.data[0].embedding

    def embed_chunks(self, chunks: List[Dict], batch_size: int = 100) -> List[Dict]:
        """Generar embeddings para fragmentos en lotes."""
        embedded_chunks = []

        for i in range(0, len(chunks), batch_size):
            batch = chunks[i:i + batch_size]
            texts = [chunk['content'] for chunk in batch]

            # Solicitud de embedding en lote
            response = self.client.embeddings.create(
                model=self.model,
                input=texts
            )

            # Agregar embeddings a fragmentos
            for chunk, embedding_data in zip(batch, response.data):
                chunk_with_embedding = chunk.copy()
                chunk_with_embedding['embedding'] = embedding_data.embedding
                embedded_chunks.append(chunk_with_embedding)

            print(f"Embebidos {len(embedded_chunks)}/{len(chunks)} fragmentos")

        return embedded_chunks

# Uso
embedder = DocumentEmbedder()
embedded_chunks = embedder.embed_chunks(chunks)

Consideraciones de Costo

Precios de Embeddings de OpenAI:

• text-embedding-3-small: $0.02 / 1M tokens
• Ejemplo: 1,000 fragmentos × 500 tokens = 500,000 tokens = $0.01

Costo total de embedding para la mayoría de la documentación: $0.01 - $0.10 único.

Consejo Pro: Solo re-embebe documentos cuando cambien. Cachea embeddings para evitar costos repetidos.

Paso 3: Configurar Base de Datos Vectorial

Pinecone es una base de datos vectorial gestionada perfecta para RAG. Es rápida, escalable y tiene un nivel gratuito generoso.

Inicializar Pinecone

# src/retrieval/vector_store.py
from typing import List, Dict, Tuple
import pinecone
import os
from dotenv import load_dotenv

load_dotenv()

class VectorStore:
    """Gestionar almacenamiento y recuperación de vectores con Pinecone."""

    def __init__(self, index_name: str = None):
        # Inicializar Pinecone
        pinecone.init(
            api_key=os.getenv("PINECONE_API_KEY"),
            environment=os.getenv("PINECONE_ENVIRONMENT")
        )

        self.index_name = index_name or os.getenv("PINECONE_INDEX_NAME")

        # Crear índice si no existe
        if self.index_name not in pinecone.list_indexes():
            pinecone.create_index(
                name=self.index_name,
                dimension=1536,  # Debe coincidir con la dimensión del embedding
                metric="cosine"  # Similitud coseno para búsqueda semántica
            )

        self.index = pinecone.Index(self.index_name)

    def upsert_chunks(self, chunks: List[Dict], batch_size: int = 100):
        """Subir fragmentos embebidos a Pinecone."""
        vectors = []

        for i, chunk in enumerate(chunks):
            vector_id = f"chunk_{i}"

            vectors.append({
                'id': vector_id,
                'values': chunk['embedding'],
                'metadata': {
                    'content': chunk['content'],
                    'source': chunk['source'],
                    'filename': chunk['filename'],
                    'chunk_index': chunk['chunk_index']
                }
            })

            # Upsert en lotes
            if len(vectors) >= batch_size:
                self.index.upsert(vectors=vectors)
                print(f"Insertados {len(vectors)} vectores")
                vectors = []

        # Upsert restantes
        if vectors:
            self.index.upsert(vectors=vectors)
            print(f"Insertados {len(vectors)} vectores")

        print(f"✓ Total de vectores en índice: {self.index.describe_index_stats()['total_vector_count']}")

    def search(self, query_embedding: List[float], top_k: int = 5) -> List[Dict]:
        """Buscar fragmentos similares."""
        results = self.index.query(
            vector=query_embedding,
            top_k=top_k,
            include_metadata=True
        )

        # Formatear resultados
        chunks = []
        for match in results['matches']:
            chunks.append({
                'content': match['metadata']['content'],
                'source': match['metadata']['source'],
                'score': match['score']
            })

        return chunks

# Uso
vector_store = VectorStore()
vector_store.upsert_chunks(embedded_chunks)

Alternativa: Usar Chroma (Base de Datos Vectorial Local)

Si prefieres una solución local, Chroma es excelente:

# Alternativa: Usar Chroma
import chromadb
from chromadb.config import Settings

client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="./chroma_db"
))

collection = client.create_collection(name="docs")
collection.add(
    embeddings=[chunk['embedding'] for chunk in embedded_chunks],
    documents=[chunk['content'] for chunk in embedded_chunks],
    metadatas=[{'source': chunk['source']} for chunk in embedded_chunks],
    ids=[f"chunk_{i}" for i in range(len(embedded_chunks))]
)

Pinecone vs. Chroma:

• Pinecone: Alojado en la nube, escalable, gestionado, nivel gratuito: 1 índice + 100K vectores
• Chroma: Local, gratuito, excelente para desarrollo, requiere auto-alojamiento para producción

Paso 4: Construir el Pipeline RAG

Ahora conectamos todo: consulta → recuperar → generar respuesta.

Crear el Sistema RAG

# src/generation/rag.py
from typing import List, Dict
from anthropic import Anthropic
import os
from dotenv import load_dotenv

load_dotenv()

class RAGChatbot:
    """Chatbot de documentación potenciado con RAG usando Claude."""

    def __init__(self, vector_store, embedder):
        self.vector_store = vector_store
        self.embedder = embedder
        self.claude = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-3-5-sonnet-20241022"

    def retrieve_context(self, question: str, top_k: int = 5) -> List[Dict]:
        """Recuperar fragmentos de documentación relevantes."""
        # Embeber la pregunta
        question_embedding = self.embedder.embed_text(question)

        # Buscar en base de datos vectorial
        results = self.vector_store.search(question_embedding, top_k=top_k)

        return results

    def generate_answer(self, question: str, context_chunks: List[Dict]) -> Dict[str, any]:
        """Generar respuesta usando Claude con contexto recuperado."""

        # Formatear contexto
        context_text = "\n\n---\n\n".join([
            f"Fuente: {chunk['source']}\n{chunk['content']}"
            for chunk in context_chunks
        ])

        # Construir prompt
        prompt = f"""Eres un asistente de documentación útil. Responde la pregunta del usuario basándote en los extractos de documentación proporcionados.

Contexto de Documentación:
{context_text}

Pregunta del Usuario: {question}

Instrucciones:
- Responde SOLO basándote en la documentación proporcionada
- Si la documentación no contiene suficiente información, dilo
- Incluye referencias a las fuentes en tu respuesta
- Sé conciso pero completo
- Usa ejemplos de los documentos cuando sea útil

Respuesta:"""

        # Llamar a la API de Claude
        response = self.claude.messages.create(
            model=self.model,
            max_tokens=2048,
            temperature=0.3,  # Temperatura más baja para precisión factual
            messages=[{
                "role": "user",
                "content": prompt
            }]
        )

        return {
            'answer': response.content[0].text,
            'sources': [chunk['source'] for chunk in context_chunks],
            'model': self.model,
            'usage': {
                'input_tokens': response.usage.input_tokens,
                'output_tokens': response.usage.output_tokens
            }
        }

    def answer_question(self, question: str, top_k: int = 5) -> Dict:
        """RAG de extremo a extremo: recuperar + generar."""
        # Paso 1: Recuperar docs relevantes
        context_chunks = self.retrieve_context(question, top_k=top_k)

        # Paso 2: Generar respuesta
        result = self.generate_answer(question, context_chunks)

        return result

# Uso
chatbot = RAGChatbot(vector_store, embedder)
result = chatbot.answer_question("¿Cómo me autentico con la API de pagos?")

print(f"Respuesta: {result['answer']}\n")
print(f"Fuentes: {', '.join(result['sources'])}")
print(f"Tokens: {result['usage']['input_tokens']} entrada, {result['usage']['output_tokens']} salida")

Probar el Chatbot

# test_chatbot.py
def test_chatbot():
    questions = [
        "¿Cómo me autentico con la API?",
        "¿Cuál es el patrón de manejo de errores?",
        "¿Cómo despliego a producción?",
        "¿Qué base de datos usamos?",
        "¿Cómo ejecuto las pruebas?"
    ]

    for question in questions:
        print(f"\n{'='*60}")
        print(f"P: {question}")
        print(f"{'='*60}")

        result = chatbot.answer_question(question)

        print(f"R: {result['answer']}")
        print(f"\nFuentes: {', '.join(result['sources'])}")
        print(f"Costo: ~${calculate_cost(result['usage']):.4f}")

def calculate_cost(usage):
    # Precios de Claude 3.5 Sonnet (Nov 2024)
    INPUT_COST = 3.00 / 1_000_000   # $3 por millón de tokens
    OUTPUT_COST = 15.00 / 1_000_000  # $15 por millón de tokens

    return (usage['input_tokens'] * INPUT_COST +
            usage['output_tokens'] * OUTPUT_COST)

test_chatbot()

Paso 5: Construir el Servidor API

Envolvamos nuestro chatbot en una API lista para producción usando FastAPI.

Crear la API

# src/main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Optional
import uvicorn
from src.generation.rag import RAGChatbot
from src.retrieval.vector_store import VectorStore
from src.embeddings.embedder import DocumentEmbedder

app = FastAPI(title="API de Chatbot de Documentación")

# Inicializar componentes
vector_store = VectorStore()
embedder = DocumentEmbedder()
chatbot = RAGChatbot(vector_store, embedder)

class QuestionRequest(BaseModel):
    question: str
    top_k: Optional[int] = 5

class AnswerResponse(BaseModel):
    answer: str
    sources: List[str]
    confidence: str
    usage: dict

@app.post("/ask", response_model=AnswerResponse)
async def ask_question(request: QuestionRequest):
    """Responder una pregunta sobre la documentación."""
    try:
        result = chatbot.answer_question(
            question=request.question,
            top_k=request.top_k
        )

        # Determinar confianza basada en relevancia de fuentes
        confidence = "alta" if len(result['sources']) >= 3 else "media"

        return AnswerResponse(
            answer=result['answer'],
            sources=result['sources'],
            confidence=confidence,
            usage=result['usage']
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    """Endpoint de verificación de salud."""
    return {"status": "saludable"}

@app.get("/stats")
async def get_stats():
    """Obtener estadísticas del chatbot."""
    stats = vector_store.index.describe_index_stats()
    return {
        "total_documents": stats['total_vector_count'],
        "index_name": vector_store.index_name,
        "status": "operacional"
    }

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Probar la API

# Iniciar el servidor
python -m src.main

# Probar en otra terminal
curl -X POST "http://localhost:8000/ask" \
  -H "Content-Type: application/json" \
  -d '{"question": "¿Cómo me autentico con la API?"}'

Agregar Manejo de Errores

# Manejo de errores mejorado
from functools import wraps
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def handle_errors(func):
    @wraps(func)
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            logger.error(f"Error en {func.__name__}: {str(e)}")
            raise HTTPException(
                status_code=500,
                detail=f"Ocurrió un error: {str(e)}"
            )
    return wrapper

@app.post("/ask")
@handle_errors
async def ask_question(request: QuestionRequest):
    # Implementación...
    pass

Paso 6: Despliegue

Despeguemos esto a producción.

Opción 1: Desplegar en Railway

Railway ofrece despliegues de configuración cero perfectos para APIs.

# Instalar Railway CLI
npm i -g @railway/cli

# Iniciar sesión
railway login

# Inicializar proyecto
railway init

# Desplegar
railway up

Crear railway.json:

{
  "build": {
    "builder": "NIXPACKS"
  },
  "deploy": {
    "startCommand": "uvicorn src.main:app --host 0.0.0.0 --port $PORT",
    "restartPolicyType": "ON_FAILURE",
    "restartPolicyMaxRetries": 10
  }
}

Opción 2: Desplegar en Vercel (Serverless)

# api/ask.py - Función serverless de Vercel
from fastapi import FastAPI
from mangum import Mangum
from src.main import app

handler = Mangum(app)

vercel.json:

{
  "builds": [
    {
      "src": "api/ask.py",
      "use": "@vercel/python"
    }
  ],
  "routes": [
    {
      "src": "/(.*)",
      "dest": "api/ask.py"
    }
  ]
}

Desplegar:

vercel --prod

Opción 3: Despliegue con Docker

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Instalar dependencias
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copiar aplicación
COPY . .

# Exponer puerto
EXPOSE 8000

# Ejecutar aplicación
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

Construir y ejecutar:

docker build -t doc-chatbot .
docker run -p 8000:8000 --env-file .env doc-chatbot

Desglose de Costos

Hablemos de números. Esto es lo que cuesta ejecutar este chatbot en producción:

Costos Únicos de Configuración

Generación de embeddings:

• 1,000 fragmentos de docs × 500 tokens = 500K tokens
• Embeddings de OpenAI: $0.02 / 1M tokens
• Total: ~$0.01

Costos Operativos Mensuales

Escenario: 1,000 consultas/mes

API de Claude (respuestas):

• Promedio: 2,000 tokens de entrada + 500 tokens de salida por consulta
• Entrada: 1,000 × 2,000 = 2M tokens × $3/M = $6
• Salida: 1,000 × 500 = 500K tokens × $15/M = $7.50
• Total Claude: $13.50/mes

OpenAI (embeddings de consultas):

• 1,000 consultas × 50 tokens = 50K tokens
• Embeddings: $0.001/mes

Pinecone (base de datos vectorial):

• Nivel gratuito: 1 índice, 100K vectores
• Nivel de pago: $70/mes (si es necesario)
• BD Vectorial: $0 - $70/mes

Total: $13.51 - $83.51/mes para 1,000 consultas

Costo por consulta: $0.014 - $0.084

Consejos de Optimización de Costos

1. Cachear consultas comunes: Ahorra 30-50% en preguntas repetidas
2. Reducir top_k: Menos fragmentos recuperados = menores costos de Claude
3. Usar Claude Haiku para consultas simples: 10x más económico para preguntas directas
4. Procesamiento por lotes: Actualizar embeddings semanalmente, no en tiempo real
5. Auto-alojar vectores: Usar Chroma en lugar de Pinecone para <10K docs

Mejorar Tu Chatbot

Agregar Memoria de Conversación

class ConversationalRAG(RAGChatbot):
    """Chatbot RAG con historial de conversación."""

    def __init__(self, vector_store, embedder):
        super().__init__(vector_store, embedder)
        self.conversations = {}  # Almacenar por session_id

    def answer_with_history(self,
                            question: str,
                            session_id: str,
                            top_k: int = 5) -> Dict:
        """Responder con contexto de conversación."""

        # Obtener historial de conversación
        history = self.conversations.get(session_id, [])

        # Recuperar contexto
        context_chunks = self.retrieve_context(question, top_k)

        # Construir mensajes con historial
        messages = history + [{
            "role": "user",
            "content": self._build_prompt(question, context_chunks)
        }]

        # Generar respuesta
        response = self.claude.messages.create(
            model=self.model,
            max_tokens=2048,
            messages=messages
        )

        # Actualizar historial
        history.append({
            "role": "user",
            "content": question
        })
        history.append({
            "role": "assistant",
            "content": response.content[0].text
        })

        self.conversations[session_id] = history[-10:]  # Mantener últimos 5 intercambios

        return {
            'answer': response.content[0].text,
            'sources': [c['source'] for c in context_chunks],
            'session_id': session_id
        }

Agregar Respuestas en Streaming

@app.post("/ask/stream")
async def ask_question_stream(request: QuestionRequest):
    """Transmitir respuesta mientras se genera."""

    async def generate():
        context_chunks = chatbot.retrieve_context(request.question)
        prompt = chatbot._build_prompt(request.question, context_chunks)

        with chatbot.claude.messages.stream(
            model=chatbot.model,
            max_tokens=2048,
            messages=[{"role": "user", "content": prompt}]
        ) as stream:
            for text in stream.text_stream:
                yield f"data: {text}\n\n"

    return StreamingResponse(generate(), media_type="text/event-stream")

Agregar Ciclo de Retroalimentación

class FeedbackStore:
    """Almacenar retroalimentación del usuario para mejorar respuestas."""

    def __init__(self):
        self.feedback = []

    def add_feedback(self, question: str, answer: str, rating: int, comment: str = ""):
        """Almacenar retroalimentación para análisis."""
        self.feedback.append({
            'timestamp': datetime.now(),
            'question': question,
            'answer': answer,
            'rating': rating,  # 1-5
            'comment': comment
        })

    def get_poor_responses(self) -> List[Dict]:
        """Identificar respuestas que necesitan mejora."""
        return [f for f in self.feedback if f['rating'] <= 2]

@app.post("/feedback")
async def submit_feedback(question: str, rating: int, comment: str = ""):
    """Enviar retroalimentación sobre una respuesta."""
    feedback_store.add_feedback(question, rating, comment)
    return {"status": "recibido"}

Solución de Problemas Comunes

Problema 1: Calidad Pobre de Respuestas

Síntomas: El chatbot da respuestas genéricas o incorrectas.

Soluciones:

1. Revisar calidad de fragmentos: Verificar que los fragmentos contengan pensamientos completos
2. Aumentar top_k: Recuperar más contexto (probar 7-10)
3. Ajustar tamaño de fragmento: Fragmentos más grandes (1024 tokens) para docs conceptuales
4. Mejorar prompt: Agregar ejemplos de buenas respuestas
5. Filtrar resultados de baja confianza: Ignorar coincidencias con puntaje < 0.7

def retrieve_context(self, question: str, top_k: int = 5, min_score: float = 0.7):
    """Recuperar con filtrado de confianza."""
    results = self.vector_store.search(question_embedding, top_k=top_k)

    # Filtrar resultados de baja confianza
    filtered = [r for r in results if r['score'] >= min_score]

    if len(filtered) < 2:
        logger.warning(f"Resultados de baja confianza para: {question}")

    return filtered

Problema 2: Tiempos de Respuesta Lentos

Síntomas: Las consultas toman >5 segundos.

Soluciones:

1. Usar Claude Haiku: 3x más rápido para consultas simples
2. Reducir max_tokens: Bajar a 1024 si las respuestas son verbosas
3. Cachear embeddings: No re-embeber consultas
4. Optimizar búsqueda vectorial: Usar vecinos más cercanos aproximados
5. Agregar streaming de respuestas: Comenzar a retornar antes de completar

Problema 3: Costos Altos

Síntomas: Facturas de API más altas de lo esperado.

Soluciones:

1. Implementar caché: Cachear consultas comunes
2. Cambiar modelos: Usar Haiku para 50% de consultas
3. Reducir contexto: Bajar top_k a 3
4. Actualizaciones por lotes: No re-embeber docs sin cambios
5. Establecer límites de uso: Limitar tokens por solicitud

# Agregar caché simple
from functools import lru_cache
import hashlib

def cache_key(question: str) -> str:
    return hashlib.md5(question.lower().encode()).hexdigest()

@lru_cache(maxsize=1000)
def cached_answer(question_hash: str, question: str) -> Dict:
    return chatbot.answer_question(question)

# Uso
result = cached_answer(cache_key(question), question)

Problema 4: Contexto Faltante

Síntomas: "No tengo información sobre eso" para temas conocidos.

Soluciones:

1. Verificar embeddings: Revisar que todos los docs fueron embebidos
2. Probar búsqueda directamente: Consultar Pinecone directamente para depurar
3. Mejorar preprocesamiento de documentos: Eliminar ruido, preservar estructura
4. Agregar filtrado de metadatos: Buscar secciones específicas de docs
5. Expandir sinónimos: Manejar variaciones de terminología

# Agregar filtrado de metadatos
def search_with_filters(self, query_embedding: List[float],
                        filter_dict: Dict = None) -> List[Dict]:
    """Buscar con filtros de metadatos."""
    results = self.index.query(
        vector=query_embedding,
        top_k=5,
        filter=filter_dict,  # ej., {"source": {"$eq": "api-docs.md"}}
        include_metadata=True
    )
    return results

Mejores Prácticas de Producción

1. Monitoreo y Registro

import logging
from datetime import datetime

class ChatbotMonitor:
    """Monitorear rendimiento y uso del chatbot."""

    def __init__(self):
        self.metrics = {
            'queries': 0,
            'errors': 0,
            'total_tokens': 0,
            'avg_response_time': 0
        }

    def log_query(self, question: str, result: Dict, duration: float):
        """Registrar métricas de consulta."""
        self.metrics['queries'] += 1
        self.metrics['total_tokens'] += (
            result['usage']['input_tokens'] +
            result['usage']['output_tokens']
        )

        # Actualizar tiempo de respuesta promedio
        self.metrics['avg_response_time'] = (
            (self.metrics['avg_response_time'] * (self.metrics['queries'] - 1) + duration)
            / self.metrics['queries']
        )

        logging.info({
            'timestamp': datetime.now().isoformat(),
            'question': question,
            'duration': duration,
            'tokens': result['usage'],
            'sources': result['sources']
        })

2. Límite de Tasa

from fastapi import Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address

limiter = Limiter(key_func=get_remote_address)
app.state.limiter = limiter
app.add_exception_handler(429, _rate_limit_exceeded_handler)

@app.post("/ask")
@limiter.limit("10/minute")  # 10 solicitudes por minuto por IP
async def ask_question(request: Request, question_request: QuestionRequest):
    # Implementación...
    pass

3. Verificaciones de Salud y Reintentos

from tenacity import retry, stop_after_attempt, wait_exponential

class RobustRAGChatbot(RAGChatbot):
    """Chatbot RAG con lógica de reintentos."""

    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=2, max=10)
    )
    def generate_answer(self, question: str, context_chunks: List[Dict]) -> Dict:
        """Generar respuesta con reintentos automáticos."""
        return super().generate_answer(question, context_chunks)

4. Seguridad

from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from fastapi import Security

security = HTTPBearer()

def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)):
    """Verificar token de API."""
    if credentials.credentials != os.getenv("API_TOKEN"):
        raise HTTPException(status_code=401, detail="Token inválido")
    return credentials

@app.post("/ask")
async def ask_question(
    request: QuestionRequest,
    credentials: HTTPAuthorizationCredentials = Security(verify_token)
):
    # Implementación...
    pass

Conclusión

Ahora tienes un chatbot de documentación listo para producción potenciado por Claude y RAG. Esta arquitectura está probada en batalla, es rentable y escala desde startup hasta empresa.

Conclusiones clave:

• RAG combina recuperación de documentos con generación de IA para respuestas precisas respaldadas por fuentes
• Las bases de datos vectoriales permiten búsqueda semántica sobre tu documentación
• Claude sobresale sintetizando contexto en respuestas claras y útiles
• Los costos son predecibles: ~$0.01-0.08 por consulta para la mayoría de casos de uso
• Comienza simple, luego agrega características como memoria de conversación y streaming
• Monitorea el uso, implementa caché y optimiza basándote en datos reales

¿Qué Sigue?

Mejoras inmediatas:

1. Agrega la documentación de tu empresa y prueba con preguntas reales
2. Implementa caché para reducir costos en 30-50%
3. Configura monitoreo y alertas
4. Despliega a producción con autenticación adecuada

Mejoras avanzadas:

1. Agregar soporte multi-idioma para equipos internacionales
2. Integrar con Slack/Discord para fácil acceso
3. Construir ciclo de retroalimentación para mejorar continuamente las respuestas
4. Agregar auto-actualización de documentos cuando cambien
5. Implementar búsqueda híbrida (semántica + palabras clave)

Temas relacionados para explorar:

• Fine-tuning de embeddings para terminología específica del dominio
• RAG multi-modal (buscar imágenes, diagramas, código)
• RAG agéntico (chatbot que puede tomar acciones)
• Métricas de evaluación y benchmarking de RAG

Lecturas Adicionales

• Documentación de la API de Claude de Anthropic
• Guía de RAG de Pinecone
• Mejores Prácticas de Embeddings de OpenAI
• Mejores Prácticas de Producción de FastAPI

Recursos

Herramientas Utilizadas:

• API de Claude
• Base de Datos Vectorial Pinecone
• API de Embeddings de OpenAI
• FastAPI

---

¿Construiste un chatbot de documentación? ¡Comparte tus resultados y cualquier mejora que hayas hecho!

¿Preguntas o atascado en un paso? ¡Deja un comentario abajo o abre un issue en GitHub!

¿Quieres ver más tutoriales de RAG? Déjanos saber qué casos de uso te interesan—soporte al cliente, asistentes de investigación, documentación de código?