The mental model for prompt chaining is a Unix pipe. Each LLM call is a process with stdin and stdout. The output from one process becomes the input to the next, and each process does exactly one thing well. The difference from a Unix pipe is that LLM outputs are unstructured prose by default, so you need explicit contracts at each boundary: usually JSON schemas, structured extraction, or constrained output formats that let you parse and validate before passing downstream.
Here is a real-world scenario that illustrates why this matters. Suppose you are building a legal document review tool. A user uploads a 40-page contract. You cannot fit the whole document into a single prompt with your analysis instructions and still have room for a useful response. Even if you could, asking the model to simultaneously identify risky clauses, summarize obligations, flag missing standard provisions, and draft a memo would produce mediocre results on all four tasks. A pro would chain it: first a map step that processes each page or section and extracts structured findings (clause type, risk level, verbatim text), then a reduce step that aggregates findings across sections into a ranked risk list, then a final generation step that drafts the attorney memo from that structured list. The map step uses a cheap, fast model like gpt-4o-mini. The final memo step uses a more capable model. Each step's output is JSON you can inspect, log, and unit-test.
The main tradeoff against single-shot prompting is latency and cost. Three serial LLM calls at 500ms each is 1.5 seconds of minimum wall time before you factor in your own processing. If you need sub-500ms response times, chaining is the wrong tool unless you can parallelize branches or pre-compute stages. The alternative, a single large context prompt, is faster but less reliable for multi-part tasks and wastes tokens on instructions the model partially ignores. Retrieval-augmented generation (RAG) pipelines are a specific form of chaining where the first stage is a retrieval call (vector search or keyword search) and the second stage is generation with retrieved context injected. The same boundary-contract principles apply.
At 10 users, you run chains sequentially and log to stdout. Nobody cares. At 10,000 users, you need async execution (Python asyncio with async OpenAI client), per-chain timeouts so a stuck step does not block a thread forever, and a queue-based architecture so spikes do not overwhelm your rate limits. You also need step-level observability: which stage failed, what was the input, what was the output, how many tokens did each step consume. At 10 million users, individual chain steps are microservices or serverless functions, you cache deterministic intermediate outputs (extracted entities from the same document do not change), and you run A/B tests on individual steps in isolation using prompt versioning tooling covered in the aidev-prompt-patterns-versioning subtopic.
Cost and reliability deserve a specific call-out. A three-step chain that runs gpt-4o-mini for steps 1 and 2 and gpt-4o only for step 3 can be 80% cheaper than running gpt-4o for all three steps, with comparable or better output quality because each step is scoped narrowly. For reliability, design every chain to be resumable: persist intermediate outputs to a database or cache with the original request ID so that if step 3 fails, you retry step 3 alone rather than restarting from step 1. This matters especially when step 1 involves expensive document processing or external API calls.
Key Takeaways
- Split complex tasks into focused single-responsibility LLM calls that pipe output forward.
- Validate and transform outputs at each chain boundary before passing to the next step.
- Match model choice and temperature to each stage, not a one-size-fits-all setting.
- Design chains so any single step can fail, retry, or be replaced without breaking the whole pipeline.
Pro tips
- Define a typed schema (Pydantic model or TypedDict) for every chain boundary before writing any prompt. The schema IS the contract. If you cannot define what step N produces, the prompt for step N is not ready.
- Run steps 1 through N-1 with your cheapest fast model and only promote to a larger model for the final synthesis or generation step. Token costs compound quickly in multi-step chains, and intermediate steps rarely need frontier model quality.
- Persist intermediate outputs keyed by (request_id, step_name). When step 3 fails at 2 AM, you retry step 3 alone with the saved step 2 output instead of burning tokens and latency re-running everything from scratch.
- Parallel branch chains (fan-out/fan-in) cut wall-clock latency dramatically when steps are independent. If step 2a and step 2b do not depend on each other, run them concurrently with asyncio.gather and merge results before step 3.
Common pitfalls
- Mistake: Passing raw LLM text directly into the next prompt without validation. Fix: Parse and validate every chain boundary. If you expect JSON, parse it and catch the exception before the next step consumes garbage.
- Mistake: Building chains where every step uses the same large model at temperature 0.7. Fix: Match model and temperature to the step's job. Extraction needs low temp and a cheap model; creative generation can use higher temp and a better model.
- Mistake: No step-level observability, so when a chain fails you cannot tell which step produced bad output. Fix: Log step name, model, token counts, latency, and a truncated output hash at every step from day one.
- Mistake: Chains that grow unbounded in context by appending all previous outputs. Fix: Pass only what the next step needs. Summarize or extract from prior outputs rather than concatenating entire responses into a growing prompt.
When to use prompt chaining vs alternatives
| Option | Use when | Avoid when |
|---|---|---|
| Single-shot prompt | Task is well-scoped, fits comfortably in context, and latency budget is tight. | Task has multiple distinct sub-goals, or output quality on any one part is unacceptably low. |
| Prompt chain (sequential) | Each step's output is a required input to the next step and you need per-step auditability. | Steps are fully independent (use parallel fan-out instead) or total latency exceeds budget. |
| Parallel fan-out / fan-in chain | Multiple independent analyses need to run on the same input, then merge results. | Steps have data dependencies on each other, or your rate limit cannot absorb concurrent calls. |
| Agent loop (dynamic chaining) | Number of steps is not known in advance and the model must decide what to do next. | The workflow is deterministic and known ahead of time. Agents add latency and unpredictability unnecessarily. |
| RAG pipeline (retrieval + generation chain) | The generation step requires factual grounding from external documents or a knowledge base. | All needed context fits in a static system prompt and retrieval latency is not acceptable. |
Code Example
# openai>=1.0.0
import os
from openai import OpenAI
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
def llm(prompt: str, system: str = "", model: str = "gpt-4o-mini") -> str:
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
resp = client.chat.completions.create(model=model, messages=messages)
return resp.choices[0].message.content.strip()
raw_review = "Battery dies after 2 hours. Screen is gorgeous though. Returned it."
# Step 1: Extract sentiment and key facts
extracted = llm(
f"Extract: sentiment (positive/negative/mixed), key issues, key positives. JSON only.\n\n{raw_review}"
)
# Step 2: Pipe extracted data into a response draft
response_draft = llm(
f"Given this product review analysis:\n{extracted}\n\nWrite a 2-sentence empathetic customer service reply."
)
print(response_draft)How this code works
This code demonstrates "prompt chaining" and "output piping," a powerful technique where the result of one AI interaction is used as input for a subsequent one, allowing for more complex tasks. The core idea is to break down a big problem into smaller, manageable steps for the AI. It starts with a raw_review — a typical customer comment. A helper function, llm, handles communication with the OpenAI API, sending a prompt and optionally a system message to an AI model and returning its text response. Notice that model="gpt-4o-mini" is specified as a default; this selects a particular AI model for its balance of capability and efficiency, a common practice to control cost and speed.
The process unfolds in two steps. First, the extracted variable captures the AI's analysis of the raw_review, asking it to identify sentiment, issues, and positives in a structured "JSON only" format. This structured output is crucial for reliability. Second, this extracted data is piped directly into another llm call to generate a response_draft. Here, the AI leverages the analysis from the previous step to craft an empathetic customer service reply. Finally, print(response_draft) displays the refined, two-sentence response, showcasing how a multi-stage approach produces a more thoughtful and relevant output than a single, complex prompt might achieve.
Production-grade example
Async calls, per-step retries with backoff, token logging, JSON validation, and graceful degradation on parse failure.
# openai>=1.0.0, tenacity>=8.2.0
import asyncio
import json
import logging
import os
import time
from typing import Any
from openai import AsyncOpenAI, RateLimitError, APITimeoutError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger(__name__)
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"], timeout=15.0)
@retry(
stop=stop_after_attempt(3),
wait=wait_exponential(multiplier=1, min=2, max=10),
retry=retry_if_exception_type((RateLimitError, APITimeoutError)),
reraise=True,
)
async def llm_call(
step_name: str,
messages: list[dict],
model: str = "gpt-4o-mini",
response_format: dict | None = None,
) -> str:
start = time.monotonic()
kwargs: dict[str, Any] = {"model": model, "messages": messages}
if response_format:
kwargs["response_format"] = response_format
resp = await client.chat.completions.create(**kwargs)
elapsed = time.monotonic() - start
usage = resp.usage
log.info(
"step=%s model=%s prompt_tokens=%d completion_tokens=%d latency_ms=%d",
step_name, model,
usage.prompt_tokens if usage else -1,
usage.completion_tokens if usage else -1,
int(elapsed * 1000),
)
return resp.choices[0].message.content.strip()
async def review_analysis_chain(raw_review: str) -> dict:
# Step 1: Extract structured facts (cheap model, JSON output)
try:
extracted_raw = await llm_call(
step_name="extract",
messages=[
{"role": "system", "content": "Return valid JSON only. No prose."},
{"role": "user", "content": (
'Extract sentiment (positive/negative/mixed), '
'a list of issues, and a list of positives from this review.\n\n'
f'{raw_review}'
)},
],
response_format={"type": "json_object"},
)
extracted: dict = json.loads(extracted_raw)
except (json.JSONDecodeError, Exception) as exc:
log.error("extract step failed: %s", exc)
# Graceful degradation: pass minimal context to next step
extracted = {"sentiment": "unknown", "issues": [], "positives": []}
# Step 2: Draft customer service reply (better model, prose output)
reply = await llm_call(
step_name="draft_reply",
messages=[
{"role": "system", "content": "You are a professional customer service agent. Be concise and empathetic."},
{"role": "user", "content": (
f"Review analysis:\n{json.dumps(extracted, indent=2)}\n\n"
"Write a 2-sentence reply addressing the customer's main concern."
)},
],
model="gpt-4o",
)
return {"analysis": extracted, "reply": reply}
if __name__ == "__main__":
review = "Battery dies after 2 hours. Screen is gorgeous though. Returned it."
result = asyncio.run(review_analysis_chain(review))
print(json.dumps(result, indent=2))How this code works
This code demonstrates prompt chaining by processing a raw customer review through sequential AI model calls. Its job is to first analyze the review to extract structured facts and then use those facts to generate a personalized customer service reply. The central llm_call function acts as a robust wrapper for interacting with the OpenAI API, featuring a built-in retry mechanism for common issues like RateLimitError or APITimeoutError. This ensures the individual steps are resilient to temporary network or API problems, and it logs useful metrics like prompt_tokens and latency_ms for each call.
The review_analysis_chain function orchestrates this two-step process. In the first step, it uses llm_call with response_format={"type": "json_object"} to instruct the AI to "extract" sentiment and other details from the raw_review as structured JSON, which is then parsed by json.loads. A subtle but critical detail is the try...except block around this extraction; if the JSON parsing fails (perhaps the model didn't return perfect JSON), the code gracefully handles the error by providing default empty values, ensuring the entire chain doesn't break. The second step then "pipes" this extracted data as context into another llm_call, using a more capable model="gpt-4o" to "draft_reply", producing a concise customer service message. Finally, the function returns a dictionary containing both the analysis and the reply.
Practice & master
Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.
Exercise
Build a two-step prompt chain that takes a raw GitHub issue title and body, extracts a structured summary (issue type, severity, affected component) in step 1, then generates a triage comment for the engineering team in step 2. Use JSON output for step 1. Log token usage for each step.
# openai>=1.0.0
import json
import os
from openai import OpenAI
client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
RAW_ISSUE = """
Title: Login button unresponsive after OAuth redirect on mobile Safari
Body: Users on iOS 17 / Safari report that after returning from the OAuth
provider, the login button does nothing. Hard refresh fixes it. Desktop
browsers are unaffected. Started after the v2.4.1 deploy on Friday.
"""
def llm_call(messages, model="gpt-4o-mini", response_format=None):
# TODO: call client.chat.completions.create with the right params
# TODO: log model, prompt_tokens, completion_tokens
pass
def run_chain(issue_text: str) -> dict:
# TODO: Step 1 - extract issue_type, severity, affected_component as JSON
extracted_raw = None # replace with llm_call
extracted = {} # TODO: parse JSON, handle parse errors gracefully
# TODO: Step 2 - generate a 3-sentence triage comment for the eng team
triage_comment = None # replace with llm_call
return {"extracted": extracted, "triage_comment": triage_comment}
if __name__ == "__main__":
result = run_chain(RAW_ISSUE)
print(json.dumps(result, indent=2))
Quick check
Why should you validate and parse a chain step's output before passing it to the next step?
You have a chain where step 2 and step 3 both depend only on step 1's output, not on each other. What is the best approach?
A 5-step chain fails at step 4. What design decision lets you retry only step 4 without re-running steps 1-3?