Prompt Chaining Techniques for Complex AI Workflows

A single prompt can answer a question or summarize a paragraph. But when you need an LLM to research a topic, extract structured data, validate it, and then generate a report, cramming everything into one prompt produces unreliable results. Prompt chaining solves this by breaking a complex task into a sequence of focused steps, where each step’s output feeds into the next.

This approach mirrors how experienced developers decompose problems. Instead of writing one function that does everything, you write small, testable functions that compose together. The same principle applies to LLM workflows.

Why Single Prompts Break Down

Consider asking an LLM to “analyze this customer feedback dataset, identify the top themes, rank them by severity, and write an executive summary with recommendations.” A single prompt attempting all of this will often skip steps, hallucinate themes, or produce shallow analysis.

The core issues with monolithic prompts:

• Cognitive overload — LLMs produce better output when focused on one task at a time
• No intermediate validation — you cannot check whether the extracted themes are accurate before the summary is written
• Debugging difficulty — when the final output is wrong, you cannot tell which reasoning step failed
• Token waste — retrying the entire prompt wastes tokens even when only the last step needs correction

Pattern 1: Sequential Chain

The simplest chain passes output from one prompt directly to the next. Each step has a clear, single responsibility.

from openai import OpenAI

client = OpenAI()

def run_step(system: str, user: str, model: str = "gpt-4o") -> str:
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        temperature=0.3,
    )
    return response.choices[0].message.content

def analyze_feedback(raw_feedback: str) -> dict:
    # Step 1: Extract themes
    themes = run_step(
        system="Extract the main themes from customer feedback. "
               "Return a numbered list of themes, each on its own line.",
        user=raw_feedback,
    )

    # Step 2: Classify severity
    classified = run_step(
        system="For each theme, classify severity as Critical, High, "
               "Medium, or Low. Return as 'Theme | Severity' per line.",
        user=f"Themes:\n{themes}\n\nOriginal feedback:\n{raw_feedback}",
    )

    # Step 3: Generate summary
    summary = run_step(
        system="Write a concise executive summary with actionable "
               "recommendations based on the classified themes.",
        user=f"Classified themes:\n{classified}",
    )

    return {"themes": themes, "classified": classified, "summary": summary}

Each step produces a focused output. If the severity classification looks wrong, you rerun only step 2 without touching extraction or summarization.

Pattern 2: Chain with Validation Gates

Adding validation between steps catches errors before they propagate. A validation gate inspects the intermediate output and either passes it forward or triggers a retry.

import json

def run_step_json(system: str, user: str) -> dict:
    response = run_step(
        system=system + "\n\nRespond with valid JSON only.",
        user=user,
    )
    return json.loads(response)

def extract_with_validation(text: str, max_retries: int = 2) -> list[dict]:
    system = (
        "Extract all product mentions from the text. Return a JSON array "
        "where each item has 'product', 'sentiment', and 'quote' fields."
    )

    for attempt in range(max_retries + 1):
        try:
            result = run_step_json(system=system, user=text)

            # Validate structure
            assert isinstance(result, list), "Result must be a list"
            for item in result:
                assert "product" in item, "Missing 'product' field"
                assert "sentiment" in item, "Missing 'sentiment' field"
                assert item["sentiment"] in (
                    "positive", "negative", "neutral"
                ), f"Invalid sentiment: {item['sentiment']}"

            return result

        except (json.JSONDecodeError, AssertionError) as e:
            if attempt == max_retries:
                raise ValueError(f"Validation failed after retries: {e}")
            # Retry with error context
            text = f"{text}\n\nPrevious attempt failed: {e}. Fix the issue."

    return []

The validation gate checks both JSON structure and business rules. This pattern is essential for production systems where downstream code depends on specific data shapes.

Pattern 3: Map-Reduce Chain

When processing large inputs, split them into chunks, process each chunk independently, then combine results. This pattern works well for long documents, multiple files, or batch analysis.

def chunk_text(text: str, max_chars: int = 3000) -> list[str]:
    paragraphs = text.split("\n\n")
    chunks, current = [], ""
    for para in paragraphs:
        if len(current) + len(para) > max_chars and current:
            chunks.append(current.strip())
            current = ""
        current += para + "\n\n"
    if current.strip():
        chunks.append(current.strip())
    return chunks

def map_reduce_summarize(document: str) -> str:
    chunks = chunk_text(document)

    # Map: summarize each chunk
    chunk_summaries = []
    for i, chunk in enumerate(chunks):
        summary = run_step(
            system=f"Summarize this section (part {i+1} of {len(chunks)}). "
                   "Preserve key facts, numbers, and conclusions.",
            user=chunk,
        )
        chunk_summaries.append(summary)

    # Reduce: combine summaries into final output
    combined = "\n---\n".join(chunk_summaries)
    final = run_step(
        system="Combine these section summaries into one coherent summary. "
               "Remove redundancy and maintain logical flow.",
        user=combined,
    )
    return final

For better throughput, run the map phase concurrently using asyncio.gather or a thread pool. The reduce step must wait for all map results.

Pattern 4: Branching Chain

Some workflows require different processing paths based on intermediate results. A classifier step routes the input to specialized prompts.

def route_support_ticket(ticket: str) -> dict:
    # Step 1: Classify the ticket
    category = run_step(
        system="Classify this support ticket into exactly one category: "
               "billing, technical, account, or general. "
               "Respond with only the category name.",
        user=ticket,
    ).strip().lower()

    # Step 2: Route to specialized handler
    handlers = {
        "billing": "You are a billing specialist. Extract the billing "
                   "issue, affected amount, and suggest a resolution.",
        "technical": "You are a technical support engineer. Identify the "
                     "technical issue, likely root cause, and steps to fix.",
        "account": "You are an account manager. Identify the account "
                   "action needed and any verification requirements.",
        "general": "You are a support agent. Summarize the request and "
                   "suggest the best course of action.",
    }

    system = handlers.get(category, handlers["general"])
    analysis = run_step(system=system, user=ticket)

    return {"category": category, "analysis": analysis}

The branching pattern keeps each specialized prompt focused. The billing handler never sees technical jargon instructions, and the technical handler does not waste tokens on billing logic.

Managing Context Between Steps

The most common mistake in prompt chains is losing important context between steps. Two strategies help.

Explicit context passing includes relevant earlier outputs in later prompts. This is simple but increases token usage:

step2_result = run_step(
    system="Refine the analysis with additional detail.",
    user=f"Original input:\n{original}\n\nStep 1 result:\n{step1_result}",
)

Structured state objects keep a running state dictionary that each step reads from and writes to:

state = {"input": raw_text, "steps": {}}
state["steps"]["extraction"] = extract_entities(state["input"])
state["steps"]["validation"] = validate_entities(state["steps"]["extraction"])
state["steps"]["report"] = generate_report(state)

Use structured state when chains have more than three steps or when later steps need results from non-adjacent earlier steps.

When Not to Chain

Prompt chaining adds latency and cost. Every step is a separate API call. Use a single prompt when the task is straightforward, the output is short, and intermediate validation is unnecessary. A good rule of thumb: if you can reliably get the right answer in one call with a well-crafted prompt, do not chain.

Chain when you need intermediate validation, when the task has distinct phases, when inputs exceed context windows, or when different steps benefit from different models or temperatures.

Summary

Prompt chaining transforms unreliable monolithic prompts into debuggable, validated pipelines. Start with the sequential pattern for simple multi-step tasks. Add validation gates when output structure matters. Use map-reduce for large inputs and branching for classification-driven workflows. Keep context explicit between steps and resist the urge to chain when a single prompt suffices.