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

A developer detailed the orchestration patterns behind a production multi-agent pipeline built on the Claude API, emphasizing tiered model selection, structured outputs, and prompt caching to ensure reliability and cost efficiency. The system, which powers the developer's own company, uses separate models for extraction, validation, and synthesis, with Pydantic-validated responses and cached shared context to avoid common failure modes.

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. python 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 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. python 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. 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. 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. php 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 = 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. 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.