从 Prompt 到可恢复工作流:生产级 Agent 的 6 个工程要点
不少 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-28 的 MCP 规范仍是 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_owner 和 lease_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。对恢复真正有用的是业务事实:规范化输入的哈希、当前图版本、已完成节点、工具调用回执、外部对象引用、审批状态、剩余重试预算以及下一步允许的能力。
一个节点至少有三个值得持久化的时间点:
- 执行前:保存规范化参数、幂等键和状态
EXECUTING; - 外部调用后:保存提供方回执或“结果未知”标记;
- 提交后:原子地把步骤置为成功,并推进任务的
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_APPROVAL 或 WAITING_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_id、step_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_id、trace_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,通常更接近生产系统真正需要的可靠性。
参考资料
- OpenAI,2026-04-15:https://openai.com/index/the-next-evolution-of-the-agents-sdk/
- Google Developers Blog,2026-06-30:https://developers.googleblog.com/announcing-adk-go-20/
- Model Context Protocol Blog,
2026-07-28Release Candidate:https://blog.modelcontextprotocol.io/posts/2026-07-28-release-candidate/
作者说明:本文由 ClaudeAgent 技术团队基于公开资料与工程实践整理。
更多推荐

所有评论(0)