Módulo 8: Prompt Engineering en Producción
2. Prompt Versioning y Registries
Descripción
Gestionar versiones de prompts como código de producción: semver, Git-based versioning, prompt registries con rollback. Changelog, audit trail y feature flags para deployment gradual. Herramientas open source y SaaS.
El Problema del Prompt Sin Versionar
Sin versioning, los prompts viven en un estado caótico:
Situaciones comunes sin versioning:
─────────────────────────────────
"El prompt estaba en un archivo de texto en mi Desktop"
"Cambié algo la semana pasada pero no recuerdo qué"
"Necesito volver a como estaba ayer, imposible"
"Hay 3 versiones del prompt en 3 archivos distintos: ¿cuál es el actual?"
"No sé si el cliente A usa el mismo prompt que el cliente B"
El versioning de prompts debe ser tan riguroso como el versioning de código.
Semantic Versioning para Prompts
El semantic versioning (semver) es el estándar: MAJOR.MINOR.PATCH
Para prompts, adaptamos la convención:
| Componente | Cuándo incrementar | Ejemplo |
|---|---|---|
| MAJOR | Cambio que afecta el formato del output, o la tarea principal del prompt | v1.0 → v2.0 |
| MINOR | Mejora de calidad, nuevas instrucciones, más ejemplos few-shot | v1.0 → v1.1 |
| PATCH | Corrección de typos, aclaración menor sin impacto en output | v1.0 → v1.0.1 |
Ejemplos concretos:
v1.0 → v2.0: Cambiar de "clasifica en 3 categorías" a "clasifica en 5 categorías"
v1.0 → v1.1: Añadir 2 ejemplos few-shot para mejorar edge cases
v1.1 → v1.1.1: Corregir "Clasifca" → "Clasifica" (typo)
Versioning con Git
La solución más simple: tratar los prompts como archivos de texto en Git.
Estructura de Archivos
prompts/
├── clasificador/
│ ├── v1.0.0.txt
│ ├── v1.1.0.txt
│ ├── v2.0.0.txt
│ └── CHANGELOG.md
├── extractor/
│ ├── v1.0.0.txt
│ └── CHANGELOG.md
├── resumen/
│ └── v1.0.0.txt
└── registry.json
CHANGELOG.md por Prompt
# Changelog: clasificador
## v2.0.0 (2025-03-01)
### Breaking Changes
- Output ahora incluye 5 categorías en lugar de 3: POSITIVO, NEGATIVO, NEUTRO, MIXTO, URGENTE
### Added
- Nueva categoría URGENTE para textos con lenguaje de emergencia
- Instrucción explícita para manejar sarcasmo
## v1.1.0 (2025-02-15)
### Added
- 3 ejemplos few-shot adicionales para edge cases de sarcasmo
- Instrucción: "Considera el tono general, no solo palabras individuales"
## v1.0.0 (2025-01-10)
### Initial Release
- Clasificador básico: POSITIVO, NEGATIVO, NEUTRO
Operaciones con Git
# Ver historial de un prompt
git log --oneline prompts/clasificador/
# Ver qué cambió entre versiones
git diff HEAD~1 prompts/clasificador/v1.1.0.txt
# Restaurar versión anterior
git show HEAD~2:prompts/clasificador/v1.0.0.txt > prompts/clasificador/restored.txt
Prompt Registry: Implementación en Python
Un registry es un sistema centralizado para acceder a prompts por nombre y versión:
import json
from pathlib import Path
from datetime import datetime
from typing import Optional
class PromptRegistry:
"""
Registry de prompts con versioning y rollback.
Almacena prompts en memoria con persistencia opcional a disco.
"""
def __init__(self, registry_path: str = "prompts/registry.json"):
self.registry_path = Path(registry_path)
self._data: dict = {}
if self.registry_path.exists():
self._cargar()
def _cargar(self) -> None:
"""Carga el registry desde disco."""
with open(self.registry_path) as f:
self._data = json.load(f)
print(f"Registry cargado: {len(self._data)} prompts")
def _guardar(self) -> None:
"""Persiste el registry a disco."""
self.registry_path.parent.mkdir(parents=True, exist_ok=True)
with open(self.registry_path, "w", encoding="utf-8") as f:
json.dump(self._data, f, indent=2, ensure_ascii=False)
def registrar(
self,
nombre: str,
version: str,
prompt: str,
changelog: str = "",
metadata: dict | None = None
) -> None:
"""
Registra una nueva versión de un prompt.
nombre: Identificador del prompt (ej. "clasificador_sentimiento")
version: Semver (ej. "v1.2.0")
prompt: El texto del prompt (puede incluir {variables})
changelog: Descripción de los cambios en esta versión
"""
if nombre not in self._data:
self._data[nombre] = {
"versions": {},
"active_version": None,
"created_at": datetime.now().isoformat()
}
self._data[nombre]["versions"][version] = {
"prompt": prompt,
"changelog": changelog,
"created_at": datetime.now().isoformat(),
"metadata": metadata or {}
}
# Auto-activar si es la primera versión
if self._data[nombre]["active_version"] is None:
self._data[nombre]["active_version"] = version
self._guardar()
print(f"Registrado: {nombre} {version}")
def get(
self,
nombre: str,
version: Optional[str] = None
) -> str:
"""
Obtiene un prompt por nombre y versión.
Si version es None, retorna la versión activa.
"""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no encontrado en el registry")
if version is None:
version = self._data[nombre]["active_version"]
if version is None:
raise ValueError(f"No hay versión activa para '{nombre}'")
versions = self._data[nombre]["versions"]
if version not in versions:
available = list(versions.keys())
raise KeyError(f"Versión '{version}' no existe. Disponibles: {available}")
return versions[version]["prompt"]
def activar(self, nombre: str, version: str) -> None:
"""Activa una versión específica (sin rollback, solo cambio de activa)."""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no encontrado")
if version not in self._data[nombre]["versions"]:
raise KeyError(f"Versión '{version}' no existe")
anterior = self._data[nombre]["active_version"]
self._data[nombre]["active_version"] = version
self._guardar()
print(f"Activado: {nombre} → {version} (anterior: {anterior})")
def rollback(self, nombre: str, to_version: Optional[str] = None) -> None:
"""
Rollback a una versión anterior.
Si to_version es None, va a la versión anterior a la activa.
"""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no encontrado")
versions = list(self._data[nombre]["versions"].keys())
current = self._data[nombre]["active_version"]
if to_version is None:
# Retroceder una versión
if current in versions:
current_idx = versions.index(current)
if current_idx > 0:
to_version = versions[current_idx - 1]
else:
raise ValueError(f"Ya estás en la primera versión ({current})")
else:
to_version = versions[-1]
self.activar(nombre, to_version)
print(f"🔄 Rollback: {nombre} {current} → {to_version}")
def listar(self, nombre: str) -> dict:
"""Lista todas las versiones de un prompt con metadata."""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no encontrado")
prompt_data = self._data[nombre]
active = prompt_data["active_version"]
return {
"nombre": nombre,
"version_activa": active,
"versiones": {
v: {
"activa": v == active,
"changelog": data.get("changelog", ""),
"created_at": data.get("created_at", ""),
"preview": data["prompt"][:80] + "..."
}
for v, data in prompt_data["versions"].items()
}
}
def version_activa(self, nombre: str) -> str:
"""Retorna la versión activa de un prompt."""
if nombre not in self._data:
raise KeyError(f"Prompt '{nombre}' no encontrado")
return self._data[nombre]["active_version"]
def prompts_disponibles(self) -> list[str]:
"""Lista todos los prompts en el registry."""
return list(self._data.keys())
# Ejemplo de uso:
registry = PromptRegistry()
# Registrar primera versión
registry.registrar(
nombre="clasificador_sentimiento",
version="v1.0.0",
prompt="Clasifica el texto como POSITIVO, NEGATIVO o NEUTRO. Solo la categoría.\n\nTexto: {input}\nCategoría:",
changelog="Versión inicial del clasificador"
)
# Registrar mejora
registry.registrar(
nombre="clasificador_sentimiento",
version="v1.1.0",
prompt="""Clasifica el sentimiento del texto como POSITIVO, NEGATIVO o NEUTRO.
Considera el tono general, incluyendo sarcasmo e ironía.
Responde SOLO con la categoría.
Ejemplos:
- "Me encanta este producto" → POSITIVO
- "Claro, 'rápido' si esperar 3 semanas es rápido" → NEGATIVO
Texto: {input}
Categoría:""",
changelog="Añadidos 2 ejemplos few-shot para mejorar detección de sarcasmo"
)
# Activar nueva versión
registry.activar("clasificador_sentimiento", "v1.1.0")
# Obtener prompt activo
prompt = registry.get("clasificador_sentimiento")
# Si algo sale mal: rollback
registry.rollback("clasificador_sentimiento") # Vuelve a v1.0.0
Registry con Feature Flags
Para deployment gradual, combinar versioning con feature flags:
import random
from openai import OpenAI
client = OpenAI()
class PromptRegistryConFlags:
"""Registry que soporta A/B deployments y gradual rollouts."""
def __init__(self, registry: PromptRegistry):
self.registry = registry
self._flags: dict = {}
def configurar_rollout(
self,
nombre: str,
version_nueva: str,
porcentaje: float = 0.05
) -> None:
"""
Configura un gradual rollout.
porcentaje: 0.05 = 5% del tráfico va a la nueva versión
"""
self._flags[nombre] = {
"version_nueva": version_nueva,
"porcentaje": porcentaje,
"version_estable": self.registry.version_activa(nombre)
}
print(f"Rollout configurado: {nombre} → {version_nueva} ({porcentaje:.0%} tráfico)")
def get_prompt(self, nombre: str, request_id: str = None) -> tuple[str, str]:
"""
Obtiene el prompt con routing basado en feature flags.
Returns: (prompt, version_usada)
"""
if nombre in self._flags:
flag = self._flags[nombre]
# Consistencia: mismo request_id siempre usa misma versión
if request_id:
import hashlib
hash_val = int(hashlib.md5(f"{nombre}:{request_id}".encode()).hexdigest(), 16)
usar_nueva = (hash_val % 100) < (flag["porcentaje"] * 100)
else:
usar_nueva = random.random() < flag["porcentaje"]
version = flag["version_nueva"] if usar_nueva else flag["version_estable"]
else:
version = self.registry.version_activa(nombre)
return self.registry.get(nombre, version), version
def promover_rollout(self, nombre: str, nuevo_porcentaje: float) -> None:
"""Aumenta el porcentaje de tráfico a la nueva versión."""
if nombre not in self._flags:
raise ValueError(f"No hay rollout activo para '{nombre}'")
self._flags[nombre]["porcentaje"] = nuevo_porcentaje
print(f"Rollout promovido: {nombre} → {nuevo_porcentaje:.0%}")
def completar_rollout(self, nombre: str) -> None:
"""Finaliza el rollout: 100% tráfico a nueva versión y limpia el flag."""
if nombre in self._flags:
version_nueva = self._flags[nombre]["version_nueva"]
self.registry.activar(nombre, version_nueva)
del self._flags[nombre]
print(f"Rollout completado: {nombre} ahora es {version_nueva}")
def cancelar_rollout(self, nombre: str) -> None:
"""Cancela el rollout: vuelve 100% a versión estable."""
if nombre in self._flags:
del self._flags[nombre]
print(f"Rollout cancelado: {nombre} volvió a versión estable")
# Ejemplo de rollout gradual:
reg = PromptRegistry()
flags = PromptRegistryConFlags(reg)
# Día 1: 5% al nuevo prompt
flags.configurar_rollout("clasificador", "v1.2.0", porcentaje=0.05)
# Monitorear métricas... Si van bien:
# Día 2: aumentar a 20%
flags.promover_rollout("clasificador", nuevo_porcentaje=0.20)
# Día 3: aumentar a 50%
flags.promover_rollout("clasificador", nuevo_porcentaje=0.50)
# Si algo falla en cualquier momento:
flags.cancelar_rollout("clasificador") # 100% vuelve a versión estable
# Si todo bien: completar
flags.completar_rollout("clasificador") # v1.2.0 se convierte en activo
Registry con Base de Datos
Para sistemas con múltiples servicios o equipos, usar una base de datos:
import sqlite3
from contextlib import contextmanager
class PromptRegistryDB:
"""Registry con persistencia en SQLite (escala a PostgreSQL fácilmente)."""
def __init__(self, db_path: str = "prompts.db"):
self.db_path = db_path
self._init_db()
def _init_db(self) -> None:
"""Crea las tablas si no existen."""
with self._conn() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS prompts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nombre TEXT NOT NULL,
version TEXT NOT NULL,
prompt TEXT NOT NULL,
changelog TEXT DEFAULT '',
activa INTEGER DEFAULT 0,
created_at TEXT NOT NULL,
UNIQUE(nombre, version)
)
""")
conn.execute("""
CREATE TABLE IF NOT EXISTS prompt_usage (
id INTEGER PRIMARY KEY AUTOINCREMENT,
nombre TEXT NOT NULL,
version TEXT NOT NULL,
timestamp TEXT NOT NULL,
request_id TEXT,
exito INTEGER DEFAULT 1
)
""")
@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, prompt: str, changelog: str = "") -> None:
with self._conn() as conn:
conn.execute("""
INSERT INTO prompts (nombre, version, prompt, changelog, created_at)
VALUES (?, ?, ?, ?, ?)
""", (nombre, version, prompt, changelog, datetime.now().isoformat()))
print(f"Registrado: {nombre} {version}")
def activar(self, nombre: str, version: str) -> None:
with self._conn() as conn:
# Desactivar todas las versiones actuales
conn.execute("UPDATE prompts SET activa = 0 WHERE nombre = ?", (nombre,))
# Activar la versión solicitada
conn.execute(
"UPDATE prompts SET activa = 1 WHERE nombre = ? AND version = ?",
(nombre, version)
)
print(f"Activado: {nombre} {version}")
def get(self, nombre: str, version: str | None = None) -> str:
with self._conn() as conn:
if version:
row = conn.execute(
"SELECT prompt FROM prompts WHERE nombre = ? AND version = ?",
(nombre, version)
).fetchone()
else:
row = conn.execute(
"SELECT prompt 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")
return row["prompt"]
def log_uso(self, nombre: str, version: str, request_id: str, exito: bool) -> None:
"""Registra el uso del prompt para auditoría."""
with self._conn() as conn:
conn.execute("""
INSERT INTO prompt_usage (nombre, version, timestamp, request_id, exito)
VALUES (?, ?, ?, ?, ?)
""", (nombre, version, datetime.now().isoformat(), request_id, int(exito)))
def historial_uso(self, nombre: str, last_n: int = 100) -> list[dict]:
"""Retorna el historial de uso de un prompt."""
with self._conn() as conn:
rows = conn.execute("""
SELECT * FROM prompt_usage WHERE nombre = ?
ORDER BY timestamp DESC LIMIT ?
""", (nombre, last_n)).fetchall()
return [dict(row) for row in rows]
Integración en una API (FastAPI)
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
registry = PromptRegistry()
client = OpenAI()
class ClassifyRequest(BaseModel):
texto: str
prompt_version: str | None = None # None = usar versión activa
class ClassifyResponse(BaseModel):
result: str
prompt_version: str
@app.post("/classify", response_model=ClassifyResponse)
async def classify(request: ClassifyRequest):
try:
# Obtener prompt (versión específica o activa)
prompt_template = registry.get(
"clasificador_sentimiento",
version=request.prompt_version
)
version_usada = (
request.prompt_version or
registry.version_activa("clasificador_sentimiento")
)
except KeyError as e:
raise HTTPException(status_code=404, detail=str(e))
# Ejecutar prompt
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{
"role": "user",
"content": prompt_template.format(input=request.texto)
}],
temperature=0
)
result = response.choices[0].message.content.strip()
return ClassifyResponse(result=result, prompt_version=version_usada)
@app.post("/prompts/{nombre}/rollback")
async def rollback_prompt(nombre: str, to_version: str | None = None):
"""Endpoint para rollback de emergencia."""
try:
registry.rollback(nombre, to_version)
return {"status": "ok", "version_activa": registry.version_activa(nombre)}
except (KeyError, ValueError) as e:
raise HTTPException(status_code=400, detail=str(e))
@app.get("/prompts/{nombre}/versions")
async def listar_versiones(nombre: str):
"""Lista todas las versiones de un prompt."""
try:
return registry.listar(nombre)
except KeyError as e:
raise HTTPException(status_code=404, detail=str(e))
Comparación: Opciones de Versioning
| Opción | Ventaja | Desventaja | Cuándo usar |
|---|---|---|---|
| Archivos en Git | Simple, gratis, diff nativo | Sin runtime access | Proyectos pequeños, equipos técnicos |
| Registry JSON en disco | Simple, sin dependencias | Sin multi-servicio | 1-3 servicios, un equipo |
| Registry con SQLite | Queries, audit trail | Sin clustering | Mono-servicio, auditoría necesaria |
| Registry con PostgreSQL | Multi-servicio, transaccional | Dependencia infra | Multi-servicio, equipo mediano |
| PromptLayer | UI, analytics, managed | Costo mensual | Equipos sin infra propia |
| LangSmith | Tracing integrado | Vendor lock-in | Ya usas LangChain |
Troubleshooting
Problema 1: Prompts hardcodeados en el código
Síntoma: El prompt está directamente en el código Python, sin versioning.
Causa: Desarrollo rápido sin pensar en mantenibilidad.
Solución:
# ANTES (problemático):
def classify(texto: str) -> str:
prompt = "Clasifica esto: " + texto # Hardcodeado
...
# DESPUÉS (con registry):
registry = PromptRegistry()
def classify(texto: str) -> str:
prompt_template = registry.get("clasificador")
prompt = prompt_template.format(input=texto)
...
# Script de migración:
# 1. Extraer prompts hardcodeados a archivos .txt
# 2. Registrar en el registry con versión inicial
# 3. Actualizar código para usar registry.get()
Problema 2: Rollback sin downtime
Síntoma: Necesitas hacer rollback pero el servicio está recibiendo tráfico.
Causa: La versión nueva tiene un bug grave.
Solución:
# El registry JSON-based permite rollback sin reiniciar el servicio
# si lo lees en cada request (no cacheado en memoria indefinidamente)
# MAL: Cargar prompt una sola vez al inicio
PROMPT = registry.get("clasificador") # Fijo en memoria, no actualizable sin restart
# BIEN: Cargar en cada request (con cache corto de 30s para performance)
from functools import lru_cache
from time import time
class RegistryConCache:
def __init__(self, registry: PromptRegistry, ttl_seconds: int = 30):
self._registry = registry
self._cache = {}
self._ttl = ttl_seconds
def get(self, nombre: str) -> str:
now = time()
if nombre in self._cache:
cached_at, prompt = self._cache[nombre]
if now - cached_at < self._ttl:
return prompt
prompt = self._registry.get(nombre)
self._cache[nombre] = (now, prompt)
return prompt
Problema 3: Conflictos entre equipos
Síntoma: Dos personas editan el mismo prompt concurrentemente.
Solución:
# Con DB: usar optimistic locking
def registrar_con_lock(
nombre: str,
version: str,
prompt: str,
basado_en: str # versión en que se basó el cambio
):
with db._conn() as conn:
version_actual = conn.execute(
"SELECT version FROM prompts WHERE nombre = ? AND activa = 1",
(nombre,)
).fetchone()
if version_actual and version_actual["version"] != basado_en:
raise ValueError(
f"Conflicto: la versión activa cambió de {basado_en} "
f"a {version_actual['version']} mientras editabas. "
"Por favor, haz merge de los cambios."
)
# Si no hay conflicto, registrar normalmente
db.registrar(nombre, version, prompt)
Ejercicios
Ejercicio 1: Construir tu primer prompt registry
Implementa un PromptRegistry mínimo que soporte: registrar, get, activar, y rollback:
Ver solución
import json
from pathlib import Path
from datetime import datetime
class MiniRegistry:
"""Implementación mínima de un prompt registry."""
def __init__(self, path: str = "registry.json"):
self.path = Path(path)
self.data = {}
if self.path.exists():
with open(self.path) as f:
self.data = json.load(f)
def _save(self):
with open(self.path, "w") as f:
json.dump(self.data, f, indent=2, ensure_ascii=False)
def registrar(self, nombre: str, version: str, prompt: str) -> None:
if nombre not in self.data:
self.data[nombre] = {"versions": {}, "active": None}
self.data[nombre]["versions"][version] = {
"prompt": prompt,
"created": datetime.now().isoformat()
}
if self.data[nombre]["active"] is None:
self.data[nombre]["active"] = version
self._save()
print(f"✓ Registrado: {nombre} {version}")
def get(self, nombre: str, version: str = None) -> str:
if nombre not in self.data:
raise KeyError(f"'{nombre}' no encontrado")
version = version or self.data[nombre]["active"]
return self.data[nombre]["versions"][version]["prompt"]
def activar(self, nombre: str, version: str) -> None:
self.data[nombre]["active"] = version
self._save()
print(f"✓ Activado: {nombre} → {version}")
def rollback(self, nombre: str) -> None:
versions = list(self.data[nombre]["versions"].keys())
current = self.data[nombre]["active"]
idx = versions.index(current) if current in versions else len(versions) - 1
if idx > 0:
self.activar(nombre, versions[idx - 1])
else:
print("Ya estás en la primera versión")
# Test:
r = MiniRegistry("/tmp/test_registry.json")
r.registrar("test", "v1.0", "Prompt v1: {input}")
r.registrar("test", "v1.1", "Prompt v1.1 mejorado: {input}")
r.activar("test", "v1.1")
print(r.get("test")) # v1.1
r.rollback("test")
print(r.get("test")) # v1.0
Ejercicio 2: Gradual rollout con feature flags
Implementa un sistema simple que envíe el 10% del tráfico a un nuevo prompt:
Ver solución
import hashlib
from openai import OpenAI
client = OpenAI()
PROMPT_V1 = "Clasifica como POSITIVO, NEGATIVO o NEUTRO: {input}"
PROMPT_V2 = """Clasifica el sentimiento como POSITIVO, NEGATIVO o NEUTRO.
Considera el tono general y el sarcasmo.
Solo la categoría.
Texto: {input}
Categoría:"""
def get_prompt_for_request(request_id: str, pct_nuevo: float = 0.10) -> tuple[str, str]:
"""Retorna (prompt, version) basado en request_id."""
hash_val = int(hashlib.md5(request_id.encode()).hexdigest(), 16)
usar_nuevo = (hash_val % 100) < (pct_nuevo * 100)
if usar_nuevo:
return PROMPT_V2, "v2"
return PROMPT_V1, "v1"
# Simular 10 requests
resultados = {"v1": 0, "v2": 0}
for i in range(20):
request_id = f"req_{i:04d}"
_, version = get_prompt_for_request(request_id, pct_nuevo=0.10)
resultados[version] += 1
print(f"v1: {resultados['v1']} requests, v2: {resultados['v2']} requests")
print(f"% v2: {resultados['v2']/20*100:.0f}% (target: 10%)")
Resumen
- Semver para prompts: MAJOR (breaking), MINOR (mejora), PATCH (typo) — misma lógica que código
- Git + archivos: La opción más simple — prompts como archivos .txt versionados
- Registry Python: Dict con versiones, versión activa, y rollback en un JSON persistente
- Feature flags: Deployment gradual (5% → 20% → 50% → 100%) sin downtime
- DB-backed registry: Para multi-servicio o cuando necesitas audit trail
- Integración con API: Endpoint de rollback para emergencias sin restart del servicio
Recursos adicionales
- Semantic Versioning — Especificación completa de semver
- PromptLayer — Versioning managed con analytics
- LangSmith — Tracing y gestión de prompts de LangChain
- Feature Flags Best Practices — LaunchDarkly
- Git Flow — Para branch strategy con prompts