Prompt Chaining & Pipelines

Not every task fits in one prompt. Complex workflows need multiple LLM calls, each one taking the output of the previous as input. Prompt chaining breaks a task into sequential steps; pipelines add branching and parallel execution. This lesson covers designing chains that are debuggable, recoverable, and testable.

What you'll learn

• Prompt chaining splits a complex task into sequential LLM calls, each step does one thing well
• Error handling at each step means a failure in step 3 doesn't force you to re-run steps 1 and 2
• Pipelines add branching (if-then) and parallelism (fan-out) on top of chains

The problem

You need to analyze a legal document: extract key clauses, classify each by risk level, summarize the findings, and format them as a report. One prompt for all of that produces a mediocre result, the model does a shallow pass on each subtask. Chaining breaks it into four focused prompts, each with a single job, each producing output the next step relies on.

Build it

Step 1: A simple two-step chain

Break complex objectives down into a sequential pipeline. The first step extracts necessary facts, while the subsequent prompt evaluates the parsed output.

# Step 1: Extract key information
extraction_prompt = """Extract the following from the document below:
- All dates mentioned
- All monetary amounts
- All named parties
Return as JSON.

Document: {document}"""

extraction_result = call_llm(extraction_prompt.format(document=doc))

# Step 2: Classify risk based on extraction
risk_prompt = """Based on the extracted information below, classify the document risk as LOW, MEDIUM, or HIGH.
Consider: number of parties, total monetary exposure, and timeframe.

{extraction}"""

risk_result = call_llm(risk_prompt.format(extraction=extraction_result))

Step 2: Add error handling and retry

Wrap each chain step in a validation loop to catch hallucinated outputs early. Automatically prompt the model to correct its mistakes before moving on.

def chain_step(prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            result = call_llm(prompt)
            validated = validate_output(result)  # Pydantic or regex
            return validated
        except ValidationError:
            if attempt == max_retries - 1:
                raise
            prompt = f"{prompt}\n\nPrevious attempt failed validation. Try again."

Step 3: Fan-out pipeline (parallel)

Execute independent tasks concurrently to reduce latency. This fan-out pattern processes multiple streams at once before merging them into a final synthesis.

import concurrent.futures

topics = ["security", "performance", "style", "documentation"]

def review_topic(topic):
    prompt = f"Review the following code for {topic} issues:\n\n{code}"
    return call_llm(prompt)

with concurrent.futures.ThreadPoolExecutor() as executor:
    reviews = list(executor.map(review_topic, topics))

# Merge step
merge_prompt = f"Synthesize these four code reviews into one report:\n" + "\n---\n".join(reviews)
final_report = call_llm(merge_prompt)

Step 4: Conditional branching

Use an initial triage prompt to determine the next logical step in the flow. This routes the input to specialized prompts based on the detected category.

sentiment = call_llm("Classify the sentiment of this customer message: " + message)

if sentiment == "ANGRY":
    response = call_llm(ESCALATION_PROMPT.format(message=message))
elif sentiment == "CONFUSED":
    response = call_llm(CLARIFICATION_PROMPT.format(message=message))
else:
    response = call_llm(STANDARD_PROMPT.format(message=message))

What goes wrong

Mistake	How you notice it	The fix
Error in step 3 requires re-running steps 1-2	Wasted API calls and latency	Cache intermediate results. Store each step's output so you can resume from any point
Fan-out overwhelms rate limits	429 errors on parallel calls	Use a semaphore or rate limiter. Batch parallel calls to stay under provider limits
Accumulated errors across steps	Final output is subtly wrong, hard to trace	Log every step's input and output. Add validation at each step boundary
Too many steps, diminishing returns	Each additional step adds latency but minimal quality improvement	Start with 2-3 steps. Only add steps when a quality metric shows improvement

Confirm it worked

Test the sequential pipeline by processing a sample document through extraction, classification, and summarization. This confirms data flows correctly across prompt boundaries.

# Test: a 3-step chain that extracts, analyzes, and summarizes
doc = "Contract between Acme Corp and Beta Inc for $50,000, signed 2026-01-15."
step1 = call_llm(f"Extract parties, amounts, and dates as JSON: {doc}")
step2 = call_llm(f"Classify risk: {step1}")
step3 = call_llm(f"Summarize in one sentence: Risk={step2} | {step1}")
assert "Acme" in step3 and "Beta" in step3

Next: Retrieval-Augmented Prompting