Gestión de errores en cadenas de prompts
Valide los resultados intermedios y recupérese de fallos en la cadena.
Gestión de errores en cadenas de prompts es una lección gratuita de AI Prompt Engineering en CoddyKit. Esta es la lección 4 de 4. Puedes leer la lección completa abajo gratuitamente — luego la practicas en el navegador con un editor de código integrado y un tutor de IA 24/7. Forma parte de la ruta de aprendizaje de AI Prompt Engineering, y tu progreso se sincroniza en la web y la app de CoddyKit. El curso de AI Prompt Engineering incluye 4 lecciones en total.
Por qué fallan las cadenas
Las cadenas de prompts introducen nuevos modos de fallo que no existen en los sistemas basados en un solo prompt. Cada paso puede fallar de una manera distinta y los fallos se acumulan: una salida incorrecta del paso 2 corrompe todos los pasos posteriores.
Modos de fallo comunes:
- El modelo devuelve JSON con un formato incorrecto que no se puede analizar
- El modelo no comprende la tarea y produce una salida semánticamente incorrecta
- Los límites de frecuencia o los tiempos de espera de la API provocan fallos en los pasos
- Se supera la ventana de contexto en una cadena larga
- El modelo inventa datos que los pasos posteriores tratan como hechos
Validación de la salida después de cada paso
La primera línea de defensa consiste en validar la salida inmediatamente después de cada paso, antes de pasarla al siguiente. Nunca dé por hecho que el modelo ha devuelto lo que se le pidió.
import json
def validate_json_output(raw_text, required_fields):
'Parse and validate that required fields are present in model output.'
try:
data = json.loads(raw_text.strip())
except json.JSONDecodeError as e:
raise ValueError(f'Invalid JSON: {e}. Raw: {raw_text[:200]}')
missing = [f for f in required_fields if f not in data]
if missing:
raise ValueError(f'Missing required fields: {missing}. Got: {list(data.keys())}')
return data
# Usage after a chain step
raw = '{"sentiment": "positive", "priority": "high"}'
validated = validate_json_output(raw, required_fields=['sentiment', 'priority'])
print('Valid:', validated)Lógica de reintentos para fallos transitorios
Los fallos de API, como los límites de frecuencia, los tiempos de espera y los errores del servidor, son transitorios. Implemente una lógica de reintentos con espera exponencial para los fallos a nivel de red:
import anthropic, time
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
def call_with_retry(prompt, max_retries=3, base_delay=1.0):
last_error = None
for attempt in range(max_retries):
try:
r = client.messages.create(
model='claude-opus-4-5', max_tokens=500,
messages=[{'role': 'user', 'content': prompt}]
)
return r.content[0].text
except anthropic.RateLimitError as e:
wait = base_delay * (2 ** attempt)
print(f'Rate limited. Waiting {wait}s before retry {attempt+1}/{max_retries}...')
time.sleep(wait)
last_error = e
except anthropic.APIError as e:
last_error = e
if attempt < max_retries - 1:
time.sleep(base_delay)
raise RuntimeError(f'All retries exhausted: {last_error}')Validación semántica
Algunos fallos son válidos desde el punto de vista estructural, pero incorrectos desde el punto de vista semántico: el modelo devuelve JSON válido, pero con valores incorrectos. Use un paso de validación ligera para comprobar la corrección semántica:
def semantic_validate(data, schema_rules):
'Apply semantic validation rules to parsed output.'
errors = []
for field, rules in schema_rules.items():
value = data.get(field)
if rules.get('required') and value is None:
errors.append(f'{field} is required but missing')
continue
if 'allowed_values' in rules and value not in rules['allowed_values']:
errors.append(f'{field} must be one of {rules["allowed_values"]}, got: {value}')
if 'min_length' in rules and isinstance(value, list) and len(value) < rules['min_length']:
errors.append(f'{field} must have at least {rules["min_length"]} items, got {len(value)}')
if errors:
raise ValueError('Semantic validation failed: ' + '; '.join(errors))
return data
rules = {'sentiment': {'allowed_values': ['positive', 'negative', 'mixed']}, 'issues': {'min_length': 1}}
data = {'sentiment': 'positive', 'issues': ['login bug']}
print(semantic_validate(data, rules))Prompts alternativos
Cuando un paso no supera la validación después de varios reintentos, un prompt alternativo puede producir una salida más sencilla pero utilizable, en lugar de hacer que falle toda la cadena:
import json
def call_with_fallback(primary_prompt, fallback_prompt, required_fields):
# Try primary prompt
try:
raw = call_with_retry(primary_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Primary prompt failed: {e}. Trying fallback...')
# Try simpler fallback prompt
try:
raw = call_with_retry(fallback_prompt)
return validate_json_output(raw, required_fields)
except (ValueError, RuntimeError) as e:
print(f'Fallback also failed: {e}. Returning safe default.')
# Return safe default — chain continues with minimal data
return {field: None for field in required_fields}
# Usage
primary = 'Analyze this review. Return JSON with 10 fields: {...}'
fallback = 'Classify this review. Return JSON: {"sentiment": "positive|negative|neutral"}'
result = call_with_fallback(primary, fallback, ['sentiment'])
print(result)Cortacircuitos
Un cortacircuitos evita que una cadena con fallos desperdicie llamadas a la API. Después de N fallos consecutivos, abre el circuito y devuelve un error inmediatamente sin realizar más llamadas a la API:
class CircuitBreaker:
def __init__(self, failure_threshold=3, recovery_timeout=60):
self.failure_count = 0
self.threshold = failure_threshold
self.state = 'closed' # closed = normal, open = blocking
self.opened_at = None
def call(self, fn, *args, **kwargs):
import time
if self.state == 'open':
elapsed = time.time() - self.opened_at
if elapsed > 60: # recovery_timeout
self.state = 'half-open'
else:
raise RuntimeError('Circuit open — skipping API call')
try:
result = fn(*args, **kwargs)
self.failure_count = 0
self.state = 'closed'
return result
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.threshold:
self.state = 'open'
self.opened_at = time.time()
print(f'Circuit opened after {self.failure_count} failures.')
raise e
cb = CircuitBreaker(failure_threshold=3)
print('Circuit breaker initialized.')Puntos de control para cadenas largas
Para cadenas con muchos pasos o pasos costosos, use puntos de control para guardar los resultados intermedios. Si falla un paso posterior, reanude la ejecución desde el punto de control en lugar de reiniciarla desde el paso 1:
import json, os
CHECKPOINT_DIR = '/tmp/chain_checkpoints'
os.makedirs(CHECKPOINT_DIR, exist_ok=True)
def save_checkpoint(run_id, step_id, data):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
with open(path, 'w') as f:
json.dump(data, f)
print(f'Checkpoint saved: step {step_id}')
def load_checkpoint(run_id, step_id):
path = os.path.join(CHECKPOINT_DIR, f'{run_id}_step{step_id}.json')
if os.path.exists(path):
with open(path) as f:
return json.load(f)
return None
def run_with_checkpoints(run_id, input_data):
step1 = load_checkpoint(run_id, 1) or json.loads(call(f'Step 1 processing: {input_data}'))
save_checkpoint(run_id, 1, step1)
step2 = load_checkpoint(run_id, 2) or json.loads(call(f'Step 2 processing: {step1}'))
save_checkpoint(run_id, 2, step2)
return step2
print('Checkpointing system defined.')Degradación controlada
Cuando un paso de la cadena falla y no se puede recuperar, la degradación controlada permite que la cadena continúe con datos parciales en lugar de fallar por completo:
def process_with_degradation(tickets):
results = []
for ticket in tickets:
try:
# Full chain: extract -> classify -> respond
extracted = json.loads(call(f'Extract issue from ticket. Return JSON: {{"issue": str}}\n\n{ticket}'))
classified = json.loads(call(f'Classify priority. Return JSON: {{"priority": str}}\n\n{extracted["issue"]}'))
response = call(f'Draft response for {classified["priority"]} priority: {extracted["issue"]}')
results.append({'ticket': ticket, 'response': response, 'degraded': False})
except Exception as e:
print(f'Chain failed for ticket, using fallback: {e}')
# Fallback: simple direct response without classification
simple_response = call(f'Respond to this support ticket:\n{ticket}')
results.append({'ticket': ticket, 'response': simple_response, 'degraded': True})
return results
print('Graceful degradation pipeline defined.')Registro estructurado de errores
Registre los errores con suficiente contexto para diagnosticar qué paso falló, cuál era la entrada y qué devolvió el modelo:
import logging, traceback
from datetime import datetime
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('chain')
def logged_step(step_name, prompt, validator=None):
start = datetime.utcnow()
try:
raw = call_with_retry(prompt)
result = validator(raw) if validator else raw
logger.info(f'[{step_name}] SUCCESS in {(datetime.utcnow()-start).total_seconds():.2f}s')
return result
except Exception as e:
logger.error(f'[{step_name}] FAILED after {(datetime.utcnow()-start).total_seconds():.2f}s')
logger.error(f'[{step_name}] PROMPT: {prompt[:200]}')
logger.error(f'[{step_name}] ERROR: {traceback.format_exc()}')
raise
print('Structured error logging defined.')Pruebas de escenarios de error
Pruebe explícitamente la gestión de errores inyectando fallos. Use objetos mock para simular errores de API y salidas con formato incorrecto:
from unittest.mock import patch, MagicMock
def test_fallback_on_json_error():
with patch('__main__.call') as mock_call:
# First call returns malformed JSON, fallback returns valid JSON
mock_call.side_effect = [
'This is not JSON at all',
'{"sentiment": "positive"}'
]
result = call_with_fallback(
primary_prompt='Analyze review with 10 fields',
fallback_prompt='Just classify sentiment as JSON',
required_fields=['sentiment']
)
assert result['sentiment'] == 'positive'
print('PASS: fallback activated correctly on JSON parse error')
def test_circuit_breaker_opens():
cb = CircuitBreaker(failure_threshold=2)
for i in range(2):
try:
cb.call(lambda: (_ for _ in ()).throw(RuntimeError('API fail')))
except RuntimeError:
pass
assert cb.state == 'open'
print('PASS: circuit breaker opened after 2 failures')
print('Error handling tests defined.')Supervisión del estado de la cadena en producción
En producción, realice un seguimiento de las métricas del estado de la cadena para detectar la degradación antes de que los usuarios la perciban:
- Tasa de éxito por paso: porcentaje de ejecuciones en las que cada paso se completa en el primer intento
- Tasa de activación del prompt alternativo: ¿Con qué frecuencia se usa el prompt alternativo?
- Tasa de degradación: ¿Qué proporción de las ejecuciones de la cadena se completa en modo degradado?
- Latencia por paso: realice un seguimiento de la latencia p50/p95 de cada paso; una latencia elevada indica problemas de complejidad del prompt
- Tasa de fallos de validación: una tasa elevada indica que el prompt necesita ajustes
Comprobación rápida
¿Cuál es la finalidad de un cortacircuitos en una cadena de prompts?
Gestión de errores en cadenas: ideas clave
Una gestión de errores sólida es lo que diferencia los prototipos de cadenas de los sistemas listos para producción:
- Valide la salida de cada paso antes de pasarla al siguiente; nunca dé por hecho que el modelo ha devuelto datos correctos
- Reintente los fallos transitorios de la API con una espera exponencial; los límites de frecuencia y los tiempos de espera se pueden recuperar
- Use prompts alternativos para obtener una salida más sencilla cuando el prompt principal no supere la validación semántica
- Los cortacircuitos evitan el desperdicio de llamadas a la API después de fallos repetidos
- Establezca puntos de control en los pasos costosos para que las cadenas largas puedan reanudarse después de un fallo en un paso posterior
- La degradación controlada mantiene el pipeline en funcionamiento con datos parciales en lugar de provocar un fallo
- Realice un seguimiento en producción de la tasa de éxito por paso, la tasa de uso de prompts alternativos y la tasa de degradación
Preguntas frecuentes
¿La lección «Gestión de errores en cadenas de prompts» es gratis?
Sí — el texto completo de «Gestión de errores en cadenas de prompts» es gratis para leer aquí en la web. Para practicarla de forma interactiva (editor de código integrado y tutor de IA 24/7) y desbloquear el resto del curso de AI Prompt Engineering, actualiza a CoddyKit PRO. El curso de AI Prompt Engineering incluye 4 lecciones en total.
¿Qué aprenderé en «Gestión de errores en cadenas de prompts»?
Valide los resultados intermedios y recupérese de fallos en la cadena. Practicas AI Prompt Engineering con código real que ejecutas directamente en el navegador, y un tutor de IA 24/7 responde tus preguntas mientras trabajas en la lección.
¿Necesito experiencia previa para empezar AI Prompt Engineering?
No se requiere experiencia previa. AI Prompt Engineering en CoddyKit está estructurado para principiantes hasta estudiantes avanzados, así que puedes empezar aquí o desde el inicio y avanzar a tu ritmo. Esta es la lección 4 de 4.
¿Cuánto tiempo toma la lección «Gestión de errores en cadenas de prompts»?
La mayoría de las lecciones de CoddyKit toman alrededor de 5–10 minutos. Cada una es compacta e interactiva, así que avanzas constantemente y retomas exactamente por donde dejaste en la web y la app.
¿Puedo escribir y ejecutar código en esta lección de AI Prompt Engineering?
Sí. Cada lección de AI Prompt Engineering incluye un editor de código integrado, así que escribes y ejecutas código real directamente en tu navegador y obtienes retroalimentación instantánea de IA — sin configuración local necesaria.
Todas las lecciones de este curso
- ¿Qué es el encadenamiento de prompts?
- Patrones de salida a entrada
- Cadenas de transformación secuencial
- Gestión de errores en cadenas de prompts