Prompt Chaining: Build Multi-Step LLM Workflows

A single LLM call can handle simple tasks like summarization or classification. But real-world applications often need multi-step reasoning: analyze a document, extract key entities, look up related data, then generate a report. Prompt chaining breaks these complex tasks into a sequence of focused LLM calls, where each step’s output feeds into the next step’s input.

This approach is more reliable than cramming everything into one massive prompt. Each step is simpler, easier to debug, and independently testable.

Why Chain Prompts

There are several reasons to split a complex task into multiple LLM calls instead of one.

Better accuracy. Smaller, focused prompts produce more reliable outputs than long, multi-instruction prompts. The model is less likely to skip steps or confuse requirements when each call has a single clear objective.

Debuggability. When something goes wrong in a five-step chain, you can inspect the output of each step to find where the problem occurred. With a single monolithic prompt, you only see the final (wrong) result.

Mixed models. You can use a cheaper, faster model for simple steps like classification and reserve a more capable model for complex steps like analysis. This optimizes both cost and latency.

Conditional logic. Some steps only need to run based on the output of previous steps. Chaining lets you add branching logic that a single prompt cannot express.

Sequential Chains

The simplest chain pattern passes the output of one step directly to the next.

import openai
import json
from dataclasses import dataclass

client = openai.OpenAI()

def llm_call(system: str, user: str, model: str = "gpt-4o-mini", temperature: float = 0) -> str:
    """Helper for a single LLM call."""
    response = client.chat.completions.create(
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user}
        ],
        temperature=temperature
    )
    return response.choices[0].message.content

def analyze_customer_feedback(feedback: str) -> dict:
    """Multi-step analysis of customer feedback."""

    # Step 1: Classify the feedback
    category = llm_call(
        system="Classify customer feedback into exactly one category. Respond with only the category name.",
        user=f"Categories: bug_report, feature_request, complaint, praise, question\n\nFeedback: {feedback}"
    ).strip().lower()

    # Step 2: Extract key details based on the category
    details = llm_call(
        system=f"Extract structured details from this {category}. Return JSON.",
        user=f"""Feedback: {feedback}

Extract:
- subject: what the feedback is about (1-2 words)
- sentiment_score: float from -1.0 (very negative) to 1.0 (very positive)
- urgency: low, medium, or high
- key_phrases: list of important phrases from the text

Return only valid JSON.""",
        model="gpt-4o"
    )

    # Step 3: Generate a response draft
    details_parsed = json.loads(details)
    response_draft = llm_call(
        system="You are a customer service agent. Write a brief, empathetic response to the customer.",
        user=f"""Category: {category}
Subject: {details_parsed['subject']}
Sentiment: {details_parsed['sentiment_score']}
Original feedback: {feedback}

Write a 2-3 sentence response acknowledging their feedback and stating next steps."""
    )

    return {
        "category": category,
        "details": details_parsed,
        "response_draft": response_draft
    }

Each step is simple enough that it rarely fails, and the overall result is more reliable than asking one prompt to do all three tasks simultaneously.

Branching Chains

Sometimes you need different processing paths based on intermediate results. A branching chain routes to different prompts based on the output of an earlier step.

def process_support_ticket(ticket: str) -> dict:
    """Route and process a support ticket through different paths."""

    # Step 1: Triage
    triage = llm_call(
        system="Classify this support ticket. Respond with only: billing, technical, account, or other.",
        user=ticket
    ).strip().lower()

    # Step 2: Branch based on triage result
    if triage == "billing":
        analysis = llm_call(
            system="""You are a billing specialist. Analyze the billing issue and extract:
- issue_type: overcharge, refund_request, payment_failed, subscription_change
- amount_mentioned: dollar amount if any, or null
- account_action_needed: what needs to happen
Return JSON only.""",
            user=ticket
        )
    elif triage == "technical":
        analysis = llm_call(
            system="""You are a technical support engineer. Analyze the technical issue and extract:
- product_area: which part of the product is affected
- error_messages: any error messages mentioned
- steps_to_reproduce: summarized steps if mentioned
- severity: p1_critical, p2_major, p3_minor
Return JSON only.""",
            user=ticket,
            model="gpt-4o"  # Use a stronger model for technical analysis
        )
    elif triage == "account":
        analysis = llm_call(
            system="""You are an account specialist. Analyze the account issue and extract:
- issue_type: access, permissions, deletion, profile_update
- requires_verification: true if identity verification is needed
- action_needed: what needs to happen
Return JSON only.""",
            user=ticket
        )
    else:
        analysis = llm_call(
            system="Summarize this support request in JSON with fields: summary, suggested_department, priority.",
            user=ticket
        )

    return {
        "triage": triage,
        "analysis": json.loads(analysis)
    }

Validation Loops

A validation loop runs a step, checks the output, and retries if the output does not meet quality criteria. This is one of the most powerful chaining patterns.

from pydantic import BaseModel, ValidationError
from typing import Optional

class CodeReview(BaseModel):
    has_bugs: bool
    bug_descriptions: list[str]
    suggestions: list[str]
    quality_score: int  # 1-10
    approved: bool

def review_with_validation(code: str, max_retries: int = 3) -> CodeReview:
    """Review code with automatic retry on invalid output."""
    messages = [
        {
            "role": "system",
            "content": """You are a code reviewer. Analyze the code and return a JSON review with:
- has_bugs (bool): whether you found bugs
- bug_descriptions (list[str]): description of each bug found
- suggestions (list[str]): improvement suggestions
- quality_score (int): 1-10 rating
- approved (bool): true if quality_score >= 7 and no bugs"""
        },
        {"role": "user", "content": f"Review this code:\n```\n{code}\n```"}
    ]

    for attempt in range(max_retries):
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=messages,
            response_format={"type": "json_object"},
            temperature=0
        )
        raw = response.choices[0].message.content

        try:
            return CodeReview.model_validate_json(raw)
        except ValidationError as e:
            messages.append({"role": "assistant", "content": raw})
            messages.append({
                "role": "user",
                "content": f"Validation error: {e}. Fix the JSON and try again."
            })

    raise ValueError(f"Failed after {max_retries} attempts")

Building a Chain Framework

For production applications, build a lightweight framework that handles common concerns like logging, error handling, and state management.

from dataclasses import dataclass, field
from typing import Any, Callable
import time
import logging

logger = logging.getLogger(__name__)

@dataclass
class ChainStep:
    name: str
    system_prompt: str
    user_prompt_template: str  # Can reference {variables} from state
    model: str = "gpt-4o-mini"
    output_key: str = ""  # Key to store result in state
    parser: Callable[[str], Any] = lambda x: x  # Post-process the output
    condition: Callable[[dict], bool] = lambda state: True  # Skip if False

@dataclass
class ChainResult:
    success: bool
    state: dict
    steps_executed: list[str]
    total_time: float
    errors: list[str] = field(default_factory=list)

class PromptChain:
    def __init__(self, steps: list[ChainStep]):
        self.steps = steps
        self.client = openai.OpenAI()

    def run(self, initial_state: dict) -> ChainResult:
        state = dict(initial_state)
        steps_executed = []
        errors = []
        start_time = time.time()

        for step in self.steps:
            # Check condition
            if not step.condition(state):
                logger.info(f"Skipping step '{step.name}' (condition not met)")
                continue

            logger.info(f"Running step '{step.name}'")

            try:
                user_prompt = step.user_prompt_template.format(**state)

                response = self.client.chat.completions.create(
                    model=step.model,
                    messages=[
                        {"role": "system", "content": step.system_prompt},
                        {"role": "user", "content": user_prompt}
                    ],
                    temperature=0
                )
                raw_output = response.choices[0].message.content
                parsed_output = step.parser(raw_output)
                state[step.output_key or step.name] = parsed_output
                steps_executed.append(step.name)

            except Exception as e:
                logger.error(f"Step '{step.name}' failed: {e}")
                errors.append(f"{step.name}: {str(e)}")
                return ChainResult(
                    success=False,
                    state=state,
                    steps_executed=steps_executed,
                    total_time=time.time() - start_time,
                    errors=errors
                )

        return ChainResult(
            success=True,
            state=state,
            steps_executed=steps_executed,
            total_time=time.time() - start_time
        )

# Usage
chain = PromptChain([
    ChainStep(
        name="classify",
        system_prompt="Classify the text as: positive, negative, or neutral. Respond with one word.",
        user_prompt_template="Text: {input_text}",
        output_key="sentiment"
    ),
    ChainStep(
        name="extract_topics",
        system_prompt="Extract the main topics discussed. Return a JSON list of strings.",
        user_prompt_template="Text: {input_text}",
        output_key="topics",
        parser=json.loads
    ),
    ChainStep(
        name="generate_summary",
        system_prompt="Write a one-paragraph summary incorporating the sentiment and topics.",
        user_prompt_template="Sentiment: {sentiment}\nTopics: {topics}\nOriginal: {input_text}",
        output_key="summary",
        model="gpt-4o"
    ),
])

result = chain.run({"input_text": "The new update is fantastic! The dark mode feature and improved search are exactly what we needed."})
print(result.state["summary"])

Parallel Chains

When steps are independent of each other, run them in parallel to reduce latency.

import asyncio

async def async_llm_call(system: str, user: str, model: str = "gpt-4o-mini") -> str:
    """Async version of LLM call."""
    response = await asyncio.to_thread(
        client.chat.completions.create,
        model=model,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user}
        ],
        temperature=0
    )
    return response.choices[0].message.content

async def parallel_analysis(text: str) -> dict:
    """Run multiple analyses in parallel, then combine."""

    # These three analyses are independent, so run them in parallel
    sentiment_task = async_llm_call(
        "Analyze sentiment. Return: positive, negative, or neutral.",
        text
    )
    topic_task = async_llm_call(
        "Extract main topics as a JSON list of strings.",
        text
    )
    language_task = async_llm_call(
        "Detect the language. Return the ISO 639-1 code only.",
        text
    )

    sentiment, topics, language = await asyncio.gather(
        sentiment_task, topic_task, language_task
    )

    # This step depends on the parallel results
    summary = await async_llm_call(
        "Write a summary incorporating all the analysis.",
        f"Text: {text}\nSentiment: {sentiment}\nTopics: {topics}\nLanguage: {language}",
        model="gpt-4o"
    )

    return {
        "sentiment": sentiment.strip(),
        "topics": json.loads(topics),
        "language": language.strip(),
        "summary": summary
    }

When to Chain vs Single Prompt

Not every task needs chaining. Here is a guide for when each approach works best.

Use a single prompt when:

• The task is simple and well-defined (classification, summarization, translation)
• Latency is critical and you need a single round trip
• The task does not require conditional logic

Use chaining when:

• The task has multiple distinct phases (analyze, then decide, then act)
• Different steps need different models or temperatures
• You need conditional branching based on intermediate results
• You need validated, structured output at each step
• The overall task is too complex for a single prompt to handle reliably

Wrapping Up

Prompt chaining transforms LLMs from single-shot question answerers into components of larger workflows. By breaking complex tasks into focused steps, you get better accuracy, easier debugging, and the ability to mix models and add conditional logic. Start with simple sequential chains and add complexity as needed. The chain framework pattern gives you a clean abstraction for building and maintaining multi-step LLM workflows in production, with consistent error handling and logging across all your chains.