🤖 Developer Cookbook - FASE 6: Integración de IA en Aplicaciones
Recetas prácticas para integrar LLMs, RAG y vector databases en tus apps
📚 Tabla de Contenidos
- Receta 6.8: APIs de LLMs - OpenAI, Anthropic
- Receta 6.9: RAG (Retrieval Augmented Generation)
- Receta 6.10: Vector Databases - Pinecone, Chroma
- Receta 6.11: MCP (Model Context Protocol)
- Receta 6.12: Fine-tuning vs Embeddings vs Prompting
Integración de IA en Aplicaciones
Receta 6.8: APIs de LLMs - OpenAI, Anthropic
OpenAI API (GPT-4):
import openai
from typing import Optional, List, Dict
import os
class OpenAIClient:
"""
Cliente wrapper para OpenAI API
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("OPENAI_API_KEY")
openai.api_key = self.api_key
def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4",
temperature: float = 0.7,
max_tokens: int = 500
) -> str:
"""
Llamada básica a Chat Completions API
"""
try:
response = openai.ChatCompletion.create(
model=model,
messages=messages,
temperature=temperature,
max_tokens=max_tokens
)
return response.choices[0].message.content
except openai.error.RateLimitError:
print("⚠️ Rate limit alcanzado, esperando...")
import time
time.sleep(60)
return self.chat_completion(messages, model, temperature, max_tokens)
except openai.error.APIError as e:
print(f"❌ API Error: {e}")
raise
def streaming_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4"
):
"""
Streaming response para UX mejor
"""
response = openai.ChatCompletion.create(
model=model,
messages=messages,
stream=True
)
for chunk in response:
if chunk.choices[0].delta.get("content"):
content = chunk.choices[0].delta.content
yield content
def function_calling(
self,
messages: List[Dict[str, str]],
functions: List[Dict],
model: str = "gpt-4"
):
"""
Function calling para integración con herramientas
"""
response = openai.ChatCompletion.create(
model=model,
messages=messages,
functions=functions,
function_call="auto"
)
message = response.choices[0].message
# Si el modelo quiere llamar una función
if message.get("function_call"):
return {
"type": "function_call",
"function": message.function_call
}
else:
return {
"type": "message",
"content": message.content
}
# Uso básico
client = OpenAIClient()
# Conversación simple
messages = [
{"role": "system", "content": "Eres un asistente experto en Python"},
{"role": "user", "content": "¿Cómo implemento un singleton en Python?"}
]
response = client.chat_completion(messages)
print(response)
# Streaming
print("\n🔄 Streaming response:")
for chunk in client.streaming_completion(messages):
print(chunk, end="", flush=True)
# Function calling example
functions = [
{
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name, e.g. San Francisco"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"]
}
},
"required": ["location"]
}
}
]
user_msg = [{"role": "user", "content": "What's the weather in Madrid?"}]
result = client.function_calling(user_msg, functions)
if result["type"] == "function_call":
print(f"\n🔧 Function to call: {result['function']['name']}")
print(f"Arguments: {result['function']['arguments']}")Anthropic API (Claude):
from anthropic import Anthropic, HUMAN_PROMPT, AI_PROMPT
from typing import List, Dict, Optional
import os
class AnthropicClient:
"""
Cliente wrapper para Anthropic Claude API
"""
def __init__(self, api_key: Optional[str] = None):
self.api_key = api_key or os.getenv("ANTHROPIC_API_KEY")
self.client = Anthropic(api_key=self.api_key)
def create_message(
self,
messages: List[Dict[str, str]],
model: str = "claude-sonnet-4-20250514",
max_tokens: int = 1000,
temperature: float = 0.7,
system: Optional[str] = None
) -> str:
"""
Crear mensaje con Claude (Messages API)
"""
try:
response = self.client.messages.create(
model=model,
max_tokens=max_tokens,
temperature=temperature,
system=system,
messages=messages
)
return response.content[0].text
except Exception as e:
print(f"❌ Error: {e}")
raise
def streaming_message(
self,
messages: List[Dict[str, str]],
model: str = "claude-sonnet-4-20250514",
max_tokens: int = 1000
):
"""
Streaming response
"""
with self.client.messages.stream(
model=model,
max_tokens=max_tokens,
messages=messages
) as stream:
for text in stream.text_stream:
yield text
def with_tools(
self,
messages: List[Dict[str, str]],
tools: List[Dict],
model: str = "claude-sonnet-4-20250514"
):
"""
Tool use (equivalente a function calling)
"""
response = self.client.messages.create(
model=model,
max_tokens=4096,
tools=tools,
messages=messages
)
return response
# Uso
claude_client = AnthropicClient()
# Mensaje simple
messages = [
{
"role": "user",
"content": "Explica qué es un closure en JavaScript con un ejemplo"
}
]
system_prompt = "Eres un instructor experto en programación. Explicas conceptos de forma clara y concisa con buenos ejemplos."
response = claude_client.create_message(
messages=messages,
system=system_prompt,
temperature=0.3 # Más determinista para explicaciones técnicas
)
print(response)
# Streaming
print("\n🔄 Streaming:")
for chunk in claude_client.streaming_message(messages):
print(chunk, end="", flush=True)
# Tool use
tools = [
{
"name": "get_user_info",
"description": "Retrieve user information from database",
"input_schema": {
"type": "object",
"properties": {
"user_id": {
"type": "string",
"description": "The user's unique identifier"
}
},
"required": ["user_id"]
}
}
]
tool_messages = [
{
"role": "user",
"content": "What's the email for user abc123?"
}
]
tool_response = claude_client.with_tools(tool_messages, tools)
print("\n🔧 Tool use response:")
print(tool_response)Comparación OpenAI vs Anthropic:
| Feature | OpenAI GPT-4 | Anthropic Claude |
|---|---|---|
| Context window | 8K - 128K | 200K |
| Function calling | ✅ Sí | ✅ Sí (Tools) |
| Streaming | ✅ Sí | ✅ Sí |
| Vision | ✅ GPT-4V | ✅ Claude 3+ |
| JSON mode | ✅ Sí | ❌ No (usar prompting) |
| Precio (prompt) | $$ | $ |
| Velocidad | 🟡 Media | 🟢 Rápida |
Manejo de errores y reintentos:
import time
from functools import wraps
from typing import Callable
def retry_with_backoff(
max_retries: int = 3,
base_delay: float = 1.0,
backoff_factor: float = 2.0
):
"""
Decorator para reintentar llamadas API con backoff exponencial
"""
def decorator(func: Callable):
@wraps(func)
def wrapper(*args, **kwargs):
retries = 0
delay = base_delay
while retries < max_retries:
try:
return func(*args, **kwargs)
except Exception as e:
retries += 1
if retries >= max_retries:
print(f"❌ Max retries reached: {e}")
raise
print(f"⚠️ Error (attempt {retries}/{max_retries}): {e}")
print(f"⏳ Waiting {delay}s before retry...")
time.sleep(delay)
delay *= backoff_factor
return wrapper
return decorator
# Uso
class RobustLLMClient:
"""Cliente LLM con manejo robusto de errores"""
def __init__(self):
self.openai_client = OpenAIClient()
self.claude_client = AnthropicClient()
@retry_with_backoff(max_retries=3, base_delay=2.0)
def call_with_fallback(
self,
messages: List[Dict[str, str]],
primary_provider: str = "openai"
) -> str:
"""
Llamar LLM con fallback automático
"""
try:
if primary_provider == "openai":
return self.openai_client.chat_completion(messages)
else:
return self.claude_client.create_message(messages)
except Exception as e:
print(f"⚠️ Primary provider failed: {e}")
print("🔄 Trying fallback provider...")
# Fallback al otro provider
if primary_provider == "openai":
return self.claude_client.create_message(messages)
else:
return self.openai_client.chat_completion(messages)
# Uso
robust_client = RobustLLMClient()
messages = [{"role": "user", "content": "Hello, world!"}]
response = robust_client.call_with_fallback(messages, primary_provider="openai")
print(response)Receta 6.9: RAG (Retrieval Augmented Generation)
¿Qué es RAG? Técnica que combina recuperación de información + generación de LLM para reducir alucinaciones y proporcionar respuestas basadas en hechos.
Arquitectura RAG:
User Query
↓
1. RETRIEVE: Buscar documentos relevantes
↓
2. AUGMENT: Añadir contexto al prompt
↓
3. GENERATE: LLM genera respuesta
↓
ResponseImplementación básica de RAG:
from typing import List, Dict, Tuple
import numpy as np
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
class SimpleRAG:
"""
Sistema RAG básico sin embeddings complejos
"""
def __init__(self, documents: List[str]):
self.documents = documents
self.vectorizer = TfidfVectorizer()
self.doc_vectors = self.vectorizer.fit_transform(documents)
def retrieve(self, query: str, top_k: int = 3) -> List[Tuple[int, str, float]]:
"""
Recuperar top K documentos más relevantes
"""
# Vectorizar query
query_vector = self.vectorizer.transform([query])
# Calcular similitud
similarities = cosine_similarity(query_vector, self.doc_vectors)[0]
# Obtener top K indices
top_indices = np.argsort(similarities)[::-1][:top_k]
# Retornar (index, document, score)
results = [
(idx, self.documents[idx], similarities[idx])
for idx in top_indices
]
return results
def generate_answer(
self,
query: str,
llm_client,
top_k: int = 3
) -> Dict:
"""
Pipeline completo RAG
"""
# 1. RETRIEVE
retrieved_docs = self.retrieve(query, top_k)
# 2. AUGMENT - Construir contexto
context = "\n\n".join([
f"Document {i+1} (relevance: {score:.2f}):\n{doc}"
for i, (idx, doc, score) in enumerate(retrieved_docs)
])
augmented_prompt = f"""
Basándote ÚNICAMENTE en los siguientes documentos, responde la pregunta.
Si la información no está en los documentos, di "No puedo responder basándome en la información proporcionada".
DOCUMENTOS:
{context}
PREGUNTA: {query}
RESPUESTA:
"""
# 3. GENERATE
messages = [{"role": "user", "content": augmented_prompt}]
response = llm_client.chat_completion(messages, temperature=0.3)
return {
"query": query,
"answer": response,
"sources": [
{"doc_id": idx, "score": score, "text": doc[:200]}
for idx, doc, score in retrieved_docs
]
}
# Ejemplo de uso
documents = [
"Python es un lenguaje de programación interpretado. Fue creado por Guido van Rossum en 1991.",
"JavaScript es el lenguaje principal para desarrollo web frontend. Fue creado por Brendan Eich en 1995.",
"Python se usa comúnmente para ciencia de datos, machine learning y desarrollo backend.",
"React es una librería de JavaScript para construir interfaces de usuario, creada por Facebook.",
"FastAPI es un framework moderno de Python para crear APIs web rápidas.",
]
rag = SimpleRAG(documents)
# Recuperar documentos relevantes
query = "¿Qué lenguaje se usa para ciencia de datos?"
results = rag.retrieve(query, top_k=2)
print("📚 Documentos recuperados:")
for idx, doc, score in results:
print(f" Score: {score:.3f} - {doc}")
# Pipeline completo
from anthropic import Anthropic
llm_client = OpenAIClient() # o AnthropicClient()
answer = rag.generate_answer(
query="¿Quién creó Python y en qué año?",
llm_client=llm_client
)
print(f"\n❓ Query: {answer['query']}")
print(f"✅ Answer: {answer['answer']}")
print(f"\n📖 Sources:")
for source in answer['sources']:
print(f" - Doc {source['doc_id']} (score: {source['score']:.3f})")RAG con embeddings (mejor calidad):
from sentence_transformers import SentenceTransformer
import faiss
import numpy as np
class AdvancedRAG:
"""
RAG con embeddings semánticos (mejor que TF-IDF)
"""
def __init__(self, documents: List[str], model_name: str = "all-MiniLM-L6-v2"):
self.documents = documents
self.model = SentenceTransformer(model_name)
# Generar embeddings
print("🔄 Generando embeddings...")
self.doc_embeddings = self.model.encode(documents, show_progress_bar=True)
# Crear índice FAISS para búsqueda rápida
dimension = self.doc_embeddings.shape[1]
self.index = faiss.IndexFlatL2(dimension)
self.index.add(self.doc_embeddings.astype('float32'))
print(f"✅ Índice creado con {len(documents)} documentos")
def retrieve(self, query: str, top_k: int = 5) -> List[Tuple[int, str, float]]:
"""
Búsqueda semántica con embeddings
"""
# Generar embedding del query
query_embedding = self.model.encode([query])[0]
# Buscar vecinos más cercanos
distances, indices = self.index.search(
query_embedding.reshape(1, -1).astype('float32'),
top_k
)
# Convertir distancia a score de similitud
results = []
for idx, distance in zip(indices[0], distances[0]):
# Convertir L2 distance a similarity score (0-1)
similarity = 1 / (1 + distance)
results.append((int(idx), self.documents[idx], float(similarity)))
return results
def add_documents(self, new_documents: List[str]):
"""
Añadir nuevos documentos al índice
"""
new_embeddings = self.model.encode(new_documents)
self.index.add(new_embeddings.astype('float32'))
self.documents.extend(new_documents)
print(f"✅ Añadidos {len(new_documents)} documentos")
def hybrid_search(
self,
query: str,
top_k: int = 5,
semantic_weight: float = 0.7
) -> List[Tuple[int, str, float]]:
"""
Búsqueda híbrida: semántica + keyword matching
"""
# Búsqueda semántica
semantic_results = self.retrieve(query, top_k)
# Búsqueda keyword (simple)
keyword_scores = {}
query_lower = query.lower()
for idx, doc in enumerate(self.documents):
# Score basado en palabras coincidentes
doc_lower = doc.lower()
matches = sum(1 for word in query_lower.split() if word in doc_lower)
keyword_scores[idx] = matches / len(query_lower.split())
# Combinar scores
combined = {}
for idx, doc, sem_score in semantic_results:
kw_score = keyword_scores.get(idx, 0)
combined[idx] = (
semantic_weight * sem_score +
(1 - semantic_weight) * kw_score
)
# Ordenar por score combinado
sorted_results = sorted(
combined.items(),
key=lambda x: x[1],
reverse=True
)[:top_k]
return [
(idx, self.documents[idx], score)
for idx, score in sorted_results
]
# Uso
knowledge_base = [
"FastAPI es un framework web moderno de Python basado en type hints.",
"Flask es un micro-framework web de Python, simple y flexible.",
"Django es un framework web completo de Python con ORM incluido.",
"Express.js es un framework minimalista para Node.js/JavaScript.",
"Next.js es un framework de React para aplicaciones full-stack.",
]
advanced_rag = AdvancedRAG(knowledge_base)
# Búsqueda semántica
query = "framework para APIs rápidas"
results = advanced_rag.retrieve(query, top_k=3)
print(f"🔍 Query: '{query}'\n")
for idx, doc, score in results:
print(f"Score: {score:.3f}")
print(f"Doc: {doc}\n")
# Añadir más documentos
advanced_rag.add_documents([
"Streamlit permite crear apps de data science sin frontend.",
"Gradio es una librería para crear demos de ML rápidamente."
])
# Búsqueda híbrida
hybrid_results = advanced_rag.hybrid_search(
"crear aplicaciones de machine learning",
top_k=3
)
print("🔀 Búsqueda híbrida:")
for idx, doc, score in hybrid_results:
print(f" {score:.3f}: {doc}")Receta 6.10: Vector Databases - Pinecone, Chroma
¿Qué son Vector Databases? Bases de datos especializadas en almacenar y buscar vectores (embeddings) eficientemente.
Comparación:
| DB | Tipo | Mejor para | Complejidad |
|---|---|---|---|
| Pinecone | Cloud | Producción, escala | 🟡 Media |
| Chroma | Local/Cloud | Desarrollo, prototipos | 🟢 Baja |
| Weaviate | Cloud/Self-hosted | Features avanzadas | 🔴 Alta |
| Milvus | Self-hosted | Control total | 🔴 Alta |
Chroma DB (Local, fácil):
import chromadb
from chromadb.config import Settings
from chromadb.utils import embedding_functions
class ChromaRAGSystem:
"""
Sistema RAG con ChromaDB
"""
def __init__(self, collection_name: str = "knowledge_base"):
# Inicializar cliente
self.client = chromadb.Client(Settings(
chroma_db_impl="duckdb+parquet",
persist_directory="./chroma_db"
))
# Función de embeddings (usa sentence-transformers)
self.embedding_function = embedding_functions.SentenceTransformerEmbeddingFunction(
model_name="all-MiniLM-L6-v2"
)
# Crear/obtener colección
try:
self.collection = self.client.create_collection(
name=collection_name,
embedding_function=self.embedding_function,
metadata={"description": "Knowledge base for RAG"}
)
except:
self.collection = self.client.get_collection(
name=collection_name,
embedding_function=self.embedding_function
)
def add_documents(
self,
documents: List[str],
metadatas: List[Dict] = None,
ids: List[str] = None
):
"""
Añadir documentos a la colección
"""
if ids is None:
ids = [f"doc_{i}" for i in range(len(documents))]
if metadatas is None:
metadatas = [{"source": "unknown"} for _ in documents]
self.collection.add(
documents=documents,
metadatas=metadatas,
ids=ids
)
print(f"✅ Añadidos {len(documents)} documentos")
def query(
self,
query_text: str,
n_results: int = 5,
where: Dict = None
) -> Dict:
"""
Buscar documentos similares
"""
results = self.collection.query(
query_texts=[query_text],
n_results=n_results,
where=where # Filtros metadata
)
return {
"documents": results["documents"][0],
"distances": results["distances"][0],
"metadatas": results["metadatas"][0],
"ids": results["ids"][0]
}
def update_document(self, doc_id: str, new_document: str, metadata: Dict = None):
"""
Actualizar documento existente
"""
self.collection.update(
ids=[doc_id],
documents=[new_document],
metadatas=[metadata] if metadata else None
)
def delete_documents(self, ids: List[str]):
"""
Eliminar documentos
"""
self.collection.delete(ids=ids)
def get_stats(self):
"""
Obtener estadísticas de la colección
"""
count = self.collection.count()
return {
"total_documents": count,
"collection_name": self.collection.name
}
# Uso
chroma_rag = ChromaRAGSystem("tech_docs")
# Añadir documentos con metadata
documents = [
"Python es excelente para data science y machine learning.",
"JavaScript es el lenguaje de la web, corre en navegadores.",
"Rust es un lenguaje de sistemas con memory safety.",
"Go es simple, concurrente, y compilado, ideal para microservicios.",
]
metadatas = [
{"category": "data_science", "language": "Python"},
{"category": "web", "language": "JavaScript"},
{"category": "systems", "language": "Rust"},
{"category": "backend", "language": "Go"},
]
chroma_rag.add_documents(documents, metadatas)
# Búsqueda simple
results = chroma_rag.query("lenguaje para análisis de datos", n_results=2)
print("🔍 Búsqueda simple:")
for doc, distance, metadata in zip(
results["documents"],
results["distances"],
results["metadatas"]
):
print(f" Distance: {distance:.3f}")
print(f" Category: {metadata['category']}")
print(f" Doc: {doc}\n")
# Búsqueda con filtros
filtered_results = chroma_rag.query(
"lenguaje para backend",
n_results=3,
where={"category": "backend"} # Solo docs de categoria "backend"
)
print("🎯 Búsqueda filtrada (category=backend):")
for doc in filtered_results["documents"]:
print(f" - {doc}")
# Estadísticas
stats = chroma_rag.get_stats()
print(f"\n📊 Stats: {stats}")Pinecone (Cloud, escalable):
import pinecone
from typing import List, Dict, Tuple
import os
class PineconeRAGSystem:
"""
Sistema RAG con Pinecone (cloud vector DB)
"""
def __init__(
self,
index_name: str,
dimension: int = 384, # Dimension del modelo de embeddings
api_key: str = None,
environment: str = "us-west1-gcp"
):
# Inicializar Pinecone
api_key = api_key or os.getenv("PINECONE_API_KEY")
pinecone.init(api_key=api_key, environment=environment)
# Crear índice si no existe
if index_name not in pinecone.list_indexes():
pinecone.create_index(
name=index_name,
dimension=dimension,
metric="cosine" # o "euclidean", "dotproduct"
)
print(f"✅ Índice '{index_name}' creado")
self.index = pinecone.Index(index_name)
# Modelo de embeddings
from sentence_transformers import SentenceTransformer
self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
def upsert_documents(
self,
documents: List[str],
metadatas: List[Dict] = None,
namespace: str = "default"
):
"""
Insertar/actualizar documentos
"""
# Generar embeddings
embeddings = self.embedding_model.encode(documents)
# Preparar vectors para Pinecone
vectors = []
for i, (doc, emb) in enumerate(zip(documents, embeddings)):
metadata = metadatas[i] if metadatas else {}
metadata["text"] = doc # Guardar texto original
vectors.append({
"id": f"doc_{i}_{hash(doc)}",
"values": emb.tolist(),
"metadata": metadata
})
# Upsert en batches
batch_size = 100
for i in range(0, len(vectors), batch_size):
batch = vectors[i:i+batch_size]
self.index.upsert(vectors=batch, namespace=namespace)
print(f"✅ Upserted {len(documents)} documentos")
def query(
self,
query_text: str,
top_k: int = 5,
namespace: str = "default",
filter: Dict = None
) -> List[Dict]:
"""
Buscar documentos similares
"""
# Generar embedding del query
query_embedding = self.embedding_model.encode([query_text])[0]
# Query Pinecone
results = self.index.query(
vector=query_embedding.tolist(),
top_k=top_k,
namespace=namespace,
filter=filter,
include_metadata=True
)
# Formatear resultados
formatted = []
for match in results["matches"]:
formatted.append({
"id": match["id"],
"score": match["score"],
"text": match["metadata"].get("text", ""),
"metadata": match["metadata"]
})
return formatted
def delete(self, ids: List[str], namespace: str = "default"):
"""
Eliminar documentos por IDs
"""
self.index.delete(ids=ids, namespace=namespace)
def get_stats(self):
"""
Estadísticas del índice
"""
return self.index.describe_index_stats()
# Uso
# pinecone_rag = PineconeRAGSystem(
# index_name="my-knowledge-base",
# api_key="your-pinecone-api-key"
# )
# documents = [
# "FastAPI es un framework moderno de Python para APIs.",
# "React es una librería de JavaScript para UIs.",
# "PostgreSQL es una base de datos relacional robusta.",
# ]
# metadatas = [
# {"category": "backend", "language": "Python"},
# {"category": "frontend", "language": "JavaScript"},
# {"category": "database", "type": "SQL"},
# ]
# pinecone_rag.upsert_documents(documents, metadatas)
# # Búsqueda
# results = pinecone_rag.query("framework para crear APIs", top_k=2)
# for result in results:
# print(f"Score: {result['score']:.3f}")
# print(f"Text: {result['text']}")
# print(f"Metadata: {result['metadata']}\n")
# # Búsqueda con filtros
# filtered = pinecone_rag.query(
# "database",
# top_k=3,
# filter={"type": {"$eq": "SQL"}}
# )Comparación Chroma vs Pinecone:
class VectorDBComparison:
"""
Comparar diferentes vector databases
"""
@staticmethod
def feature_comparison():
features = {
"Chroma": {
"Tipo": "Embebido/Cliente-Servidor",
"Hosting": "Local/Cloud",
"Escalabilidad": "🟡 Media (millones)",
"Latencia": "🟢 Baja (local)",
"Costo": "🟢 Gratis",
"Setup": "🟢 Muy fácil",
"Producción": "🟡 Sí (limitado)",
"Mejor para": "Desarrollo, MVPs",
},
"Pinecone": {
"Tipo": "Serverless Cloud",
"Hosting": "Solo Cloud",
"Escalabilidad": "🟢 Alta (billones)",
"Latencia": "🟡 Media (network)",
"Costo": "🟡 Pago (free tier limitado)",
"Setup": "🟢 Fácil",
"Producción": "🟢 Sí",
"Mejor para": "Producción, escala",
},
"Weaviate": {
"Tipo": "Self-hosted/Cloud",
"Hosting": "Ambos",
"Escalabilidad": "🟢 Alta",
"Latencia": "🟡 Variable",
"Costo": "🟡 Free tier + paid",
"Setup": "🟡 Moderado",
"Producción": "🟢 Sí",
"Mejor para": "Features avanzadas",
}
}
import pandas as pd
df = pd.DataFrame(features).T
print(df.to_markdown())
@staticmethod
def when_to_use():
"""
Guía de cuándo usar cada DB
"""
guide = """
🎯 GUÍA DE SELECCIÓN:
Usa CHROMA si:
✅ Estás prototipando o en fase MVP
✅ Quieres algo simple y local
✅ Menos de 1M vectores
✅ No necesitas alta disponibilidad
Usa PINECONE si:
✅ Aplicación en producción
✅ Necesitas escalar a millones/billones
✅ Quieres managed service (sin DevOps)
✅ Necesitas baja latencia global
Usa WEAVIATE si:
✅ Necesitas GraphQL/APIs complejas
✅ Quieres híbrido (keyword + vector)
✅ Control total sobre infraestructura
✅ Features avanzadas (multi-tenancy, etc)
Usa MILVUS si:
✅ Máximo control y customización
✅ On-premise deployment requerido
✅ Casos de uso muy específicos
✅ Equipo DevOps fuerte
"""
print(guide)
# Ver comparación
VectorDBComparison.feature_comparison()
VectorDBComparison.when_to_use()Receta 6.11: MCP (Model Context Protocol)
¿Qué es MCP? Protocol desarrollado por Anthropic para que LLMs puedan interactuar con herramientas externas y fuentes de datos de forma estandarizada.
Conceptos clave:
"""
MCP Architecture:
┌─────────────┐
│ Client │ (Tu aplicación)
│ (LLM app) │
└──────┬──────┘
│
│ MCP Protocol
│
┌──────▼──────┐
│ MCP Server │ (Provee herramientas/datos)
│ (Tools) │
└──────┬──────┘
│
│
┌──────▼──────────────┐
│ External Systems │
│ (DB, APIs, Files) │
└─────────────────────┘
"""
# Componentes:
# 1. Resources: Datos que el LLM puede leer (files, DB)
# 2. Prompts: Templates pre-definidos
# 3. Tools: Funciones que el LLM puede ejecutarEjemplo conceptual de MCP Server:
from typing import Dict, List, Any
import json
class MCPServer:
"""
Servidor MCP básico (conceptual)
"""
def __init__(self, name: str):
self.name = name
self.tools = {}
self.resources = {}
self.prompts = {}
def register_tool(self, tool_name: str, tool_func, description: str, parameters: Dict):
"""
Registrar herramienta disponible para LLM
"""
self.tools[tool_name] = {
"function": tool_func,
"description": description,
"parameters": parameters
}
def register_resource(self, resource_id: str, resource_getter):
"""
Registrar recurso (datos) accesibles
"""
self.resources[resource_id] = resource_getter
def list_tools(self) -> List[Dict]:
"""
Listar herramientas disponibles (para LLM)
"""
return [
{
"name": name,
"description": tool["description"],
"parameters": tool["parameters"]
}
for name, tool in self.tools.items()
]
def execute_tool(self, tool_name: str, arguments: Dict) -> Any:
"""
Ejecutar herramienta solicitada por LLM
"""
if tool_name not in self.tools:
raise ValueError(f"Tool '{tool_name}' not found")
tool = self.tools[tool_name]
return tool["function"](**arguments)
def get_resource(self, resource_id: str) -> Any:
"""
Obtener recurso solicitado por LLM
"""
if resource_id not in self.resources:
raise ValueError(f"Resource '{resource_id}' not found")
return self.resources[resource_id]()
# Ejemplo: MCP Server para gestión de tareas
task_database = []
def add_task(title: str, priority: str = "medium") -> Dict:
"""Añadir nueva tarea"""
task = {
"id": len(task_database) + 1,
"title": title,
"priority": priority,
"completed": False
}
task_database.append(task)
return task
def list_tasks(filter_by: str = "all") -> List[Dict]:
"""Listar tareas"""
if filter_by == "completed":
return [t for t in task_database if t["completed"]]
elif filter_by == "pending":
return [t for t in task_database if not t["completed"]]
return task_database
def complete_task(task_id: int) -> Dict:
"""Marcar tarea como completada"""
for task in task_database:
if task["id"] == task_id:
task["completed"] = True
return task
raise ValueError(f"Task {task_id} not found")
# Crear servidor MCP
mcp_server = MCPServer("TaskManagementServer")
# Registrar herramientas
mcp_server.register_tool(
"add_task",
add_task,
"Add a new task to the task list",
{
"type": "object",
"properties": {
"title": {"type": "string", "description": "Task title"},
"priority": {"type": "string", "enum": ["low", "medium", "high"]}
},
"required": ["title"]
}
)
mcp_server.register_tool(
"list_tasks",
list_tasks,
"List all tasks or filter by status",
{
"type": "object",
"properties": {
"filter_by": {"type": "string", "enum": ["all", "completed", "pending"]}
}
}
)
mcp_server.register_tool(
"complete_task",
complete_task,
"Mark a task as completed",
{
"type": "object",
"properties": {
"task_id": {"type": "integer", "description": "ID of task to complete"}
},
"required": ["task_id"]
}
)
# Registrar recursos
mcp_server.register_resource(
"all_tasks",
lambda: {"tasks": task_database, "count": len(task_database)}
)
# Uso con LLM
print("🛠️ Available tools:")
for tool in mcp_server.list_tools():
print(f" - {tool['name']}: {tool['description']}")
# Simular LLM llamando herramientas
print("\n📝 Adding tasks...")
result1 = mcp_server.execute_tool("add_task", {"title": "Write documentation", "priority": "high"})
result2 = mcp_server.execute_tool("add_task", {"title": "Review code"})
print(f"Added: {result1}")
print(f"Added: {result2}")
print("\n📋 Listing tasks...")
tasks = mcp_server.execute_tool("list_tasks", {"filter_by": "pending"})
print(f"Pending tasks: {tasks}")
print("\n✅ Completing task...")
completed = mcp_server.execute_tool("complete_task", {"task_id": 1})
print(f"Completed: {completed}")
print("\n📊 Resource: all_tasks")
resource_data = mcp_server.get_resource("all_tasks")
print(resource_data)Integración MCP con Claude:
from anthropic import Anthropic
from typing import List, Dict, Any
class MCPClaude Integration:
"""
Integrar MCP Server con Claude
"""
def __init__(self, mcp_server: MCPServer, anthropic_api_key: str):
self.mcp_server = mcp_server
self.client = Anthropic(api_key=anthropic_api_key)
def chat_with_tools(
self,
user_message: str,
conversation_history: List[Dict] = None
) -> str:
"""
Chat con Claude usando herramientas MCP
"""
if conversation_history is None:
conversation_history = []
# Añadir mensaje del usuario
conversation_history.append({
"role": "user",
"content": user_message
})
# Convertir herramientas MCP a formato Anthropic
tools = self._convert_tools_to_anthropic_format()
# Llamar a Claude
response = self.client.messages.create(
model="claude-sonnet-4-20250514",
max_tokens=2000,
tools=tools,
messages=conversation_history
)
# Procesar respuesta
if response.stop_reason == "tool_use":
# Claude quiere usar una herramienta
tool_results = []
for content_block in response.content:
if content_block.type == "tool_use":
tool_name = content_block.name
tool_input = content_block.input
print(f"🔧 Claude wants to use: {tool_name}")
print(f" Arguments: {tool_input}")
# Ejecutar herramienta
try:
result = self.mcp_server.execute_tool(tool_name, tool_input)
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": str(result)
})
except Exception as e:
tool_results.append({
"type": "tool_result",
"tool_use_id": content_block.id,
"content": f"Error: {str(e)}",
"is_error": True
})
# Añadir resultados al historial
conversation_history.append({
"role": "assistant",
"content": response.content
})
conversation_history.append({
"role": "user",
"content": tool_results
})
# Recursivamente obtener respuesta final
return self.chat_with_tools("", conversation_history)
else:
# Respuesta final de Claude
return response.content[0].text
def _convert_tools_to_anthropic_format(self) -> List[Dict]:
"""
Convertir herramientas MCP a formato Anthropic
"""
anthropic_tools = []
for tool_name, tool_info in self.mcp_server.tools.items():
anthropic_tools.append({
"name": tool_name,
"description": tool_info["description"],
"input_schema": tool_info["parameters"]
})
return anthropic_tools
# Uso
# integration = MCPClaudeIntegration(mcp_server, "your-anthropic-api-key")
# response = integration.chat_with_tools(
# "Add a high priority task called 'Fix production bug' and then show me all pending tasks"
# )
# print("💬 Claude's response:")
# print(response)Receta 6.12: Fine-tuning vs Embeddings vs Prompting
Comparación de estrategias:
| Estrategia | Costo | Tiempo | Casos de uso | Mantenimiento |
|---|---|---|---|---|
| Prompting | 🟢 Bajo | 🟢 Minutos | Mayoría de casos | 🟢 Fácil |
| Embeddings + RAG | 🟡 Medio | 🟡 Horas | Knowledge base | 🟡 Medio |
| Fine-tuning | 🔴 Alto | 🔴 Días | Estilo/formato muy específico | 🔴 Complejo |
Árbol de decisión:
class AIStrategySelector:
"""
Seleccionar estrategia de IA apropiada
"""
@staticmethod
def select_strategy(use_case: Dict) -> str:
"""
Decidir entre prompting, RAG, o fine-tuning
Args:
use_case: {
"has_custom_data": bool,
"data_changes_frequently": bool,
"needs_specific_format": bool,
"budget": "low"|"medium"|"high",
"latency_sensitive": bool
}
"""
# PASO 1: ¿Tienes datos privados/custom?
if not use_case.get("has_custom_data", False):
return "PROMPTING (zero/few-shot)"
# PASO 2: ¿Los datos cambian frecuentemente?
if use_case.get("data_changes_frequently", False):
return "RAG (embeddings + vector DB)"
# PASO 3: ¿Necesitas formato muy específico o estilo único?
if use_case.get("needs_specific_format", False):
if use_case.get("budget") == "high":
return "FINE-TUNING"
else:
return "RAG con ejemplos (few-shot in context)"
# PASO 4: Default
return "RAG (mejor balance costo/beneficio)"
@staticmethod
def explain_strategy(strategy: str) -> Dict:
"""
Explicar estrategia seleccionada
"""
explanations = {
"PROMPTING (zero/few-shot)": {
"descripción": "Usar LLM base con buenos prompts",
"pros": [
"Costo más bajo",
"Implementación inmediata",
"Fácil de iterar",
"No requiere datos de entrenamiento"
],
"cons": [
"Limitado por context window",
"Puede alucinar sin datos correctos",
"No aprende patrones complejos"
],
"ejemplo": "Clasificación simple, Q&A general, generación de texto"
},
"RAG (embeddings + vector DB)": {
"descripción": "Buscar info relevante y pasarla al LLM",
"pros": [
"Usa datos privados/actuales",
"Reduce alucinaciones",
"Fácil actualizar datos",
"Provee fuentes",
"Balance costo/beneficio"
],
"cons": [
"Requiere vector DB",
"Calidad depende de retrieval",
"Latencia adicional (búsqueda)",
"No cambia comportamiento del modelo"
],
"ejemplo": "Chatbot con docs empresa, Q&A sobre productos, soporte técnico"
},
"FINE-TUNING": {
"descripción": "Re-entrenar modelo con tus datos",
"pros": [
"Modelo aprende patrones específicos",
"Mejor para formato/estilo único",
"Puede reducir costo de inferencia (prompts más cortos)",
"Mejora en tareas muy específicas"
],
"cons": [
"Muy costoso (tiempo + $)",
"Requiere dataset grande (1000s ejemplos)",
"Difícil de mantener actualizado",
"Riesgo de overfitting",
"No añade conocimiento nuevo (usa data de entrenamiento)"
],
"ejemplo": "Estilo de escritura muy específico, formato JSON complejo, comportamiento único"
}
}
return explanations.get(strategy, {"descripción": "Estrategia no reconocida"})
# Uso
use_cases = [
{
"name": "Customer support chatbot",
"has_custom_data": True,
"data_changes_frequently": True,
"needs_specific_format": False,
"budget": "medium",
"latency_sensitive": True
},
{
"name": "Code generator",
"has_custom_data": False,
"data_changes_frequently": False,
"needs_specific_format": True,
"budget": "low",
"latency_sensitive": True
},
{
"name": "Legal document analyzer",
"has_custom_data": True,
"data_changes_frequently": False,
"needs_specific_format": True,
"budget": "high",
"latency_sensitive": False
}
]
for use_case in use_cases:
print(f"\n📌 Use case: {use_case['name']}")
strategy = AIStrategySelector.select_strategy(use_case)
print(f"✅ Recommended: {strategy}")
explanation = AIStrategySelector.explain_strategy(strategy)
print(f"\n📖 Why?")
print(f" {explanation['descripción']}")
print(f"\n Pros: {', '.join(explanation['pros'][:2])}")Cuándo usar cada estrategia - Guía rápida:
def strategy_quick_guide():
"""
Guía rápida de decisión
"""
guide = """
┌─────────────────────────────────────────────────────────────┐
│ GUÍA RÁPIDA: PROMPTING vs RAG vs FINE-TUNING │
└─────────────────────────────────────────────────────────────┘
🟢 USA PROMPTING si:
✓ No tienes datos privados
✓ Tarea general (traducción, resumen, clasificación simple)
✓ Quieres empezar rápido
✓ Presupuesto limitado
Ejemplos:
- "Traduce este texto al español"
- "Resume este artículo en 3 puntos"
- "Clasifica el sentimiento de este comentario"
🟡 USA RAG si:
✓ Tienes docs/datos privados o actualizados
✓ Necesitas reducir alucinaciones
✓ Quieres citar fuentes
✓ Datos cambian frecuentemente
Ejemplos:
- Chatbot con docs de empresa
- Q&A sobre productos actuales
- Búsqueda en base de conocimiento
- "¿Qué dice nuestra política sobre X?"
🔴 USA FINE-TUNING si:
✓ Necesitas formato MUY específico consistentemente
✓ Tienes 1000s de ejemplos de calidad
✓ Presupuesto alto
✓ Estilo de output muy particular
✓ Ninguna otra estrategia funciona
Ejemplos:
- Generar SQL queries en dialecto específico
- Estilo de escritura corporativo único
- Formato JSON complejo con reglas específicas
- Chatbot con personalidad muy definida
⚡ COMBINACIONES:
✓ RAG + Prompting: Más común, mejor balance
✓ Fine-tuning + RAG: Para casos muy especializados
✓ Prompting → RAG → Fine-tuning: Ruta de evolución típica
🎯 REGLA GENERAL:
Empieza con PROMPTING → si no es suficiente, añade RAG
→ solo si todavía no funciona, considera FINE-TUNING
"""
print(guide)
strategy_quick_guide()