Prompt Chaining

The Core Idea

Trying to solve a complex task in a single Large Language Model (LLM) call is like asking one person to build an entire car. Prompt chaining breaks the problem into a sequence of smaller, focused steps — each one a separate LLM call, where the output of one becomes the input to the next.

The Danger: Error Propagation

The same dependency that makes chaining powerful also makes it fragile. If an early step produces flawed output, that error compounds through every step that follows.

This is the fundamental tension: chaining gives you better step-by-step accuracy, but introduces dependency risk. The solution: validation gates between every step.

Validation Gates

Quality control checkpoints between each step in the chain. Before an output is passed to the next step, a gate checks whether it's accurate, relevant, and correctly formatted. Catch errors early, before they cascade.

Four Types of Validation

Validation in Code

Each validation type translates to a concrete gate function that sits between chain steps. (chat() is the thin LLM wrapper used throughout this series — it sends a prompt and returns the text reply.)

import json

def validate_json_output(output):
    """Gate: is it valid JSON with required keys?"""
    try:
        data = json.loads(output)
    except json.JSONDecodeError:
        return False, "Output is not valid JSON"

    required = ["category", "urgency", "summary"]
    missing = [k for k in required if k not in data]
    if missing:
        return False, f"Missing keys: {missing}"

    return True, data

def validate_with_llm(original_input, agent_output):
    """Gate: does the output actually address the input?"""
    verdict = chat(
        user_prompt=f"""INPUT:\n{original_input}\n\nOUTPUT:\n{agent_output}\n\n
Does the output accurately and completely address the input?
Reply with exactly PASS or FAIL followed by a one-line reason.""",
        system_prompt="""You are a strict quality reviewer. Evaluate whether
            the output faithfully addresses the input.""",
        temperature=0,
        max_tokens=60,
    )
    passed = verdict.strip().upper().startswith("PASS")
    return passed, verdict

Production note: Parsing raw text with .startswith("PASS") is fragile — an LLM might return "Based on my analysis, PASS." which starts with "B", not "P", and your gate silently fails. In production, use structured outputs (e.g. response_format={"type": "json_object"} in the OpenAI Application Programming Interface (API), or libraries like instructor / Pydantic) to force the model to return an exact schema like {"passed": true, "reason": "..."} — no string parsing needed.

def validate_email_rules(email_text):
    """Gate: does the email meet business format requirements?"""
    rules = {
        "has greeting":  any(email_text.lower().startswith(g)
                          for g in ["dear", "hi", "hello"]),
        "has closing":   any(c in email_text.lower()
                          for c in ["regards", "sincerely", "thank you"]),
        "under 500 words": len(email_text.split()) < 500,
    }
    failures = [r for r, ok in rules.items() if not ok]
    return len(failures) == 0, failures

def validate_confidence(agent_output, threshold=0.7):
    """Gate: ask the LLM to self-rate, reject below threshold."""
    score_text = chat(
        user_prompt=f"""Rate your confidence in this output on a scale of
0.0 to 1.0. Reply with ONLY the number.\n\n{agent_output}""",
        system_prompt="You are a calibration judge. Output a single float.",
        temperature=0,
        max_tokens=10,
    )
    try:
        score = float(score_text.strip())
    except ValueError:
        return False, "Could not parse confidence score"
    return score >= threshold, score

Using Gates in a Chain

# Between step 1 and step 2: validate then pass or retry
step1_output = classifier_agent(ticket)

passed, detail = validate_json_output(step1_output)
if not passed:
    print(f"Gate failed: {detail} -- retrying step 1")
    step1_output = classifier_agent(ticket)  # retry

step2_output = responder_agent(step1_output)

def run_with_gate(agent_fn, gate_fn, input_data, max_retries=2):
    """Run an agent, validate output, re-prompt with feedback on failure."""
    output = agent_fn(input_data)

    for attempt in range(max_retries):
        passed, detail = gate_fn(output)
        if passed:
            return output

        # Re-prompt with structured feedback
        output = agent_fn(
            f"--- ORIGINAL INPUT ---\n{input_data}\n\n"
            f"--- GATE FEEDBACK ---\n"
            f"Previous attempt rejected: {detail}. Fix the issues."
        )

    return output  # fallback: return best effort after retries

# Usage
classification = run_with_gate(classifier_agent, validate_json_output, ticket)
response = responder_agent(classification)

Cost note: Every retry re-runs the full agent call — you pay for the tokens again. Set max_retries conservatively (1–2 is typical) and log every retry so you can monitor spend. An infinite retry loop against a stubborn validation gate will drain your API budget fast.

When Validation Fails: Error Handling

A gate catches a bad output. Now what? Five strategies, from simplest to most sophisticated:

Context Management

Each step needs context from previous outputs, but too much context degrades performance — the LLM gets distracted or hits token limits. Two strategies to balance this:

Example: Customer Support Escalation Chain

A three-step chain that classifies a support ticket, drafts a response, and translates it. Watch what each step actually needs from the previous one:

Worked Example: Research → Drafter Chain

The simplest real prompt chain: a researcher gathers structured facts, a drafter turns them into a polished article. Two function-based agents, each powered by an LLM, connected by a single variable.

Notice how agents are plain functions. The chaining happens in run_simple_chain — the researcher's return value is passed directly into the drafter's prompt.

def researcher_agent(topic):
    return chat(
        user_prompt=f"Research the following topic thoroughly:\n\n{topic}",
        system_prompt="""You are a research specialist who provides structured
            information. Always format with: # OVERVIEW # KEY POINTS # DETAILS""",
        max_tokens=600,
        temperature=0.3,  # low = factual, focused
    )

def drafter_agent(topic, research_results):
    return chat(
        user_prompt=f"Write an engaging article about: {topic}\n\n"
                    f"Use this research as your source material:\n\n{research_results}",
        system_prompt="""You are a content drafter who creates engaging material
            from research. Create a well-structured article...""",
        max_tokens=800,
        temperature=0.7,  # higher = creative prose
    )

def run_simple_chain(topic):
    research = researcher_agent(topic)
    article  = drafter_agent(topic, research)  # ← the chain
    return {"research": research, "article": article}

Worked Example: Startup Pitch Evaluation

A more ambitious chain with four specialised agents and a key new pattern: fan-in — one step receiving inputs from two or more earlier steps. The final recommender synthesises insights from two separate analysis streams.

# Agent 1: Understand the pitch
def pitch_analyst_agent(startup_pitch):
    return chat(
        user_prompt=f"Analyse this startup pitch: {startup_pitch}",
        system_prompt="""You are a startup analyst who evaluates pitch decks.
            Summarise the proposition, target market, revenue model...""",
        max_tokens=600, temperature=0.2,
    )

# Agent 2: Assess the market
def market_assessor_agent(pitch_analysis):
    return chat(
        user_prompt=f"""Based on this pitch analysis, assess the market:\n\n
                    {pitch_analysis}""",
        system_prompt="""You are a market research specialist. Evaluate TAM,
            competitive landscape, timing, and growth potential...""",
        max_tokens=600, temperature=0.2,
    )

# Agent 3: Identify risks
def risk_analyst_agent(market_assessment):
    return chat(
        user_prompt=f"Identify key risks based on this assessment: {market_assessment}",
        system_prompt="""You are a risk analyst specialising in early-stage
            investments. Rate each risk as high, medium, or low...""",
        max_tokens=600, temperature=0.3,
    )

# Agent 4: Synthesise — receives inputs from agents 2 AND 3
def investment_recommender_agent(market_assessment, risk_analysis):
    return chat(
        user_prompt=f"""Given the following market assessment:
--- MARKET ASSESSMENT ---
{market_assessment}
--- END MARKET ASSESSMENT ---

And the following risk analysis:
--- RISK ANALYSIS ---
{risk_analysis}
--- END RISK ANALYSIS ---

Provide a clear invest or pass recommendation with rationale.""",
        system_prompt="You are a venture capital investment committee adviser...",
        max_tokens=700, temperature=0.3,
    )

def run_pitch_evaluation(startup_pitch):
    analysis = pitch_analyst_agent(startup_pitch)
    market   = market_assessor_agent(analysis)
    risks    = risk_analyst_agent(market)
    # fan-in: takes outputs from BOTH step 2 and step 3
    return investment_recommender_agent(market, risks)

Orchestration Topologies

A topology is simply the shape of a chain — how its steps connect and feed one another. The two worked examples showed two shapes; there's a third that becomes important in later lessons:

* Implementation note: "Latency = slowest branch" in the parallel topology only holds if branches execute concurrently (e.g. asyncio.gather in Python). With synchronous code, parallel branches run sequentially and the latency equals the sum of all branches. True parallelism requires async orchestration.

Prompt Chaining vs. Agentic Systems

A common source of confusion: prompt chaining and agentic systems both involve multiple LLM calls, but they are fundamentally different architectures.

Lesson Recap