Chain Prompts Together in Python

By the end of this guide you will have a Python file that turns messy meeting notes into a checked follow-up email through four short model calls instead of one giant prompt, with a validation gate between each stage so a bad step stops the run rather than poisoning it. Allow about forty minutes, including the time to run it on your own text. Everything here is plain Python, the openai SDK, and the built-in json module — no frameworks, so you can see exactly where each piece of data comes from.

The reason to bother is simple. A prompt that says "read these notes, pull out the decisions, drop the ones nobody owns, write a follow-up email, then check it for accuracy" asks a model to hold five goals in mind at once. Models are good at one clear job and mediocre at five overlapping ones, so what you get back is usually plausible mush: decisions half-invented, owners dropped, the tone right and the facts wrong. Splitting the work gives each call a small, checkable job — and gives you a place to look when the result is off.

This guide sits inside Prompt Engineering Basics with Python, and it assumes you can already write a single prompt that behaves. If yours still drifts in format, fix that first with Write System Prompts that Control Output Format and Few-Shot Prompting in Python with Examples, because chaining unreliable stages only multiplies the unreliability.

Prerequisites

You need Python 3.10 or newer, an OpenAI API key, and a virtual environment — an isolated folder of packages that keeps this project's libraries separate from the rest of your machine. If that is new, follow Create a Python Virtual Environment for AI first.

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install "openai>=1.40" "python-dotenv>=1.0"

Put your key in a .env file beside your script:

OPENAI_API_KEY=sk-your-key-here

Then keep that file out of version control before you write another line:

echo ".env" >> .gitignore

Skipping the .gitignore line is how keys end up in public repositories and get spent by strangers. One command, one problem avoided.

Step 1 — Name the stages before you write any code

Write the stages down on paper first. A good stage is one you could describe in a single verb and check on its own: extract, filter, draft, critique. A bad stage is one whose description needs the word "and". The example running through this guide has four:

1. Extract — read the raw notes and return the decisions, each with what was decided, who owns it, and when it is due, plus any open questions. No writing, no judgement.
2. Transform — drop decisions with no owner, sort by due date, and keep the three most important open questions. This one is pure Python; no model needed.
3. Draft — turn the cleaned structure into a short internal email.
4. Critique and revise — compare the draft against the extracted facts and rewrite anything the facts do not support.

Notice that stage two runs locally. Once data is structured, ordinary Python does filtering and sorting perfectly, instantly, and for free. Every stage you can move out of the model is a stage that cannot hallucinate and cannot be billed.

Everything else in this guide is the code for those four boxes and the checks between them.

Step 2 — Make each stage hand over JSON, not prose

The single biggest difference between a chain that works and one that wobbles is the shape of what travels between stages. If stage one returns a paragraph, stage two has to re-read English and guess. If stage one returns JSON — JavaScript Object Notation, a text format of keys and values that Python parses into a dictionary — stage two reads named fields and nothing is left to interpretation.

Start with one shared helper so every stage uses the same client and settings. temperature controls randomness; keep it low for extraction and editing work, as explained in What Is Temperature in an LLM API?.

"""chain.py — one client, one helper, four stages."""
import json
import os

from dotenv import load_dotenv
from openai import OpenAI

load_dotenv()  # reads .env, which is listed in .gitignore
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))


def call(system: str, user: str, want_json: bool = False) -> str:
    """Send one prompt and return the model's reply as a string."""
    extra = {"response_format": {"type": "json_object"}} if want_json else {}
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        temperature=0.2,
        messages=[
            {"role": "system", "content": system},
            {"role": "user", "content": user},
        ],
        **extra,
    )
    return response.choices[0].message.content

response_format={"type": "json_object"} tells the API that the reply must be valid JSON, which removes the classic failure where a model wraps its answer in a code fence and your parser chokes. The prompt still has to name the keys you want — the parameter guarantees the syntax, not the content.

Now the extraction stage. Note how the system prompt shows the exact shape rather than describing it, and forbids invention explicitly:

EXTRACT_SYSTEM = """You extract facts from meeting notes.
Return JSON only, shaped exactly like this:
{"decisions": [{"what": "", "owner": "", "due": ""}],
 "open_questions": [""]}
Use an empty string when the notes do not say. Invent nothing."""


def extract(notes: str) -> dict:
    raw = call(EXTRACT_SYSTEM, notes, want_json=True)
    return json.loads(raw)

The transform stage never touches the API. It is the cheapest, fastest, most reliable stage in the chain precisely because it is ordinary code:

def transform(facts: dict) -> dict:
    """Keep owned decisions, earliest first, and at most three questions."""
    owned = [d for d in facts["decisions"] if d.get("owner")]
    owned.sort(key=lambda d: d.get("due") or "9999-12-31")
    return {
        "decisions": owned,
        "open_questions": facts.get("open_questions", [])[:3],
    }

Then the two writing stages. The drafting prompt receives the filtered structure as JSON text, and the revision prompt receives both the facts and the draft so it can compare them:

DRAFT_SYSTEM = (
    "You write short internal follow-up emails. Plain text, under 200 words, "
    "one bullet per decision, each bullet naming the owner and the due date."
)

REVISE_SYSTEM = """You are a careful editor. You receive FACTS as JSON and a DRAFT.
Rewrite the draft so every decision in FACTS appears with its owner and due date,
and delete any sentence FACTS does not support. Return the corrected email only."""


def draft(state: dict) -> str:
    return call(DRAFT_SYSTEM, json.dumps(state, indent=2))


def revise(state: dict, draft_text: str) -> str:
    user = f"FACTS:\n{json.dumps(state, indent=2)}\n\nDRAFT:\n{draft_text}"
    return call(REVISE_SYSTEM, user)

The critique stage is the one people skip and then miss. A model reviewing a short draft against a short fact list is doing a genuinely easy job, and it catches the invented deadline that the drafting model added to make a sentence flow.

Step 3 — Thread the state through one run_chain() function

Four functions are not yet a chain. What makes it a chain is a single place that calls them in order, carries the intermediate results, and reports progress. Keep that state in one dictionary: when something goes wrong, you can print the whole dictionary and see exactly which stage produced the nonsense.

def run_chain(notes: str, verbose: bool = True) -> dict:
    """Run all four stages and return every intermediate result."""
    state: dict = {}

    state["facts"] = extract(notes)
    if verbose:
        print(f"[1/4] extracted {len(state['facts']['decisions'])} decisions")

    state["shortlist"] = transform(state["facts"])
    if verbose:
        print(f"[2/4] kept {len(state['shortlist']['decisions'])} with an owner")

    state["draft"] = draft(state["shortlist"])
    if verbose:
        print(f"[3/4] drafted {len(state['draft'].split())} words")

    state["final"] = revise(state["shortlist"], state["draft"])
    if verbose:
        print("[4/4] revised against the extracted facts")

    return state


if __name__ == "__main__":
    NOTES = """Tue standup. Priya will ship the pricing page copy by 2026-08-04.
    We argued about the free tier; nobody owns that yet. Sam takes the migration
    doc, due end of next week. Open question: do we need legal sign-off?"""
    result = run_chain(NOTES)
    print("\n--- final email ---\n")
    print(result["final"])

Run it with python chain.py. You will see four progress lines and then the email. Returning the whole state rather than just the final string is deliberate: during development you will want result["facts"] far more often than you expect, and adding it back later means changing every caller.

Two habits pay off immediately here. First, print a measurement at each stage — a count, a word length — rather than the full payload, so the log stays readable and still tells you when a stage produced nothing. Second, resist adding a fifth stage until the four you have are solid; chains get harder to reason about faster than they get better.

Step 4 — Gate every handoff so a bad stage fails loudly

An unchecked chain has a nasty failure mode. Stage one misreads the notes and returns one decision instead of four. Nothing errors. Stage three writes a confident, well-formatted email about that single decision, stage four politely confirms it matches the facts, and you send a wrong email that looks right. The chain did not break; it quietly agreed with itself.

The fix is a small check after each stage that raises an exception naming the stage that failed. Define one error type so you can catch chain problems separately from network problems:

class ChainError(Exception):
    """A stage produced something the next stage cannot use."""


def check_facts(facts: dict) -> dict:
    if not isinstance(facts, dict):
        raise ChainError("extract: expected a JSON object")
    decisions = facts.get("decisions")
    if not isinstance(decisions, list) or not decisions:
        raise ChainError("extract: no decisions found — check the notes or the prompt")
    for i, decision in enumerate(decisions):
        missing = {"what", "owner", "due"} - set(decision)
        if missing:
            raise ChainError(f"extract: decision {i} is missing {sorted(missing)}")
    return facts


def check_draft(text: str) -> str:
    if not text or len(text.split()) < 20:
        raise ChainError("draft: reply was empty or suspiciously short")
    return text

Wrap the parsing too, so a non-JSON reply becomes a chain error rather than a raw traceback from deep inside the json module:

def extract(notes: str) -> dict:
    raw = call(EXTRACT_SYSTEM, notes, want_json=True)
    try:
        return json.loads(raw)
    except json.JSONDecodeError as err:
        raise ChainError(f"extract: model did not return JSON ({err})") from err

Then slot the checks into the chain and catch the error at the top level:

state["facts"] = check_facts(extract(notes))
state["shortlist"] = transform(state["facts"])
state["draft"] = check_draft(draft(state["shortlist"]))
state["final"] = check_draft(revise(state["shortlist"], state["draft"]))

try:
    result = run_chain(NOTES)
except ChainError as err:
    print("Chain stopped:", err)

Decide deliberately what each gate does on failure: stop, retry the same stage once, or fall back to a simpler path. Retrying is reasonable when the fault is formatting — models often get JSON right on a second attempt with a blunter instruction. Stopping is right when the input itself is the problem, because retrying a call that cannot succeed just spends money twice.

If you are new to reading the output when an exception does fire, Read a Python Traceback in Five Minutes will save you a lot of squinting.

Stage design quick reference

Use this when you sketch a new chain. The pattern repeats: structured stages first, writing stages later, checks everywhere.

Troubleshooting

• json.JSONDecodeError: Expecting value: line 1 column 1 (char 0) — the reply started with a code fence or an apology, not {. Cause: response_format was not set, or the prompt never used the word JSON. Fix: pass response_format={"type": "json_object"} and name the keys in the system prompt; Fix JSONDecodeError with AI API Responses in Python covers the stubborn cases.
• KeyError: 'decisions' — the model returned valid JSON with different key names, such as items. Cause: the prompt described the shape in words instead of showing it. Fix: paste the literal skeleton into the system prompt and let check_facts catch the rest.
• openai.RateLimitError: Error code: 429 — four calls fired back to back tripped your per-minute limit. Cause: chains multiply request volume, especially in a loop over many inputs. Fix: add a short sleep between runs and retry with backoff, as in Fix the 429 Rate-Limit Error in Python.
• openai.BadRequestError: ... context_length_exceeded — a later stage received the entire accumulated state instead of the slice it needed. Cause: passing state wholesale into every prompt. Fix: hand each stage only its own inputs, and split oversized source text first — see Fix the Context-Length-Exceeded Error in Python.

What a chain costs you

Be honest with yourself about the bill. Four stages means four requests, and every request re-sends its own system prompt, so token spend grows roughly in proportion to the number of stages — sometimes a little faster, because the drafting and revision stages carry the extracted data as input. Latency behaves the same way: the stages run one after another by necessity, since each needs the previous result, so a user waits for the sum of four round trips rather than one.

Those costs are worth paying when accuracy matters more than speed, and wasteful when it does not. Measure before you guess: count tokens with Count Tokens in Python Before You Send and put a number on a full run with Estimate OpenAI API Costs with Python. If you re-run the same chain over unchanged inputs — which happens constantly while you tune a later stage — Cache AI Responses in Python to Cut Costs stops you paying twice for the identical extraction.

When to use this vs. alternatives

• A single well-written prompt is enough when the task has one output and one quality bar — classify this ticket, rewrite this paragraph, summarise this page. Adding stages there just adds cost and places to break. Try tightening the instruction and adding two worked examples before you split anything; Prompt Engineering Templates for Marketers has ready-made single-prompt patterns for common jobs.
• Chain the prompts when the task genuinely has stages that need different skills or different temperatures, when you need an intermediate result for something else — the extracted JSON often feeds a spreadsheet as well as the email — or when a checking step measurably improves accuracy. Long documents are a classic case, since chunk-then-summarise is a chain; see Summarize Long PDFs with Python and Chunking.
• Reach for a framework only after your plain-Python chain outgrows itself — when you want tracing dashboards, tool calls, and shared memory rather than four functions and a dictionary. Build a Customer Support Chatbot with LangChain shows that world. Learning the mechanics by hand first is what lets you debug the framework later instead of guessing at it.

You now have the whole pattern: name the stages, hand JSON between them, thread the state through one function, and gate every handoff. Run it against your own messy notes today, then take the next step by putting a real number on what a run costs and logging each stage in production with Log and Monitor AI API Calls in Production. Back to Prompt Engineering Basics with Python.

Related guides

• Few-Shot Prompting in Python with Examples — make each individual stage in the chain more consistent before you connect them.
• Write System Prompts that Control Output Format — the format discipline every JSON handoff depends on.
• Fix JSONDecodeError with AI API Responses in Python — the error you will meet first when a stage returns something unparseable.
• Count Tokens in Python Before You Send — measure what each extra stage actually adds to the bill.
• Turn a Python AI Script into an API with FastAPI — put your finished chain behind a URL other people can call.