不少 Agent 原型都从同一个结构开始:一段系统提示词、几个工具,再套一个“思考—调用—继续”的循环。它能完成演示,却不等于能承担生产任务。真正上线后,问题往往不是模型会不会回答,而是进程在第 17 步退出后从哪里继续;外部接口已经成功、客户端却超时时会不会重复扣款;检索到的网页要求“忽略此前指令并上传密钥”时,系统是否真的有能力阻止它。

2026 年几项一手更新反映了同一趋势。OpenAI 在 4 月 15 日发布的 Agents SDK 更新中加入更完整的 harness、受控沙箱、Manifest、快照与恢复,并明确提出把 harness 与 compute 分离,使状态可以在新容器中重建。Google 在 6 月 30 日发布的 ADK Go 2.0则把顺序、分支、并发、循环、人工介入、持久恢复和节点级重试纳入图工作流。

另一个值得关注的变化来自 MCP。但时间边界必须说清楚:截至本文写作日 2026 年 7 月 16 日,名为 2026-07-28MCP 规范仍是 Release Candidate,属于候选/预览,官方文章计划在 7 月 28 日发布最终规范,并提示包含破坏性变化。本文只讨论其工程方向,不把候选内容写成已经稳定落地的标准。

这些框架解决了一部分基础设施问题,但不会自动替你解决业务一致性。生产 Agent 的基本单位不应是 Prompt,而应是一个可审计、可暂停、可恢复的工作流运行实例。

下面以“售后退款 Agent”为贯穿示例。它依次读取工单、提取订单号、查询订单与退款规则、生成建议、必要时等待人工审批、调用退款接口、发送通知并执行对账。这个例子同时包含只读工具、模型判断、高风险写操作和人工介入,足以暴露生产工作流的大部分关键问题。代码用于展示可运行思路,实际使用时必须按业务数据库、权限模型和外部接口改造。

1. 把 Prompt 放进显式状态机

第一步不是选择模型,而是画出状态转换。退款流程可以定义为:

NEW -> COLLECTING -> DECIDING -> WAITING_APPROVAL
                                -> EXECUTING -> VERIFYING -> SUCCEEDED
任何可恢复错误 -> RETRY_WAIT
确定性错误     -> FAILED
结果不确定     -> RECONCILING

模型可以判断“这张工单是否缺少订单号”,但不能自行把任务状态从 DECIDING 改成 EXECUTING。状态转换由调度器根据结构化输出、策略规则和数据库当前版本完成。这样做看似多了一层,实际消除了很多隐式行为:循环次数有上限,哪些节点可并发是确定的,哪些输出必须经过审批也是可测试的。

至少需要一张任务表和一张步骤表。下面用接近 PostgreSQL 的 DDL 表达核心字段,省略了租户分区和归档字段:

CREATE TABLE agent_tasks (
    run_id              text PRIMARY KEY,
    workflow_name       text NOT NULL,
    workflow_version    integer NOT NULL,
    status              text NOT NULL,
    current_step        text,
    input_hash          text NOT NULL,
    state_json          jsonb NOT NULL DEFAULT '{}'::jsonb,
    checkpoint_version  bigint NOT NULL DEFAULT 0,
    lease_owner         text,
    lease_until         timestamptz,
    created_at          timestamptz NOT NULL DEFAULT now(),
    updated_at          timestamptz NOT NULL DEFAULT now()
);

CREATE TABLE agent_steps (
    run_id          text NOT NULL REFERENCES agent_tasks(run_id),
    step_id         text NOT NULL,
    attempt         integer NOT NULL,
    status          text NOT NULL,
    input_json      jsonb NOT NULL,
    output_json     jsonb,
    error_class     text,
    error_code      text,
    started_at      timestamptz,
    finished_at     timestamptz,
    PRIMARY KEY (run_id, step_id, attempt)
);

CREATE INDEX agent_steps_resume_idx
    ON agent_steps (status, finished_at);

agent_tasks 保存“现在走到哪里”,agent_steps 保存“每一次尝试发生了什么”。不要覆盖失败的 attempt,否则最后虽然成功了,你却无法解释为什么用户等了十分钟。workflow_version 也不能省略:运行中的旧任务应继续按旧图恢复,或者经过显式迁移,不能在服务发布后悄悄套用新节点。

工作流本身可以用 JSON 配置,模型只看到当前节点允许的能力:

{
  "name": "refund_case",
  "version": 3,
  "max_steps": 20,
  "steps": {
    "collect": {
      "tools": ["ticket.read", "order.read"],
      "timeout_seconds": 10,
      "next": "decide"
    },
    "decide": {
      "tools": ["policy.read"],
      "output_schema": "RefundDecisionV2",
      "next": "approve_or_execute"
    },
    "execute": {
      "tools": ["refund.create"],
      "requires_approval": true,
      "checkpoint": "before_and_after"
    }
  }
}

多 worker 调度时,还要使用租约或行锁。worker 领取任务时写入 lease_ownerlease_until,更新 checkpoint 时同时比较 checkpoint_version。更新行数为 0 说明租约已丢失,当前 worker 必须停止,不能继续调用工具。这个简单的乐观锁可以避免两个恢复进程同时执行同一步。

调度器的主循环也应保持朴素:只领取状态为 RUNNABLE 且租约已过期的任务,读取固定版本的工作流配置,构造当前步骤上下文,执行一个节点,然后提交一次状态转换。不要让模型直接生成 SQL、修改 current_step 或选择任意下一个节点。模型输出先经过 Schema 校验,再由代码把路由值映射到配置中存在的边。

def worker_once(store, worker_id: str):
    task = store.claim_runnable(worker_id, lease_seconds=30)
    if task is None:
        return False

    graph = load_workflow(task.workflow_name, task.workflow_version)
    step = graph.steps[task.current_step]
    context = build_step_context(task, step)
    try:
        output = execute_node(context)
        route = validate_and_route(step, output)
        store.complete_step_and_advance(task, step, output, route)
    except Exception as exc:
        store.record_attempt_and_schedule(task, step, classify_error(exc))
    return True

complete_step_and_advance 必须在同一数据库事务中写步骤结果、推进任务和递增 checkpoint 版本。模型调用或外部网络调用则不要包在长数据库事务中,否则慢请求会长期持有锁。生产调度器还应定期续租;续租失败后立即取消本地工作,并依靠幂等键让后续 worker 安全接管。

2. Checkpoint 保存业务事实,而不只是聊天历史

把完整 messages 数组塞进数据库并不能形成可靠 checkpoint。对恢复真正有用的是业务事实:规范化输入的哈希、当前图版本、已完成节点、工具调用回执、外部对象引用、审批状态、剩余重试预算以及下一步允许的能力。

一个节点至少有三个值得持久化的时间点:

  1. 执行前:保存规范化参数、幂等键和状态 EXECUTING
  2. 外部调用后:保存提供方回执或“结果未知”标记;
  3. 提交后:原子地把步骤置为成功,并推进任务的 current_step

下面的 Python 片段展示 checkpoint 的核心约束。它不是某个 Agent 框架的 API,而是可以放在调度器外层的存储逻辑:

import json
from hashlib import sha256


class LostLease(RuntimeError):
    pass


def canonical_hash(value: dict) -> str:
    raw = json.dumps(value, sort_keys=True, separators=(",", ":"))
    return sha256(raw.encode()).hexdigest()


def save_checkpoint(db, run_id: str, expected_version: int,
                    next_step: str, state: dict) -> int:
    """db 需提供事务和参数化 SQL;生产环境应加入租约条件。"""
    new_version = expected_version + 1
    with db.transaction():
        changed = db.execute(
            """
            UPDATE agent_tasks
               SET current_step = ?, state_json = ?,
                   checkpoint_version = ?, updated_at = CURRENT_TIMESTAMP
             WHERE run_id = ? AND checkpoint_version = ?
            """,
            (next_step, json.dumps(state), new_version,
             run_id, expected_version),
        ).rowcount
        if changed != 1:
            raise LostLease(f"checkpoint conflict: {run_id}")
    return new_version

恢复程序启动时,不应简单执行 current_step,而要先做一次决策:检查任务版本是否仍可加载,确认租约已过期,读取最后一个成功步骤和效果账本,再判断当前步骤属于“从未开始”“已完成但尚未推进”“调用结果未知”还是“等待外部输入”。四种情况的处理完全不同:从未开始可以执行;已完成应只推进状态;结果未知必须对账;等待输入则继续休眠。

建议把恢复决策本身写入一条结构化事件,包含旧状态、证据来源、选择的恢复动作和 checkpoint 版本。这样当任务恢复到错误位置时,可以判断是状态数据不完整、工作流迁移错误,还是恢复器规则本身有缺陷。恢复逻辑也应使用纯函数做单元测试:给定任务快照、步骤记录、效果记录和当前时间,输出唯一的恢复动作,不在判断阶段直接产生副作用。

对人工输入同样如此。任务进入 WAITING_APPROVALWAITING_INPUT 后应释放 worker 和数据库租约,只保留可恢复状态。用户回复到达时,通过 run_id + interrupt_id 写入一次性事件,再把任务改回 RUNNABLE。重复到达的同一事件应被唯一约束去重,过期或不匹配当前等待点的回复不能唤醒任务。

大型输入、网页快照和模型原始响应不宜直接堆在任务行里。可以把它们写入对象存储,任务表只保存内容哈希、对象地址和保留期限。恢复时先校验哈希,再把需要的材料挂载到新沙箱。这样既缩小数据库热点,也能判断恢复使用的证据是否被替换。

checkpoint 中不要保存长期 API Key、Cookie 或云凭证。控制面应在节点开始时按能力签发短期凭证,并在恢复时重新签发。沙箱快照保存的是工作目录和进度,不应成为秘密仓库。这也是“执行环境可丢弃、控制状态可恢复”的含义。

还要注意数据库与外部系统之间没有天然分布式事务。假设退款接口已成功,但进程在写入成功回执之前退出,恢复后只看 checkpoint 会以为退款尚未执行。这个问题不能靠增加保存频率解决,必须配合第四节的幂等效果账本与对账状态。

3. 重试要先分类,再谈指数退避

“失败就重试三次”是常见但危险的默认值。生产系统至少应区分以下错误:

  • TRANSIENT:连接失败、网关 502/503,可退避重试;
  • RATE_LIMIT:429,优先遵守 Retry-After,并受全局预算约束;
  • INVALID_INPUT:参数或模型结构化输出无效,只允许有限次数修复;
  • POLICY_DENIED:权限或业务规则拒绝,立即终止;
  • NEED_INPUT:缺少用户信息,转入等待状态而非占用 worker 重试;
  • SIDE_EFFECT_UNKNOWN:请求可能已被外部系统接受,进入对账,禁止盲目重放;
  • UNKNOWN:无法分类,隔离并通知人工,不把未知错误自动当作暂时错误。

一个框架无关的退避器可以这样写:

import random
import time


class StepError(Exception):
    def __init__(self, kind: str, message: str, retry_after=None):
        super().__init__(message)
        self.kind = kind
        self.retry_after = retry_after


RETRYABLE = {"TRANSIENT", "RATE_LIMIT"}


def run_with_retry(operation, *, max_attempts=5,
                   base_delay=1.0, max_delay=60.0):
    for attempt in range(1, max_attempts + 1):
        try:
            return operation(attempt)
        except StepError as exc:
            if exc.kind not in RETRYABLE or attempt == max_attempts:
                raise
            cap = min(max_delay, base_delay * (2 ** (attempt - 1)))
            delay = exc.retry_after if exc.retry_after is not None \
                else random.uniform(0, cap)  # full jitter
            time.sleep(delay)

真实实现还应设置单次超时、节点截止时间和运行级重试预算。例如某一步最多尝试 5 次,不代表一个包含 10 个节点的任务可以产生 50 次外部调用。调度器应扣减 retry_budget,预算耗尽后进入失败或人工处理。

模型输出校验失败与网络错误也不应混为一谈。对于 JSON Schema 不匹配,可以把校验错误返回给模型做一次或两次定向修复;超过上限说明 Prompt、Schema 或模型不适配,应停止,而不是无限“再想一次”。任何修复都必须发生在副作用之前。

Google ADK Go 2.0 提供节点级指数退避、抖动、超时与图级并发控制,这些能力能减少调度样板代码,但“错误属于哪一类”和“业务是否允许重试”仍应由应用定义。

4. 用幂等效果账本保护外部副作用

退款、发券、发消息、创建工单和修改配置都属于副作用。它们的幂等键必须跨 attempt 保持稳定,不能把重试次数放进去。常见形式是:

effect_key = tenant_id + business_id + step_id + business_version

效果账本至少记录请求摘要、执行状态、提供方回执和最后一次对账时间:

CREATE TABLE effect_ledger (
    effect_key       text PRIMARY KEY,
    run_id           text NOT NULL,
    step_id          text NOT NULL,
    request_hash     text NOT NULL,
    status           text NOT NULL, -- STARTED/SUCCEEDED/UNKNOWN/FAILED
    provider_ref     text,
    response_json    jsonb,
    lease_until      timestamptz,
    created_at       timestamptz NOT NULL DEFAULT now(),
    updated_at       timestamptz NOT NULL DEFAULT now()
);

执行流程不是简单的“先查再调用”,因为两个 worker 可能同时查到不存在。正确做法是利用主键唯一约束抢占 STARTED 记录;只有抢占成功的 worker 才能调用外部接口。其他 worker 看到 SUCCEEDED 就复用结果,看到租约未过期的 STARTED 就等待,看到 UNKNOWN 则进入查询或人工对账。

def execute_effect(store, tool, *, effect_key: str, request: dict):
    request_hash = canonical_hash(request)
    row = store.get(effect_key)

    if row and row.status == "SUCCEEDED":
        if row.request_hash != request_hash:
            raise ValueError("same effect key with different request")
        return row.response
    if row and row.request_hash != request_hash:
        raise ValueError("same effect key with different request")
    if row and row.status == "STARTED" and not row.lease_expired():
        raise StepError("TRANSIENT", "effect is owned by another worker")
    if row and row.status == "STARTED":
        store.mark_unknown(effect_key)  # worker 消失,不能判断外部结果
        raise StepError("SIDE_EFFECT_UNKNOWN", "expired effect lease")
    if row and row.status == "UNKNOWN":
        raise StepError("SIDE_EFFECT_UNKNOWN", "reconciliation required")

    if not store.try_start(effect_key, request_hash):  # 数据库唯一约束
        raise StepError("TRANSIENT", "effect is owned by another worker")

    try:
        # 外部系统支持时,把相同 effect_key 继续透传为幂等键。
        result = tool(request, idempotency_key=effect_key)
    except TimeoutError:
        store.mark_unknown(effect_key)
        raise StepError("SIDE_EFFECT_UNKNOWN", "timeout after send")
    else:
        store.mark_succeeded(effect_key, result.reference, result.safe_json)
        return result

如果提供方支持幂等键和按幂等键查询,应同时使用;本地账本不能替代外部保证。如果提供方完全不支持,则应优先采用事务性 outbox、业务唯一号或补偿流程。最危险的做法是在超时后立即创建第二笔退款,因为“没有收到响应”不等于“对方没有执行”。

可以单独部署 reconciliation worker 处理 UNKNOWN。它只做三件事:用业务唯一号向提供方查询;找到唯一结果时补写 provider_ref 并推进任务;无法确认时保持挂起并升级人工。对账任务需要自己的退避和截止时间,但不能调用“再次创建”接口。若提供方既不能按业务号查询,也不提供幂等能力,应在接入评审时把这一限制标为业务风险,而不是让 Agent 猜测执行结果。

补偿操作也不等于数据库回滚。例如退款通知发送失败,不应撤销已经完成的退款;正确状态可能是“退款成功、通知待重试”。工作流需要为每类副作用定义可补偿、不可补偿和仅可对账三种语义,决定后续节点,而不是用一个全局 FAILED 掩盖部分成功。

效果账本也必须验证 request_hash。同一个幂等键对应不同金额时,应拒绝而不是复用旧结果。幂等保护的是同一业务意图的重复交付,不是让不同操作共享结果。

5. 审批令牌与工具网关共同形成权限边界

人工审批不能只保存一个 approved=true。审批必须绑定 run_idstep_id、工作流版本、规范化参数哈希、审批人、策略版本、有效期和一次性 nonce。恢复后只要退款金额、收款对象或执行参数发生变化,旧审批就应失效。

下面是简化的签名令牌。示例使用 HMAC 便于说明;生产环境应使用密钥管理服务、轮换机制或非对称签名,并把 nonce 的使用状态写入数据库。

import base64
import hashlib
import hmac
import json
import time


def issue_approval(secret: bytes, claims: dict) -> str:
    body = json.dumps(claims, sort_keys=True, separators=(",", ":")).encode()
    sig = hmac.new(secret, body, hashlib.sha256).digest()
    return base64.urlsafe_b64encode(body).decode() + "." + \
        base64.urlsafe_b64encode(sig).decode()


def verify_approval(secret: bytes, token: str, expected: dict) -> dict:
    body64, sig64 = token.split(".", 1)
    body = base64.urlsafe_b64decode(body64)
    supplied = base64.urlsafe_b64decode(sig64)
    wanted = hmac.new(secret, body, hashlib.sha256).digest()
    if not hmac.compare_digest(supplied, wanted):
        raise PermissionError("invalid approval signature")
    claims = json.loads(body)
    if claims["exp"] < int(time.time()):
        raise PermissionError("approval expired")
    for key in ("run_id", "step_id", "workflow_version", "args_hash"):
        if claims[key] != expected[key]:
            raise PermissionError(f"approval mismatch: {key}")
    return claims

签名验证通过后,还要原子消费 nonce,防止同一审批被重放。审批页面应展示经过规范化的关键参数和差异,而不是只显示模型生成的一段自然语言总结。

Prompt Injection 的处理同样不能依赖“检测出恶意句子”。网页、邮件、代码仓库和工单内容都应视为不可信数据。即使检测器漏报,工具网关仍必须让攻击无法越权:

STEP_CAPABILITIES = {
    "collect": {"ticket.read", "order.read"},
    "decide": {"policy.read"},
    "execute": {"refund.create"},
}


def call_tool(ctx, name: str, args: dict):
    if name not in STEP_CAPABILITIES.get(ctx.step_id, set()):
        raise PermissionError("tool is not allowed in this step")
    validate_json_schema(name, args)
    enforce_tenant_scope(ctx.tenant_id, name, args)
    if name == "refund.create":
        verify_and_consume_approval(ctx, args_hash=canonical_hash(args))
    return TOOL_GATEWAY.invoke(
        name=name,
        args=args,
        credential_scope=(ctx.tenant_id, name),
        egress_policy="allowlisted",
    )

模型进程不应持有长期凭证,也不应拥有任意 shell、任意 URL 请求和整库读取能力。只读节点挂载只读目录,写节点只暴露特定结构化工具;网络出口按域名和方法允许;工具返回值去除密钥、Cookie 和内部错误栈。高风险动作在工具网关再次验权,而不是相信模型声称“用户已经同意”。

这套边界的目标不是让模型永不受注入影响,而是即使模型被影响,也只能在当前步骤的最小能力范围内行动。

6. 可观测性要回答“能否安全恢复”

传统 HTTP 日志只能告诉你某个请求返回了 500,无法解释一个长任务经历了哪些节点。每次运行应贯穿 run_idtrace_id,每次尝试使用独立 attempt_id,模型调用、工具调用、checkpoint 和审批形成同一条 trace。

建议结构化日志至少包含以下字段:

{
  "timestamp": "2026-07-16T08:10:20Z",
  "run_id": "run_01...",
  "workflow": "refund_case",
  "workflow_version": 3,
  "step_id": "execute",
  "attempt": 2,
  "transition": "EXECUTING->RECONCILING",
  "checkpoint_version": 8,
  "trace_id": "4f...",
  "model": "configured-model-id",
  "tool": "refund.create",
  "request_hash": "sha256:...",
  "effect_key_hash": "sha256:...",
  "approval_id": "approval_01...",
  "latency_ms": 1840,
  "prompt_tokens": 1320,
  "completion_tokens": 210,
  "error_class": "SIDE_EFFECT_UNKNOWN",
  "error_code": "UPSTREAM_TIMEOUT_AFTER_SEND"
}

原始 Prompt、完整工具参数、密钥和个人信息不应默认进入日志。可以保存哈希、脱敏摘要和受访问控制的对象引用。run_id 适合日志检索,却不适合直接做监控标签,否则会造成高基数。

指标应围绕运行健康度而不是只看模型响应时间,例如:

  • agent_runs_total{workflow,version,status}:运行终态分布;
  • agent_step_attempts_total{step,error_class}:步骤尝试与错误分类;
  • agent_step_duration_seconds{step}:节点耗时分布;
  • agent_waiting_approval_age_seconds:等待审批时长;
  • agent_checkpoint_conflicts_total:并发恢复冲突;
  • agent_effects_pending_reconcile:结果未知、等待对账的副作用;
  • agent_retry_budget_exhausted_total:重试预算耗尽次数。

运行目标也应围绕业务状态定义。例如“可恢复运行在故障后多长时间重新变为 RUNNABLE”“处于 UNKNOWN 的效果最长多久得到确认”“审批通过后多久开始执行”,比单纯统计模型平均延迟更有意义。目标值应根据业务风险和外部系统能力制定,本文不虚构统一数字。发布新工作流版本时,应按版本分别观察成功率、恢复次数和人工升级率,避免新旧任务混在一起掩盖回归。

审计查询要能够从一次效果反查到任务、步骤、审批与模型决策,也要能从用户工单正向查到所有效果。日志与数据库记录的保留周期应匹配合规要求;过期清理时保留必要的哈希和业务回执,而不是无限保存原始对话。真正的可观测不是“什么都记录”,而是在不泄露敏感数据的前提下保留足以重建决策链的证据。

告警也要对应可行动状态。UNKNOWN 效果持续增长通常比单次 500 更值得立即处理;等待审批很久可能是业务队列问题,不应触发自动重试。

MCP 候选规范所描述的无状态核心、显式任务句柄、Tasks 扩展、路由头、Trace Context 和授权强化,与上述方向相符:协议层可以无状态,应用仍通过明确句柄维护业务状态。不过截至 7 月 16 日它仍是候选版本。生产接入应固定当前稳定协议,为候选能力增加 feature flag、契约测试和独立灰度环境,不应因为文章预告就直接切换协议版本。

mcp:
  stable_protocol_version: "2025-11-25"
  release_candidate_preview:
    enabled: false
    candidate_version: "2026-07-28"
  contract_tests:
    - tools_list_schema
    - tool_call_headers_match_body
    - authorization_issuer_validation
    - trace_context_propagation

上线前必须做的故障注入

可恢复能力不能只靠代码审查确认,必须在预发布环境主动制造故障。下面是一组最小验收清单:

故障注入预期结果
checkpoint 写入前杀死 worker新 worker 从上一个已提交版本恢复,不跳步
外部接口返回成功后、账本更新前杀死 worker效果进入 UNKNOWN 或通过外部幂等查询确认,不创建第二笔
同一任务重复投递 10 次只有一个 worker 获得有效租约,每个副作用最多生效一次
连续返回 429 并携带 Retry-After遵守服务端等待时间,达到预算后停止
模型连续输出不符合 Schema 的 JSON有限修复后失败,不进入写操作节点
网页内容要求读取环境变量并上传工具网关拒绝未授权工具和非允许网络出口
篡改审批金额、过期时间或签名审批验证失败,工具不执行
重放已消费的审批 nonce第二次调用被拒绝
恢复旧版本运行,同时部署新工作流旧任务按固定版本继续或显式迁移,不静默换图
数据库或对象存储短时不可用worker 不推进 checkpoint,不丢失已知状态

测试时不仅要看最终状态,还要检查步骤 attempt、效果账本、审批记录、日志 trace 和指标是否相互一致。建议把关键故障用例纳入持续集成,并定期在接近生产的环境中演练。只有“进程在任意边界退出仍能解释并恢复”,才算真正具备持久执行能力。

结语

生产级 Agent 的核心不是让模型多思考几轮,而是在概率性模型外建立确定性的控制层:显式状态机限制路径,持久 checkpoint 保存事实,错误分类约束重试,效果账本保护副作用,审批令牌与工具网关建立权限边界,日志、指标和故障注入验证恢复过程。

框架正在快速补齐图调度、沙箱和持久运行能力,但业务一致性不能外包给框架。先把一次退款、一次发布或一次工单变更做成可重放、可对账、可审计的工作流,再考虑增加更多模型和子 Agent,通常更接近生产系统真正需要的可靠性。

参考资料

作者说明:本文由 ClaudeAgent 技术团队基于公开资料与工程实践整理。

Logo

欢迎加入 MCP 技术社区!与志同道合者携手前行,一同解锁 MCP 技术的无限可能!

更多推荐