Prompt chaining

Prompt chaining splits one task into a fixed sequence of steps, and hands each step’s output to the next. Anthropic’s own description is direct: it “decomposes a task into a sequence of steps, where each LLM call processes the output of the previous one”[1]. A model call can sit inside any step, but the sequence itself, and what happens between steps, is fixed by your code before the chain ever runs.

The step between two model calls is usually a gate: ordinary code that checks the output so far before letting the chain continue. Anthropic gives two examples: generate marketing copy, then translate it, or write a document outline, check it against a rule, then write the document from that outline[1].

Prompt chaining sits at level 3, workflows. The model fills in the content of each step; your code decides how many steps there are, what order they run in, and what gate sits between them. The line to level 4 falls where the model’s output starts selecting what runs next: where it is offered a tool and can choose to call it.

This page provides primary references and illustrative examples. The examples are scripted, not measured model runs; source references do not establish the correctness of every implementation or outcome.

"Is the DR-520 vent length still 35 feet?"

LEVEL = 3
MAX_QUERIES = 3
PER_QUERY_K = 2
REWRITE_SYSTEM = (
    "Break the user's question into 1 to 3 short search queries over Halvorsen appliance "
    "documents, one per line, plain text, no numbering."
)
DRAFT_SYSTEM = (
    "You answer questions about Halvorsen appliances using only the numbered sources below. "
    "If the sources do not contain the answer, say so instead of guessing. End your answer with "
    "a line starting 'Sources:' listing the citations, like 'dw300-manual#3', that you used."
)


def _rewrite_queries(question: str, model: Model, tracer: Tracer) -> list[str]:
    completion = model.complete([Message(role="system", content=REWRITE_SYSTEM), Message(role="user", content=question)], max_tokens=150)
    queries = [line.strip() for line in completion.text.splitlines() if line.strip()][:MAX_QUERIES] or [question]
    tracer.record(
        kind="model",
        decided_by="code",
        title="Rewrite into search queries",
        detail="; ".join(queries),
        tokens_in=completion.tokens_in,
        tokens_out=completion.tokens_out,
        ms=completion.ms,
    )
    return queries


def _retrieve(queries: list[str], sections: dict[str, Section], tracer: Tracer) -> list[Section]:
    seen: dict[str, Section] = {}
    for query in queries:
        for section, score in bm25_search(sections, query, k=PER_QUERY_K):
            if score > 0:
                seen[section.cite] = section
    sources = list(seen.values())
    tracer.record(kind="code", decided_by="code", title="Retrieve for each query", detail=", ".join(seen.keys()) or "none")
    return sources


def _draft(question: str, sources: list[Section], model: Model, tracer: Tracer):
    blocks = "\n\n".join(f"[{s.cite}] {s.title}\n{s.text}" for s in sources)
    prompt = f"Sources:\n\n{blocks}\n\nQuestion: {question}"
    completion = model.complete([Message(role="system", content=DRAFT_SYSTEM), Message(role="user", content=prompt)], max_tokens=500)
    tracer.record(
        kind="model",
        decided_by="code",
        title="Draft answer from sources",
        detail=completion.text[:200],
        tokens_in=completion.tokens_in,
        tokens_out=completion.tokens_out,
        ms=completion.ms,
    )
    return completion


def _check_citations(draft_text: str, sources: list[Section], tracer: Tracer) -> list[str]:
    retrieved = {s.cite for s in sources}
    claimed = set(cited_sources(draft_text))
    grounded = sorted(claimed & retrieved)
    dropped = sorted(claimed - retrieved)
    tracer.record(
        kind="code",
        decided_by="code",
        title="Check citations against retrieval",
        detail=f"kept {grounded}" + (f", dropped ungrounded {dropped}" if dropped else ""),
    )
    return grounded


def run(question: str, model: Model, embedder: Embedder | None, tracer: Tracer, *, corpus_dir: Path = DEFAULT_CORPUS_DIR) -> Answer:
    del embedder  # level 3 retrieves by keyword, not by vector
    sections = load_sections(corpus_dir)
    queries = _rewrite_queries(question, model, tracer)
    sources = _retrieve(queries, sections, tracer)
    completion = _draft(question, sources, model, tracer)
    citations = _check_citations(completion.text, sources, tracer)
    return Answer(text=completion.text, citations=citations, retrieved_sources=[s.cite for s in sources])

python -m examples.prompt_chaining --model stub:scripted

Failure modes

Cost and latency

Illustrative, not measured: no result file exists for this technique yet, so every number below is a worked example. See docs/EVALS.md for how a real run is recorded and scored.

How to Evaluate It

Scored on the same 60-question set as every other technique, over the appliance documents in evals/corpus/. The citation check gives prompt chaining an extra number RAG does not have on its own: how often a citation the draft claims was actually something retrieval found, tracked separately from whether the final answer was correct.

Conflicting-source and multi-hop questions are where the extra query rewrite step is expected to earn its cost: a single retrieval pass over “is the vent length still 35 feet” can miss the correcting service bulletin entirely, where a second, differently worded query aimed at it has a chance to find it. No result file exists yet (see docs/EVALS.md), so this page cannot say whether that expectation holds. Run python scripts/eval_run.py --example prompt_chaining --model <spec> --dry to project the cost of a real run before spending anything on one.

Run it

Try it

1. Use it

   Find a multi-step automation you use: an email rule, a form that files a ticket, a scheduled report. Write its steps down in order. Which one, if it silently got something wrong, would nobody downstream catch?

2. Build it

   Run python -m examples.prompt_chaining --model stub:scripted from the repo root. The rewrite turns one question into three queries, retrieval brings back five sections, and the citation check keeps service-bulletin#2, the bulletin correcting the manual. Now set MAX_QUERIES in examples/prompt_chaining/run.py from 3 to 1: one query returns two sections, neither the bulletin, and the check drops the citation the draft still claims.

3. Either lane

   Take the gate that never fails, above, and cause it on purpose: find a check in a process you run that has never once rejected anything, and work out whether that is because nothing bad has arrived or because the check cannot see the thing that would be bad.

4. Either lane

   Take a short requirements list you actually have and chain it by hand: propose a test for each requirement, build the traceability table, then check that every requirement has a test and every test names a requirement. Where does the chain, not the requirements, turn out to be the hard part?