AIPaths Academy
•20 de octubre de 2025
•15 min de lectura
Construye un Chatbot de Documentación con Claude y RAG
claudetutorialragvector-databaseautomatizacion
Table of Contents(15 sections)
On This Page
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:
- Recupera documentos relevantes de tu base de conocimiento
- Aumenta el prompt de IA con ese contexto
- 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
.enval 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
- Cachear consultas comunes: Ahorra 30-50% en preguntas repetidas
- Reducir top_k: Menos fragmentos recuperados = menores costos de Claude
- Usar Claude Haiku para consultas simples: 10x más económico para preguntas directas
- Procesamiento por lotes: Actualizar embeddings semanalmente, no en tiempo real
- 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:
- Revisar calidad de fragmentos: Verificar que los fragmentos contengan pensamientos completos
- Aumentar top_k: Recuperar más contexto (probar 7-10)
- Ajustar tamaño de fragmento: Fragmentos más grandes (1024 tokens) para docs conceptuales
- Mejorar prompt: Agregar ejemplos de buenas respuestas
- 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:
- Usar Claude Haiku: 3x más rápido para consultas simples
- Reducir max_tokens: Bajar a 1024 si las respuestas son verbosas
- Cachear embeddings: No re-embeber consultas
- Optimizar búsqueda vectorial: Usar vecinos más cercanos aproximados
- 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:
- Implementar caché: Cachear consultas comunes
- Cambiar modelos: Usar Haiku para 50% de consultas
- Reducir contexto: Bajar top_k a 3
- Actualizaciones por lotes: No re-embeber docs sin cambios
- 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:
- Verificar embeddings: Revisar que todos los docs fueron embebidos
- Probar búsqueda directamente: Consultar Pinecone directamente para depurar
- Mejorar preprocesamiento de documentos: Eliminar ruido, preservar estructura
- Agregar filtrado de metadatos: Buscar secciones específicas de docs
- 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:
- Agrega la documentación de tu empresa y prueba con preguntas reales
- Implementa caché para reducir costos en 30-50%
- Configura monitoreo y alertas
- Despliega a producción con autenticación adecuada
Mejoras avanzadas:
- Agregar soporte multi-idioma para equipos internacionales
- Integrar con Slack/Discord para fácil acceso
- Construir ciclo de retroalimentación para mejorar continuamente las respuestas
- Agregar auto-actualización de documentos cuando cambien
- 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:
¿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?
Was this helpful?
Share this content
0comments