Module 6: Prompt Composition and Chaining
4. Multi-Stage Pipelines
Overview
A multi-stage pipeline is a processing system where multiple prompts run in sequence, passing state between them, with per-stage error handling, automatic retry, and monitoring. It's the central architectural pattern for building production LLM systems.
In this capsule you'll learn to design robust pipelines with state management, smart retry, logging, and fallback strategies that guarantee the system keeps working even when individual parts fail.
The Difference Between a Simple Chain and a Pipeline
Simple chain:
A → B → C
If B fails, everything fails.
No state management.
No retry.
No logging.
Production pipeline:
[Step A] → [Validate] → [Step B] → [Validate] → [Step C]
↑ ↑ ↑
Retry(2) Retry(2) Retry(2)
↑ ↑ ↑
Logging Logging Logging
↑
Fallback
State: {original_input, output_A, output_B, output_C, errors, timestamps}
Base Pipeline Architecture
from openai import OpenAI
from typing import Callable, Any, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
import time
import logging
client = OpenAI()
logger = logging.getLogger(__name__)
class StageStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped" # When the fallback is used
@dataclass
class StageMetrics:
"""Execution metrics for one stage."""
start_time: float = 0.0
end_time: float = 0.0
attempts: int = 0
input_tokens: int = 0
output_tokens: int = 0
status: StageStatus = StageStatus.PENDING
error: Optional[str] = None
@property
def duration(self) -> float:
return self.end_time - self.start_time
@dataclass
class PipelineState:
"""
Complete state of the pipeline during its execution.
Holds the original input, every intermediate output,
and the execution metrics.
"""
original_input: str
outputs: dict[str, Any] = field(default_factory=dict)
metrics: dict[str, StageMetrics] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
def get_output(self, stage_name: str, default: str = "") -> Any:
"""Gets the output of a stage, or default if it doesn't exist."""
return self.outputs.get(stage_name, default)
def get_last_output(self) -> str:
"""Returns the output of the last successful stage."""
for name in reversed(list(self.outputs.keys())):
if name != "_error":
return self.outputs[name]
return self.original_input
def metrics_summary(self) -> dict:
"""Builds a summary of the pipeline metrics."""
completed_stages = sum(1 for m in self.metrics.values() if m.status == StageStatus.COMPLETED)
failed_stages = sum(1 for m in self.metrics.values() if m.status == StageStatus.FAILED)
total_time = sum(m.duration for m in self.metrics.values())
return {
"total_stages": len(self.metrics),
"completed_stages": completed_stages,
"failed_stages": failed_stages,
"total_time_seconds": total_time,
"success_rate": completed_stages / len(self.metrics) if self.metrics else 0
}
@dataclass
class StageConfig:
"""Configuration of one pipeline stage."""
name: str
fn: Callable[[PipelineState], str] # Function that runs the stage
max_retries: int = 2 # Retries on error
fallback_fn: Optional[Callable] = None # Fallback function if it fails
validate_fn: Optional[Callable[[str], bool]] = None # Output validator
timeout_seconds: float = 30.0 # Stage timeout
critical: bool = True # If it fails with no fallback, stop the pipeline?The Pipeline Engine
class Pipeline:
"""
Execution engine for LLM pipelines.
Features:
- Automatic retry per stage
- Fallback when retries run out
- Output validation
- Detailed logging
- Metrics monitoring
- State management
"""
def __init__(self, name: str = "pipeline"):
self.name = name
self.stages: list[StageConfig] = []
def add_stage(self, config: StageConfig) -> 'Pipeline':
"""Adds a stage to the pipeline (fluent interface)."""
self.stages.append(config)
return self
def run(
self,
initial_input: str,
verbose: bool = True
) -> PipelineState:
"""
Runs the complete pipeline.
Args:
initial_input: The input of the first stage
verbose: If True, prints the progress
Returns:
PipelineState with every output and metric
"""
state = PipelineState(original_input=initial_input)
if verbose:
print(f"\n{'='*60}")
print(f"Pipeline: {self.name}")
print(f"Stages: {[s.name for s in self.stages]}")
print(f"{'='*60}")
for config in self.stages:
metrics = StageMetrics(
start_time=time.time(),
status=StageStatus.IN_PROGRESS
)
state.metrics[config.name] = metrics
if verbose:
print(f"\n[{config.name}] Starting...")
success = False
last_error = None
# Retry cycle
for attempt in range(config.max_retries + 1):
metrics.attempts = attempt + 1
try:
output = config.fn(state)
# Optional validation
if config.validate_fn and not config.validate_fn(output):
raise ValueError(f"Validation failed for stage '{config.name}'")
# Success
state.outputs[config.name] = output
metrics.status = StageStatus.COMPLETED
metrics.end_time = time.time()
success = True
if verbose:
print(f" ✓ Completed in {metrics.duration:.2f}s (attempt {attempt+1})")
print(f" Output: {str(output)[:100]}...")
break
except Exception as e:
last_error = str(e)
logger.warning(f"Attempt {attempt+1} failed in '{config.name}': {e}")
if verbose:
print(f" ⚠ Error on attempt {attempt+1}: {str(e)[:80]}")
if not success:
metrics.error = last_error
# Try the fallback
if config.fallback_fn:
try:
fallback_output = config.fallback_fn(state)
state.outputs[config.name] = fallback_output
metrics.status = StageStatus.SKIPPED
metrics.end_time = time.time()
if verbose:
print(f" ⚡ Using fallback for '{config.name}'")
except Exception as fb_error:
metrics.status = StageStatus.FAILED
if verbose:
print(f" ✗ The fallback failed too: {fb_error}")
if config.critical:
raise RuntimeError(
f"Critical stage '{config.name}' failed after {config.max_retries+1} attempts: {last_error}"
)
else:
metrics.status = StageStatus.FAILED
if config.critical:
raise RuntimeError(
f"Critical stage '{config.name}' failed: {last_error}"
)
if verbose:
summary = state.metrics_summary()
print(f"\n{'='*60}")
print(f"Pipeline complete")
print(f" Total time: {summary['total_time_seconds']:.2f}s")
print(f" Success: {summary['completed_stages']}/{summary['total_stages']} stages")
print(f"{'='*60}")
return stateComplete Example: Document Analysis Pipeline
def build_analysis_pipeline() -> Pipeline:
"""
Builds a pipeline for document analysis with:
- Extraction of key information
- Sentiment and context analysis
- Generation of a structured report
- Formatting of the final output
"""
# Functions for each stage
def step_preprocess(state: PipelineState) -> str:
"""Normalizes and prepares the document."""
doc = state.original_input
# If the document is very long, summarize it first
if len(doc) > 8000:
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Summarize this document in 2000 words, preserving all the important information:\n\n{doc[:15000]}"}],
temperature=0,
max_tokens=600
).choices[0].message.content
return doc
def step_extract(state: PipelineState) -> str:
"""Extracts structured information from the document."""
doc = state.get_output("preprocess", state.original_input)
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Extract the key information from the following document:
1. Document type
2. Main entities (people, organizations, places)
3. Important dates
4. Key numbers and metrics
5. Main topics (3-5)
Document: {doc[:6000]}
Respond in JSON with the key "entities" for the entities."""}],
temperature=0,
max_tokens=600,
response_format={"type": "json_object"}
).choices[0].message.content
def step_analyze(state: PipelineState) -> str:
"""Deep analysis based on the extraction."""
extraction = state.get_output("extract")
doc = state.get_output("preprocess", state.original_input)
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Analyze this document based on the extraction:
Extraction: {extraction[:1000]}
Document: {doc[:4000]}
Generate:
1. Context and relevance analysis
2. Most important insights (3-5)
3. Risks or points of attention
4. Opportunities, if any"""}],
temperature=0,
max_tokens=600
).choices[0].message.content
def step_synthesize(state: PipelineState) -> str:
"""Synthesizes extraction + analysis into conclusions."""
analysis = state.get_output("analyze")
extraction = state.get_output("extract")
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Based on this complete analysis:
Extraction: {extraction[:500]}
Analysis: {analysis[:800]}
Generate:
1. The 5 most important conclusions (ordered by relevance)
2. The 3 recommended actions
3. One paragraph of executive conclusion
Be concise and actionable."""}],
temperature=0.1,
max_tokens=500
).choices[0].message.content
def step_format(state: PipelineState) -> str:
"""Formats the final output as structured markdown."""
synthesis = state.get_output("synthesize")
extraction_raw = state.get_output("extract", "{}")
try:
extraction = json.loads(extraction_raw)
entities = extraction.get("entities", {})
except Exception:
entities = {}
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"""Format this analysis as a professional markdown report.
Include:
- Descriptive title (H1)
- An "Executive Summary" section (1 paragraph)
- A "Key Entities" section (if there is data: {str(entities)[:200]})
- A "Conclusions" section
- A "Recommendations" section
Analysis to format:
{synthesis}"""}],
temperature=0.2,
max_tokens=700
).choices[0].message.content
# Fallbacks
def fallback_preprocess(state: PipelineState) -> str:
"""If preprocessing fails, use the original document directly."""
return state.original_input[:6000] # Truncate if needed
def fallback_synthesize(state: PipelineState) -> str:
"""If the synthesis fails, concatenate extraction and analysis."""
return f"Extraction:\n{state.get_output('extract')}\n\nAnalysis:\n{state.get_output('analyze')}"
# Validators
def validate_json(output: str) -> bool:
try:
json.loads(output)
return True
except Exception:
return False
def validate_not_empty(output: str) -> bool:
return len(output.strip()) > 50
# Build the pipeline
pipeline = Pipeline(name="document_analysis")
pipeline.add_stage(StageConfig(
name="preprocess",
fn=step_preprocess,
max_retries=1,
fallback_fn=fallback_preprocess,
critical=False # Not critical; we can use the original document
))
pipeline.add_stage(StageConfig(
name="extract",
fn=step_extract,
max_retries=2,
validate_fn=validate_json,
critical=True # Critical: without the extraction it can't continue properly
))
pipeline.add_stage(StageConfig(
name="analyze",
fn=step_analyze,
max_retries=2,
validate_fn=validate_not_empty,
critical=True
))
pipeline.add_stage(StageConfig(
name="synthesize",
fn=step_synthesize,
max_retries=1,
fallback_fn=fallback_synthesize,
critical=False
))
pipeline.add_stage(StageConfig(
name="format",
fn=step_format,
max_retries=1,
critical=False # If the formatting fails, the analysis is already complete
))
return pipeline
# Usage example:
if __name__ == "__main__":
document = """
Quarterly Report Q4 2025 - TechStartup S.L.
Executive Summary:
The fourth quarter of 2025 was a period of accelerated growth for TechStartup.
Revenue reached €2.4M, a 45% increase over Q4 2024.
Active users reached 125,000, with a retention rate of 87%.
R&D investment:
23% of revenue (€552K) was allocated to research and development,
focused mainly on developing custom AI models...
"""
pipeline = build_analysis_pipeline()
state = pipeline.run(document, verbose=True)
print("\n=== FINAL OUTPUT ===")
print(state.get_output("format", state.get_last_output()))
print("\n=== METRICS ===")
for name, metrics in state.metrics.items():
print(f"{name}: {metrics.status.value} ({metrics.duration:.2f}s, {metrics.attempts} attempts)")Retry with Exponential Backoff
import random
def run_with_backoff(
fn: Callable,
state: PipelineState,
max_retries: int = 3,
backoff_base: float = 1.0,
jitter: bool = True
) -> str:
"""
Runs a function with retry and exponential backoff.
Exponential backoff prevents thundering herd problems
when multiple pipelines fail simultaneously.
Args:
fn: Function to run
state: Pipeline state
max_retries: Maximum number of attempts
backoff_base: Base of the wait time (seconds)
jitter: If True, adds random variation to the wait time
Returns:
Output of the function
"""
for attempt in range(max_retries):
try:
return fn(state)
except Exception as e:
if attempt == max_retries - 1:
raise # Re-raise on the last attempt
# Compute the wait time: base * 2^attempt
wait_time = backoff_base * (2 ** attempt)
if jitter:
# Add jitter: ±50% of the wait time
wait_time *= (1 + random.uniform(-0.5, 0.5))
logger.warning(f"Attempt {attempt+1} failed: {e}. Retrying in {wait_time:.1f}s")
time.sleep(wait_time)
raise RuntimeError("Max retries reached without success")Monitoring and Observability
class PipelineMonitor:
"""
Monitoring system for LLM pipelines.
Records metrics, errors, and performance.
"""
def __init__(self):
self.runs: list[dict] = []
def record_run(self, state: PipelineState, pipeline_name: str):
"""Records the metrics of one run."""
summary = state.metrics_summary()
entry = {
"pipeline": pipeline_name,
"timestamp": time.time(),
"input_length": len(state.original_input),
**summary,
"stage_details": {
name: {
"status": m.status.value,
"duration": m.duration,
"attempts": m.attempts,
"error": m.error
}
for name, m in state.metrics.items()
}
}
self.runs.append(entry)
return entry
def stats(self) -> dict:
"""Computes aggregate statistics across every run."""
if not self.runs:
return {}
times = [r["total_time_seconds"] for r in self.runs]
success_rates = [r["success_rate"] for r in self.runs]
return {
"n_runs": len(self.runs),
"average_time": sum(times) / len(times),
"p95_time": sorted(times)[int(len(times) * 0.95)],
"average_success_rate": sum(success_rates) / len(success_rates),
"n_runs_with_fallback": sum(
1 for r in self.runs
if any(d["status"] == "skipped" for d in r["stage_details"].values())
)
}
def slowest_stages(self, n: int = 3) -> list[tuple[str, float]]:
"""Identifies the stages with the highest average latency."""
times_per_stage: dict[str, list[float]] = {}
for run in self.runs:
for stage, detail in run["stage_details"].items():
if stage not in times_per_stage:
times_per_stage[stage] = []
times_per_stage[stage].append(detail["duration"])
averages = [(stage, sum(ts)/len(ts)) for stage, ts in times_per_stage.items()]
return sorted(averages, key=lambda x: x[1], reverse=True)[:n]
# Using the monitor:
monitor = PipelineMonitor()
pipeline = build_analysis_pipeline()
state = pipeline.run("Test document...", verbose=False)
monitor.record_run(state, "document_analysis")
print(monitor.stats())Pipeline with Intermediate Cache
import hashlib
import shelve
from pathlib import Path
class CachedPipeline(Pipeline):
"""
Pipeline with a cache so expensive stages don't get re-run
when the input hasn't changed.
"""
def __init__(self, name: str, cache_dir: str = ".pipeline_cache"):
super().__init__(name)
self.cache_dir = Path(cache_dir)
self.cache_dir.mkdir(exist_ok=True)
def _cache_key(self, stage_name: str, input_data: str) -> str:
"""Builds a unique key for the cache."""
return hashlib.md5(f"{stage_name}:{input_data}".encode()).hexdigest()
def run_with_cache(
self,
initial_input: str,
cacheable_stages: list[str] = None,
ttl_hours: int = 24,
verbose: bool = True
) -> PipelineState:
"""
Runs the pipeline with a cache for the specified stages.
Cached stages are not re-run if the output exists in the cache.
"""
state = PipelineState(original_input=initial_input)
cache_file = str(self.cache_dir / f"{self.name}.shelve")
with shelve.open(cache_file) as cache:
for config in self.stages:
stage_input = state.get_last_output()
cache_key = self._cache_key(config.name, stage_input)
# Check the cache
use_cache = (
cacheable_stages and
config.name in cacheable_stages and
cache_key in cache
)
if use_cache:
cache_entry = cache[cache_key]
age_hours = (time.time() - cache_entry["timestamp"]) / 3600
if age_hours < ttl_hours:
state.outputs[config.name] = cache_entry["output"]
if verbose:
print(f"[{config.name}] ✓ Cache hit ({age_hours:.1f}h old)")
continue
# Run the stage normally
try:
output = config.fn(state)
state.outputs[config.name] = output
# Store in the cache if it's cacheable
if cacheable_stages and config.name in cacheable_stages:
cache[cache_key] = {"output": output, "timestamp": time.time()}
if verbose:
print(f"[{config.name}] ✓ Completed (stored in the cache)")
except Exception as e:
if verbose:
print(f"[{config.name}] ✗ Error: {e}")
raise
return stateTroubleshooting
Problem 1: State explosion (the state grows too large)
Symptom: The state accumulates the outputs of every stage and becomes enormous, causing memory problems and excessive tokens in later prompts.
Solution:
class OptimizedPipelineState(PipelineState):
"""State with a size limit for outputs."""
MAX_OUTPUT_CHARS = 2000
def set_output(self, name: str, output: str, full_output: str = None):
"""Stores the output with automatic truncation."""
if len(output) > self.MAX_OUTPUT_CHARS:
# Store the truncated version for use in prompts
self.outputs[name] = output[:self.MAX_OUTPUT_CHARS] + "\n[...truncated...]"
# Optionally store the full version in the metadata
if full_output:
self.metadata[f"{name}_full"] = full_output
else:
self.outputs[name] = outputProblem 2: High latency in sequential pipelines
Symptom: The pipeline takes 45 seconds for 5 stages of 10s each.
Solution: Parallelize the independent stages:
import asyncio
async def pipeline_with_auto_parallelization(
stage_groups: list[list[StageConfig]],
initial_input: str
) -> PipelineState:
"""
Runs groups of stages in parallel whenever possible.
stage_groups: [[stage1, stage2_parallel], [stage3_dependent], [stage4]]
"""
from openai import AsyncOpenAI
client_async = AsyncOpenAI()
state = PipelineState(original_input=initial_input)
for group in stage_groups:
if len(group) == 1:
# A single stage: run it normally
output = group[0].fn(state)
state.outputs[group[0].name] = output
else:
# Multiple stages: run them in parallel
async def run_stage(config):
loop = asyncio.get_event_loop()
output = await loop.run_in_executor(None, config.fn, state)
return config.name, output
tasks = [run_stage(c) for c in group]
results = await asyncio.gather(*tasks)
for name, output in results:
state.outputs[name] = output
return stateProblem 3: Infinite retry loop
Symptom: The pipeline retries a stage indefinitely when it's never going to succeed.
Solution: Implement a circuit breaker:
class CircuitBreaker:
"""
Implements the Circuit Breaker pattern for pipelines.
After N consecutive failures it opens the circuit and
allows no further attempts for T seconds.
"""
def __init__(self, failure_threshold: int = 3, reset_time: float = 60.0):
self.failure_threshold = failure_threshold
self.reset_time = reset_time
self.failures = 0
self.last_failure = 0
self.is_open = False
def can_run(self) -> bool:
if not self.is_open:
return True
# Check whether the reset time has already elapsed
if time.time() - self.last_failure > self.reset_time:
self.is_open = False
self.failures = 0
return True
return False
def record_success(self):
self.failures = 0
self.is_open = False
def record_failure(self):
self.failures += 1
self.last_failure = time.time()
if self.failures >= self.failure_threshold:
self.is_open = TrueExercises
Exercise 1: Email processing pipeline
Build a 3-stage pipeline to process support emails:
- Classify the urgency and type
- Extract the key information (customer, problem, context)
- Generate a personalized automatic reply
Include retry and a generic-reply fallback.
See solution
def email_support_pipeline(email: str) -> str:
def step_classify(state: PipelineState) -> str:
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Classify this email:\nurgency: high/medium/low\ntype: bug/question/complaint/other\nJSON: {{\"urgency\": str, \"type\": str}}\n\nEmail: {state.original_input}"}],
temperature=0, response_format={"type": "json_object"}
).choices[0].message.content
def step_extract(state: PipelineState) -> str:
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Extract from the email: customer_name, main_problem, context. JSON.\n\nEmail: {state.original_input}"}],
temperature=0, response_format={"type": "json_object"}
).choices[0].message.content
def step_reply(state: PipelineState) -> str:
classification = json.loads(state.get_output("classify", "{}"))
data = json.loads(state.get_output("extract", "{}"))
urgency = classification.get("urgency", "medium")
return client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": f"Generate an empathetic reply for a support email (urgency: {urgency}). Customer: {data.get('customer_name', 'valued customer')}. Problem: {data.get('main_problem', 'their question')}. 2-3 sentences."}],
temperature=0.2, max_tokens=150
).choices[0].message.content
def fallback_reply(state: PipelineState) -> str:
return "Dear customer, we have received your message and our team will look into it as soon as possible. Thank you for contacting us."
p = Pipeline("email_support")
p.add_stage(StageConfig("classify", step_classify, max_retries=2, critical=False))
p.add_stage(StageConfig("extract", step_extract, max_retries=2, critical=False))
p.add_stage(StageConfig("reply", step_reply, max_retries=1, fallback_fn=fallback_reply, critical=True))
state = p.run(email, verbose=False)
return state.get_output("reply")
test_email = "I've been unable to access my account for 3 days. I've lost important data and I need an urgent solution. I'm a premium customer."
print(email_support_pipeline(test_email))Exercise 2: Add monitoring
Instrument the document analysis pipeline built in this capsule with PipelineMonitor. Run 3 different documents and show the statistics.
See solution
monitor = PipelineMonitor()
pipeline = build_analysis_pipeline()
documents = [
"Q1 2026 sales report...",
"Service agreement between company A and company B...",
"Article about technology trends in 2026..."
]
for doc in documents:
state = pipeline.run(doc, verbose=False)
monitor.record_run(state, "document_analysis")
stats = monitor.stats()
print(f"Runs: {stats['n_runs']}")
print(f"Average time: {stats['average_time']:.2f}s")
print(f"Success rate: {stats['average_success_rate']:.0%}")
print(f"With fallback: {stats['n_runs_with_fallback']}")
print("\nSlowest stages:")
for stage, t in monitor.slowest_stages():
print(f" {stage}: {t:.2f}s average")Summary
- PipelineState: A centralized object that keeps the original input, every output, and the metrics of each stage
- StageConfig: Declarative configuration: function, max_retries, fallback, validator, timeout
- Smart retry: Exponential backoff with jitter to avoid the thundering herd
- Circuit Breaker: Prevents infinite loops on stages that systematically fail
- Monitoring: Record time, attempts and errors per stage so you can optimize
- Cache: For expensive stages that produce the same output for the same input