Embedding y recuperación vectorial

Cómo funcionan los modelos de Embedding

Los modelos de Embedding convierten texto en vectores de alta dimensión, de modo que textos semánticamente similares queden cerca en el espacio vectorial.

"Me gusta comer manzanas"   →  [0.12, -0.34, 0.56, ..., 0.78]  (1024 dimensiones)
"Me encanta comer manzanas"  →  [0.11, -0.33, 0.55, ..., 0.79]  ← muy cercanos
"Manual de reparación de autos" →  [-0.78, 0.45, -0.23, ..., 0.01] ← muy lejanos

El entrenamiento de modelos de Embedding suele incluir una etapa de aprendizaje contrastivo (Contrastive Learning): la entrada consiste en muchos pares de texto etiquetados como relevantes/irrelevantes, y el objetivo de entrenamiento es maximizar la similitud vectorial de los pares relevantes y minimizar la de los irrelevantes.

# Calcular la similitud coseno entre dos vectores de texto
import numpy as np

def cosine_similarity(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# Ejemplo
query_vec = embedding_model.encode("¿Cómo solicitar vacaciones anuales?")
doc_vec = embedding_model.encode("Procedimiento de solicitud de vacaciones anuales para empleados")

similarity = cosine_similarity(query_vec, doc_vec)
print(f"Similitud: {similarity:.4f}")  # 0.92 — altamente relevante

Selección de base de datos vectorial

La base de datos vectorial es la infraestructura central de un sistema RAG. Al elegir, hay que sopesar:

# Ejemplo usando Chroma (base vectorial local ligera)
import chromadb
from chromadb.utils import embedding_functions

client = chromadb.PersistentClient(path="./chroma_db")
collection = client.create_collection(
    name="company_docs",
    embedding_function=embedding_functions.OpenAIEmbeddingFunction(
        api_key="your-api-key",
        model_name="text-embedding-3-small"
    )
)

# Añadir documentos
collection.add(
    documents=["Procedimiento de solicitud de vacaciones anuales...", "Normas de reembolso de viajes..."],
    metadatas=[{"source": "hr_policy.pdf"}, {"source": "finance_policy.pdf"}],
    ids=["doc_1", "doc_2"]
)

# Recuperar
results = collection.query(
    query_texts=["¿Cómo solicitar vacaciones anuales?"],
    n_results=3
)

Estrategias de segmentación de documentos

La contradicción fundamental de la segmentación

La efectividad de la recuperación en un sistema RAG depende en gran medida de la calidad de la segmentación de documentos. La contradicción central es:

Segmentos demasiado grandes → introducen demasiado ruido en la recuperación, la atención del modelo se diluye
Segmentos demasiado pequeños → la información clave puede truncarse, se pierde contexto

No existe una estrategia de segmentación óptima universal. Debes elegir según el tipo de documento, el escenario de recuperación y la capacidad del modelo.

Cinco métodos principales de segmentación

Segmentación por Token

Segmenta por número fijo de tokens, adecuado para escenarios que requieren control preciso del consumo de tokens.

from llama_index.core.node_parser import TokenTextSplitter

splitter = TokenTextSplitter(
    chunk_size=256,     # número de tokens por segmento
    chunk_overlap=30    # tokens de solapamiento entre segmentos adyacentes
)

nodes = splitter.get_nodes_from_documents(documents)

Ventajas: control preciso del tamaño del contexto, adecuado para modelos con ventanas de contexto más pequeñas. Desventajas: puede cortar en medio de una oración, rompiendo la integridad semántica.

Segmentación por oración

Mantiene la integridad de las oraciones; es la opción por defecto en la mayoría de escenarios.

from llama_index.core.node_parser import SentenceSplitter

splitter = SentenceSplitter(
    chunk_size=512,
    chunk_overlap=50
)

Ventajas: preserva la integridad de las unidades semánticas del lenguaje natural. Desventajas: no percibe la estructura del documento, puede separar párrafos relacionados en distintos segmentos.

Segmentación con ventana de oración

Usa diferente granularidad en indexación y recuperación: la indexación usa granularidad fina para coincidencia precisa, y al recuperar se devuelve el segmento junto con su ventana de contexto adyacente.

from llama_index.core.node_parser import SentenceWindowNodeParser

parser = SentenceWindowNodeParser(
    window_size=3,          # expandir 3 oraciones adyacentes al recuperar
    window_metadata_key="window",
    original_text_metadata_key="original"
)

Ventaja principal: equilibra precisión de recuperación e integridad contextual.

Segmentación semántica

Elige los puntos de corte de forma adaptativa según la relevancia semántica, manteniendo la continuidad semántica del documento.

from llama_index.core.node_parser import SemanticSplitterNodeParser

splitter = SemanticSplitterNodeParser(
    buffer_size=1,
    breakpoint_percentile_threshold=95,  # cortar cuando la similitud cae por debajo de este umbral
    embed_model=embed_model
)

Caso de uso: documentos largos con buena estructura lógica y contenido especializado.

Segmentación por Markdown

Optimizada específicamente para documentos estructurados en Markdown, segmenta según los niveles de encabezados.

from llama_index.core.node_parser import MarkdownNodeParser

parser = MarkdownNodeParser()
# Reconoce automáticamente los niveles de encabezado #, ##, ###, etc., cortando en cada sección

Mejor práctica: convertir documentos de PDF/Word a Markdown antes de segmentar, aprovechando la estructura de encabezados para mejorar la precisión de la recuperación.

Guía de selección de estrategia de segmentación

Pipeline de generación aumentada por recuperación

Arquitectura en dos fases de RAG

RAG (Retrieval-Augmented Generation) es la arquitectura central para resolver la “falta de conocimiento” de los modelos grandes. Divide el proceso en dos fases:

Fase uno: Construcción del índice

1. Parseo de documentos: convertir PDF, Word, Markdown y otros formatos a texto plano
2. Segmentación de texto: dividir el documento en párrafos según la estrategia elegida
3. Vectorización: transformar cada segmento en vector usando el modelo de Embedding
4. Almacenamiento del índice: guardar los vectores en la base de datos vectorial y construir el índice

Fase dos: Recuperación y generación

┌──────────┐    ┌──────────┐    ┌──────────┐    ┌──────────┐
│ Pregunta  │ → │ Recuperación │ → │ Ensamblado  │ → │ Generación │
│ del usuario│    │ vectorial │    │ del Prompt│    │ del modelo │
└──────────┘    └──────────┘    └──────────┘    └──────────┘

1. Pregunta del usuario: recibir la pregunta
2. Recuperación vectorial: vectorizar la pregunta y buscar los segmentos más similares en la base de datos vectorial
3. Ensamblado del Prompt: combinar los fragmentos de conocimiento recuperados + la pregunta original + instrucciones en un Prompt completo
4. Generación del modelo: el modelo grande genera la respuesta basándose en el contexto aumentado

Ejemplo completo de pipeline RAG

from openai import OpenAI
import numpy as np

client = OpenAI()

class SimpleRAG:
    def __init__(self, embed_model="text-embedding-3-small"):
        self.embed_model = embed_model
        self.documents = []      # almacenar textos de documentos
        self.embeddings = []     # almacenar vectores de documentos

    def add_documents(self, docs: list[str]):
        """Construir índice: vectorizar y almacenar documentos"""
        for doc in docs:
            vec = self._embed(doc)
            self.documents.append(doc)
            self.embeddings.append(vec)

    def _embed(self, text: str) -> np.ndarray:
        resp = client.embeddings.create(
            model=self.embed_model,
            input=text
        )
        return np.array(resp.data[0].embedding)

    def retrieve(self, query: str, top_k: int = 3) -> list[str]:
        """Recuperar los fragmentos de documento más relevantes"""
        query_vec = self._embed(query)
        similarities = [
            np.dot(query_vec, doc_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(doc_vec))
            for doc_vec in self.embeddings
        ]
        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.documents[i] for i in top_indices]

    def query(self, question: str) -> str:
        """Consulta RAG completa"""
        contexts = self.retrieve(question)
        prompt = f"""Responde la siguiente pregunta basándote en la información de referencia:

Información de referencia:
{' '.join(contexts)}

Pregunta: {question}

Si la información de referencia no es suficiente para responder, indícalo explícitamente."""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": prompt}]
        )
        return resp.choices[0].message.content

Reescritura de preguntas en diálogo multi-turno con RAG

Implementar diálogo multi-turno en un escenario RAG presenta desafíos únicos. Si en la segunda ronda el usuario dice “¿Quién es su supervisor?”, usar esta frase directamente para recuperar fallará por completo: el sistema no sabe a quién se refiere “su”.

Solución: reescritura de preguntas (Query Rewriting)

def rewrite_query(conversation_history: list, current_query: str) -> str:
    """Usar el modelo grande para transformar una pregunta dependiente del contexto en una pregunta independiente"""
    rewrite_prompt = f"""
    Basándote en el historial de la conversación, transforma la pregunta actual en una pregunta independiente que no dependa del contexto.

    Historial de conversación:
    {format_history(conversation_history)}

    Pregunta actual: {current_query}

    Pregunta reescrita:"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": rewrite_prompt}],
        temperature=0.1
    )
    return resp.choices[0].message.content

# Ejemplo
# Historial: usuario pregunta "¿Dónde está el puesto de trabajo de Zhang San?"
#             asistente responde "En el edificio A, planta 5"
# Actual: "¿Quién es su supervisor?"
# Reescrita: "¿Quién es el supervisor de Zhang San?"

Patrones avanzados de RAG

HyDE (Hypothetical Document Embeddings)

La idea central de HyDE: primero hacer que el modelo “invente” una respuesta hipotética y luego usar esa respuesta hipotética para recuperar, en lugar de usar la pregunta original. La intuición es que una respuesta hipotética es semánticamente más cercana a los documentos reales que la pregunta.

def hyde_retrieve(query: str, top_k: int = 3) -> list[str]:
    """Recuperar usando el método HyDE"""
    # Paso 1: generar respuesta hipotética
    hyde_prompt = f"""
    Question: {query}
    Please write a passage that answers this question.
    Passage:"""

    hyde_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": hyde_prompt}]
    )
    hypothetical_doc = hyde_resp.choices[0].message.content

    # Paso 2: recuperar usando la respuesta hipotética, no la pregunta original
    query_vec = embed(hypothetical_doc)
    results = vector_db.search(query_vec, top_k=top_k)
    return results

HyDE es especialmente adecuado para escenarios donde la consulta original es muy corta pero semánticamente compleja, ya que la respuesta hipotética proporciona más pistas semánticas.

Reordenamiento (Re-Ranking)

La recuperación vectorial inicial (fase gruesa) es rápida pero tiene precisión limitada. Se puede introducir un modelo de reordenamiento (fase fina) para ordenar de nuevo los segmentos candidatos:

from sentence_transformers import CrossEncoder

reranker = CrossEncoder('BAAI/bge-reranker-v2-m3')

def rerank(query: str, candidates: list[str], top_k: int = 3):
    """Reordenar los segmentos candidatos"""
    pairs = [(query, doc) for doc in candidates]
    scores = reranker.predict(pairs)

    # Ordenar por puntuación de relevancia
    ranked = sorted(
        zip(candidates, scores),
        key=lambda x: x[1],
        reverse=True
    )
    return [doc for doc, _ in ranked[:top_k]]

Panorama de optimización de estrategias de recuperación

Estrategia de preparación de documentos para RAG

La clave para construir un sistema RAG de alta calidad es la preparación de documentos. Debes entender la relación entre el “espacio de intención” y el “espacio de conocimiento”:

Espacio de intención           Espacio de conocimiento
(lo que el usuario puede preguntar)  (lo que la base de conocimiento cubre)
     ┌─────────┐
     │  Zona de   │ ← RAG puede responder
     │solapamiento│
     └─────────┘
     ↑            ↑
  Intenciones no    Conocimiento no
  cubiertas →       aprovechado →
  añadir conocimiento  optimizar recall

Principios fundamentales:

• Antes de optimizar el algoritmo, completa el conocimiento faltante
• Antes de mejorar el recall, mejora la calidad de los documentos
• Recopila continuamente las intenciones de los usuarios para formar un ciclo cerrado de “recopilación de datos - actualización de conocimiento - validación experta”

El protocolo Function Calling

AutoGen Studio — el constructor no-code de flujos multi-agente de Microsoft — vía microsoft/autogen

DSPy — programa LLMs declarativamente (no con prompts); usado por la pipeline de auto-evolución de Hermes — vía stanfordnlp/dspy

Qué es Function Calling

Function Calling (invocación de funciones, también llamado Tool Calling) es una capacidad estándar que ofrecen las API de modelos grandes. Permite que el modelo, cuando sea necesario, emita instrucciones estructuradas de invocación de herramientas en lugar de respuestas de texto plano.

El flujo de trabajo es el siguiente:

Definición de herramientas con JSON Schema

# Definir lista de herramientas
tools = [
    {
        "type": "function",
        "function": {
            "name": "search_knowledge_base",
            "description": "Buscar en la base de conocimiento interna de la empresa para obtener documentos de políticas y guías de procedimientos",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Palabra clave o pregunta de búsqueda"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["hr", "it", "finance", "general"],
                        "description": "Categoría de conocimiento"
                    }
                },
                "required": ["query"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "send_email",
            "description": "Enviar un correo electrónico",
            "parameters": {
                "type": "object",
                "properties": {
                    "to": {
                        "type": "string",
                        "description": "Dirección de correo del destinatario"
                    },
                    "subject": {
                        "type": "string",
                        "description": "Asunto del correo"
                    },
                    "body": {
                        "type": "string",
                        "description": "Cuerpo del correo"
                    }
                },
                "required": ["to", "subject", "body"]
            }
        }
    }
]

Ciclo completo de Function Calling

from openai import OpenAI
import json

client = OpenAI()

def execute_function_call(tool_call):
    """Ejecutar la invocación de herramienta y devolver el resultado"""
    func_name = tool_call.function.name
    args = json.loads(tool_call.function.arguments)

    if func_name == "search_knowledge_base":
        # Lógica real de búsqueda
        result = knowledge_base.search(args["query"])
        return json.dumps(result)
    elif func_name == "send_email":
        # Lógica real de envío de correo
        result = email_service.send(
            to=args["to"],
            subject=args["subject"],
            body=args["body"]
        )
        return json.dumps({"status": "sent" if result else "failed"})
    else:
        return json.dumps({"error": f"Unknown function: {func_name}"})

def chat_with_tools(user_message: str, messages: list = None):
    """Diálogo con soporte para Function Calling"""
    if messages is None:
        messages = [
            {"role": "system", "content": "Eres un asistente empresarial que puede buscar en la base de conocimiento y enviar correos."}
        ]

    messages.append({"role": "user", "content": user_message})

    # Primera llamada: el modelo decide si usar herramientas
    response = client.chat.completions.create(
        model="gpt-4",
        messages=messages,
        tools=tools
    )

    assistant_msg = response.choices[0].message

    # Si el modelo quiere invocar herramientas
    if assistant_msg.tool_calls:
        messages.append(assistant_msg)

        for tool_call in assistant_msg.tool_calls:
            # Ejecutar herramienta
            result = execute_function_call(tool_call)
            # Devolver resultado al modelo
            messages.append({
                "role": "tool",
                "tool_call_id": tool_call.id,
                "content": result
            })

        # Segunda llamada: el modelo genera la respuesta final basándose en los resultados
        final_response = client.chat.completions.create(
            model="gpt-4",
            messages=messages
        )
        return final_response.choices[0].message.content

    # Si el modelo responde directamente
    return assistant_msg.content

Mejores prácticas para la definición de herramientas

1. Descripciones precisas: el modelo decide cuándo invocar una herramienta basándose en description; una descripción ambigua provoca invocaciones erróneas
2. Parámetros con restricciones: usa enum, required y restricciones de tipo para reducir errores en los parámetros
3. Funciones de responsabilidad única: no agrupes múltiples operaciones en una sola función
4. Resultados estructurados: la salida de la herramienta debe ser fácil de entender para el modelo; se recomienda JSON

# ❌ Mala descripción de herramienta
{"name": "do_stuff", "description": "ejecutar operación", "parameters": {...}}

# ✅ Descripción precisa de herramienta
{
    "name": "cancel_meeting",
    "description": "Cancelar una reunión especificada; requiere ID de reunión y motivo de cancelación",
    "parameters": {
        "properties": {
            "meeting_id": {"type": "string", "description": "Identificador único de la reunión"},
            "reason": {"type": "string", "description": "Motivo de cancelación, se notificará a todos los asistentes"}
        },
        "required": ["meeting_id"]
    }
}

El ciclo ReAct de razonamiento-acción

La idea central de ReAct

ReAct (Reasoning + Acting) es un patrón que hace que el modelo alterne entre razonamiento (Thought) y acción (Action). No genera una respuesta final de una sola vez, sino que resuelve problemas mediante un ciclo de “pensar → actuar → observar → pensar…”.

Un proceso típico de ejecución ReAct:

Usuario: "Busca el departamento de Zhang San y envía un correo a su supervisor"

Thought 1: Necesito buscar primero la información del departamento de Zhang San
Action 1: search_knowledge_base(query="Zhang San departamento")
Observation 1: "Zhang San pertenece al Departamento de I+D, supervisor es Li Si ([email protected])"

Thought 2: He obtenido la información, ahora necesito escribir un correo a Li Si
Action 2: send_email(to="[email protected]", subject="Sobre Zhang San",
                      body="...")
Observation 2: {"status": "sent"}

Thought 3: Tarea completada
Final Answer: "He encontrado que Zhang San está en el Departamento de I+D y he enviado un correo a su supervisor Li Si."

Implementación manual de un ReAct Agent

class ReActAgent:
    def __init__(self, tools: dict, max_iterations: int = 10):
        self.tools = tools
        self.max_iterations = max_iterations

    def run(self, task: str) -> str:
        messages = [
            {"role": "system", "content": self._build_system_prompt()},
            {"role": "user", "content": task}
        ]

        for i in range(self.max_iterations):
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self._format_tools()
            )

            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                # El modelo ha dado la respuesta final
                return msg.content

            if msg.tool_calls:
                # Añadir la invocación de herramienta del asistente al historial
                messages.append(msg)

                for tc in msg.tool_calls:
                    tool_name = tc.function.name
                    args = json.loads(tc.function.arguments)

                    # Ejecutar herramienta
                    result = self.tools[tool_name](**args)

                    # Añadir el resultado de la observación al historial
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tc.id,
                        "content": json.dumps(result)
                    })
                    print(f"  [Tool: {tool_name}({args}) → {result}]")

        return "Ciclo ReAct alcanzó el máximo de iteraciones"

    def _build_system_prompt(self) -> str:
        return """Eres un asistente inteligente capaz de usar herramientas.
Sigue el patrón ReAct: primero piensa, luego actúa, observa el resultado y decide el siguiente paso.
Si la tarea está completada, da directamente la respuesta final."""

    def _format_tools(self) -> list:
        return [
            {
                "type": "function",
                "function": {
                    "name": name,
                    "description": func.__doc__ or "",
                    "parameters": get_schema(func)
                }
            }
            for name, func in self.tools.items()
        ]

Ventajas y limitaciones de ReAct

Ventajas:

• Observable: cada paso de pensamiento y acción deja rastro
• Corregible: al observar un resultado erróneo, puede ajustar la estrategia
• Componible: combina automáticamente múltiples herramientas en una solución

Limitaciones:

• Número de ciclos impredecible (puede entrar en bucle infinito)
• Múltiples llamadas API aumentan latencia y coste
• Depende de la calidad de las observaciones devueltas por las herramientas

Protocolo MCP y ecosistema de herramientas

Por qué necesitamos MCP

Function Calling tiene un problema fundamental: la definición y el consumo de herramientas están acoplados. Cada desarrollador de Agent necesita codificar manualmente el JSON Schema de las herramientas en su propio código. Cuando la API de una herramienta se actualiza, todos los Agents que la integran deben actualizarse manualmente.

Modelo tradicional de Function Calling:
  Agent A ──Schema hardcodeado──→ web_search v1
  Agent B ──Schema hardcodeado──→ web_search v1  ← definición repetida
  Agent C ──Schema hardcodeado──→ web_search v1  ← definición repetida

Modelo MCP:
  Agent A ──┐
  Agent B ──┼── MCP Client ──→ MCP Server (web_search)
  Agent C ──┘                    ↑
                          El proveedor de la herramienta define el Schema

La idea central de MCP (Model Context Protocol) es “quien proporciona la herramienta, define la herramienta”. Transfiere la responsabilidad de la definición de herramientas del Agent (consumidor) al servicio de herramientas (proveedor).

Roles en la arquitectura MCP

Construcción de MCP Server y Client

Ejemplo de MCP Server:

from mcp.server import Server, stdio_server
from mcp.types import Tool, TextContent

app = Server("web-search")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="web_search",
            description="Buscar en internet para obtener información actualizada",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "Palabra clave de búsqueda"},
                    "num_results": {"type": "integer", "default": 5}
                },
                "required": ["query"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if name == "web_search":
        results = search_engine.search(
            arguments["query"],
            num=arguments.get("num_results", 5)
        )
        return [TextContent(type="text", text=json.dumps(results))]
    raise ValueError(f"Unknown tool: {name}")

# Iniciar Server vía stdio
async def main():
    async with stdio_server() as streams:
        await app.run(streams[0], streams[1], app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Ejemplo de integración de MCP Client:

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def run_with_mcp_tools(user_query: str):
    server_params = StdioServerParameters(
        command="python",
        args=["web_search_server.py"]
    )

    async with stdio_client(server_params) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Obtener definiciones de herramientas del MCP Server
            tools_result = await session.list_tools()
            tools = tools_result.tools

            # Convertir al formato tools de OpenAI
            openai_tools = [
                {
                    "type": "function",
                    "function": {
                        "name": tool.name,
                        "description": tool.description,
                        "parameters": tool.inputSchema
                    }
                }
                for tool in tools
            ]

            # Flujo estándar de Function Calling
            response = client.chat.completions.create(
                model="gpt-4",
                messages=[{"role": "user", "content": user_query}],
                tools=openai_tools
            )

            # Si el modelo quiere invocar herramientas, ejecutar vía MCP Client
            if response.choices[0].message.tool_calls:
                for tc in response.choices[0].message.tool_calls:
                    result = await session.call_tool(
                        tc.function.name,
                        json.loads(tc.function.arguments)
                    )
                    # ... devolver resultado al modelo

Valor de ingeniería de MCP

• Desacoplamiento: la definición de herramientas y la implementación del servicio se separan, iterando de forma independiente
• Descubrimiento dinámico: el Agent obtiene automáticamente la lista de herramientas más reciente al iniciarse, con coste de mantenimiento cero
• Efecto ecosistema: terceros pueden proporcionar MCP Servers estandarizados; los desarrolladores de Agents solo necesitan integrar un MCP Client
• Múltiples protocolos de transporte: soporta stdio (comunicación entre procesos locales) y HTTP/SSE (comunicación remota)

Reflexión y autocorrección

Por qué necesitamos reflexión

El contenido generado por el modelo grande no siempre es utilizable. Puede que:

• “Corrija” silenciosamente nombres de variables en el código causando errores de ejecución
• Continúe razonando sobre “hechos” erróneos, produciendo errores en cascada
• En una salida muy larga, olvide las restricciones anteriores

La idea central de la reflexión (Reflection) es: dar al modelo la oportunidad de examinar y evaluar el contenido completo que ya ha generado, para así descubrir y corregir errores.

Dos modos de autoevaluación

Modo uno: reflexión en una sola instrucción

En una sola llamada, indicar al modelo mediante el Prompt que genere la respuesta y al mismo tiempo reflexione:

prompt_with_reflection = """
## Tarea
1. Mejora la expresión lingüística del siguiente contenido del curso, mostrando el texto completo mejorado.
2. Reflexiona sobre el contenido mejorado:
   - ¿Cumple con las normas de redacción?
   - Aparte de la expresión lingüística, ¿se ha modificado accidentalmente algún otro contenido?
   Muestra el resultado de la reflexión y las sugerencias de modificación.
3. Modifica el curso según las sugerencias y muestra el texto completo modificado.

## Borrador del curso
{original_content}
"""

Ventajas: implementación simple, se completa en una sola llamada. Desventajas: el modelo tiende a autoverificarse con el mismo sesgo de pensamiento, cayendo en “autoconfirmación”.

Modo dos: “generación-evaluación” en dos pasos

Separar la generación y la revisión en dos llamadas independientes:

def generate_and_review(content: str) -> str:
    # Primer paso: generar
    draft_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": "Eres un redactor de cursos. Mejora el siguiente contenido para hacerlo más atractivo."
        }, {
            "role": "user",
            "content": content
        }]
    )
    draft = draft_resp.choices[0].message.content

    # Segundo paso: revisar (¡con un system prompt diferente!)
    review_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{
            "role": "system",
            "content": """Eres un revisor técnico riguroso.
Compara el [contenido original] y el [contenido mejorado]:
- Si solo se ha modificado la expresión del texto y el contenido técnico como el código es completamente idéntico → responde "Aprobado"
- Si el contenido técnico ha sido modificado → responde "No aprobado", indicando la ubicación concreta"""
        }, {
            "role": "user",
            "content": f"Contenido original:\n{content}\n\nContenido mejorado:\n{draft}"
        }]
    )

    # Tercer paso: si no se aprueba, enviar la revisión al Agent redactor para corregir
    review = review_resp.choices[0].message.content
    if "No aprobado" in review:
        # ... reenviar la retroalimentación al Agent redactor para corrección
        pass

    return draft

Ventaja principal: la perspectiva del Agent revisor es diferente a la del Agent redactor, evitando el sesgo de rol. Incluso se pueden configurar múltiples Agents revisores especializados: revisión de hechos, revisión de lógica, revisión de estilo, revisión de seguridad.

Retroalimentación externa

La autoevaluación tiene una limitación natural: el modelo no puede verificar la corrección del contenido en un entorno real. La idea de la retroalimentación externa es ejecutar el resultado generado en un entorno real y usar hechos objetivos para verificarlo.

def generate_code_and_validate(spec: str) -> str:
    # Primer paso: generar código
    code_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": f"Escribe código Python según los siguientes requisitos:\n{spec}"}]
    )
    code = extract_code(code_resp.choices[0].message.content)

    # Segundo paso: validación por ejecución externa
    import subprocess, tempfile
    with tempfile.NamedTemporaryFile(suffix=".py", mode="w") as f:
        f.write(code)
        f.flush()
        result = subprocess.run(
            ["python", f.name],
            capture_output=True,
            text=True,
            timeout=30
        )

    # Tercer paso: enviar el error al modelo para corrección
    if result.returncode != 0:
        fix_prompt = f"""El siguiente código produce un error al ejecutarse:

Código:
{code}

Mensaje de error:
{result.stderr}

Corrige el código y muestra la versión completa corregida."""
        fix_resp = client.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": fix_prompt}]
        )
        return fix_resp.choices[0].message.content

    return code

Escenarios de aplicación de la retroalimentación externa:

• Validación por ejecución de código: ejecutar código con un intérprete y capturar errores en tiempo de ejecución
• Validación de JSON Schema: verificar salidas estructuradas con librerías como Pydantic
• Validación de cálculos numéricos: verificar resultados matemáticos con herramientas de cálculo
• Validación de renderizado visual: tras generar un gráfico, hacer que el modelo “vea” el resultado del renderizado para inspección visual

Patrón Plan & Execute

Por qué necesitamos planificación explícita

Cuando se le pide directamente a un Agent que ejecute tareas complejas, los problemas comunes son:

• Olvido: al llegar a pasos posteriores “olvida” las restricciones anteriores
• Errores en cascada: los errores tempranos se convierten en la base del razonamiento posterior, amplificándose exponencialmente
• Pérdida de estructura: el modelo tiende al procesamiento lineal y no puede identificar relaciones de paralelismo/dependencia entre tareas

La idea central del patrón Plan & Execute es: primero planificar y luego ejecutar — primero elaborar un plan de acción completo, confirmarlo mediante revisión y luego ejecutarlo paso a paso.

Implementación del Plan Mode

from typing import List
from pydantic import BaseModel

class PlanStep(BaseModel):
    step_id: int
    description: str
    dependencies: List[int] = []  # IDs de pasos de los que depende
    tool: str = ""               # herramienta a utilizar
    expected_output: str = ""    # salida esperada

class ExecutionPlan(BaseModel):
    goal: str
    steps: List[PlanStep]

def plan_and_execute(task: str) -> str:
    # Fase 1: Elaborar plan
    plan_prompt = f"""
    Eres un experto en planificación de proyectos. Elabora un plan de ejecución detallado para la siguiente tarea.

    Requisitos:
    1. Descomponer la tarea en pasos concretos
    2. Indicar las dependencias entre pasos
    3. Cada paso debe indicar la salida esperada

    Tarea: {task}

    Proporciona el plan en formato JSON."""

    plan_resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": plan_prompt}],
        response_format={"type": "json_object"}
    )
    plan = ExecutionPlan.model_validate_json(
        plan_resp.choices[0].message.content
    )

    # Fase 2: Ejecutar según dependencias
    results = {}
    executed = set()

    while len(executed) < len(plan.steps):
        for step in plan.steps:
            if step.step_id in executed:
                continue
            # Verificar que todas las dependencias se han ejecutado
            if all(dep in executed for dep in step.dependencies):
                # Ejecutar paso
                result = execute_step(step, results)
                results[step.step_id] = result
                executed.add(step.step_id)

    # Fase 3: Consolidar resultados
    return summarize_results(plan, results)

Flujo de trabajo consolidado: patrón Pipeline

Cuando los pasos de una tarea son deterministas y repetibles, deben consolidarse en un pipeline:

Input → Step 1 → Step 2 → Step 3 → ... → Output

class Pipeline:
    """Pipeline fijo: la salida de cada paso es la entrada del siguiente"""

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

    def add_step(self, name: str, func):
        self.steps.append({"name": name, "func": func})

    def run(self, input_data):
        result = input_data
        for step in self.steps:
            print(f"  [Ejecutando] {step['name']}")
            result = step["func"](result)
        return result

# Ejemplo: pipeline de procesamiento de documentos
pipeline = Pipeline()
pipeline.add_step("Parsear PDF", parse_pdf_to_text)
pipeline.add_step("Limpiar texto", clean_text)
pipeline.add_step("Segmentar", split_sections)
pipeline.add_step("Vectorizar", vectorize_chunks)
pipeline.add_step("Almacenar índice", store_to_vectordb)

pipeline.run("document.pdf")

Patrones de orquestación de flujos de trabajo

Cinco patrones fundamentales de flujo de trabajo

Las tareas complejas requieren organizar los nodos Agent en topologías específicas. Estos son los cinco patrones fundamentales:

Enrutamiento por rama (Branching/Router)

En el nodo de entrada se determina el tipo de tarea y se deriva a diferentes rutas de procesamiento.

def router_agent(user_input: str):
    """Derivar a diferentes pipelines según la intención"""
    classify_prompt = f"""
    Analiza el tipo de la siguiente solicitud del usuario, responde solo una palabra:
    - code_review: revisar/verificar código
    - style_review: mejorar/optimizar lenguaje
    - fact_check: verificar precisión de hechos/conceptos

    Solicitud: {user_input}
    Tipo:"""

    intent = client.chat.completions.create(
        model="gpt-4o-mini",  # usar modelo ligero para ahorrar costes
        messages=[{"role": "user", "content": classify_prompt}],
        temperature=0
    ).choices[0].message.content.strip()

    pipelines = {
        "code_review": code_review_pipeline,
        "style_review": style_review_pipeline,
        "fact_check": fact_check_pipeline
    }
    return pipelines.get(intent, default_pipeline)(user_input)

Ejecución en paralelo (Parallel)

Las subtareas que no dependen entre sí se distribuyen simultáneamente y se consolidan al final.

              ┌→ Revisión de código ──┐
Entrada → Split ─┼→ Verificación de hechos ──┼→ Merge → Informe consolidado
              └→ Revisión de estilo ──┘

import asyncio

async def parallel_review(notebook_content: str):
    """Realizar tres revisiones en paralelo sobre el contenido del curso"""
    tasks = [
        asyncio.create_task(check_code(notebook_content)),
        asyncio.create_task(check_facts(notebook_content)),
        asyncio.create_task(check_style(notebook_content))
    ]

    code_result, fact_result, style_result = await asyncio.gather(*tasks)

    # Consolidar
    return generate_summary_report(code_result, fact_result, style_result)

Mezcla de expertos (Mixture-of-Agents, MoA)

Múltiples modelos distintos procesan la misma tarea y un agregador sintetiza el mejor resultado.

             ┌→ Modelo A (experto en razonamiento) ──┐
Pregunta → Split ─┼→ Modelo B (experto en creatividad) ──┼→ Aggregator → Respuesta óptima
             └→ Modelo C (experto en precisión) ──┘

El hallazgo central de MoA es la “colaboratividad” (Collaborativeness) de los modelos: cuando un modelo puede consultar las salidas de otros modelos, a menudo genera respuestas de mayor calidad.

def mixture_of_agents(task: str):
    """Implementación MOA: múltiples modelos + agregación"""
    # Primera capa: los proponentes generan en paralelo
    proposers = ["gpt-4", "claude-3-opus", "gemini-pro"]
    proposals = []

    for model in proposers:
        resp = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": task}]
        )
        proposals.append(resp.choices[0].message.content)

    # Segunda capa: el agregador sintetiza
    aggregator_prompt = f"""
    A continuación hay {len(proposals)} respuestas a la misma pregunta. Sintetiza sus ventajas
    y genera una respuesta óptima.

    {format_proposals(proposals)}

    Respuesta sintetizada:"""

    final = client.chat.completions.create(
        model="gpt-4",  # usar el modelo más potente para agregar
        messages=[{"role": "user", "content": aggregator_prompt}]
    )
    return final.choices[0].message.content

Colaboración humano-máquina (Human-in-the-Loop, HITL)

Introducir revisión humana en nodos clave, formando un ciclo de “IA ejecuta → humano aprueba → IA continúa”.

def hitl_workflow(task: str):
    """Flujo de trabajo con humano en el bucle"""
    plan = generate_plan(task)
    print(f"Plan de ejecución:\n{format_plan(plan)}")

    approval = input("¿Aprueba este plan? (s/n): ")
    if approval.lower() != 's':
        return "Tarea cancelada"

    for step in plan.steps:
        result = execute_step(step)
        print(f"Paso {step.step_id} completado: {result['summary']}")

        if step.get("requires_review"):
            review = input(f"Revise el resultado del paso (aprobar/modificar/rechazar): ")
            if review == "rechazar":
                print("Paso rechazado, re-ejecutando...")
                result = execute_step(step, feedback=review)

    return generate_final_output()

Metodología de selección de patrones

La mejor práctica es el modo mixto “explorar-consolidar”: primero usar Plan & Execute para explorar y encontrar la solución óptima, luego consolidarla en un Pipeline para producción a gran escala.

Patrón de colaboración jerárquica

Arquitectura Leader-Worker

La colaboración jerárquica (Hierarchical/Team Leader Pattern) es el patrón de colaboración multi-agente más intuitivo. Simula la estructura organizativa de “jefe de proyecto + miembros del equipo”:

El Leader Agent se encarga de:

1. Recibir y comprender la tarea de alto nivel
2. Descomponerla en subtareas y asignarlas al Worker adecuado
3. Hacer seguimiento del progreso global
4. Consolidar los resultados de los Workers

Cada Worker Agent posee experiencia en un dominio específico y se concentra en ejecutar las subtareas asignadas.

Implementación de colaboración jerárquica mediante Handoff

class LeaderWorkerSystem:
    """Implementación del sistema de colaboración jerárquica"""

    def __init__(self):
        self.workers = {
            "instructional_designer": self._create_worker(
                "Eres diseñador instruccional, experto en diseñar esquemas de cursos y rutas de aprendizaje."
            ),
            "data_scientist": self._create_worker(
                "Eres científico de datos, experto en escribir código Python de análisis de datos y casos prácticos."
            ),
            "content_writer": self._create_worker(
                "Eres redactor de contenido, experto en transformar contenido técnico en guiones de curso atractivos."
            )
        }
        self.leader = self._create_leader()

    def _create_leader(self):
        return {
            "system_prompt": """Eres el director del proyecto del curso. Tus responsabilidades son:
1. Analizar los requisitos y descomponerlos en subtareas
2. Asignar tareas al experto adecuado
3. Integrar los resultados de cada experto en un curso completo
Expertos disponibles: instructional_designer, data_scientist, content_writer""",
            "tools": [
                {
                    "type": "function",
                    "function": {
                        "name": "delegate_to_worker",
                        "description": "Asignar una subtarea al experto especificado",
                        "parameters": {
                            "type": "object",
                            "properties": {
                                "worker": {
                                    "type": "string",
                                    "enum": list(self.workers.keys()),
                                    "description": "Experto que recibe la tarea"
                                },
                                "task": {
                                    "type": "string",
                                    "description": "Descripción concreta de la subtarea"
                                }
                            },
                            "required": ["worker", "task"]
                        }
                    }
                }
            ]
        }

    def run(self, project_brief: str) -> str:
        """Ejecutar un proyecto completo de desarrollo de curso"""
        messages = [
            {"role": "system", "content": self.leader["system_prompt"]},
            {"role": "user", "content": project_brief}
        ]

        # Bucle del Leader
        while True:
            response = client.chat.completions.create(
                model="gpt-4",
                messages=messages,
                tools=self.leader["tools"]
            )
            msg = response.choices[0].message

            if msg.content and not msg.tool_calls:
                return msg.content  # Salida final consolidada

            if msg.tool_calls:
                messages.append(msg)
                for tc in msg.tool_calls:
                    if tc.function.name == "delegate_to_worker":
                        args = json.loads(tc.function.arguments)
                        # Llamar al Worker para ejecutar la subtarea
                        worker_result = self._run_worker(
                            args["worker"], args["task"]
                        )
                        messages.append({
                            "role": "tool",
                            "tool_call_id": tc.id,
                            "content": worker_result
                        })

    def _run_worker(self, worker_name: str, task: str) -> str:
        worker = self.workers[worker_name]
        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": worker},
                {"role": "user", "content": task}
            ]
        )
        return resp.choices[0].message.content

Ventajas y desventajas de la colaboración jerárquica

Ventajas:

• Estructura clara, cada Agent tiene responsabilidades bien definidas
• El Leader controla la visión global, sin desviarse del objetivo
• Cada Worker posee una ventana de contexto independiente, más enfocado
• Soporta asignación de tareas en paralelo

Desventajas:

• Los Workers no se comunican directamente entre sí, hay retraso/distorsión en la transmisión de información
• El Leader se convierte en un punto único de cuello de botella
• La combinación de módulos puede carecer de fluidez global

Patrón de colaboración por pizarra

Co-creación descentralizada

El patrón de pizarra (Blackboard/Co-creation Pattern) simula la forma de trabajar de “expertos reunidos alrededor de una pizarra haciendo tormenta de ideas”. No tiene un coordinador centralizado; todos los Agents leen y escriben de forma equitativa en un espacio compartido:

Implementación del patrón de pizarra

class BlackboardSystem:
    """Sistema de colaboración por pizarra"""

    def __init__(self, agents: dict, max_rounds: int = 3):
        self.agents = agents
        self.max_rounds = max_rounds
        self.blackboard = []  # espacio compartido

    def run(self, problem: str) -> str:
        # Escribir el problema en la pizarra
        self.blackboard.append({"source": "user", "content": problem})

        for round_num in range(self.max_rounds):
            print(f"\n=== Ronda {round_num + 1} ===")
            new_contributions = []

            # Todos los Agents leen la pizarra en paralelo y contribuyen
            for name, agent_config in self.agents.items():
                contribution = self._agent_contribute(
                    name, agent_config, self.blackboard
                )
                if contribution:
                    new_contributions.append({
                        "source": name,
                        "content": contribution
                    })

            # Escribir las nuevas contribuciones en la pizarra
            self.blackboard.extend(new_contributions)

            # Verificar si se ha alcanzado consenso
            if self._check_consensus():
                break

        return self._synthesize_final_answer()

    def _agent_contribute(self, name, config, blackboard):
        """Cada Agent lee la pizarra y aporta sus propias ideas"""
        board_text = self._format_blackboard(blackboard)

        prompt = f"""Eres un {config['role']}.

Contenido actual en la pizarra compartida:
{board_text}

Basándote en la discusión existente, aporta tus ideas, complementos, cuestionamientos o nuevas propuestas.
Si la solución existente ya es completa, puedes expresar tu acuerdo y explicar por qué."""

        resp = client.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": config["system_prompt"]},
                {"role": "user", "content": prompt}
            ]
        )
        return resp.choices[0].message.content

    def _format_blackboard(self, blackboard):
        return "\n\n".join([
            f"[{entry['source']}]: {entry['content']}"
            for entry in blackboard
        ])

    def _check_consensus(self):
        """Verificar si las últimas contribuciones en la pizarra han formado consenso"""
        # Implementar lógica de detección de consenso
        pass

    def _synthesize_final_answer(self):
        """Sintetizar la solución final a partir del contenido de la pizarra"""
        pass

Patrón de pizarra vs patrón jerárquico

Metodología de selección de patrones de colaboración

Aprender del mundo real

El diseño de sistemas multi-agente excelentes surge de la observación y destilación del trabajo en equipo en el mundo real. En lugar de memorizar nombres abstractos de patrones, es mejor adentrarse en el negocio y observar cómo los equipos de expertos humanos realizan tareas similares.

Tres dimensiones de observación:

Diseño de patrones híbridos

En proyectos reales es raro usar un solo patrón. Lo común es el diseño híbrido:

                    ┌─────────────┐
                    │   Leader    │  ← patrón jerárquico
                    └──────┬──────┘
           ┌───────────────┼───────────────┐
           ▼               ▼               ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ Diseñador │    │ Redactor  │    │ Revisor   │
    │instrucc.  │    │de contenido│    │  Leader   │
    │  Worker   │    │  Worker   │    └─────┬────┘
    └──────────┘    └──────────┘           │
                                  ┌────────┼────────┐
                                  ▼        ▼        ▼
                             ┌──────┐ ┌──────┐ ┌──────┐
                             │Rev.  │ │Rev.  │ │Rev.  │  ← patrón paralelo
                             │código│ │hechos│ │estilo│
                             └──────┘ └──────┘ └──────┘
                                 │        │        │
                                 └────────┼────────┘
                                          ▼
                                    ┌──────────┐
                                    │ Informe  │
                                    │consolidado│
                                    └──────────┘

Conciencia del coste

El consumo de tokens de un sistema multi-agente suele ser de 3 a 5 veces el de un solo Agent. Al diseñar, hay que sopesar:

def estimate_cost(num_agents: int, avg_tokens_per_agent: int,
                  rounds: int = 1, price_per_1k: float = 0.01):
    """Estimar el coste en tokens de un sistema multi-agente"""
    total_tokens = num_agents * avg_tokens_per_agent * rounds
    return total_tokens * price_per_1k / 1000

# Ejemplo: 5 Agents, 2000 tokens cada uno, 3 rondas de discusión en pizarra
cost = estimate_cost(5, 2000, 3)
print(f"Coste estimado: ${cost:.2f}")
# La cifra real puede ser mayor, porque además hay que incluir
# la retransmisión del contenido de la pizarra

Gestión de memoria a corto plazo

Sin estado: la raíz del problema

Los modelos de lenguaje grandes son inherentemente sin estado (Stateless). Cada llamada a la API es independiente: no recuerda el contenido de la ronda anterior de conversación, tus preferencias ni los consensos alcanzados previamente.

# Estas dos llamadas son completamente independientes entre sí
response1 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Me llamo Zhang San"}]
)
# Respuesta: "¡Hola Zhang San! ¿En qué puedo ayudarte?"

response2 = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "¿Cómo me llamo?"}]
)
# Respuesta: "Lo siento, no sé cómo te llamas, porque no hemos conversado antes."

Solución: mantener una lista de historial de conversación y enviar el historial completo en cada llamada.

class ConversationBuffer:
    """La memoria a corto plazo más simple: guardar el historial completo de conversación"""

    def __init__(self, system_prompt: str = ""):
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})

    def chat(self, user_input: str) -> str:
        self.messages.append({"role": "user", "content": user_input})
        response = client.chat.completions.create(
            model="gpt-4",
            messages=self.messages
        )
        reply = response.choices[0].message.content
        self.messages.append({"role": "assistant", "content": reply})
        return reply

La presión de la ventana de contexto

A medida que aumentan las rondas de conversación, el enfoque de historial completo enfrenta tres problemas fatales:

1. Exceder la ventana de contexto: la longitud del historial supera el límite del modelo → error del programa
2. Coste fuera de control: cada llamada reenvía todo el historial → consumo de tokens crece linealmente
3. Dilución de la atención: en contextos largos, la capacidad del modelo para procesar información de la parte central disminuye significativamente

Tres estrategias de gestión de memoria

Estrategia uno: truncamiento por ventana fija (Context Truncation)

Conservar solo las últimas N rondas de conversación o N tokens.

class TruncationMemory:
    def __init__(self, max_tokens: int = 4000):
        self.max_tokens = max_tokens
        self.messages = []

    def add_and_truncate(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        # Eliminar desde los mensajes más antiguos hasta que el total de tokens esté dentro del límite
        while self._total_tokens() > self.max_tokens:
            self.messages.pop(0)  # eliminar el mensaje no-system más antiguo

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

Ventajas: implementación extremadamente simple, bajo coste computacional. Desventajas: si la información clave está en las primeras conversaciones, al ser truncada el Agent “pierde la memoria”.

Estrategia dos: resumen progresivo (Rolling Summary)

Antes de olvidar, extraer primero los puntos clave.

Historial de conversación: [msg1, msg2, msg3, msg4, msg5, msg6, msg7, msg8]
                          ↓ comprimir la primera mitad
         [Resumen(m1-m4), msg5, msg6, msg7, msg8]
                          ↓ seguir comprimiendo
         [Resumen(m1-m6), msg7, msg8]

class RollingSummaryMemory:
    def __init__(self, summary_trigger_tokens: int = 3000):
        self.summary_trigger = summary_trigger_tokens
        self.messages = []
        self.summary = ""

    def add_message(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})

        if self._total_tokens() > self.summary_trigger:
            self._compress()

    def _compress(self):
        """Comprimir la primera mitad de la conversación en un resumen"""
        split_point = len(self.messages) // 2
        to_compress = self.messages[:split_point]
        remaining = self.messages[split_point:]

        compress_prompt = f"""
        Resume el siguiente historial de conversación en un párrafo conciso,
        conservando la información clave:

        Conversación:
        {format_messages(to_compress)}

        Resumen:"""

        resp = client.chat.completions.create(
            model="gpt-4o-mini",  # usar modelo ligero para resumir
            messages=[{"role": "user", "content": compress_prompt}]
        )
        self.summary = resp.choices[0].message.content

        # Reemplazar los mensajes comprimidos con el resumen
        self.messages = [
            {"role": "system", "content": f"Resumen del historial de conversación:\n{self.summary}"}
        ] + remaining

    def _total_tokens(self):
        return sum(count_tokens(m["content"]) for m in self.messages)

Ventajas: conserva la información central mientras comprime la longitud, manteniendo coherencia a largo plazo. Desventajas: coste adicional de llamadas API; la calidad del resumen afecta directamente las conversaciones posteriores.

Estrategia tres: recuperación vectorizada (Vector-based Retrieval)

La forma más inteligente: almacenar el historial de conversación en una base de datos vectorial y recuperar los recuerdos más relevantes bajo demanda.

class VectorBasedMemory:
    def __init__(self):
        self.conversations = []  # registro completo de conversaciones
        self.embeddings = []     # vector de cada ronda de conversación
        self.embed_model = "text-embedding-3-small"

    def store_conversation(self, user_msg: str, assistant_msg: str):
        """Almacenar una ronda de conversación y vectorizarla"""
        conversation_text = f"User: {user_msg}\nAssistant: {assistant_msg}"
        self.conversations.append(conversation_text)

        vec = client.embeddings.create(
            model=self.embed_model,
            input=conversation_text
        )
        self.embeddings.append(vec.data[0].embedding)

    def retrieve_relevant(self, current_query: str, top_k: int = 5):
        """Recuperar las conversaciones históricas más relevantes para la pregunta actual"""
        query_vec = client.embeddings.create(
            model=self.embed_model,
            input=current_query
        ).data[0].embedding

        # Calcular similitud
        similarities = [
            np.dot(query_vec, mem_vec) /
            (np.linalg.norm(query_vec) * np.linalg.norm(mem_vec))
            for mem_vec in self.embeddings
        ]

        top_indices = np.argsort(similarities)[-top_k:][::-1]
        return [self.conversations[i] for i in top_indices]

Ventajas: se libera fundamentalmente de la limitación de longitud de la ventana de contexto, coincidencia semántica precisa. Desventajas: mayor complejidad del sistema, introduce modelo de Embedding y base de datos vectorial.

Guía de selección de estrategias

Memoria a largo plazo y almacenamiento vectorial

De contexto pasivo a gestión activa de memoria

Un Agent verdaderamente inteligente no debería limitarse a recibir pasivamente un contexto procesado, sino que debería poder gestionar activamente su propia memoria: decidir por sí mismo cuándo recordar algo y cuándo recuperarlo.

Esto requiere proporcionar al Agent dos herramientas fundamentales:

# Herramientas de gestión de memoria disponibles para el Agent
memory_tools = [
    {
        "type": "function",
        "function": {
            "name": "record_to_memory",
            "description": "Almacenar información importante en la memoria a largo plazo. Usar cuando el usuario exprese preferencias explícitas, proporcione información clave o tome decisiones importantes.",
            "parameters": {
                "type": "object",
                "properties": {
                    "content": {
                        "type": "string",
                        "description": "Contenido a recordar"
                    },
                    "category": {
                        "type": "string",
                        "enum": ["preference", "fact", "decision", "context"],
                        "description": "Categoría del recuerdo"
                    }
                },
                "required": ["content"]
            }
        }
    },
    {
        "type": "function",
        "function": {
            "name": "retrieve_from_memory",
            "description": "Recuperar información relevante de la memoria a largo plazo. Usar cuando sea necesario recordar preferencias del usuario, decisiones históricas o contenido discutido anteriormente.",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "Consulta de recuperación"
                    },
                    "category": {
                        "type": "string",
                        "description": "Opcional, limitar la categoría de recuerdos a recuperar"
                    }
                },
                "required": ["query"]
            }
        }
    }
]

Memoria a corto plazo vs memoria a largo plazo

┌─────────────────────────────────────────────────────┐
│                Arquitectura del sistema de memoria    │
├─────────────────────────────────────────────────────┤
│                                                     │
│  Memoria a corto plazo (Short-term Memory)            │
│  ┌───────────────────────────────────────────────┐  │
│  │ Almacenamiento: buffer de diálogo (lista de    │  │
│  │   messages)                                   │  │
│  │ Gestión: truncamiento / resumen               │  │
│  │ Ciclo de vida: sesión actual                  │  │
│  │ Función: mantener coherencia de sesión,        │  │
│  │   recordar "de qué se estaba hablando"         │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  Memoria a largo plazo (Long-term Memory)             │
│  ┌───────────────────────────────────────────────┐  │
│  │ Almacenamiento: BD vectorial + metadatos       │  │
│  │ Gestión: recuperación vectorizada + invocación │  │
│  │   como herramienta                            │  │
│  │ Ciclo de vida: entre sesiones                 │  │
│  │ Función: persistir información clave, soportar │  │
│  │   recuperación inteligente entre sesiones      │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
└─────────────────────────────────────────────────────┘

Mejores prácticas de gestión de memoria

1. Recordar con criterio: recordar más no siempre es mejor. La información de bajo valor interfiere en las recuperaciones posteriores. Establece un mecanismo de admisión de escritura: solo escribir cuando el usuario lo solicite explícitamente o cuando la importancia de la información supere un umbral.

2. Gobernanza continua: la memoria es un activo de datos dinámico. Limpia periódicamente la información obsoleta, fusiona entradas duplicadas y verifica la precisión de los hechos. Proporciona interfaces para que el usuario gestione su memoria (consultar, modificar, eliminar).

3. Aplicación contextualizada: distintos escenarios requieren diferentes necesidades de memoria. En flujos de trabajo de documentación de cursos no se deben registrar preferencias personalizadas; para información factual de productos (parámetros de API, limitaciones de funcionalidad, etc.) sí se debe registrar y revisar periódicamente su vigencia.

Diseño del sistema de Skill

De Prompt a Skill: la evolución

Un Prompt cuidadosamente elaborado tiene un gran valor, pero solo surte efecto en la sesión actual. Al terminar la sesión, el conocimiento se dispersa. Skill supone la evolución del Prompt a un módulo de conocimiento profesional reutilizable, versionable y compartible en equipo.

Ruta evolutiva:

Prompt improvisado ("Revisa este curso para mí")
    ↓ Problema: hay que reescribirlo cada vez, criterios inconsistentes
Prompt fijo (guardado en el historial de chat)
    ↓ Problema: disperso, difícil de encontrar, no colaborativo
Archivo independiente (course-review.md)
    ↓ Problema: los archivos se hinchan, difícil de mantener
Directorio de base de conocimiento (course-review/)
    ↓ Problema: aún hay que decir manualmente al Agent qué hacer
Skill (SKILL.md + archivos de recursos + scripts)

Estructura de un Skill

Un Skill bien formado se compone de frontmatter YAML + cuerpo Markdown:

---
name: course-review
description: |
  Revisar la precisión técnica, corrección del código y calidad pedagógica del contenido del curso.
  Usar esta skill cuando el usuario solicite revisar, auditar o evaluar cursos o materiales de formación existentes.
---

# Skill de revisión de cursos

## Flujo de revisión
1. Extraer la estructura del directorio del Notebook para comprender la organización general de los capítulos
2. Revisar cada capítulo sección por sección, contrastando con las siguientes dimensiones:
   - Ejecutabilidad del código (ver [code-quality.md](code-quality.md))
   - Precisión del contenido (ver [content-accuracy.md](content-accuracy.md))
   - Estilo de explicación (ver [style-guide.md](style-guide.md))
   - API obsoletas (ver [outdated-api.md](outdated-api.md))
3. Consolidar los resultados de la revisión y generar el informe según el formato de salida

## Lista de anti-patrones
- ❌ No modificar nombres de variables ni números de versión de API en el código
- ❌ No omitir ninguna verificación
- ❌ No introducir nuevos conceptos técnicos durante la revisión

## Formato de salida
Para cada verificación, indicar: Aprobado/No aprobado/Requiere revisión humana + ubicación + sugerencia de modificación

Estructura del directorio:

course-review/
├── SKILL.md              # Punto de entrada principal de instrucciones
├── code-quality.md       # Elementos de verificación de ejecutabilidad del código
├── content-accuracy.md   # Elementos de verificación de precisión factual
├── style-guide.md        # Ejemplos positivos y negativos de estilo de explicación
├── outdated-api.md       # Tabla de referencia de API obsoletas
└── scripts/
    ├── extract_toc.py    # Extraer el directorio del Notebook
    └── validate_code.py  # Ejecutar validación automática de código

Skill vs RAG

Mucha gente confunde Skill y RAG. Su diferencia fundamental es:

Método de cinco pasos para escribir Skills de alta calidad

Paso 1: Determinar si vale la pena
  ├─ ¿Esta tarea tiene "intuición experta"? (un experto la hace bien pero un novato omite fácilmente condiciones de contorno)
  ├─ ¿Es suficientemente compleja? (si se puede hacer con GUI en 3 pasos, no)
  └─ ¿Se ejecutará repetidamente? (si es una sola vez, no)

Paso 2: Extraer qué escribir
  ├─ Extraer el árbol de decisión del experto, no solo los pasos
  ├─ Inyectar verificación de anti-patrones ("qué trampas evitar a toda costa")
  ├─ Patrón Template: proporcionar plantillas de salida estandarizadas
  └─ Patrón Examples: usar ejemplos en lugar de descripciones textuales

Paso 3: Redactar bien las instrucciones
  ├─ Concisión: cada frase debe justificar su coste en tokens
  ├─ Graduación de libertad: el grado de restricción debe ajustarse al riesgo de la tarea
  │   ├─ Baja libertad (migración de BD): scripts precisos
  │   ├─ Libertad media (generación de informes): pseudocódigo/parametrizado
  │   └─ Alta libertad (revisión de código): instrucciones textuales
  └─ Divulgación progresiva: el archivo principal se mantiene limpio, los detalles se cargan bajo demanda

Paso 4: Equipar con herramientas
  ├─ Flujo de trabajo: Checklist rastreable
  ├─ Bucle de retroalimentación: ejecutar → verificar → corregir → repetir
  ├─ Operaciones de alto riesgo: validar el plan antes de ejecutar
  └─ Scripts amigables para IA: estado estructurado + pistas de reparación + degradación elegante + seguridad idempotente

Paso 5: Validar e iterar
  ├─ Fase uno: establecer la línea base de evaluación (primero la evaluación, luego escribir el Skill)
  ├─ Fase dos: extraer el Skill (usar IA para resumir la información proporcionada repetidamente)
  └─ Fase tres: prueba iterativa con dos Agents (diseñador vs usuario)

Skill as Code

Tratar los Skills como código, beneficiándose de las metodologías de ingeniería de software:

• Control de versiones: el directorio del Skill se incluye en Git, cada modificación tiene historial
• Revisión de código: los Skills nuevos o modificados requieren revisión mediante PR
• CI/CD: los cambios en Skills disparan pipelines de evaluación para asegurar que no haya regresiones
• Compartición comunitaria: compartir y reutilizar Skills como librerías open-source

Divulgación progresiva y Skill as Code

Filosofía de diseño de la divulgación progresiva

La contradicción central a la que se enfrenta la ingeniería de prompts tradicional: demasiada información → contexto abarrotado, atención diluida; muy poca información → el Agent carece de conocimiento suficiente para tomar decisiones.

La divulgación progresiva (Progressive Disclosure) es un patrón de diseño que resuelve esta contradicción:

Nivel 1: Lista de Skills (visible al iniciar el Agent)
  ↓ El Agent elige activar un Skill
Nivel 2: SKILL.md (cargado tras la activación del Skill)
  ↓ El Agent ejecuta un paso determinado
Nivel 3: Subarchivos (cargar recursos correspondientes bajo demanda)
  ↓ Necesidad de verificar un detalle técnico concreto
Nivel 4: Ejecución de scripts (proporciona resultados deterministas)

Principios de diseño:

• Mantener estructura plana, evitar anidamiento profundo de referencias
• SKILL.md enlaza directamente todos los archivos de recursos, asegurando accesibilidad en “un solo paso”
• No meter todo el contenido en SKILL.md: es solo el punto de entrada

Diseño de scripts amigables para IA

La calidad de la salida de los scripts de herramientas afecta directamente al rendimiento del Agent. Cuatro principios clave:

1. Retroalimentación de estado estructurada: emitir JSON en lugar de texto libre:

# ❌ Mala salida
print("Error: exit code 1")
print("Could not find module 'openpyxl'")

# ✅ Salida amigable para IA
print(json.dumps({
    "status": "failed",
    "error_code": "MODULE_NOT_FOUND",
    "missing_module": "openpyxl",
    "fix_hint": "Ejecuta pip install openpyxl para instalar la dependencia faltante",
    "fallback_available": True,
    "affected_cells": [42, 43, 44]
}))

2. Los mensajes de error incluyen pistas de reparación: decir al Agent “cómo arreglarlo”, no solo “qué falló”.

3. Degradación elegante en lugar de colapso: cuando sea posible, proporcionar valores por defecto y continuar.

4. Idempotencia y seguridad: soportar ejecución repetida sin efectos secundarios:

Diseño del marco de evaluación

Por qué “me da buena sensación” no es fiable

Optimizar un Agent basándose solo en sensaciones subjetivas conduce a:

• Difícil de cuantificar: “se siente mejor” no sirve como base para decisiones de ingeniería
• Falta de estándar: distintos evaluadores o distintos momentos pueden tener criterios cambiantes
• Irreproducible: no se pueden hacer pruebas de regresión sistemáticas para asegurar que los nuevos cambios no rompen funcionalidades anteriores

El desarrollo guiado por evaluación (Evaluation-Driven Development) eleva la evaluación desde el final del flujo de desarrollo hasta una posición central:

         ┌──────────────────┐
         │  Evaluación =    │
         │  patrón de calidad│
         └────────┬─────────┘
                  │
    ┌─────────────┼─────────────┐
    ▼             ▼             ▼
Lo que se puede   Cuanto más rápida   Determina el
medir se puede    y precisa sea la    límite superior
mejorar           retroalimentación,  de la capacidad
                  más eficiente la    del producto
                  mejora

Evaluación end-to-end (End-to-End Evaluation)

La evaluación end-to-end se centra en la salida final y responde a la pregunta “¿este Agent le funciona bien al usuario?”.

Los indicadores de evaluación se dividen en dos tipos:

Evaluación de caja blanca (White-box Evaluation)

Cuando el flujo del Agent se vuelve complejo, la evaluación end-to-end no permite localizar problemas concretos. La evaluación de caja blanca propone profundizar en el interior del sistema, diseñando sistemas de evaluación independientes para los componentes clave.

Evaluación end-to-end (solo mira la salida final):
  Agent completo → puntuación 4.2/5 → pero no se sabe dónde está el problema

Evaluación de caja blanca (revisa los pasos intermedios):
  ┌────────────┐    ┌────────────┐    ┌────────────┐
  │ Explicación │    │ Generación  │    │ Ajuste de  │
  │ conceptual  │    │ de código   │    │ estilo     │
  │ 3.1/5      │    │ 4.8/5      │    │ 4.5/5      │
  └────────────┘    └────────────┘    └────────────┘
       ↑ ¡El cuello de botella está aquí!

Ventajas principales de la evaluación de caja blanca:

• Señal clara: señal de mejora sin interferencias, enfocada en el verdadero cuello de botella
• Iteración rápida: solo hay que probar un componente, sin ejecutar todo el flujo
• Optimización precisa: el efecto de cada cambio se puede medir con precisión

LLM-as-Judge

Usar modelos grandes como evaluadores

Hacer que otro modelo grande actúe como “experto evaluador”, puntuando automáticamente según indicadores y rúbricas definidos.

def llm_judge_evaluate(generated_output: str, criteria: dict):
    """Usar LLM como evaluador"""
    judge_prompt = f"""
    Eres un evaluador profesional de calidad de cursos. Puntúa según los siguientes criterios:

    Criterios de evaluación:
    {json.dumps(criteria, indent=2, ensure_ascii=False)}

    Contenido a evaluar:
    {generated_output}

    Proporciona el resultado de la puntuación en formato JSON:
    {{
        "scores": {{
            "accuracy": <1-5>,
            "clarity": <1-5>,
            "engagement": <1-5>
        }},
        "overall": <1-5>,
        "comments": "Valoración global"
    }}"""

    resp = client.chat.completions.create(
        model="gpt-4",
        messages=[{"role": "user", "content": judge_prompt}],
        response_format={"type": "json_object"}
    )
    return json.loads(resp.choices[0].message.content)

Sesgos del evaluador LLM

Al usar un LLM como evaluador, hay que estar alerta ante sus sesgos inherentes:

Mejor práctica: en las fases iniciales, usar expertos humanos para establecer un “conjunto de pruebas dorado” y usarlo para calibrar al evaluador LLM. Verificar periódicamente la consistencia de la evaluación automática mediante muestreo manual.

Desglose de indicadores de evaluación

Descomponer objetivos de evaluación difusos en reglas concretas que se puedan verificar una a una:

# Ejemplo de desglose de evaluación de calidad de contenido
dimensión_de_evaluación: "Calidad del contenido"
reglas:
  - id: "pain_point"
    description: "¿Comienza con un punto de dolor concreto?"
    type: "boolean"
  - id: "theory_depth"
    description: "¿Señala claramente las limitaciones del enfoque inicial e introduce la teoría central?"
    type: "boolean"
  - id: "code_relevance"
    description: "¿El ejemplo de código está estrechamente relacionado con la teoría explicada y es suficientemente simplificado?"
    type: "boolean"
  - id: "anti_pattern_check"
    description: "¿Evita frases cursis como '¡felicidades por desbloquear una nueva habilidad!'?"
    type: "boolean"

Iteración guiada por evaluación

El ciclo cerrado de evaluación

La evaluación no es algo puntual, sino un motor que impulsa la mejora continua:

    ┌──────────────────────────────────┐
    │                                  │
    ▼                                  │
┌─────────┐   ┌──────────┐   ┌─────────┐
│Construir│ → │Detectar  │ → │Refinar  │
│  MVP    │   │problemas │   │indicadores│
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  ▼
┌─────────┐   ┌──────────┐   ┌─────────┐
│Desplegar│ ← │Prueba de │ ← │Optimizar│
│a prod.  │   │regresión │   │y mejorar│
└─────────┘   └──────────┘   └─────────┘
                                  │
                                  └────→ (ciclo)

Expertos de negocio liderando los criterios de evaluación

Los indicadores de evaluación (especialmente los subjetivos) deben ser definidos por los expertos de negocio más veteranos:

1. Movilizar con objetivos de negocio: no decir “ayúdanos a definir indicadores de evaluación”, sino “este Agent te va a ayudar a reducir el ciclo de producción de cursos de 2 semanas a 3 días, manteniendo más del 90% de satisfacción de usuarios”.

2. Proporcionar herramientas estructuradas para bajar la barrera: plantillas de escalas de puntuación, herramientas de anotación de casos, preguntas guía como “si solo pudieras mirar tres indicadores para juzgar la calidad de un curso, ¿cuáles elegirías?”.

3. Establecer mecanismos de colaboración continua: en las reuniones semanales de revisión, los expertos miran los datos y el equipo técnico ajusta el sistema, tomando decisiones conjuntas.

La palanca de eficiencia de la evaluación

No toda evaluación necesita ser completamente automática y de cobertura total. Empieza por el método más simple:

Nivel 1: Muestreo manual → "Copiar el código y ejecutarlo"
Nivel 2: Scripts automatizados → "Escribir un script para ejecución por lotes"
Nivel 3: Pipeline de evaluación integrado → Integración CI/CD, activación automática en cada PR
Nivel 4: Monitorización continua → Monitorización en tiempo real de indicadores clave en producción

Cada nivel tiene un retorno de inversión diferente. En las fases iniciales, el Nivel 1 tiene el mayor retorno: detecta problemas más rápido y con el menor coste de implementación. A medida que el sistema madura, se evoluciona gradualmente hacia niveles superiores.

Estrategia de despliegue de modelos

Marco de análisis de requisitos de negocio

Al publicar una aplicación de modelo grande en producción, el primer paso no es la selección tecnológica, sino el análisis de requisitos:

┌────────────────────────────────────────────┐
│           Matriz de análisis de requisitos   │
│                de negocio                   │
├────────────────────────────────────────────┤
│                                            │
│  Requisitos funcionales (qué hacer):         │
│  ├─ Procesamiento de lenguaje natural → LLM general │
│  ├─ Generación de código → LLM optimizado para código │
│  ├─ Razonamiento matemático → LLM fine-tuned en mates │
│  ├─ Comprensión visual → Modelo multimodal  │
│  └─ Procesamiento de voz → Modelo de voz    │
│                                            │
│  Requisitos no funcionales (cómo hacerlo):   │
│  ├─ Rendimiento: TTFT < 500ms, TPOT < 50ms │
│  ├─ Coste: llamada individual < $0.01       │
│  ├─ Estabilidad: 99.9% disponibilidad       │
│  ├─ Seguridad: filtrado de contenido,       │
│  │   protección de privacidad               │
│  └─ Cumplimiento: requisitos regulatorios   │
│                                            │
└────────────────────────────────────────────┘

Estrategia de selección de modelos

No todos los escenarios requieren el modelo más grande. La selección de modelos sigue el principio de “mínimo viable”:

Complejidad de la tarea
    ↑
    │  ┌──────────────────────────┐
    │  │ Modelo grande (GPT-4, Claude)│
    │  │ - Razonamiento complejo   │
    │  │ - Planificación multi-paso│
    │  │ - Generación creativa     │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ Modelo medio (GPT-4o-mini) │
    │  │ - Reconocimiento de intención│
    │  │ - Extracción estructurada │
    │  │ - Generación de resúmenes │
    │  └──────────────────────────┘
    │  ┌──────────────────────────┐
    │  │ Modelo pequeño / destilado│
    │  │ - Clasificación de texto  │
    │  │ - Coincidencia de keywords│
    │  │ - Validación de formato   │
    │  └──────────────────────────┘
    └─────────────────────────────────→ Frecuencia de invocación

Destilación: dotar a modelos pequeños de capacidad profesional

La idea central de la destilación: “copiar” la capacidad de juicio del modelo grande al modelo pequeño.

Modelo profesor (GPT-4)      Modelo estudiante (0.6B parámetros)
      │                         │
      │  Generar datos anotados   │
      ├─────────────────────────→│
      │  "Comprende la intención  │  Aprender los patrones
      │   de la solicitud"        │  de comportamiento
      │  [pares entrada→salida]   │  del profesor
      │                         │
      │  Resultado: el modelo     │
      │  pequeño se acerca al     │
      │  nivel del profesor en    │
      │  tareas específicas       │

Destilación vs fine-tuning:

Tres rutas de destilación:

Optimización de inferencia

Marco de optimización de rendimiento

La optimización de inferencia de LLM se divide en cuatro direcciones:

Procesar solicitudes más rápido

• Miniaturización del modelo: elegir variantes con menos parámetros
• Cuantización: cuantización INT4/INT8/FP16 para reducir requisitos de recursos computacionales
• Poda: eliminar pesos redundantes, reducir complejidad del modelo
• Destilación de conocimiento: entrenar modelos pequeños con datos del modelo grande

Comparativa de precisión de cuantización:
  FP32 (precisión completa) → rendimiento base, mayor coste computacional
  FP16 (media precisión)   → ~2x aceleración, precisión casi sin pérdida
  INT8                     → ~4x aceleración, pequeña pérdida de precisión
  INT4                     → ~8x aceleración, requiere evaluar cuidadosamente el impacto en precisión

Reducir el número de solicitudes procesadas

• Caché de contexto (Context Cache): cachear el prefijo común de diálogos multi-turno para reducir cálculos repetidos
• Procesamiento por lotes (Batching): combinar múltiples solicitudes en un lote para mejorar la utilización del hardware
• Caché de resultados: devolver directamente resultados cacheados para preguntas frecuentes idénticas

# Aplicación típica de caché de contexto
# En diálogos multi-turno, el System Prompt + documentos de conocimiento histórico son el prefijo común
# Primera ronda: cálculo completo (precio completo)
# Rondas posteriores: la parte cacheada se factura al 20% del precio

Reducir tokens de entrada y salida

• Lado de entrada: simplificar la entrada, eliminar información redundante, generar primero un resumen de documentos largos
• Lado de salida: guiar respuestas concisas mediante Prompt, establecer max_tokens razonable

La filosofía de diseño de max_tokens: es una válvula de seguridad, no un mecanismo de control de contenido. Las respuestas breves y semánticamente completas deben guiarse mediante Prompt; max_tokens es más adecuado como última línea de defensa para el control de costes.

Procesamiento en paralelo

La inferencia de modelos grandes es esencialmente cálculo matricial a gran escala. Comprender las diferencias entre CPU y GPU:

Estrategias de paralelización en GPU:

• Paralelismo de datos: distribuir fragmentos de datos entre múltiples GPUs
• Paralelismo de modelo: distribuir distintas capas del modelo entre diferentes dispositivos
• Paralelismo de pipeline: dividir el proceso de cálculo en etapas ejecutadas secuencialmente

No dependas por defecto del modelo grande

En muchos escenarios, enfoques más simples resultan más eficientes:

Barreras de seguridad

Amenazas de seguridad para los modelos grandes

Las aplicaciones con modelos grandes enfrentan amenazas de seguridad en múltiples niveles, que requieren estrategias de defensa sistemáticas:

Matriz de estrategias de defensa

Implementación de ingeniería de barreras de seguridad

class SafetyGuard:
    """Implementación de barreras de seguridad multicapa"""

    def __init__(self):
        self.blocked_keywords = set()    # Palabras sensibles personalizadas
        self.rate_limits = {}            # Registro de limitación de tasa
        self.max_tool_calls = 10         # Máximo de invocaciones de herramienta del Agent
        self.dangerous_commands = {      # Lista negra de comandos peligrosos
            "rm -rf", "DROP TABLE", "DELETE FROM",
            "os.system", "subprocess.call", "eval("
        }

    def check_input(self, user_input: str) -> tuple[bool, str]:
        """Verificación de seguridad de entrada"""
        # 1. Detección de palabras sensibles
        for keyword in self.blocked_keywords:
            if keyword in user_input.lower():
                return False, f"La entrada contiene palabra sensible: {keyword}"

        # 2. Detección de inyección de comandos
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in user_input.lower():
                return False, f"Se detectó posible instrucción peligrosa: {dangerous}"

        return True, "Aprobado"

    def check_tool_call(self, tool_name: str, args: dict) -> tuple[bool, str]:
        """Auditoría previa a la invocación de herramienta"""
        # 1. Verificar frecuencia de invocación
        if self.rate_limits.get(tool_name, 0) >= self.max_tool_calls:
            return False, f"La herramienta {tool_name} excedió el límite de invocaciones"

        # 2. Verificar seguridad de parámetros
        args_str = json.dumps(args).lower()
        for dangerous in self.dangerous_commands:
            if dangerous.lower() in args_str:
                return False, f"Los parámetros de la herramienta contienen instrucción peligrosa: {dangerous}"

        self.rate_limits[tool_name] = self.rate_limits.get(tool_name, 0) + 1
        return True, "Aprobado"

    def check_output(self, output: str) -> tuple[bool, str]:
        """Revisión de contenido de salida"""
        # Detectar si contiene patrones de información sensible que no deberían mostrarse
        sensitive_patterns = [
            r'\b\d{17}[\dXx]\b',           # Número de identificación
            r'\b1[3-9]\d{9}\b',            # Número de teléfono móvil
            r'[Pp]assword\s*[:=]\s*\S+',  # Patrón de contraseña
        ]

        for pattern in sensitive_patterns:
            if re.search(pattern, output):
                return False, f"La salida podría contener información sensible: {pattern}"

        return True, "Aprobado"

Mecanismo de fusible (Circuit Breaker)

Establecer límites de recursos explícitos para cada tarea del Agent:

class CircuitBreaker:
    """Fusible del Agent: evita que los bucles descontrolados causen grandes pérdidas"""

    def __init__(self,
                 max_api_calls: int = 10,      # máximo de llamadas API
                 max_wall_time: int = 300,      # tiempo máximo de ejecución (segundos)
                 max_cost: float = 0.50):       # coste máximo (dólares)
        self.max_api_calls = max_api_calls
        self.max_wall_time = max_wall_time
        self.max_cost = max_cost
        self.reset()

    def reset(self):
        self.api_calls = 0
        self.start_time = time.time()
        self.total_cost = 0.0

    def check(self) -> tuple[bool, str]:
        """Verificar si se debe activar el fusible"""
        self.api_calls += 1
        elapsed = time.time() - self.start_time

        if self.api_calls > self.max_api_calls:
            return False, f"Límite de llamadas API excedido ({self.api_calls}/{self.max_api_calls})"
        if elapsed > self.max_wall_time:
            return False, f"Tiempo de ejecución excedido ({elapsed:.0f}s/{self.max_wall_time}s)"
        if self.total_cost > self.max_cost:
            return False, f"Coste excedido (${self.total_cost:.2f}/${self.max_cost:.2f})"

        return True, "Normal"

Harness Engineering

El plano general del entorno de producción

Harness Engineering es el conjunto completo de prácticas de ingeniería para garantizar que el sistema Agent funcione de forma estable desde el desarrollo hasta la producción. Incluye:

Observabilidad

Usar el estándar OpenTelemetry para establecer tres tipos de recolección de datos:

• Metrics (métricas): consumo de tokens, distribución de latencia, tasa de errores
• Traces (trazas): cada etapa por la que pasa una solicitud y su duración
• Logs (registros): entrada y salida de cada etapa, pilas de errores, información de auditoría

# Usar OpenTelemetry para instrumentar llamadas del Agent
from opentelemetry import trace
from opentelemetry.instrumentation.openai import OpenAIInstrumentor

# Instrumentar automáticamente llamadas a la API de OpenAI
OpenAIInstrumentor().instrument()

tracer = trace.get_tracer(__name__)

@tracer.start_as_current_span("agent_task")
def run_agent_task(task: str):
    # El span registra automáticamente duración y contexto
    with tracer.start_as_current_span("llm_call") as span:
        span.set_attribute("task", task)
        result = agent.run(task)
        span.set_attribute("tokens_used", result.usage.total_tokens)
        return result

Checklist previa al despliegue

Antes de llevar un Agent a producción, completa las siguientes verificaciones:

☐ SLO definidos
  ├─ TTFT (tiempo hasta el primer token) objetivo: _____ ms
  ├─ TPOT (tiempo por token generado) objetivo: _____ ms
  └─ Objetivo de disponibilidad: _____%

☐ Control de costes
  ├─ Presupuesto máximo por llamada: $_____
  ├─ Límite diario de consumo de tokens: _____ tokens
  └─ Umbrales de alerta configurados

☐ Protección de seguridad
  ├─ Verificación de seguridad de entrada activada
  ├─ Revisión de contenido de salida activada
  ├─ Fusible de comportamiento del Agent configurado
  └─ Alertas de monitorización de seguridad configuradas

☐ Plan de recuperación ante desastres
  ├─ Ruta de degradación de modelo definida
  ├─ Lógica de respaldo para rutas críticas validada
  └─ Simulacro de recuperación de fallos completado

☐ Línea base de evaluación
  ├─ Puntuación de evaluación end-to-end: _____
  ├─ Puntuación de evaluación por componentes: _____
  └─ Pipeline de pruebas de regresión integrado

☐ Observabilidad
  ├─ OpenTelemetry integrado
  ├─ Dashboard de indicadores clave creado
  └─ Reglas de alerta configuradas y validadas

Estrategia de despliegue progresivo

No hacer un lanzamiento completo de una sola vez. Adoptar una estrategia progresiva:

Fase 1: Pruebas internas (1-2 semanas)
  └─ Usado por el equipo, recopilar feedback inicial

Fase 2: Canary release a pequeña escala (5% de usuarios)
  └─ Comparar indicadores de evaluación entre el sistema nuevo y el antiguo

Fase 3: Ampliación del alcance (25% → 50% → 100%)
  └─ Permanecer 3-5 días en cada etapa observando indicadores

Fase 4: Lanzamiento completo
  └─ Mantener monitorización, establecer ciclo de mejora continua

Mejores prácticas de operación en producción

Gestión de línea base de evaluación:

• Establecer la versión actual en producción como línea base; cualquier versión nueva debe superarla
• Re-evaluar periódicamente (semanalmente) la línea base y las versiones candidatas con los datos más recientes
• Integrar la verificación de línea base en el pipeline de despliegue; bloquear automáticamente las versiones que no la alcancen

Estrategia de degradación por capas:

1. Modelo principal no disponible → cambiar al modelo de respaldo
2. Modelo de respaldo también no disponible → usar respuestas frecuentes cacheadas
3. Caché no acierta → devolver plantilla de respuesta de degradación predefinida

Gobernanza de costes:

• Analizar el consumo de tokens por modelo, por usuario y por tipo de tarea
• Identificar picos anómalos de coste y configurar alertas
• Revisar periódicamente: ¿hay contextos largos innecesarios, System Prompts redundantes?

Destilación de modelos: enseñar a modelos pequeños conocimiento de dominio

Por qué importa la destilación

Los modelos grandes (GPT-4, Claude, Qwen-72B) ofrecen excelente calidad pero conllevan altos costes de inferencia y latencia. Para sistemas en producción que manejan miles de solicitudes por minuto, el coste en tokens puede llegar a ser prohibitivo. La destilación de modelos ofrece una solución práctica: usar las salidas del modelo grande como datos de entrenamiento para enseñar a un modelo pequeño (7B-14B) a replicar el mismo comportamiento a una fracción del coste.

Modelo profesor (GPT-4, 175B parámetros)
  │
  ├── Generar respuestas de alta calidad para tareas de dominio
  │
  ▼
Datos de entrenamiento (pares entrada → salida_del_profesor)
  │
  ├── Fine-tuning del modelo estudiante
  │
  ▼
Modelo estudiante (7B-14B parámetros)
  │
  ├── Misma calidad, coste 10-50x menor
  └── Latencia 5-10x menor

Pipeline de destilación

Paso 1: Definir el alcance de la tarea

La destilación funciona mejor cuando la tarea está bien definida y es repetitiva. Identifica los dominios específicos donde necesitas que el modelo pequeño rinda:

# Ejemplo: clasificación de intenciones de soporte al cliente
task_definitions = [
    {
        "name": "intent_classification",
        "input_schema": {"user_message": "string", "context": "string"},
        "output_schema": {"intent": "string", "confidence": "float", "reasoning": "string"},
    },
    {
        "name": "response_generation",
        "input_schema": {"intent": "string", "knowledge": "string", "tone": "string"},
        "output_schema": {"response": "string", "sources": ["string"]},
    },
]

Paso 2: Generar datos de entrenamiento con el modelo profesor

Usar el modelo grande para generar ejemplos de alta calidad para tu dominio:

from openai import OpenAI

teacher = OpenAI(api_key="...")  # GPT-4 o similar
student = OpenAI(base_url="http://localhost:8000/v1")  # Tu modelo pequeño

def generate_training_examples(task_def: dict, n_examples: int = 1000) -> list:
    examples = []
    for i in range(n_examples):
        # Generar entradas diversas
        input_prompt = f"Genera una entrada realista para la tarea '{task_def['name']}'. Varía la complejidad y los casos extremos."
        input_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[{"role": "user", "content": input_prompt}],
        )
        user_input = input_resp.choices[0].message.content

        # Generar salida del profesor
        teacher_resp = teacher.chat.completions.create(
            model="gpt-4",
            messages=[
                {"role": "system", "content": f"Eres un experto en {task_def['name']}. Sigue este esquema de salida: {task_def['output_schema']}"},
                {"role": "user", "content": user_input},
            ],
        )
        teacher_output = teacher_resp.choices[0].message.content

        examples.append({"input": user_input, "output": teacher_output})

    return examples

training_data = generate_training_examples(task_definitions[0], n_examples=2000)

Paso 3: Fine-tuning del modelo estudiante

Usar LoRA o fine-tuning completo para entrenar el modelo pequeño con las salidas del profesor:

# Usando Hugging Face Transformers + PEFT para fine-tuning con LoRA
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import LoraConfig, get_peft_model

model_name = "meta-llama/Llama-2-7b-hf"
model = AutoModelForCausalLM.from_pretrained(model_name)
tokenizer = AutoTokenizer.from_pretrained(model_name)

# Configurar LoRA para fine-tuning eficiente
lora_config = LoraConfig(
    r=16,  # Rango de LoRA
    lora_alpha=32,
    target_modules=["q_proj", "v_proj"],
    lora_dropout=0.05,
    bias="none",
    task_type="CAUSAL_LM",
)

model = get_peft_model(model, lora_config)
model.print_trainable_parameters()
# Salida: trainable params: 4,194,304 || all params: 6,742,609,920 || trainable%: 0.0622

# Entrenar con los datos de destilación
# ... (bucle de entrenamiento estándar)

Paso 4: Evaluar e iterar

Comparar las salidas del modelo estudiante con las del profesor en un conjunto de prueba separado:

def evaluate_student_vs_teacher(test_set: list, teacher_client, student_client) -> dict:
    results = {"exact_match": 0, "semantic_similarity": 0, "total": len(test_set)}

    for example in test_set:
        student_resp = student_client.chat.completions.create(
            model="student-7b",
            messages=[{"role": "user", "content": example["input"]}],
        )
        student_output = student_resp.choices[0].message.content
        teacher_output = example["output"]

        # Verificación de coincidencia exacta
        if student_output.strip() == teacher_output.strip():
            results["exact_match"] += 1

        # Similitud semántica (usando embeddings)
        # ... (calcular similitud de coseno entre las salidas)

    results["exact_match_rate"] = results["exact_match"] / results["total"]
    return results

# Objetivo: >85% de similitud semántica con el profesor

Cuándo destilar

Mejores prácticas de producción: monitoreo, canary releases, A/B testing

Stack de observabilidad

Los sistemas Agent en producción necesitan observabilidad integral. Configurar las siguientes capas de monitoreo:

1. Métricas a nivel de aplicación

import time
from prometheus_client import Counter, Histogram, Gauge

# Métricas de solicitudes
request_counter = Counter('agent_requests_total', 'Total de solicitudes', ['model', 'intent', 'status'])
request_latency = Histogram('agent_request_duration_seconds', 'Latencia de solicitudes', ['model', 'intent'])
active_sessions = Gauge('agent_active_sessions', 'Sesiones de conversación activas')

# Métricas de uso de tokens
token_usage = Counter('agent_tokens_total', 'Uso de tokens', ['model', 'type'])  # type: input/output

# Métricas de calidad
hallucination_rate = Gauge('agent_hallucination_rate', 'Tasa de alucinaciones estimada')
user_satisfaction = Histogram('agent_user_satisfaction', 'Puntuación de satisfacción del usuario', buckets=[1, 2, 3, 4, 5])

def track_request(model: str, intent: str, status: str, latency: float, tokens_in: int, tokens_out: int):
    request_counter.labels(model=model, intent=intent, status=status).inc()
    request_latency.labels(model=model, intent=intent).observe(latency)
    token_usage.labels(model=model, type='input').inc(tokens_in)
    token_usage.labels(model=model, type='output').inc(tokens_out)

2. Trazado distribuido

Usar OpenTelemetry para trazar solicitudes a través del pipeline del Agent:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

def process_user_request(user_input: str):
    with tracer.start_as_current_span("process_request") as span:
        span.set_attribute("user.input_length", len(user_input))

        with tracer.start_as_current_span("retrieve_context") as retrieve_span:
            context = retrieve_relevant_docs(user_input)
            retrieve_span.set_attribute("docs.retrieved", len(context))

        with tracer.start_as_current_span("llm_generate") as llm_span:
            llm_span.set_attribute("model", "gpt-4")
            response = call_llm(user_input, context)
            llm_span.set_attribute("tokens.used", response.usage.total_tokens)

        return response