0Pricing
AI Prompt Engineering · Lesson

Error Handling in Prompt Chains

Validating intermediate outputs and recovering from chain failures.

Error Handling in Prompt Chains is a free AI Prompt Engineering lesson on CoddyKit — lesson 4 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Prompt Engineering learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.

Why Chains Fail

Prompt chains introduce new failure modes that single-prompt systems do not have. Each step can fail in its own way, and failures compound — a bad Step 2 output corrupts every downstream step.

Common failure modes:

  • Model returns malformed JSON that cannot be parsed
  • Model misunderstands the task and produces semantically wrong output
  • Rate limits or API timeouts cause step failures
  • Context window exceeded in a long chain
  • Model hallucinates data that subsequent steps treat as fact

Output Validation After Every Step

The first line of defense is validating output immediately after each step before passing it to the next. Never assume the model returned what you asked for.

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)

Retry Logic for Transient Failures

API failures (rate limits, timeouts, server errors) are transient. Implement exponential backoff retry logic for network-level failures:

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}')

Semantic Validation

Some failures are structurally valid but semantically wrong — the model returns valid JSON but with incorrect values. Use a lightweight validation step to check semantic correctness:

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

Fallback Prompts

When a step fails validation after retries, a fallback prompt can produce simpler but usable output instead of crashing the entire chain:

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)

Circuit Breakers

A circuit breaker prevents a failing chain from wasting API calls. After N consecutive failures, it opens the circuit and returns an error immediately without making further API calls:

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.')

Checkpointing Long Chains

For chains with many steps or expensive steps, use checkpointing to save intermediate results. If a late step fails, resume from the checkpoint instead of restarting from Step 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.')

Graceful Degradation

When a step in the chain fails and cannot be recovered, graceful degradation continues the chain with partial data rather than crashing entirely:

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.')

Structured Error Logging

Log errors with enough context to diagnose which step failed, what the input was, and what the model returned:

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.')

Testing Error Scenarios

Test your error handling explicitly by injecting failures. Use mock objects to simulate API errors and malformed outputs:

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.')

Monitoring Chain Health in Production

In production, track chain health metrics to detect degradation before users notice:

  • Step success rate: Percentage of runs where each step succeeds on first attempt
  • Fallback activation rate: How often is the fallback prompt used?
  • Degradation rate: What fraction of chain runs complete in degraded mode?
  • Step latency: Track p50/p95 latency per step — a slow step indicates prompt complexity issues
  • Validation failure rate: High rate signals prompt needs refinement

Quick Check

What is the purpose of a circuit breaker in a prompt chain?

Error Handling in Chains — Key Takeaways

Robust error handling is what separates prototype chains from production systems:

  • Validate every step's output before passing it downstream — never assume the model returned correct data
  • Retry transient API failures with exponential backoff — rate limits and timeouts are recoverable
  • Use fallback prompts for simpler output when the primary prompt fails semantic validation
  • Circuit breakers stop API call waste after repeated failures
  • Checkpoint expensive steps so long chains can resume after a late-step failure
  • Graceful degradation keeps the pipeline running with partial data instead of crashing
  • Track step success rate, fallback rate, and degradation rate in production

Frequently asked questions

Is the “Error Handling in Prompt Chains” lesson free?

Yes — the full text of “Error Handling in Prompt Chains” is free to read here on the web, and the AI Prompt Engineering course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Prompt Engineering course, upgrade to CoddyKit PRO.

What will I learn in “Error Handling in Prompt Chains”?

Validating intermediate outputs and recovering from chain failures. You practise AI Prompt Engineering with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.

Do I need any experience to start AI Prompt Engineering?

No prior experience is required. AI Prompt Engineering on CoddyKit is structured for beginners through advanced learners; this is — lesson 4 of 4, so you can start here or from the beginning and move at your own pace.

How long does the “Error Handling in Prompt Chains” lesson take?

Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.

Can I write and run code in this AI Prompt Engineering lesson?

Yes. Every AI Prompt Engineering lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.

All lessons in this course

  1. What Is Prompt Chaining?
  2. Output-to-Input Patterns
  3. Sequential Transformation Chains
  4. Error Handling in Prompt Chains
← Back to AI Prompt Engineering