5. Prompt Management Systems

Descripción

Diseñar y construir un sistema para gestionar prompts en equipo: templates con variables, versionado integrado, testing automatizado, access control, y audit trail. Comparación entre construir tu propio sistema (build) vs usar herramientas existentes (buy).

Por Qué Necesitas un Prompt Management System

Sin un sistema de gestión, los prompts se dispersan y crean problemas:

Sin Prompt Management System:
──────────────────────────────
• Prompt hardcodeado en 5 archivos diferentes
• El equipo no sabe qué versión está en producción
• Un cambio de prompt requiere un deploy completo
• No hay forma de probar un cambio antes de producción
• Dos personas editan el mismo prompt concurrentemente
• Nadie sabe quién cambió qué o cuándo

Con Prompt Management System:
──────────────────────────────
• Prompts centralizados con versioning
• La versión activa es visible y cambiable sin deploy
• Rollback instantáneo si algo falla
• Testing integrado antes de activar nueva versión
• Audit trail de todos los cambios
• Templating con variables reutilizables

Componentes del Sistema

Un prompt management system completo tiene:

1. TEMPLATE ENGINE    → Jinja2 para variables y lógica
2. REGISTRY           → Almacenamiento versionado de prompts
3. VARIABLE STORE     → Valores de variables compartidos
4. TESTING LAYER      → Regression tests antes de activar
5. ACCESS CONTROL     → Quién puede leer/escribir/activar
6. AUDIT TRAIL        → Log de todos los cambios
7. API               → Interface para servicios

Implementación: Template Engine con Jinja2

from jinja2 import Environment, FileSystemLoader, Template, TemplateError
from pathlib import Path
import json


class PromptTemplateEngine:
    """
    Motor de templates para prompts con Jinja2.
    
    Soporta: variables, condicionales, loops, filtros, includes.
    """
    
    def __init__(self, templates_dir: str = "prompts/templates"):
        self.templates_dir = Path(templates_dir)
        self.templates_dir.mkdir(parents=True, exist_ok=True)
        
        # Jinja2 environment con templates desde directorio
        self.env = Environment(
            loader=FileSystemLoader(str(self.templates_dir)),
            trim_blocks=True,      # Elimina newline después de bloques
            lstrip_blocks=True     # Elimina espacios antes de bloques
        )
        
        # Registrar filtros custom
        self.env.filters["truncar"] = lambda s, n=200: s[:n] + "..." if len(s) > n else s
        self.env.filters["mayusculas"] = lambda s: s.upper()
        self.env.filters["lista_a_texto"] = lambda lista: ", ".join(lista)
    
    def renderizar_string(self, template_str: str, variables: dict) -> str:
        """Renderiza un template string con variables."""
        try:
            template = Template(template_str)
            return template.render(**variables)
        except TemplateError as e:
            raise ValueError(f"Error en template: {e}")
    
    def renderizar_archivo(self, nombre_archivo: str, variables: dict) -> str:
        """Renderiza un template desde archivo."""
        try:
            template = self.env.get_template(nombre_archivo)
            return template.render(**variables)
        except TemplateError as e:
            raise ValueError(f"Error en template {nombre_archivo}: {e}")
    
    def validar_variables(self, template_str: str, variables_disponibles: dict) -> dict:
        """
        Valida que todas las variables requeridas están disponibles.
        
        Returns: {"valido": bool, "variables_faltantes": list, "variables_extra": list}
        """
        from jinja2 import meta
        
        env_temp = Environment()
        ast = env_temp.parse(template_str)
        variables_requeridas = meta.find_undeclared_variables(ast)
        
        faltantes = variables_requeridas - set(variables_disponibles.keys())
        extras = set(variables_disponibles.keys()) - variables_requeridas
        
        return {
            "valido": len(faltantes) == 0,
            "variables_requeridas": list(variables_requeridas),
            "variables_faltantes": list(faltantes),
            "variables_extra": list(extras),
            "variables_disponibles": list(variables_disponibles.keys())
        }


# Ejemplos de templates complejos:

TEMPLATE_CLASIFICADOR_AVANZADO = """
Eres un clasificador de {{ dominio }}.

{% if instrucciones_especiales %}
INSTRUCCIONES ESPECIALES:
{{ instrucciones_especiales }}

{% endif %}
Clasifica el siguiente texto en una de estas categorías:
{% for cat in categorias %}
- {{ cat }}
{% endfor %}

{% if ejemplos %}
EJEMPLOS:
{% for ej in ejemplos %}
- "{{ ej.input }}" → {{ ej.output }}
{% endfor %}

{% endif %}
IMPORTANTE: Responde SOLO con la categoría. Sin explicaciones.

Texto: {{ input }}
Categoría:
"""

# Renderizar:
engine = PromptTemplateEngine()

prompt = engine.renderizar_string(
    TEMPLATE_CLASIFICADOR_AVANZADO,
    variables={
        "dominio": "sentimiento de reviews",
        "categorias": ["POSITIVO", "NEGATIVO", "NEUTRO", "MIXTO"],
        "instrucciones_especiales": "Presta especial atención al sarcasmo",
        "ejemplos": [
            {"input": "Me encanta el producto", "output": "POSITIVO"},
            {"input": "Claro, 'rápido' con 3 semanas de espera", "output": "NEGATIVO"}
        ],
        "input": "El envío tardó mucho pero el producto es bueno"
    }
)
print(prompt)

Sistema Completo con Todas las Capas

import json
import sqlite3
from datetime import datetime
from pathlib import Path
from contextlib import contextmanager
from typing import Optional
from openai import OpenAI

client = OpenAI()


class PromptManagementSystem:
    """
    Sistema de gestión de prompts completo.
    
    Integra: template engine + registry + testing + audit trail + access control.
    """
    
    def __init__(self, db_path: str = "prompt_management.db"):
        self.db_path = db_path
        self.template_engine = PromptTemplateEngine()
        self._init_db()
    
    def _init_db(self) -> None:
        """Inicializa la base de datos con todas las tablas."""
        with self._conn() as conn:
            # Tabla principal de prompts
            conn.execute("""
                CREATE TABLE IF NOT EXISTS prompts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    nombre TEXT NOT NULL,
                    version TEXT NOT NULL,
                    template TEXT NOT NULL,
                    variables_schema TEXT DEFAULT '{}',
                    changelog TEXT DEFAULT '',
                    activa INTEGER DEFAULT 0,
                    creado_por TEXT DEFAULT 'system',
                    created_at TEXT NOT NULL,
                    UNIQUE(nombre, version)
                )
            """)
            
            # Tabla de variable stores (valores compartidos)
            conn.execute("""
                CREATE TABLE IF NOT EXISTS variable_store (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    nombre TEXT NOT NULL,
                    clave TEXT NOT NULL,
                    valor TEXT NOT NULL,
                    tipo TEXT DEFAULT 'string',
                    updated_at TEXT NOT NULL,
                    UNIQUE(nombre, clave)
                )
            """)
            
            # Audit trail
            conn.execute("""
                CREATE TABLE IF NOT EXISTS audit_log (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    prompt_nombre TEXT NOT NULL,
                    accion TEXT NOT NULL,
                    version_anterior TEXT,
                    version_nueva TEXT,
                    usuario TEXT NOT NULL,
                    timestamp TEXT NOT NULL,
                    detalles TEXT DEFAULT ''
                )
            """)
            
            # Test results
            conn.execute("""
                CREATE TABLE IF NOT EXISTS test_results (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    prompt_nombre TEXT NOT NULL,
                    prompt_version TEXT NOT NULL,
                    accuracy REAL,
                    pasados INTEGER,
                    fallidos INTEGER,
                    timestamp TEXT NOT NULL
                )
            """)
    
    @contextmanager
    def _conn(self):
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        try:
            yield conn
            conn.commit()
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()
    
    def registrar(
        self,
        nombre: str,
        version: str,
        template: str,
        variables_schema: dict | None = None,
        changelog: str = "",
        usuario: str = "system"
    ) -> None:
        """
        Registra una nueva versión de un prompt.
        
        variables_schema: Dict con definición de variables requeridas.
            {"input": {"tipo": "string", "requerido": True, "descripcion": "..."},
             "categorias": {"tipo": "list", "requerido": True}}
        """
        schema_json = json.dumps(variables_schema or {})
        
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO prompts (nombre, version, template, variables_schema, changelog, created_at, creado_por)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (nombre, version, template, schema_json, changelog, datetime.now().isoformat(), usuario))
            
            # Si es el primero, activarlo automáticamente
            count = conn.execute(
                "SELECT COUNT(*) as n FROM prompts WHERE nombre = ? AND activa = 1", (nombre,)
            ).fetchone()["n"]
            
            if count == 0:
                conn.execute(
                    "UPDATE prompts SET activa = 1 WHERE nombre = ? AND version = ?",
                    (nombre, version)
                )
        
        # Audit
        self._log(nombre, "REGISTRADO", version_nueva=version, usuario=usuario, detalles=changelog)
        print(f"✓ Registrado: {nombre} {version}")
    
    def activar(
        self,
        nombre: str,
        version: str,
        usuario: str = "system",
        verificar_tests: bool = True
    ) -> None:
        """
        Activa una versión de un prompt.
        
        Si verificar_tests=True, verifica que la versión pasó los tests
        antes de activar (protección contra activar versiones no testeadas).
        """
        if verificar_tests:
            with self._conn() as conn:
                test_result = conn.execute("""
                    SELECT accuracy FROM test_results
                    WHERE prompt_nombre = ? AND prompt_version = ?
                    ORDER BY timestamp DESC LIMIT 1
                """, (nombre, version)).fetchone()
            
            if not test_result:
                raise ValueError(
                    f"No hay resultados de tests para {nombre} {version}. "
                    "Ejecuta los tests antes de activar."
                )
            
            if test_result["accuracy"] < 0.80:
                raise ValueError(
                    f"La versión {version} no pasó los tests "
                    f"(accuracy: {test_result['accuracy']:.2%}). "
                    "Mínimo requerido: 80%"
                )
        
        with self._conn() as conn:
            # Obtener versión actual para audit
            actual = conn.execute(
                "SELECT version FROM prompts WHERE nombre = ? AND activa = 1",
                (nombre,)
            ).fetchone()
            version_anterior = actual["version"] if actual else None
            
            # Desactivar todas, activar nueva
            conn.execute("UPDATE prompts SET activa = 0 WHERE nombre = ?", (nombre,))
            conn.execute(
                "UPDATE prompts SET activa = 1 WHERE nombre = ? AND version = ?",
                (nombre, version)
            )
        
        self._log(
            nombre, "ACTIVADO",
            version_anterior=version_anterior,
            version_nueva=version,
            usuario=usuario
        )
        print(f"✓ Activado: {nombre} {version} (anterior: {version_anterior})")
    
    def rollback(
        self,
        nombre: str,
        to_version: Optional[str] = None,
        usuario: str = "system"
    ) -> None:
        """
        Rollback a versión anterior.
        
        Si to_version es None, retrocede una versión.
        """
        with self._conn() as conn:
            versiones = [
                row["version"] for row in conn.execute(
                    "SELECT version FROM prompts WHERE nombre = ? ORDER BY created_at ASC",
                    (nombre,)
                ).fetchall()
            ]
            
            actual = conn.execute(
                "SELECT version FROM prompts WHERE nombre = ? AND activa = 1",
                (nombre,)
            ).fetchone()
            
            if not actual:
                raise ValueError(f"No hay versión activa para '{nombre}'")
            
            version_actual = actual["version"]
            
            if to_version is None:
                idx = versiones.index(version_actual) if version_actual in versiones else len(versiones)
                if idx == 0:
                    raise ValueError("Ya estás en la primera versión")
                to_version = versiones[idx - 1]
        
        # Activar sin verificar tests (es un rollback de emergencia)
        self.activar(nombre, to_version, usuario=usuario, verificar_tests=False)
        print(f"🔄 ROLLBACK: {nombre} {version_actual} → {to_version}")
    
    def render(
        self,
        nombre: str,
        variables: dict,
        version: Optional[str] = None
    ) -> str:
        """
        Obtiene y renderiza el prompt con variables.
        
        Si version es None, usa la activa.
        """
        with self._conn() as conn:
            if version:
                row = conn.execute(
                    "SELECT template, variables_schema FROM prompts WHERE nombre = ? AND version = ?",
                    (nombre, version)
                ).fetchone()
            else:
                row = conn.execute(
                    "SELECT template, variables_schema FROM prompts WHERE nombre = ? AND activa = 1",
                    (nombre,)
                ).fetchone()
        
        if not row:
            raise KeyError(f"Prompt '{nombre}' {f'v{version}' if version else '(activo)'} no encontrado")
        
        template = row["template"]
        schema = json.loads(row["variables_schema"])
        
        # Validar variables si hay schema
        if schema:
            validacion = self.template_engine.validar_variables(template, variables)
            if not validacion["valido"]:
                raise ValueError(
                    f"Variables faltantes: {validacion['variables_faltantes']}. "
                    f"Variables disponibles: {validacion['variables_disponibles']}"
                )
        
        return self.template_engine.renderizar_string(template, variables)
    
    def ejecutar(
        self,
        nombre: str,
        variables: dict,
        model: str = "gpt-4o-mini",
        version: Optional[str] = None
    ) -> dict:
        """Renderiza el prompt, lo ejecuta y retorna resultado."""
        prompt = self.render(nombre, variables, version)
        
        response = client.chat.completions.create(
            model=model,
            messages=[{"role": "user", "content": prompt}],
            temperature=0
        )
        
        return {
            "output": response.choices[0].message.content,
            "prompt_usado": nombre,
            "version_usada": version or self.version_activa(nombre),
            "tokens": response.usage.total_tokens
        }
    
    def version_activa(self, nombre: str) -> Optional[str]:
        """Retorna la versión activa de un prompt."""
        with self._conn() as conn:
            row = conn.execute(
                "SELECT version FROM prompts WHERE nombre = ? AND activa = 1",
                (nombre,)
            ).fetchone()
        return row["version"] if row else None
    
    def guardar_resultado_test(
        self,
        nombre: str,
        version: str,
        accuracy: float,
        pasados: int,
        fallidos: int
    ) -> None:
        """Guarda el resultado de un test para el gate de activación."""
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO test_results (prompt_nombre, prompt_version, accuracy, pasados, fallidos, timestamp)
                VALUES (?, ?, ?, ?, ?, ?)
            """, (nombre, version, accuracy, pasados, fallidos, datetime.now().isoformat()))
    
    def _log(
        self,
        nombre: str,
        accion: str,
        version_nueva: Optional[str] = None,
        version_anterior: Optional[str] = None,
        usuario: str = "system",
        detalles: str = ""
    ) -> None:
        """Registra en el audit trail."""
        with self._conn() as conn:
            conn.execute("""
                INSERT INTO audit_log (prompt_nombre, accion, version_anterior, version_nueva, usuario, timestamp, detalles)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            """, (nombre, accion, version_anterior, version_nueva, usuario, datetime.now().isoformat(), detalles))
    
    def audit_trail(self, nombre: str, limit: int = 20) -> list[dict]:
        """Retorna el historial de cambios de un prompt."""
        with self._conn() as conn:
            rows = conn.execute("""
                SELECT * FROM audit_log WHERE prompt_nombre = ?
                ORDER BY timestamp DESC LIMIT ?
            """, (nombre, limit)).fetchall()
        
        return [dict(row) for row in rows]
    
    def set_variable(self, namespace: str, clave: str, valor, tipo: str = "string") -> None:
        """Guarda una variable reutilizable en el variable store."""
        with self._conn() as conn:
            conn.execute("""
                INSERT OR REPLACE INTO variable_store (nombre, clave, valor, tipo, updated_at)
                VALUES (?, ?, ?, ?, ?)
            """, (namespace, clave, json.dumps(valor), tipo, datetime.now().isoformat()))
    
    def get_variable(self, namespace: str, clave: str):
        """Obtiene una variable del variable store."""
        with self._conn() as conn:
            row = conn.execute(
                "SELECT valor, tipo FROM variable_store WHERE nombre = ? AND clave = ?",
                (namespace, clave)
            ).fetchone()
        
        if not row:
            raise KeyError(f"Variable '{namespace}.{clave}' no encontrada")
        
        return json.loads(row["valor"])
    
    def get_variables_namespace(self, namespace: str) -> dict:
        """Obtiene todas las variables de un namespace."""
        with self._conn() as conn:
            rows = conn.execute(
                "SELECT clave, valor, tipo FROM variable_store WHERE nombre = ?",
                (namespace,)
            ).fetchall()
        
        return {row["clave"]: json.loads(row["valor"]) for row in rows}

Variable Store: Variables Compartidas

# Ejemplo completo de uso del variable store:

pms = PromptManagementSystem()

# Guardar variables compartidas (configurables sin cambiar el template)
pms.set_variable("clasificador", "categorias", ["POSITIVO", "NEGATIVO", "NEUTRO", "MIXTO"])
pms.set_variable("clasificador", "ejemplos", [
    {"input": "Excelente producto", "output": "POSITIVO"},
    {"input": "Terrible servicio", "output": "NEGATIVO"},
])
pms.set_variable("clasificador", "min_score", 0.8)

# Registrar template que usa variables del store
TEMPLATE = """
Eres un clasificador de sentimiento de reviews.

Clasifica en: {{ categorias | lista_a_texto }}

{% if ejemplos %}
Ejemplos:
{% for ej in ejemplos %}
- "{{ ej.input }}" → {{ ej.output }}
{% endfor %}
{% endif %}

Review: {{ input }}
Categoría:
"""

pms.registrar(
    nombre="clasificador_reviews",
    version="v1.0.0",
    template=TEMPLATE,
    variables_schema={
        "input": {"tipo": "string", "requerido": True},
        "categorias": {"tipo": "list", "requerido": True},
        "ejemplos": {"tipo": "list", "requerido": False}
    },
    changelog="Versión inicial",
    usuario="miguel"
)

# Ejecutar combinando variables del store y del request
variables_store = pms.get_variables_namespace("clasificador")

resultado = pms.ejecutar(
    nombre="clasificador_reviews",
    variables={
        **variables_store,  # categorias, ejemplos del store
        "input": "El envío tardó pero el producto es excelente"  # del request
    }
)

print(resultado["output"])  # POSITIVO

Testing Gate: Tests Antes de Activar

def test_y_activar(
    pms: PromptManagementSystem,
    nombre: str,
    version: str,
    golden_set: list[dict],
    accuracy_minima: float = 0.85,
    usuario: str = "system"
) -> bool:
    """
    Ejecuta tests y activa si pasan.
    
    Este patrón asegura que nunca se activa un prompt sin tests verdes.
    """
    print(f"🧪 Testeando {nombre} {version}...")
    
    # Renderizar y testear
    correctos = 0
    variables_comunes = pms.get_variables_namespace(nombre.split("_")[0])
    
    for ej in golden_set:
        try:
            output = pms.ejecutar(
                nombre,
                variables={**variables_comunes, "input": ej["input"]},
                version=version
            )["output"]
            
            if output.strip().lower() == str(ej["expected_output"]).strip().lower():
                correctos += 1
        except Exception as e:
            print(f"  Error en ejemplo {ej.get('id', '?')}: {e}")
    
    n = len(golden_set)
    accuracy = correctos / n if n > 0 else 0.0
    fallidos = n - correctos
    
    # Guardar resultado de tests
    pms.guardar_resultado_test(nombre, version, accuracy, correctos, fallidos)
    
    print(f"  Accuracy: {accuracy:.2%} ({correctos}/{n}) — Mínimo: {accuracy_minima:.2%}")
    
    if accuracy >= accuracy_minima:
        # Activar si los tests pasan
        pms.activar(nombre, version, usuario=usuario)
        print(f"  ✅ Tests pasaron — {nombre} {version} activado")
        return True
    else:
        print(f"  ❌ Tests fallaron — {nombre} {version} NO activado")
        return False

Build vs Buy: Cuándo Construir vs Usar Herramientas

Recomendación:

• Startups y proyectos nuevos: Empezar con Buy (PromptLayer o LangSmith)
• Equipos con requisitos custom: Build cuando las herramientas no cubren tus necesidades
• Empresas con compliance: Build para control total de datos y audit trail

Integración con PromptLayer (Alternativa Managed)

import promptlayer
from openai import OpenAI

# PromptLayer intercepta las llamadas a OpenAI y las registra automáticamente
promptlayer.api_key = "pl_xxx"
openai = promptlayer.openai

def usar_prompt_desde_promptlayer(
    prompt_nombre: str,
    variables: dict
) -> str:
    """
    PromptLayer tiene su propio sistema de templates y versioning.
    Retorna el prompt renderizado desde su plataforma.
    """
    # Obtener template desde PromptLayer
    prompt_template = promptlayer.templates.get(prompt_nombre)
    
    # Formatear con variables
    prompt = prompt_template["prompt"]["template"].format(**variables)
    
    # Ejecutar con tracking automático
    response = openai.ChatCompletion.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0,
        pl_tags=["produccion", prompt_nombre],
        return_pl_id=True
    )
    
    return response.choices[0].message.content

Troubleshooting

Problema 1: Variables no definidas en el template

Síntoma: jinja2.UndefinedError: 'variable_name' is undefined

Causa: Se renderiza el template sin pasar todas las variables requeridas.

Solución:

# Validar antes de renderizar
def render_seguro(pms: PromptManagementSystem, nombre: str, variables: dict) -> str:
    """Renderiza con validación previa de variables."""
    # Obtener schema del prompt
    with pms._conn() as conn:
        row = conn.execute(
            "SELECT variables_schema FROM prompts WHERE nombre = ? AND activa = 1",
            (nombre,)
        ).fetchone()
    
    if row:
        schema = json.loads(row["variables_schema"])
        requeridas = {k for k, v in schema.items() if v.get("requerido", False)}
        faltantes = requeridas - set(variables.keys())
        
        if faltantes:
            raise ValueError(
                f"Variables requeridas faltantes para '{nombre}': {faltantes}\n"
                f"Variables disponibles: {set(variables.keys())}"
            )
    
    return pms.render(nombre, variables)

Problema 2: Colisiones de nombres entre equipos

Síntoma: El equipo A tiene "clasificador" y el equipo B también — se pisan.

Solución:

# Namespace por equipo/dominio en el nombre del prompt
# Formato: {equipo}/{dominio}/{nombre}

pms.registrar("equipo_a/soporte/clasificador_urgencia", "v1.0", template_a)
pms.registrar("equipo_b/ventas/clasificador_intento", "v1.0", template_b)

# Nunca se pisan:
pms.render("equipo_a/soporte/clasificador_urgencia", {...})
pms.render("equipo_b/ventas/clasificador_intento", {...})

Problema 3: Testing lento para golden sets grandes

Síntoma: El test gate tarda 10+ minutos.

Solución:

import asyncio
from openai import AsyncOpenAI

async_client = AsyncOpenAI()

async def test_prompt_async(
    pms: PromptManagementSystem,
    nombre: str,
    version: str,
    golden_set: list[dict],
    max_concurrent: int = 10
) -> float:
    """Versión async del test gate para golden sets grandes."""
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def evaluar(ej):
        async with semaphore:
            template = pms.render(nombre, {"input": ej["input"]}, version)
            response = await async_client.chat.completions.create(
                model="gpt-4o-mini",
                messages=[{"role": "user", "content": template}],
                temperature=0
            )
            output = response.choices[0].message.content.strip()
            return output.lower() == str(ej["expected_output"]).lower()
    
    resultados = await asyncio.gather(*[evaluar(ej) for ej in golden_set])
    return sum(resultados) / len(resultados)

Ejercicios

Ejercicio 1: Crear un template con variables condicionales

Crea un template para un prompt de resumen que incluya contexto opcional:

from jinja2 import Template

TEMPLATE_RESUMEN = """
{% if audiencia %}
Escribe para una audiencia de {{ audiencia }}.
{% endif %}
{% if max_palabras %}
Resumen en máximo {{ max_palabras }} palabras.
{% else %}
Resumen en máximo 3 oraciones.
{% endif %}
{% if estilo %}
Estilo: {{ estilo }}.
{% endif %}

Texto a resumir:
{{ texto }}

Resumen:
"""

engine = PromptTemplateEngine()

# Con todos los parámetros
prompt_completo = engine.renderizar_string(TEMPLATE_RESUMEN, {
    "texto": "El mercado de IA creció un 30% este año...",
    "audiencia": "ejecutivos no técnicos",
    "max_palabras": 50,
    "estilo": "formal"
})

# Solo con lo básico
prompt_basico = engine.renderizar_string(TEMPLATE_RESUMEN, {
    "texto": "El mercado de IA creció un 30% este año..."
})

print("Con parámetros:")
print(prompt_completo)
print("\nSolo texto:")
print(prompt_basico)

Ejercicio 2: Implementar testing gate simple

Crea una función que ejecute tests y bloquee la activación si accuracy < 85%:

from openai import OpenAI

client = OpenAI()

def test_gate(
    prompt_template: str,
    golden_set: list[dict],
    accuracy_minima: float = 0.85
) -> dict:
    """
    Testing gate: retorna True si el prompt pasa los tests.
    Bloquear el deploy si retorna False.
    """
    correctos = 0
    fallos = []
    
    for ej in golden_set:
        prompt = prompt_template.format(input=ej["input"])
        r = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}],
            temperature=0
        )
        output = r.choices[0].message.content.strip()
        
        if output.lower() == str(ej["expected_output"]).lower():
            correctos += 1
        else:
            fallos.append({
                "id": ej.get("id", "?"),
                "input": ej["input"][:50],
                "expected": ej["expected_output"],
                "actual": output
            })
    
    n = len(golden_set)
    accuracy = correctos / n
    
    resultado = {
        "passed": accuracy >= accuracy_minima,
        "accuracy": accuracy,
        "correctos": correctos,
        "fallidos": n - correctos,
        "fallos": fallos[:5],
        "mensaje": (
            f"✅ PASS ({accuracy:.2%} >= {accuracy_minima:.2%})"
            if accuracy >= accuracy_minima
            else f"❌ FAIL ({accuracy:.2%} < {accuracy_minima:.2%})"
        )
    }
    
    print(resultado["mensaje"])
    if fallos:
        print(f"Primeros {min(3, len(fallos))} fallos:")
        for f in fallos[:3]:
            print(f"  [{f['id']}] '{f['input']}' → expected: {f['expected']}, got: {f['actual']}")
    
    return resultado

# Uso:
GOLDEN = [
    {"id": "1", "input": "Excelente producto", "expected_output": "POSITIVO"},
    {"id": "2", "input": "Terrible", "expected_output": "NEGATIVO"},
]

resultado = test_gate(
    "Clasifica como POSITIVO, NEGATIVO o NEUTRO: {input}. Solo la categoría.",
    GOLDEN,
    accuracy_minima=0.85
)

if resultado["passed"]:
    print("✅ Deploy autorizado")
else:
    print("❌ Deploy bloqueado — corregir el prompt")

Resumen

• Template engine: Jinja2 para variables, condicionales y loops en prompts — más flexible que f-strings
• Registry: Almacenamiento versionado con activación explícita y rollback
• Variable store: Variables compartidas que se pueden actualizar sin cambiar el template
• Testing gate: Tests automáticos antes de activar — nunca activar sin tests verdes
• Audit trail: Quién cambió qué y cuándo — imprescindible para debugging en producción
• Build vs buy: Buy (PromptLayer, LangSmith) para empezar rápido; Build para control total
• Namespace: Usar prefijos para evitar colisiones entre equipos

Recursos adicionales

1. PromptLayer — Prompt management managed con UI
2. LangSmith — Observabilidad + gestión para LangChain
3. Jinja2 Documentation — Template engine
4. Langfuse — Open source LLM observability
5. Helicone — Proxy para LLMs con analytics