Inteligencia Artificial
Implementando RAG desde Cero con Pinecone y OpenAI: Guía Completa
Tutorial paso a paso para implementar Retrieval-Augmented Generation (RAG) en producción con Pinecone, OpenAI y Python. Incluye chunking, embeddings y optimización.
Equipo Nexgen
11 min de lectura
#RAG#Pinecone#OpenAI#LLM#Vector Database#Embeddings
Respuesta Directa
RAG (Retrieval-Augmented Generation) combina búsqueda semántica con generación de LLMs. Implementarlo requiere: (1) Dividir documentos en chunks, (2) Generar embeddings, (3) Almacenar en vector DB (Pinecone), (4) Buscar chunks relevantes, (5) Generar respuesta con LLM usando contexto recuperado. Tiempo total de implementación: ~2 horas para un sistema básico funcional.
¿Por Qué RAG?
El Problema con LLMs Estándar
# Sin RAG - Conocimiento limitado
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "user",
"content": "¿Cuáles son las políticas de RH de mi empresa?"
}]
)
# Respuesta: "No tengo información específica sobre tu empresa..."
Limitaciones:
- Conocimiento cortado en la fecha de entrenamiento
- No sabe información específica de tu dominio
- No puede acceder a documentos privados
- Puede alucinar información
Con RAG
# Con RAG - Conocimiento actualizado y específico
response = rag_system.query(
"¿Cuáles son las políticas de RH de mi empresa?"
)
# Respuesta con contexto de tus documentos internos:
# "Según el Manual de RH actualizado en octubre 2025, las políticas son..."
Arquitectura RAG Completa
┌──────────────┐
│ Documentos │
│ (PDF, TXT) │
└──────┬───────┘
│
▼
┌──────────────┐
│ Chunking │ ← Divide en fragmentos manejables
└──────┬───────┘
│
▼
┌──────────────┐
│ Embeddings │ ← Convierte texto a vectores
│ (OpenAI) │
└──────┬───────┘
│
▼
┌──────────────┐
│ Pinecone │ ← Almacena vectores
│ (Vector DB) │
└──────┬───────┘
│
▼
┌──────────────┐
│ User Query │
└──────┬───────┘
│
▼
┌──────────────┐
│ Similarity │ ← Busca chunks relevantes
│ Search │
└──────┬───────┘
│
▼
┌──────────────┐
│ LLM with │ ← Genera respuesta con contexto
│ Context │
└──────────────┘
Paso 1: Setup del Entorno
Instalación
pip install pinecone-client openai tiktoken pypdf2 python-dotenv
Variables de Entorno
# .env
OPENAI_API_KEY=sk-...
PINECONE_API_KEY=...
PINECONE_ENVIRONMENT=us-west1-gcp
Imports Necesarios
# rag_system.py
import os
from typing import List, Dict
import tiktoken
from openai import OpenAI
import pinecone
from pypdf2 import PdfReader
from dotenv import load_dotenv
load_dotenv()
# Inicializar clientes
openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
pinecone.init(
api_key=os.getenv("PINECONE_API_KEY"),
environment=os.getenv("PINECONE_ENVIRONMENT")
)
Paso 2: Chunking - Dividir Documentos
Estrategia de Chunking
def chunk_text(
text: str,
chunk_size: int = 500, # tokens
chunk_overlap: int = 50
) -> List[str]:
"""
Divide texto en chunks con overlap para mantener contexto.
"""
encoding = tiktoken.get_encoding("cl100k_base")
tokens = encoding.encode(text)
chunks = []
start = 0
while start < len(tokens):
# Tomar chunk_size tokens
end = start + chunk_size
chunk_tokens = tokens[start:end]
# Decodificar a texto
chunk_text = encoding.decode(chunk_tokens)
chunks.append(chunk_text)
# Mover start con overlap
start = end - chunk_overlap
return chunks
# Ejemplo de uso
text = """
La arquitectura multi-tenant es fundamental para SaaS.
Existen tres modelos principales: schema compartido,
schemas separados, y base de datos por tenant...
"""
chunks = chunk_text(text, chunk_size=200, chunk_overlap=30)
print(f"Texto dividido en {len(chunks)} chunks")
Chunking Avanzado: Semantic Chunking
def semantic_chunking(
text: str,
max_chunk_size: int = 500
) -> List[str]:
"""
Divide por párrafos para mantener coherencia semántica.
"""
# Dividir por párrafos
paragraphs = text.split('\n\n')
chunks = []
current_chunk = ""
current_size = 0
encoding = tiktoken.get_encoding("cl100k_base")
for para in paragraphs:
para_tokens = len(encoding.encode(para))
if current_size + para_tokens > max_chunk_size:
# Chunk actual está lleno, guardar
if current_chunk:
chunks.append(current_chunk.strip())
current_chunk = para
current_size = para_tokens
else:
# Agregar al chunk actual
current_chunk += "\n\n" + para
current_size += para_tokens
# Agregar último chunk
if current_chunk:
chunks.append(current_chunk.strip())
return chunks
Paso 3: Generar Embeddings
Función de Embedding
def get_embedding(text: str, model: str = "text-embedding-3-small") -> List[float]:
"""
Genera embedding de un texto usando OpenAI.
"""
# Limpiar texto
text = text.replace("\n", " ").strip()
response = openai_client.embeddings.create(
input=[text],
model=model
)
return response.data[0].embedding
# Ejemplo
chunk = "La arquitectura multi-tenant permite servir múltiples clientes..."
embedding = get_embedding(chunk)
print(f"Dimensiones del embedding: {len(embedding)}") # 1536 para text-embedding-3-small
print(f"Primeros 5 valores: {embedding[:5]}")
Batch Processing para Eficiencia
def get_embeddings_batch(
texts: List[str],
model: str = "text-embedding-3-small",
batch_size: int = 100
) -> List[List[float]]:
"""
Genera embeddings en batches para eficiencia.
"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# Limpiar textos
batch = [text.replace("\n", " ").strip() for text in batch]
response = openai_client.embeddings.create(
input=batch,
model=model
)
batch_embeddings = [item.embedding for item in response.data]
all_embeddings.extend(batch_embeddings)
print(f"Procesados {len(all_embeddings)}/{len(texts)} textos")
return all_embeddings
Paso 4: Configurar Pinecone
Crear Índice
def create_pinecone_index(
index_name: str = "rag-demo",
dimension: int = 1536, # text-embedding-3-small
metric: str = "cosine"
):
"""
Crea un índice en Pinecone.
"""
# Verificar si el índice ya existe
if index_name not in pinecone.list_indexes():
pinecone.create_index(
name=index_name,
dimension=dimension,
metric=metric,
pods=1,
pod_type="s1.x1" # Tier gratuito
)
print(f"Índice '{index_name}' creado")
else:
print(f"Índice '{index_name}' ya existe")
return pinecone.Index(index_name)
# Crear/obtener índice
index = create_pinecone_index()
Insertar Vectores
def upsert_documents(
index,
chunks: List[str],
embeddings: List[List[float]],
metadata: List[Dict] = None
):
"""
Inserta chunks y embeddings en Pinecone.
"""
# Preparar datos
vectors = []
for i, (chunk, embedding) in enumerate(zip(chunks, embeddings)):
vector_id = f"chunk_{i}"
# Metadata para cada vector
meta = {
"text": chunk,
"chunk_index": i
}
# Agregar metadata adicional si se proporciona
if metadata and i < len(metadata):
meta.update(metadata[i])
vectors.append({
"id": vector_id,
"values": embedding,
"metadata": meta
})
# Upsert en batches de 100
batch_size = 100
for i in range(0, len(vectors), batch_size):
batch = vectors[i:i + batch_size]
index.upsert(vectors=batch)
print(f"Insertados {min(i + batch_size, len(vectors))}/{len(vectors)} vectores")
print(f"Total de vectores en índice: {index.describe_index_stats()}")
Paso 5: Sistema RAG Completo
Clase RAGSystem
class RAGSystem:
def __init__(
self,
index_name: str = "rag-demo",
embedding_model: str = "text-embedding-3-small",
llm_model: str = "gpt-4-turbo-preview"
):
self.index = pinecone.Index(index_name)
self.embedding_model = embedding_model
self.llm_model = llm_model
self.openai_client = OpenAI()
def add_documents(self, documents: List[str], metadata: List[Dict] = None):
"""
Agrega documentos al sistema RAG.
"""
print("Chunking documentos...")
all_chunks = []
all_metadata = []
for i, doc in enumerate(documents):
chunks = semantic_chunking(doc, max_chunk_size=500)
all_chunks.extend(chunks)
# Metadata para cada chunk
for j in range(len(chunks)):
chunk_meta = {"document_id": i, "chunk_in_doc": j}
if metadata and i < len(metadata):
chunk_meta.update(metadata[i])
all_metadata.append(chunk_meta)
print(f"Generando embeddings para {len(all_chunks)} chunks...")
embeddings = get_embeddings_batch(all_chunks, model=self.embedding_model)
print("Insertando en Pinecone...")
upsert_documents(self.index, all_chunks, embeddings, all_metadata)
print("✅ Documentos agregados exitosamente")
def search(self, query: str, top_k: int = 5) -> List[Dict]:
"""
Busca chunks relevantes para una consulta.
"""
# Generar embedding de la consulta
query_embedding = get_embedding(query, model=self.embedding_model)
# Buscar en Pinecone
results = self.index.query(
vector=query_embedding,
top_k=top_k,
include_metadata=True
)
# Extraer resultados
matches = []
for match in results.matches:
matches.append({
"text": match.metadata["text"],
"score": match.score,
"metadata": match.metadata
})
return matches
def query(
self,
question: str,
top_k: int = 5,
temperature: float = 0.3
) -> Dict:
"""
Responde una pregunta usando RAG.
"""
# 1. Buscar contexto relevante
print(f"Buscando contexto para: '{question}'")
matches = self.search(question, top_k=top_k)
if not matches:
return {
"answer": "No encontré información relevante en la base de conocimiento.",
"sources": []
}
# 2. Construir contexto
context = "\n\n---\n\n".join([match["text"] for match in matches])
# 3. Construir prompt
prompt = f"""Responde la pregunta basándote ÚNICAMENTE en el siguiente contexto.
Si la información no está en el contexto, di que no lo sabes.
Contexto:
{context}
Pregunta: {question}
Respuesta:"""
# 4. Generar respuesta con LLM
print("Generando respuesta con LLM...")
response = self.openai_client.chat.completions.create(
model=self.llm_model,
messages=[
{
"role": "system",
"content": "Eres un asistente útil que responde preguntas basándose en el contexto proporcionado."
},
{
"role": "user",
"content": prompt
}
],
temperature=temperature
)
answer = response.choices[0].message.content
return {
"answer": answer,
"sources": [
{
"text": match["text"][:200] + "...",
"score": match["score"],
"metadata": match["metadata"]
}
for match in matches
]
}
Paso 6: Uso del Sistema
Indexar Documentos
# Inicializar sistema
rag = RAGSystem(index_name="empresa-docs")
# Documentos de ejemplo
documents = [
"""
Manual de Recursos Humanos 2025
Política de Vacaciones:
- Todos los empleados tienen derecho a 15 días de vacaciones por año.
- Las vacaciones deben solicitarse con 2 semanas de anticipación.
- Se pueden acumular hasta 30 días.
""",
"""
Política de Trabajo Remoto
- Los empleados pueden trabajar remoto 3 días por semana.
- Se requiere conexión estable y espacio adecuado.
- Reuniones importantes son presenciales.
""",
"""
Beneficios para Empleados
- Seguro médico mayor
- Vales de despensa $2,000 mensuales
- Fondo de ahorro 13%
- Capacitación anual $10,000 MXN
"""
]
metadata = [
{"category": "RH", "doc_name": "Manual RH"},
{"category": "Políticas", "doc_name": "Trabajo Remoto"},
{"category": "Beneficios", "doc_name": "Compensaciones"}
]
# Agregar documentos
rag.add_documents(documents, metadata)
Hacer Consultas
# Consulta 1
result = rag.query("¿Cuántos días de vacaciones tengo?")
print("Respuesta:", result["answer"])
print("\nFuentes:")
for i, source in enumerate(result["sources"], 1):
print(f"{i}. Score: {source['score']:.3f}")
print(f" {source['text']}")
print()
# Output:
# Respuesta: Todos los empleados tienen derecho a 15 días de vacaciones por año,
# y se pueden acumular hasta 30 días. Las vacaciones deben solicitarse con 2
# semanas de anticipación.
#
# Fuentes:
# 1. Score: 0.892
# Manual de Recursos Humanos 2025 - Política de Vacaciones...
# Consulta 2
result = rag.query("¿Puedo trabajar desde casa?")
print("Respuesta:", result["answer"])
# Los empleados pueden trabajar remoto 3 días por semana...
Optimizaciones Avanzadas
1. Re-ranking
from sentence_transformers import CrossEncoder
class RAGWithReranking(RAGSystem):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
# Modelo de re-ranking
self.reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L-6-v2')
def query(self, question: str, top_k: int = 5, rerank_top_k: int = 20):
# 1. Búsqueda inicial (más resultados)
initial_matches = self.search(question, top_k=rerank_top_k)
# 2. Re-ranking
pairs = [[question, match["text"]] for match in initial_matches]
scores = self.reranker.predict(pairs)
# 3. Reordenar por score de re-ranking
for match, score in zip(initial_matches, scores):
match["rerank_score"] = float(score)
reranked = sorted(initial_matches, key=lambda x: x["rerank_score"], reverse=True)
# 4. Tomar top_k después de re-ranking
top_matches = reranked[:top_k]
# Resto del proceso igual...
2. Hybrid Search (Semantic + Keyword)
from rank_bm25 import BM25Okapi
class HybridRAG(RAGSystem):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self.bm25_index = None
self.documents = []
def add_documents(self, documents: List[str], *args, **kwargs):
super().add_documents(documents, *args, **kwargs)
# Crear índice BM25 para keyword search
tokenized_docs = [doc.lower().split() for doc in documents]
self.bm25_index = BM25Okapi(tokenized_docs)
self.documents = documents
def hybrid_search(self, query: str, top_k: int = 5, alpha: float = 0.5):
"""
Combina semantic search (Pinecone) con keyword search (BM25).
alpha: peso para semantic (1-alpha para keyword)
"""
# 1. Semantic search
semantic_results = self.search(query, top_k=top_k * 2)
# 2. Keyword search
tokenized_query = query.lower().split()
bm25_scores = self.bm25_index.get_scores(tokenized_query)
# 3. Normalizar scores y combinar
# ...implementación de combinación de scores...
return combined_results
3. Query Transformation
def transform_query(query: str) -> List[str]:
"""
Genera múltiples versiones de la consulta para mejor retrieval.
"""
response = openai_client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "user",
"content": f"""Genera 3 reformulaciones de esta pregunta que ayuden a encontrar información relevante:
Pregunta original: {query}
Reformulaciones:
1."""
}],
temperature=0.7
)
# Parsear respuestas...
return reformulated_queries
# Uso
original_query = "¿Cuánto cuesta el producto?"
queries = transform_query(original_query)
# Buscar con todas las variantes y combinar resultados
all_results = []
for q in queries:
results = rag.search(q, top_k=3)
all_results.extend(results)
# Deduplicar y reordenar...
Métricas y Evaluación
def evaluate_rag(
rag_system: RAGSystem,
test_questions: List[str],
expected_answers: List[str]
) -> Dict:
"""
Evalúa el sistema RAG con preguntas de prueba.
"""
from rouge import Rouge
rouge = Rouge()
results = []
for question, expected in zip(test_questions, expected_answers):
result = rag_system.query(question)
actual = result["answer"]
# Calcular ROUGE score
scores = rouge.get_scores(actual, expected)[0]
results.append({
"question": question,
"rouge_1": scores["rouge-1"]["f"],
"rouge_2": scores["rouge-2"]["f"],
"rouge_l": scores["rouge-l"]["f"]
})
# Promedios
avg_rouge_1 = sum(r["rouge_1"] for r in results) / len(results)
avg_rouge_2 = sum(r["rouge_2"] for r in results) / len(results)
return {
"avg_rouge_1": avg_rouge_1,
"avg_rouge_2": avg_rouge_2,
"results": results
}
Costos Estimados
OpenAI
- Embeddings (text-embedding-3-small): $0.00002 / 1K tokens
- 1 millón de tokens ≈ $0.02
- GPT-4 Turbo: $0.01 / 1K prompt tokens, $0.03 / 1K completion tokens
- 100 consultas con 2K tokens de contexto ≈ $2-3
Pinecone
- Tier gratuito: 1 index, 100K vectores
- Starter ($70/mes): 5M vectores, 10 pods
- Scale on demand: $0.096/hora por pod adicional
Conclusión
Has aprendido a implementar RAG completo:
✅ Chunking estratégico para mejor retrieval ✅ Embeddings con OpenAI ✅ Almacenamiento en Pinecone ✅ Sistema de Q&A con contexto ✅ Optimizaciones avanzadas
En Nexgen, usamos RAG para chatbots empresariales, análisis de documentos legales, y asistentes de código. El resultado: sistemas que combinan el poder de LLMs con conocimiento específico actualizado.
Próximos Pasos
¿Necesitas implementar RAG para tu empresa? Contáctanos para una consulta técnica.
Artículos Relacionados
Inteligencia Artificial
Cómo Construir tu Primer Agente de IA con LangChain en 5 Pasos
Tutorial paso a paso para construir un agente de IA funcional usando LangChain y Python. Desde setup hasta implementación de herramientas.
4 min
Por Equipo Nexgen
Inteligencia Artificial
LangChain vs LangGraph: ¿Qué Framework Elegir para tu Agente Autónomo?
Comparativa técnica detallada de los frameworks más populares para desarrollo de agentes autónomos. Código de ejemplo, casos de uso y recomendaciones.
7 min
Por Equipo Nexgen
SaaS
Kubernetes vs Serverless para SaaS: ¿Cuál Elegir en 2025?
Comparativa exhaustiva de Kubernetes vs Serverless (AWS Lambda) para plataformas SaaS. Costos, escalabilidad, complejidad y casos de uso reales.
11 min
Por Equipo Nexgen