🤖 Developer Cookbook - FASE 6: Inteligencia Artificial y Colaboración con IA

Recetas prácticas para integrar IA en tus aplicaciones y colaborar efectivamente con LLMs

📚 Tabla de Contenidos

1. Fundamentos de IA/ML
2. Ingeniería de Prompts
3. Integración de IA en Aplicaciones

Fundamentos de IA/ML

Receta 6.1: ML vs Deep Learning vs LLMs - ¿Qué son?

Definiciones:

Jerarquía:

Inteligencia Artificial (IA)
    └── Machine Learning (ML)
            └── Deep Learning (DL)
                    └── Large Language Models (LLM)

Comparación:

Ejemplo de diferencias:

# 1. Machine Learning Tradicional (Scikit-learn)
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
import pandas as pd

# Datos de entrenamiento
df = pd.read_csv('customer_data.csv')
X = df[['age', 'income', 'purchase_history']]
y = df['will_buy']

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

# Modelo tradicional ML
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# Predicción
prediction = model.predict([[35, 75000, 5]])
print(f"¿Comprará? {prediction[0]}")

# 2. Deep Learning (TensorFlow/Keras)
import tensorflow as tf
from tensorflow import keras

# Red neuronal para clasificación de imágenes
model = keras.Sequential([
    keras.layers.Conv2D(32, (3,3), activation='relu', input_shape=(28, 28, 1)),
    keras.layers.MaxPooling2D((2,2)),
    keras.layers.Conv2D(64, (3,3), activation='relu'),
    keras.layers.MaxPooling2D((2,2)),
    keras.layers.Flatten(),
    keras.layers.Dense(64, activation='relu'),
    keras.layers.Dense(10, activation='softmax')
])

model.compile(optimizer='adam',
              loss='sparse_categorical_crossentropy',
              metrics=['accuracy'])

# Entrenamiento
model.fit(train_images, train_labels, epochs=5)

# 3. LLMs (API de OpenAI/Anthropic)
import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

# Usar LLM para clasificación de texto
message = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=100,
    messages=[{
        "role": "user",
        "content": "¿Este comentario es positivo o negativo? 'El producto es terrible, no funciona'"
    }]
)

print(message.content[0].text)

Receta 6.2: Entrenamiento, Inferencia y Fine-tuning

Definiciones:

Entrenamiento (Training):

• Proceso de enseñar al modelo usando datos
• Actualiza los pesos/parámetros del modelo
• Computacionalmente costoso (días/semanas)
• Requiere dataset grande

# Ejemplo conceptual de entrenamiento
import torch
import torch.nn as nn

# Modelo simple
model = nn.Sequential(
    nn.Linear(10, 50),
    nn.ReLU(),
    nn.Linear(50, 1)
)

optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.MSELoss()

# Loop de entrenamiento
for epoch in range(100):
    for batch_x, batch_y in train_loader:
        # Forward pass
        predictions = model(batch_x)
        loss = criterion(predictions, batch_y)
        
        # Backward pass (actualiza pesos)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    
    print(f"Epoch {epoch}, Loss: {loss.item()}")

Inferencia (Inference):

• Usar el modelo entrenado para hacer predicciones
• NO actualiza pesos
• Rápido (milisegundos/segundos)
• Solo requiere modelo pre-entrenado

# Inferencia: usar modelo ya entrenado
model.eval()  # Modo evaluación (no training)

with torch.no_grad():  # No calcular gradientes
    prediction = model(new_data)
    print(f"Predicción: {prediction}")

Fine-tuning:

• Tomar modelo pre-entrenado y “ajustarlo” con datos específicos
• Más rápido y barato que entrenar desde cero
• Requiere menos datos (cientos/miles vs millones)

# Fine-tuning conceptual
# 1. Cargar modelo pre-entrenado
from transformers import AutoModelForSequenceClassification, AutoTokenizer

model = AutoModelForSequenceClassification.from_pretrained("bert-base-uncased")
tokenizer = AutoTokenizer.from_pretrained("bert-base-uncased")

# 2. Congelar capas tempranas (opcional)
for param in model.bert.encoder.layer[:8].parameters():
    param.requires_grad = False

# 3. Entrenar solo las últimas capas con tus datos
# (código de training aquí)

Comparación:

Receta 6.3: Sesgos Algorítmicos y Ética en IA

¿Qué son los sesgos algorítmicos? Cuando un modelo de IA produce resultados sistemáticamente injustos o discriminatorios hacia ciertos grupos.

Tipos de sesgos:

1. Sesgo en datos de entrenamiento:

# Ejemplo: Dataset de contratación sesgado
training_data = [
    {"name": "John", "experience": 5, "hired": True},
    {"name": "Michael", "experience": 5, "hired": True},
    {"name": "Mary", "experience": 5, "hired": False},
    {"name": "Jennifer", "experience": 5, "hired": False},
]
# Problema: El modelo aprenderá a preferir hombres

2. Sesgo de representación:

# Dataset de reconocimiento facial
face_dataset = {
    "white_males": 10000,  # Sobre-representado
    "white_females": 8000,
    "black_males": 500,    # Sub-representado
    "black_females": 300,
}
# Resultado: Peor performance en grupos minoritarios

3. Sesgo de confirmación:

# Sistema de recomendación
def recommend_news(user_history):
    # Problema: Solo recomienda noticias similares
    # Crea cámara de eco
    return similar_articles(user_history)

Cómo detectar y mitigar sesgos:

import pandas as pd
from sklearn.metrics import confusion_matrix
import numpy as np

def analyze_fairness(model, X_test, y_test, sensitive_attribute):
    """
    Analizar si modelo es justo entre diferentes grupos
    """
    # Hacer predicciones
    y_pred = model.predict(X_test)
    
    # Agrupar por atributo sensible (ej: género, raza)
    groups = X_test[sensitive_attribute].unique()
    
    results = {}
    for group in groups:
        # Filtrar por grupo
        mask = X_test[sensitive_attribute] == group
        y_true_group = y_test[mask]
        y_pred_group = y_pred[mask]
        
        # Calcular métricas
        from sklearn.metrics import accuracy_score, precision_score
        
        results[group] = {
            'accuracy': accuracy_score(y_true_group, y_pred_group),
            'precision': precision_score(y_true_group, y_pred_group),
            'sample_size': len(y_true_group),
            'positive_rate': y_pred_group.mean()
        }
    
    # Comparar disparidad entre grupos
    df_results = pd.DataFrame(results).T
    print("\n📊 Análisis de equidad:")
    print(df_results)
    
    # Calcular disparate impact
    max_positive_rate = df_results['positive_rate'].max()
    min_positive_rate = df_results['positive_rate'].min()
    disparate_impact = min_positive_rate / max_positive_rate
    
    print(f"\n⚖️ Disparate Impact: {disparate_impact:.2f}")
    print("   (< 0.8 indica posible discriminación)")
    
    return df_results

# Uso
# analyze_fairness(model, X_test, y_test, sensitive_attribute='gender')

Principios éticos:

class EthicalAIChecklist:
    """Checklist de ética para proyectos de IA"""
    
    @staticmethod
    def evaluate_project(project_info):
        checks = {
            "transparency": "¿Los usuarios saben que están interactuando con IA?",
            "explainability": "¿Puedes explicar por qué el modelo tomó una decisión?",
            "fairness": "¿El modelo trata a todos los grupos equitativamente?",
            "privacy": "¿Proteges los datos personales adecuadamente?",
            "accountability": "¿Hay un humano responsable de las decisiones?",
            "safety": "¿Has evaluado posibles daños?",
            "human_oversight": "¿Hay supervisión humana en decisiones críticas?",
        }
        
        print("🤔 Checklist Ético de IA:\n")
        for key, question in checks.items():
            answer = input(f"{question} (s/n): ")
            if answer.lower() != 's':
                print(f"⚠️ ADVERTENCIA: Considera revisar {key}")
        
        print("\n✅ Evaluación ética completada")

# Uso
# EthicalAIChecklist.evaluate_project(my_ai_project)

Casos de uso reales con problemas éticos:

Receta 6.4: Alucinaciones y Limitaciones de LLMs

¿Qué son las alucinaciones? Cuando un LLM genera información falsa o inventada con confianza, como si fuera real.

Tipos de alucinaciones:

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# 1. Alucinaciones factuales
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=100,
    messages=[{
        "role": "user",
        "content": "¿Cuándo murió el presidente Joe Biden?"
    }]
)
# Problema: Puede inventar una fecha (Joe Biden está vivo)

# 2. Alucinaciones de fuentes
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=200,
    messages=[{
        "role": "user",
        "content": "Dame la cita exacta donde Einstein dijo 'La imaginación es más importante que el conocimiento' con la fuente"
    }]
)
# Problema: Puede inventar una fuente específica

# 3. Alucinaciones en código
response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=300,
    messages=[{
        "role": "user",
        "content": "Muéstrame cómo usar la función pandas.magical_transform()"
    }]
)
# Problema: Esta función no existe, pero puede inventar documentación

Cómo detectar alucinaciones:

def detect_hallucination_risks(prompt: str) -> dict:
    """
    Identificar si un prompt tiene riesgo alto de alucinaciones
    """
    risks = {
        'factual_claims': False,
        'specific_dates': False,
        'citations_needed': False,
        'technical_specifics': False,
        'statistics': False,
    }
    
    # Detectar patrones de riesgo
    import re
    
    # Preguntas sobre hechos específicos
    if re.search(r'(cuándo|dónde|quién|qué año)', prompt.lower()):
        risks['factual_claims'] = True
    
    # Solicitudes de fechas
    if re.search(r'\b(fecha|día|año|mes)\b', prompt.lower()):
        risks['specific_dates'] = True
    
    # Solicitudes de citas o fuentes
    if re.search(r'(cita|fuente|referencia|estudio)', prompt.lower()):
        risks['citations_needed'] = True
    
    # APIs o funciones específicas
    if re.search(r'(función|método|API|clase)\s+\w+\.\w+', prompt):
        risks['technical_specifics'] = True
    
    # Estadísticas
    if re.search(r'(porcentaje|estadística|cuántos)', prompt.lower()):
        risks['statistics'] = True
    
    # Evaluar riesgo total
    risk_count = sum(risks.values())
    
    return {
        'risks': risks,
        'risk_level': 'HIGH' if risk_count >= 3 else 'MEDIUM' if risk_count >= 1 else 'LOW',
        'recommendation': 'Verificar respuesta' if risk_count >= 1 else 'Probablemente seguro'
    }

# Uso
prompt = "¿Cuándo murió Einstein y qué dijo en su lecho de muerte?"
analysis = detect_hallucination_risks(prompt)
print(f"Riesgo: {analysis['risk_level']}")
print(f"Recomendación: {analysis['recommendation']}")

Técnicas para reducir alucinaciones:

# 1. Pedir al modelo que admita incertidumbre
prompt_safe = """
Responde la siguiente pregunta. Si no estás seguro o no tienes información
confiable, di explícitamente "No tengo información confiable sobre esto".

Pregunta: ¿Cuál es el PIB de Andorra en 2024?
"""

# 2. Proporcionar contexto (RAG - Retrieval Augmented Generation)
from typing import List

def query_with_context(question: str, context_docs: List[str]) -> str:
    """
    Usar RAG para reducir alucinaciones
    """
    # Construir prompt con contexto
    context = "\n\n".join([f"Documento {i+1}:\n{doc}" 
                           for i, doc in enumerate(context_docs)])
    
    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".

{context}

Pregunta: {question}
"""
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=500,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text

# 3. Verificación de hechos post-generación
def verify_response(response: str, knowledge_base: dict) -> dict:
    """
    Verificar afirmaciones en la respuesta
    """
    import re
    
    # Extraer afirmaciones numéricas
    numbers = re.findall(r'\d+(?:\.\d+)?', response)
    dates = re.findall(r'\b(19|20)\d{2}\b', response)
    
    verification = {
        'response': response,
        'numbers_found': numbers,
        'dates_found': dates,
        'warnings': []
    }
    
    # Verificar contra knowledge base
    for claim, truth in knowledge_base.items():
        if claim in response and truth not in response:
            verification['warnings'].append(
                f"⚠️ Posible error: se menciona '{claim}' pero debería ser '{truth}'"
            )
    
    return verification

# Uso
knowledge_base = {
    "Einstein murió": "1955",
    "Newton nació": "1643"
}

response_text = "Einstein murió en 1950"  # Incorrecto
verification = verify_response(response_text, knowledge_base)
print(verification['warnings'])

Limitaciones de los LLMs:

class LLMLimitations:
    """
    Documentar limitaciones conocidas de LLMs
    """
    
    LIMITATIONS = {
        "knowledge_cutoff": {
            "descripción": "No saben eventos después de su fecha de entrenamiento",
            "ejemplo": "No saben quién ganó elecciones recientes",
            "solución": "Usar web search o proporcionar contexto actualizado"
        },
        "arithmetic": {
            "descripción": "Pueden cometer errores en cálculos complejos",
            "ejemplo": "234 * 567 = ? (puede equivocarse)",
            "solución": "Usar herramientas externas (calculadoras, Python)"
        },
        "context_length": {
            "descripción": "Límite en tokens que pueden procesar",
            "ejemplo": "No pueden analizar libro completo de 500 páginas",
            "solución": "Chunking, summarización incremental"
        },
        "inconsistency": {
            "descripción": "Pueden dar respuestas diferentes a la misma pregunta",
            "ejemplo": "Pregunta X puede tener respuesta Y o Z",
            "solución": "Temperature=0 para más determinismo, múltiples samples"
        },
        "no_learning": {
            "descripción": "No aprenden de conversaciones pasadas (sin fine-tuning)",
            "ejemplo": "Olvidan lo que les dijiste en chats anteriores",
            "solución": "Embeddings, vector DBs para 'memoria'"
        },
        "visual_limitations": {
            "descripción": "Limitaciones en tareas visuales complejas",
            "ejemplo": "Contar objetos exactos en imagen",
            "solución": "Modelos especializados (YOLO, SAM)"
        }
    }
    
    @classmethod
    def check_limitations(cls, task_description: str):
        """
        Verificar si una tarea puede tener limitaciones
        """
        print(f"🔍 Analizando tarea: '{task_description}'\n")
        
        warnings = []
        
        # Verificar diferentes limitaciones
        if any(keyword in task_description.lower() for keyword in ['calcular', 'multiplicar', 'sumar']):
            warnings.append(cls.LIMITATIONS['arithmetic'])
        
        if any(keyword in task_description.lower() for keyword in ['reciente', 'actual', 'hoy', '2024', '2025']):
            warnings.append(cls.LIMITATIONS['knowledge_cutoff'])
        
        if any(keyword in task_description.lower() for keyword in ['libro', 'documento largo', 'análisis completo']):
            warnings.append(cls.LIMITATIONS['context_length'])
        
        if warnings:
            print("⚠️ Limitaciones detectadas:\n")
            for i, warning in enumerate(warnings, 1):
                print(f"{i}. {warning['descripción']}")
                print(f"   Ejemplo: {warning['ejemplo']}")
                print(f"   Solución: {warning['solución']}\n")
        else:
            print("✅ No se detectaron limitaciones obvias")

# Uso
LLMLimitations.check_limitations("Analiza este documento de 1000 páginas y dame las ventas de enero 2025")

Ingeniería de Prompts

Receta 6.5: Zero-shot, Few-shot, Chain-of-Thought

Definiciones:

Ejemplos prácticos:

from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

# 1. ZERO-SHOT: Sin ejemplos
def zero_shot_example():
    """Clasificación sin ejemplos"""
    prompt = """
Clasifica el siguiente comentario como POSITIVO, NEGATIVO o NEUTRAL:

"El producto llegó a tiempo pero la calidad es mediocre"
"""
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=50,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text

# 2. FEW-SHOT: Con ejemplos
def few_shot_example():
    """Clasificación con ejemplos"""
    prompt = """
Clasifica comentarios de productos como POSITIVO, NEGATIVO o NEUTRAL.

Ejemplos:

Comentario: "¡Excelente producto! Superó mis expectativas"
Clasificación: POSITIVO

Comentario: "Terrible, se rompió al primer uso"
Clasificación: NEGATIVO

Comentario: "Es un producto estándar, nada especial"
Clasificación: NEUTRAL

Comentario: "Buen precio pero la entrega tardó mucho"
Clasificación: NEUTRAL

Ahora clasifica este:
Comentario: "El producto llegó a tiempo pero la calidad es mediocre"
Clasificación:
"""
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=50,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text

# 3. CHAIN-OF-THOUGHT: Razonamiento paso a paso
def chain_of_thought_example():
    """Resolver problema con razonamiento"""
    prompt = """
Resuelve el siguiente problema PASO A PASO:

"Si un tren viaja a 120 km/h y necesita recorrer 360 km, pero hace una parada 
de 30 minutos a mitad de camino, ¿cuánto tiempo total tardará el viaje?"

Piensa paso a paso:
1. Primero, calcula el tiempo de viaje sin paradas
2. Luego, suma el tiempo de la parada
3. Finalmente, da la respuesta total
"""
    
    response = client.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=300,
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.content[0].text

# Ejecutar ejemplos
print("=== ZERO-SHOT ===")
print(zero_shot_example())

print("\n=== FEW-SHOT ===")
print(few_shot_example())

print("\n=== CHAIN-OF-THOUGHT ===")
print(chain_of_thought_example())

Comparación de efectividad:

import time

def benchmark_prompting_techniques(test_cases: list):
    """
    Comparar efectividad de diferentes técnicas
    """
    results = {
        'zero_shot': {'correct': 0, 'total': 0, 'avg_time': 0},
        'few_shot': {'correct': 0, 'total': 0, 'avg_time': 0},
        'cot': {'correct': 0, 'total': 0, 'avg_time': 0}
    }
    
    for test_case in test_cases:
        question = test_case['question']
        expected = test_case['answer']
        
        # Zero-shot
        start = time.time()
        zero_shot_answer = get_zero_shot_answer(question)
        results['zero_shot']['total'] += 1
        results['zero_shot']['avg_time'] += time.time() - start
        if zero_shot_answer == expected:
            results['zero_shot']['correct'] += 1
        
        # Few-shot
        start = time.time()
        few_shot_answer = get_few_shot_answer(question)
        results['few_shot']['total'] += 1
        results['few_shot']['avg_time'] += time.time() - start
        if few_shot_answer == expected:
            results['few_shot']['correct'] += 1
        
        # Chain-of-Thought
        start = time.time()
        cot_answer = get_cot_answer(question)
        results['cot']['total'] += 1
        results['cot']['avg_time'] += time.time() - start
        if cot_answer == expected:
            results['cot']['correct'] += 1
    
    # Calcular métricas
    for technique in results:
        if results[technique]['total'] > 0:
            results[technique]['accuracy'] = (
                results[technique]['correct'] / results[technique]['total']
            )
            results[technique]['avg_time'] /= results[technique]['total']
    
    return results

Cuándo usar cada técnica:

class PromptingStrategy:
    """
    Seleccionar estrategia de prompting automáticamente
    """
    
    @staticmethod
    def select_strategy(task_type: str, complexity: str) -> str:
        """
        Recomendar técnica de prompting basada en tarea
        """
        strategies = {
            # Tareas simples
            ('classification', 'low'): 'zero_shot',
            ('translation', 'low'): 'zero_shot',
            ('summarization', 'low'): 'zero_shot',
            
            # Tareas con formato específico
            ('extraction', 'medium'): 'few_shot',
            ('formatting', 'medium'): 'few_shot',
            ('style_transfer', 'medium'): 'few_shot',
            
            # Tareas complejas
            ('math', 'high'): 'chain_of_thought',
            ('reasoning', 'high'): 'chain_of_thought',
            ('planning', 'high'): 'chain_of_thought',
            ('analysis', 'high'): 'chain_of_thought',
        }
        
        key = (task_type, complexity)
        return strategies.get(key, 'few_shot')  # Default
    
    @staticmethod
    def build_prompt(strategy: str, task: str, examples: list = None) -> str:
        """
        Construir prompt según estrategia
        """
        if strategy == 'zero_shot':
            return f"Realiza la siguiente tarea:\n\n{task}"
        
        elif strategy == 'few_shot':
            prompt = "Aquí hay algunos ejemplos:\n\n"
            for ex in examples:
                prompt += f"Input: {ex['input']}\nOutput: {ex['output']}\n\n"
            prompt += f"Ahora realiza:\nInput: {task}\nOutput:"
            return prompt
        
        elif strategy == 'chain_of_thought':
            return f"""
Resuelve paso a paso:

{task}

Muestra tu razonamiento:
Paso 1:
Paso 2:
...
Respuesta final:
"""
        
        return task

# Uso
task_type = 'math'
complexity = 'high'

strategy = PromptingStrategy.select_strategy(task_type, complexity)
print(f"Estrategia recomendada: {strategy}")

prompt = PromptingStrategy.build_prompt(
    strategy, 
    "Si compro 3 productos a $15 cada uno y tengo un descuento del 20%, ¿cuánto pago?"
)
print(f"\nPrompt generado:\n{prompt}")

Receta 6.6: Optimización de Prompts para Código

Técnicas para mejorar generación de código:

# ❌ PROMPT MALO: Vago e impreciso
bad_prompt = "Escribe código para API"

# ✅ PROMPT BUENO: Específico y estructurado
good_prompt = """
Crea un endpoint REST API en Python usando Flask con las siguientes especificaciones:

Endpoint: POST /api/users
Funcionalidad: Crear nuevo usuario en base de datos
Input JSON:
{
  "name": "string (requerido)",
  "email": "string (requerido, debe ser email válido)",
  "age": "integer (opcional, debe ser > 0)"
}

Requisitos:
1. Validar input con pydantic
2. Hash password con bcrypt
3. Guardar en SQLite usando SQLAlchemy
4. Retornar 201 con ID del usuario creado
5. Manejar errores (email duplicado, validación)
6. Incluir docstring y type hints

Formato de respuesta exitosa:
{
  "id": 123,
  "message": "Usuario creado exitosamente"
}
"""

# Ejecutar con LLM
from anthropic import Anthropic

client = Anthropic(api_key="your-api-key")

response = client.messages.create(
    model="claude-sonnet-4-20250514",
    max_tokens=2000,
    messages=[{"role": "user", "content": good_prompt}]
)

print(response.content[0].text)

Template para prompts de código:

class CodePromptBuilder:
    """
    Constructor de prompts estructurados para generación de código
    """
    
    def __init__(self):
        self.template = """
Genera código {language} con las siguientes especificaciones:

📋 DESCRIPCIÓN:
{description}

🎯 FUNCIONALIDAD:
{functionality}

📥 INPUT:
{input_spec}

📤 OUTPUT:
{output_spec}

✅ REQUISITOS:
{requirements}

📚 DEPENDENCIAS:
{dependencies}

⚠️ MANEJO DE ERRORES:
{error_handling}

📝 ADICIONAL:
{additional_notes}
"""
    
    def build(self,
              language: str,
              description: str,
              functionality: str,
              input_spec: str,
              output_spec: str,
              requirements: list,
              dependencies: list = None,
              error_handling: str = "Manejar excepciones comunes",
              additional_notes: str = "Incluir type hints y docstrings") -> str:
        """
        Construir prompt estructurado
        """
        requirements_str = "\n".join([f"{i+1}. {req}" for i, req in enumerate(requirements)])
        dependencies_str = ", ".join(dependencies) if dependencies else "Librería estándar"
        
        return self.template.format(
            language=language,
            description=description,
            functionality=functionality,
            input_spec=input_spec,
            output_spec=output_spec,
            requirements=requirements_str,
            dependencies=dependencies_str,
            error_handling=error_handling,
            additional_notes=additional_notes
        )

# Uso
builder = CodePromptBuilder()

prompt = builder.build(
    language="Python",
    description="Sistema de cache con expiración automática",
    functionality="Almacenar valores en memoria con TTL configurable",
    input_spec="key (str), value (Any), ttl_seconds (int)",
    output_spec="None al guardar, valor o None al recuperar",
    requirements=[
        "Usar threading para expiración automática",
        "Thread-safe con locks",
        "Método get() que retorna None si expiró",
        "Método set() para guardar con TTL",
        "Método clear() para limpiar todo"
    ],
    dependencies=["threading", "time"],
    error_handling="Validar tipos de datos, manejar race conditions",
    additional_notes="Incluir tests unitarios con pytest"
)

print(prompt)

Prompts para debugging:

def generate_debug_prompt(buggy_code: str, error_message: str, expected_behavior: str) -> str:
    """
    Generar prompt para debugging de código
    """
    prompt = f"""
🐛 DEBUGGING ASSISTANCE

CÓDIGO CON ERROR:
```python
{buggy_code}

❌ ERROR RECIBIDO: {error_message}

✅ COMPORTAMIENTO ESPERADO: {expected_behavior}

Por favor:

1. Identifica el error exacto

2. Explica POR QUÉ ocurre el error

3. Proporciona el código corregido

4. Sugiere cómo prevenir este error en el futuro

5. Si es posible, añade un test para verificar la corrección """

   return prompt

Uso

buggy_code = """ def calculate_average(numbers): total = sum(numbers) return total / len(numbers)

result = calculate_average([]) print(result) """

error_msg = “ZeroDivisionError: division by zero”

debug_prompt = generate_debug_prompt( buggy_code=buggy_code, error_message=error_msg, expected_behavior=“Debería manejar listas vacías sin error” )

Enviar a LLM para debug

response = client.messages.create( model=“claude-sonnet-4-20250514”, max_tokens=1000, messages=[{“role”: “user”, “content”: debug_prompt}] )

print(”🔧 Solución:\n”) print(response.content[0].text)

**Prompts para refactoring:**

```python
def generate_refactor_prompt(code: str, issues: list, style_guide: str = "PEP 8") -> str:
    """
    Generar prompt para refactorización
    """
    issues_str = "\n".join([f"- {issue}" for issue in issues])
    
    prompt = f"""
♻️ REFACTORING REQUEST

CÓDIGO ACTUAL:
```python
{code}

🔍 PROBLEMAS IDENTIFICADOS: {issues_str}

📏 GUÍA DE ESTILO: {style_guide}

Por favor refactoriza el código:

1. Aplica principios SOLID
2. Mejora nombres de variables/funciones
3. Reduce complejidad ciclomática
4. Añade type hints
5. Documenta con docstrings
6. Elimina código duplicado
7. Mejora legibilidad

Explica los cambios realizados y por qué mejoran el código. """

return prompt

Uso

messy_code = """ def f(x): r = [] for i in x: if i > 0: r.append(i * 2) return r """

refactor_prompt = generate_refactor_prompt( code=messy_code, issues=[ “Nombres de variables no descriptivos”, “Falta documentación”, “No hay type hints”, “Lógica puede simplificarse” ] )

print(refactor_prompt)

---

## Receta 6.7: GitHub Copilot, Cursor, ChatGPT - Uso Efectivo

**GitHub Copilot:**

```python
# TIPS para usar Copilot efectivamente:

# 1. Comentarios descriptivos
# ✅ BUENO: Copilot entiende lo que quieres
def validate_email(email: str) -> bool:
    """
    Validate email address using regex.
    Returns True if valid, False otherwise.
    Must check for @ symbol, domain, and TLD.
    """
    # Copilot generará una regex apropiada
    pass

# ❌ MALO: Comentario vago
def check(x):
    # validate
    pass

# 2. Usar docstrings como guía
def calculate_compound_interest(
    principal: float,
    rate: float,
    time: int,
    compounds_per_year: int = 12
) -> float:
    """
    Calculate compound interest using the formula:
    A = P(1 + r/n)^(nt)
    
    Args:
        principal: Initial amount
        rate: Annual interest rate (as decimal, e.g., 0.05 for 5%)
        time: Time in years
        compounds_per_year: Number of times interest compounds per year
    
    Returns:
        Final amount after compound interest
    
    Example:
        >>> calculate_compound_interest(1000, 0.05, 10, 12)
        1647.01
    """
    # Copilot generará implementación basada en docstring
    pass

# 3. Nombrar funciones descriptivamente
def fetch_user_by_email_and_cache_result(email: str):
    """
    Fetch user from database by email, cache the result in Redis
    with 1 hour expiration, and return user object
    """
    # Copilot entenderá: fetch DB -> cache Redis -> return
    pass

# 4. Usar context de archivos relacionados
# Si tienes User model definido en otro archivo:
# from models import User
# 
# Copilot usará el contexto del modelo para sugerir mejor código

Cursor AI:

# TIPS para Cursor:

# 1. Usar Ctrl+K para edits en contexto
# Ejemplo: Selecciona función y pide "Add error handling and logging"

# 2. Usar @-mentions para contexto
# @web "Latest FastAPI best practices 2024"
# @docs "SQLAlchemy relationship examples"
# @codebase "How do we handle authentication?"

# 3. Composer mode para archivos múltiples
# Cursor puede editar varios archivos simultáneamente:
# "Refactor user authentication to use JWT tokens across routes.py, 
#  models.py, and auth.py"

# 4. Chat para explicaciones
# Selecciona código complejo -> Cmd+L -> "Explain this algorithm step by step"

# Ejemplo de uso efectivo:
class UserService:
    def __init__(self, db_session):
        self.db = db_session
    
    # Cursor command: "Add method to get user by email with caching"
    # Cursor generará:
    async def get_user_by_email(self, email: str) -> Optional[User]:
        """Fetch user by email with Redis caching"""
        cache_key = f"user:email:{email}"
        
        # Check cache first
        cached = await redis_client.get(cache_key)
        if cached:
            return User.from_json(cached)
        
        # Query database
        user = await self.db.query(User).filter(User.email == email).first()
        
        if user:
            # Cache for 1 hour
            await redis_client.setex(cache_key, 3600, user.to_json())
        
        return user

ChatGPT/Claude para desarrollo:

# Estrategias de prompting para desarrollo:

# 1. ITERACIÓN: Empieza simple, refina
# Primera iteración:
prompt_v1 = "Create a function to validate passwords"

# Segunda iteración (más específica):
prompt_v2 = """
Create a Python function to validate passwords with these rules:
- At least 8 characters
- At least one uppercase letter
- At least one lowercase letter
- At least one number
- At least one special character (!@#$%^&*)
Return tuple (is_valid: bool, errors: list[str])
"""

# 2. PROPORCIONAR CONTEXTO
context_prompt = """
I have a Flask app with SQLAlchemy. I need to add a feature for users to 
upload profile pictures. The images should be:
- Stored in AWS S3
- Resized to 200x200
- Original filename preserved
- URL saved in User model

Here's my current User model:
```python
class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String(80), unique=True)
    email = db.Column(db.String(120), unique=True)

Generate the complete solution including:

1. Updated User model
2. Upload endpoint
3. S3 integration
4. Image resizing logic """

3. PEDIR EXPLICACIONES

explanation_prompt = """ Explain this code line by line in simple terms:

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

Focus on:

• What is @lru_cache?
• How does it improve performance?
• What are the trade-offs? """

4. COMPARAR ALTERNATIVAS

comparison_prompt = """ Compare these two approaches for handling async file uploads in FastAPI:

Approach 1: BackgroundTasks Approach 2: Celery queue

For each, explain:

• Pros and cons
• When to use
• Sample code
• Performance implications """

5. DEBUGGING ASISTIDO

debug_prompt = """ This code is giving me unexpected results:

def remove_duplicates(numbers):
    return list(set(numbers))

result = remove_duplicates([3, 1, 4, 1, 5, 9, 2, 6, 5])
print(result)  # Expected [3,1,4,5,9,2,6] but order is random

Why is the order random and how do I preserve original order while removing duplicates? Provide solution with explanation. """

**Framework para preguntas efectivas a AI assistants:**

```python
class AIAssistantQuery:
    """
    Template para hacer preguntas efectivas a AI coding assistants
    """
    
    @staticmethod
    def build_query(
        task: str,
        context: str = "",
        constraints: list = None,
        examples: list = None,
        format_preference: str = "code + explanation"
    ) -> str:
        """
        Construir query estructurada
        """
        query = f"🎯 TASK:\n{task}\n\n"
        
        if context:
            query += f"📋 CONTEXT:\n{context}\n\n"
        
        if constraints:
            query += "⚠️ CONSTRAINTS:\n"
            query += "\n".join([f"- {c}" for c in constraints])
            query += "\n\n"
        
        if examples:
            query += "📚 EXAMPLES:\n"
            for i, ex in enumerate(examples, 1):
                query += f"Example {i}:\n{ex}\n\n"
        
        query += f"📄 FORMAT: {format_preference}\n"
        
        return query
    
    @staticmethod
    def code_generation(description: str, language: str, **kwargs) -> str:
        """Helper para generación de código"""
        return AIAssistantQuery.build_query(
            task=f"Generate {language} code for: {description}",
            format_preference="Working code with comments and example usage",
            **kwargs
        )
    
    @staticmethod
    def code_review(code: str, focus_areas: list = None) -> str:
        """Helper para code review"""
        focus = focus_areas or ["bugs", "performance", "readability", "security"]
        return AIAssistantQuery.build_query(
            task="Review this code",
            context=f"```\n{code}\n```",
            constraints=[f"Focus on: {', '.join(focus)}"],
            format_preference="Issues found + suggested improvements"
        )
    
    @staticmethod
    def explain_code(code: str, level: str = "intermediate") -> str:
        """Helper para explicaciones"""
        return AIAssistantQuery.build_query(
            task=f"Explain this code for {level} developer",
            context=f"```\n{code}\n```",
            format_preference="Line-by-line explanation + high-level summary"
        )

# Uso
query = AIAssistantQuery.code_generation(
    description="Rate limiter decorator",
    language="Python",
    constraints=[
        "Use Redis for distributed rate limiting",
        "Support sliding window algorithm",
        "Thread-safe",
        "Must work with async functions"
    ],
    examples=[
        "@rate_limit(max_calls=100, window_seconds=60)\nasync def api_endpoint():\n    pass"
    ]
)

print(query)

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:

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
    ↓
Response

Implementació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:

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 ejecutar

Ejemplo 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)