Claude API 多 Agent 流水线在生产环境的可靠实践

大多数"多 Agent"演示在一接触真实数据时就崩溃了。一个 Agent 的幻觉会成为下一个 Agent 的输入，成本飙升因为每个步骤都运行最贵的模型，整套系统变成了凌晨两点无法调试的黑箱。我运行的生产系统在自己的公司上使用分层 Agent 管道，而保持可靠性的模式是刻意做得枯燥的。以下是那些真正重要的模式的完整解析，附有针对 Claude API 的可运行 Python 代码。

我将在全文中使用一个通用示例：一个运营智能管道，从几个业务系统拉取数据，验证发现的内容，并生成每周简报。这个形状可以泛化到几乎任何"读取一堆来源、跨来源推理、生成可信输出"的问题。

按推理深度匹配模型层级

成本和可靠性最大的杠杆不是对所有任务使用一个模型。多 Agent 管道自然分成多层，每一层需要不同程度的算力：

L1 — 提取和路由。 从原始源数据中提取结构化字段，对记录分类，决定数据去向。高容量、浅层推理。这是 Haiku 的工作。

L2 — 处理和验证。 跨来源协调数据，检查一致性， enriched 数据。真正的工作，但不是深层多步推理。用 Sonnet。

L3 — 综合。 跨 L1 和 L2 生成的所有内容进行推理，撰写包含建议的实际简报。这是你为最佳模型付费并开启扩展推理的地方。用 Opus。

把所有层都默认为 Opus 是烧钱最多却没有任何质量提升的方式。把所有层都默认为 Haiku 是得到快速、便宜、却自信地错误的报告的方式。分层设计就是关键。

import anthropic

# SDK 自动重试连接错误、429 和 5xx，使用指数退避。
# 为长时间运行的管道调高这个值，这样短暂的抖动不会中途杀死一次运行。
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：强制结构化，不要解析字符串

让管道变得脆弱最快的方式是在 prompt 中要求模型输出 JSON，然后用 json.loads() 解析返回的任何内容。使用结构化输出，让形状在 API 层就有保证。Python SDK 会为你验证响应是否符合 Pydantic 模型。

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 是一个经过验证的 ExtractedRecord，不是你必须信任的字典
    return response.parsed_output

messages.parse() 返回一个类型化对象。没有脆弱的正则表达式，没有"模型又把 JSON 包在代码围栏里了"的午夜 bug。如果你用的是原始 HTTP 或者想要内联 schema，等效的方式是在 messages.create() 上用 output_config={"format": {"type": "json_schema", "schema": {...}}}——但 Pydantic 路径仅凭验证这一点就值得用。

缓存每个 Agent 共享的上下文

在真实的管道中，每个 Agent 都需要相同的背景：业务结构、术语定义、格式规则、上周期的数据。这个上下文通常很大，而且在一 single run 的数百次 L1/L2 调用中是完全相同的。每次都非缓存地发送是纯粹的浪费。

Prompt 缓存是前缀匹配：把稳定的内容放在前面，用 cache_control 标记，随后的每次调用以大约十分之一的输入价格读取它，而不是支付全额。

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()}",
        }],
    )
    # 确认你真的得到了缓存命中——如果一次运行中这个值是 0，
    # 说明前缀中混入了某种不稳定的东西（时间戳、UUID、未排序的 JSON）。
    assert response.usage.cache_read_input_tokens >= 0
    return {"record": record, "text": _first_text(response)}

需要注意的失败模式：cache_read_input_tokens 在整个运行过程中保持在零。这意味着前缀中有一个静默的失效因子——系统 prompt 中的 datetime.now()、未排序的 json.dumps()、per-request 的 ID。前缀中任何字节的改变都会使之后的所有内容失效。把易变的内容放在最后，放在最后一个缓存断点之后。

L3：开启真正的推理，并流式传输

综合层是你希望模型真正思考的地方——跨每个已协调的记录进行推理，权衡它们，生成站得住脚的建议。在当前的 Opus 上，这意味着自适应思考加上 effort 设置，而不是固定的 token 预算。

这里有两个让人翻车的坑。首先，在当前的 Opus 和 Sonnet 模型上，旧的 thinking={"type": "enabled", "budget_tokens": N} 形式已经废弃——它会返回 400。自适应思考取代了它：模型决定思考多少，你用 effort 来 steering 深度对成本的权衡。其次，深度综合调用可能运行几分钟并产生长输出，所以你要流式传输它——否则你可能面临请求的 HTTP 超时。

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 是真正重要的旋钮。对于大多数综合工作，high 是最佳选择；当正确性比延迟和成本更重要时，用 xhigh 或 max；当你在做例行公事的事情想要速度时，用 medium 或 low。把昂贵的设置留给值得它们的层——L3——并保持 L1/L2 精简。

让它可调试的模式：在 L2 和 L3 之间验证

这是把演示与你可以信任的业务系统区分开来的部分。Agent 不应该互相传递自由文本。每一层把结构化的、清理过的观察写入共享存储，下一层从那个存储中读取。在生产环境中这是数据库（我使用带 pgvector 列的 Postgres 来处理需要跨引用的情况）；对于演示，一个 dict 就足够展示形状。

这很重要的原因：它给你一个来源验证检查点。在昂贵的 L3 综合运行之前，一个便宜的 L2 pass 验证每个观察是否与其声称的来源相符。未验证的观察被丢弃，不进入综合。这是整个管道中最高的杠杆可靠性动作，因为它阻止了一个糟糕的提取变成最终简报中自信的一行。

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)

每一步都写入 context_store，每个声明在被送到综合之前都被检查，每一层的模型都按其工作大小调整。当简报说了错误的内容时，你可以向后走查存储，精确找到是哪一层引入的问题。

关于密钥的说明，因为它会立刻出现

一旦这样的管道接触真实的业务系统，立刻就有人问你怎么处理凭证。让你远离麻烦的答案是：没有 Agent 持有另一个 Agent 的密钥，原始凭证从不与数据一起传递。每个集成都坐落在自己的凭证存储后面，ingestion 在敏感系统所在的网络边界内运行，只有结构化的、清理过的观察——而不是原始财务数据或 secrets——被写入 Agent 读取的共享上下文层。把它设计成即使综合 prompt 完全被攻陷也无法触及凭证，因为凭证从来就不在它的触及范围内。

什么真正保持了可靠性

剥离具体细节，可靠性来自四个习惯，没有一个是聪明的：

分层模型。 Haiku 提取，Sonnet 处理和验证，Opus 综合。不要为字符串提取支付 Opus 价格；不要把最终判断托付给 Haiku。

在 API 层强制结构化。 结构化输出意味着你永远不需要解析一个充满期望的字符串。形状是有保证的，否则调用会大声失败。

在综合之前验证。 对来源进行便宜的 L2 检查，丢弃任何不支持的内容，阻止一个糟糕的提取变成自信的错误结论。

缓存共享前缀并流式传输深度调用。 缓存削减每个 Agent 需要的上下文成本；流式传输防止长时间的综合调用超时。

多 Agent 系统听起来令人兴奋的部分很少是让它们在生产环境中工作的原因。枯燥的部分——分层、结构化契约、验证检查点、你可以审计的共享存储——才是。优先构建这些， impressive 的行为自然会从中产生。

我在 TheAIShop 构建生产 AI 系统，并用自己的公司在一个形状类似的管道上运行。如果你正在处理一个多 Agent 构建，它快速、便宜，但偶尔、自信地错误，上面的验证检查点是我会开始的地方。