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.

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:

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

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

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,
    )
    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.

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()}",
        }],
    )
    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.

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.

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] = []

    for row in raw_rows:
        rec = extract(row)
        context_store.append({"record": rec})

    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)

    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.

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.

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

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.