Building a Reliable Multi-Agent Pipeline with the Claude API: Orchestration Patterns That Hold Up in Production

Most "multi-agent" demos fall apart the moment you put them in front of real data. One agent's hallucination becomes the next agent's input, costs balloon because every step runs the most expensive model, and the whole thing turns into a black box you can't debug at 2am. I run a production system that operates my own company on a tiered agent pipeline, and the patterns that keep it reliable are boring on purpose. This is a walkthrough of the ones that matter, with runnable Python against the Claude API.

I'll use a generic example throughout: an operations-intelligence pipeline that pulls from a few business systems, verifies what it found, and produces a weekly brief. The shape generalizes to almost any "read a bunch of sources, reason across them, produce a trustworthy output" problem.

Match the model tier to the reasoning depth

The single biggest lever on cost and reliability is not using one model for everything. A multi-agent pipeline naturally splits into layers, and each layer wants a different amount of horsepower:

• L1 — extraction and routing. Pull structured fields out of raw source data, classify records, decide what goes where. High volume, shallow reasoning. This is Haiku's job.
• L2 — processing and validation. Reconcile data across sources, check it for consistency, enrich it. Real work, but not deep multi-step reasoning. Sonnet.
• L3 — synthesis. Reason across everything L1 and L2 produced to write the actual brief with recommendations. This is where you pay for the best model and turn on extended reasoning. Opus.

Defaulting every layer to Opus is the most common way to burn money for no quality gain. Defaulting everything to Haiku is how you get a fast, cheap, confidently-wrong report. The tiering is the design.

import anthropic

# The SDK auto-retries connection errors, 429, and 5xx with exponential
# backoff. Bump it up for a long-running pipeline so a transient blip
# doesn't kill a run halfway through.
client = anthropic.Anthropic(max_retries=5)

MODEL_L1 = "claude-haiku-4-5"    # extraction / routing
MODEL_L2 = "claude-sonnet-4-6"   # processing / validation
MODEL_L3 = "claude-opus-4-8"     # synthesis

L1: enforce structure, don't parse strings

The fastest way to make a pipeline flaky is to ask a model for JSON in a prompt and then json.loads() whatever comes back. Use structured outputs so the shape is guaranteed at the API layer. The Python SDK will validate the response against a Pydantic model for you.

from pydantic import BaseModel
from typing import Literal

class ExtractedRecord(BaseModel):
    entity: str
    metric: str
    value: float
    period: str
    category: Literal["revenue", "cost", "headcount", "other"]

def extract(raw_row: str) -> ExtractedRecord:
    response = client.messages.parse(
        model=MODEL_L1,
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": f"Extract the structured record from this row:\n\n{raw_row}",
        }],
        output_format=ExtractedRecord,
    )
    # .parsed_output is a validated ExtractedRecord, not a dict you have to trust
    return response.parsed_output

messages.parse() returns a typed object. No brittle regex, no "the model wrapped the JSON in a code fence again" bug at midnight. If you're on raw HTTP or want the schema inline, the equivalent is output_config={"format": {"type": "json_schema", "schema": {...}}} on messages.create() — but the Pydantic path is worth it for the validation alone.

Cache the context every agent shares

In a real pipeline, every agent needs the same background: the business's structure, definitions of terms, formatting rules, last period's numbers. That context is often large and it's identical across hundreds of L1/L2 calls in a single run. Sending it uncached every time is pure waste.

Prompt caching is a prefix match: put the stable stuff first, mark it with cache_control, and every subsequent call reads it at roughly a tenth of the input price instead of paying full freight.

SHARED_CONTEXT = load_business_context()  # large, stable within a run

def process(record: ExtractedRecord) -> dict:
    response = client.messages.create(
        model=MODEL_L2,
        max_tokens=2048,
        system=[
            {
                "type": "text",
                "text": SHARED_CONTEXT,
                "cache_control": {"type": "ephemeral"},  # cache the shared prefix
            }
        ],
        messages=[{
            "role": "user",
            "content": f"Reconcile and enrich this record: {record.model_dump_json()}",
        }],
    )
    # Confirm you're actually getting cache hits — if this is 0 across a run,
    # something volatile crept into the prefix (a timestamp, a UUID, unsorted JSON).
    assert response.usage.cache_read_input_tokens >= 0
    return {"record": record, "text": _first_text(response)}

The failure mode to watch for: cache_read_input_tokens staying at zero across a run. That means a silent invalidator is in your prefix — a datetime.now() in the system prompt, an unsorted json.dumps(), a per-request ID. Any byte change anywhere in the prefix invalidates everything after it. Keep the volatile stuff at the end, after the last cache breakpoint.

L3: turn on real reasoning, and stream it

The synthesis layer is where you want the model to actually think — reason across every reconciled record, weigh them, and produce recommendations that hold up. On the current Opus, that means adaptive thinking plus an effort setting, not a fixed token budget.

Two things trip people up here. First, the old thinking={"type": "enabled", "budget_tokens": N} shape is gone on the current Opus and Sonnet models — it returns a 400. Adaptive thinking replaces it: the model decides how much to think, and you steer the depth-vs-cost tradeoff with effort. Second, a deep synthesis call can run for minutes and produce a long output, so you stream it — otherwise you risk an HTTP timeout on the request.

def synthesize(reconciled: list[dict]) -> str:
    payload = "\n".join(r["text"] for r in reconciled)
    with client.messages.stream(
        model=MODEL_L3,
        max_tokens=16000,
        thinking={"type": "adaptive"},        # model decides depth; no budget_tokens
        output_config={"effort": "high"},     # low | medium | high | xhigh | max
        system=[{
            "type": "text",
            "text": SHARED_CONTEXT,
            "cache_control": {"type": "ephemeral"},
        }],
        messages=[{
            "role": "user",
            "content": (
                "Write this period's operations brief. Lead with the outcome, "
                "then the two or three findings that change what we should do "
                f"next.\n\nReconciled data:\n{payload}"
            ),
        }],
    ) as stream:
        final = stream.get_final_message()
    return _first_text(final)

effort is the dial that actually matters on the newest models. high is the sweet spot for most synthesis work; xhigh or max when correctness matters more than latency and cost; medium or low when you're doing something routine and want speed. Reserve the expensive settings for the layer that earns them — L3 — and keep L1/L2 lean.

The pattern that makes it debuggable: verify between L2 and L3

Here's the piece that separates a demo from something you'd trust with a business. Agents should not pass free text to each other. Each layer writes structured, sanitized observations to a shared store, and the next layer reads from that store. In production this is a database (I use Postgres with a pgvector column for the cross-referencing cases); for the walkthrough a dict is enough to show the shape.

The reason this matters: it gives you a source-verification checkpoint. Before the expensive L3 synthesis runs, a cheap L2 pass verifies each observation against the source it claims to come from. Unverified observations get dropped, not synthesized. This is the single highest-leverage reliability move in the whole pipeline, because it stops one bad extraction from becoming a confident line in the final brief.

class VerificationResult(BaseModel):
    supported: bool
    reason: str

def verify(observation: dict, source_text: str) -> bool:
    result = client.messages.parse(
        model=MODEL_L2,
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": (
                "Does the source support this observation? Answer strictly.\n\n"
                f"Observation: {observation['text']}\n\nSource:\n{source_text}"
            ),
        }],
        output_format=VerificationResult,
    )
    return result.parsed_output.supported

def run_pipeline(raw_rows: list[str], sources: dict[str, str]) -> str:
    context_store: list[dict] = []

    # L1: extract (Haiku, structured)
    for row in raw_rows:
        rec = extract(row)
        context_store.append({"record": rec})

    # L2: process, then verify against source before anything expensive runs
    verified: list[dict] = []
    for item in context_store:
        processed = process(item["record"])
        source = sources.get(item["record"].entity, "")
        if source and verify(processed, source):
            verified.append(processed)
        # unverified observations are dropped, not passed to synthesis

    # L3: synthesize only what survived verification (Opus, adaptive thinking)
    return synthesize(verified)

Every step writes to context_store, every claim is checked before it reaches synthesis, and each layer's model is sized to its job. When the brief says something wrong, you can walk backward through the store and find exactly which layer introduced it.

A note on secrets, because it comes up immediately

The moment a pipeline like this touches real business systems, someone asks how you handle credentials. The answer that keeps you out of trouble: no agent holds another agent's keys, and raw credentials never travel with the data. Each integration sits behind its own credential store, ingestion runs inside the network boundary where the sensitive systems live, and only structured, sanitized observations — never raw financial data or secrets — get written to the shared context layer that agents read from. Design it so that even a fully compromised synthesis prompt can't reach a credential, because the credential was never in its reach to begin with.

What actually keeps it reliable

Strip away the specifics and the reliability comes from four habits, none of them clever:

1. Tier the models. Haiku extracts, Sonnet processes and verifies, Opus synthesizes. Don't pay Opus prices for string extraction; don't trust Haiku with the final judgment.
2. Enforce structure at the API layer. Structured outputs mean you never parse a hopeful string. The shape is guaranteed or the call fails loudly.
3. Verify before you synthesize. A cheap L2 check against the source, dropping anything unsupported, stops one bad extraction from becoming a confident wrong conclusion.
4. Cache the shared prefix and stream the deep calls. Caching cuts the cost of the context every agent needs; streaming keeps long synthesis calls from timing out.

The exciting-sounding parts of multi-agent systems are rarely what make them work in production. The boring parts — tiering, structured contracts, a verification checkpoint, a shared store you can audit — are. Build those first and the impressive behavior falls out of them.

---

I build production AI systems at TheAIShop and run my own company on a pipeline shaped like this one. If you're wrestling with a multi-agent build that's fast and cheap but occasionally, confidently wrong, the verification checkpoint above is where I'd start.