AGI 未来之门

主题

可恢复的 AI 任务执行器 ​

“请求进来，循环调用模型直到完成”在本地演示中可行，到了生产环境就会遇到断网、限流、进程重启、人工审批等待、重复提交和费用失控。长任务的核心问题不是让模型更聪明，而是让执行过程可暂停、可恢复、可重放、可审计。

本页以“批量分析 Pull Request 并生成发布风险包”为例，设计一个不依赖具体队列产品的任务执行器。

1. 什么时候需要任务执行器 ​

满足任意两项，就不应把任务绑在一次 HTTP 请求里：

  • 总耗时可能超过 20–30 秒；
  • 包含多个模型或外部工具调用；
  • 中间需要人确认；
  • 单步失败后希望从断点继续；
  • 用户会重复提交或刷新页面；
  • 必须控制总 Token、费用、工具次数或墙钟时间；
  • 需要解释任务为何得到当前结果。

短文本分类仍可同步完成。不要为了“架构先进”把所有调用都塞进队列。

2. 把任务建模成状态，而不是一段循环 ​

ts
type JobStatus =
  | "queued"
  | "running"
  | "waiting_approval"
  | "succeeded"
  | "failed"
  | "cancelled";

type StepStatus = "pending" | "running" | "succeeded" | "failed" | "skipped";

type Job = {
  id: string;
  type: "release-risk-pack";
  status: JobStatus;
  inputRef: string;
  inputHash: string;
  workflowVersion: string;
  currentStep: string;
  budget: { maxTokens: number; maxCostCents: number; deadlineAt: string };
  consumed: { tokens: number; costCents: number; toolCalls: number };
  createdBy: string;
  createdAt: string;
  updatedAt: string;
};

每一步都有独立输入、输出、状态和尝试次数。任务状态写入数据库，Worker 随时可以退出；新的 Worker 读取状态后继续，而不是依赖内存中的对话历史。

3. 工作流要显式版本化 ​

ts
const releaseRiskWorkflow = {
  version: "2026-09-17.1",
  steps: [
    "load_change",
    "collect_repository_context",
    "classify_change",
    "analyze_impact",
    "run_deterministic_checks",
    "assemble_risk_pack",
    "request_approval",
    "publish_artifact"
  ]
} as const;

Worker 不应临时问模型“下一步做什么”来决定所有路径。固定顺序和确定性分支由工作流代码负责；只有“这个差异属于哪类变更”“哪些模块可能受影响”交给模型。

任务开始后固定 workflowVersion、提示版本和模型路由。部署新代码不能悄悄改变正在运行的旧任务，否则重放无法复现。

4. 每一步都遵守函数协议 ​

ts
type StepContext<I> = {
  jobId: string;
  attempt: number;
  input: I;
  remainingBudget: { tokens: number; costCents: number; milliseconds: number };
  signal: AbortSignal;
};

type StepResult<O> = {
  output: O;
  usage: { tokens: number; costCents: number; toolCalls: number };
  events: Array<{ type: string; payload: unknown }>;
};

interface Step<I, O> {
  name: string;
  run(context: StepContext<I>): Promise<StepResult<O>>;
}

步骤输入来自已持久化的前一步产物，而不是某个全局可变对象。输出先写对象存储或数据库，再原子地把步骤标记为成功。日志不是步骤输出的替代品。

5. 幂等键阻止重复副作用 ​

重试不可避免。所有外部写操作必须接受幂等键：

text
idempotency_key = job_id + step_name + logical_action_id

例如发布风险报告时，先在 side_effects 表中插入唯一键；如果唯一键已存在，返回第一次发布的资源 ID。不要用“步骤状态是 succeeded”作为唯一保护，因为进程可能在外部调用成功后、写状态前崩溃。

sql
create table side_effects (
  idempotency_key text primary key,
  job_id text not null,
  step_name text not null,
  external_resource_id text,
  created_at timestamptz not null default now()
);

6. 检查点只保存可继续工作的事实 ​

每个模型调用结束后都保存：结构化输出、使用的上下文引用、模型与提示版本、Token/费用、校验结果。不要只保存完整提示和自然语言回答；恢复时真正需要的是下一步的确定输入。

示例：analyze_impact 的检查点不是一段分析文字，而是：

json
{
  "change_class": "database-migration",
  "affected_components": ["billing-api", "invoice-worker"],
  "required_checks": ["migration-dry-run", "rollback-test"],
  "evidence": [
    { "path": "db/migrations/20260917_add_state.sql", "lines": "1-42" }
  ],
  "open_questions": ["旧 Worker 是否能读取新增枚举值？"]
}

恢复时重新校验检查点 Schema；不兼容则明确迁移或从上一个安全步骤重跑。

7. 错误分类决定重试策略 ​

错误类型例子处理
瞬时错误超时、限流、服务端 5xx指数退避 + 抖动，在总截止时间内重试
永久配置错误凭证无效、模型不存在立即失败并告警
输入错误差异过大、文件格式损坏标记失败，返回可操作原因
模型协议错误JSON 无法解析、字段缺失携带校验错误修复一次
业务冲突资产负责人缺失、规则互斥进入人工处理队列
预算耗尽Token、费用、步骤或时间超限结束并交付部分结果

最大尝试次数只是最后一道保险。更重要的是总截止时间与总预算，避免每一步重试三次后组合成几十次调用。

8. 预算在调用前预留、调用后结算 ​

并发 Worker 如果只在结束后累计成本，多个步骤可能同时冲破预算。调用前先做原子预留：

ts
async function reserveBudget(jobId: string, estimatedCostCents: number): Promise<boolean> {
  // 伪代码：单条条件更新，防止并发超支
  const updated = await db.execute(`
    update jobs
       set reserved_cost_cents = reserved_cost_cents + $2
     where id = $1
       and consumed_cost_cents + reserved_cost_cents + $2 <= max_cost_cents
  `, [jobId, estimatedCostCents]);
  return updated.rowCount === 1;
}

调用结束后用实际成本结算并释放差额；调用失败也释放预留。预算不足时可以降级到更小上下文、跳过可选步骤或直接交付部分结果，但降级路径必须预先定义。

9. 人工审批是一种可持久化状态 ​

审批步骤不应阻塞 Worker：

  1. 生成待审批对象及其不可变摘要哈希；
  2. 把任务设为 waiting_approval；
  3. Worker 释放资源；
  4. 审批请求携带任务版本与摘要哈希；
  5. 收到决定后重新入队，从下一步继续。
ts
type Approval = {
  jobId: string;
  artifactHash: string;
  decision: "approve" | "reject" | "request_changes";
  reason: string;
  decidedBy: string;
  decidedAt: string;
};

如果审批期间上游产物被修改，哈希不一致，旧审批自动失效。高风险动作还应校验审批人角色与职责分离。

10. 取消任务需要协作式终止 ​

取消不是直接杀掉进程。设置 cancel_requested_at，Worker 在每个工具调用前后检查取消标志，并通过 AbortSignal 中止支持取消的网络请求。已完成的外部副作用不应假装回滚；系统要记录已做到哪一步，并提供补偿动作。

text
取消结果：
- 已完成：读取差异、影响分析、测试建议
- 未执行：发布报告、创建工单
- 外部副作用：无
- 可保留产物：risk-pack-draft.json

11. 用事件时间线调试任务 ​

至少记录这些事件：

text
job.created
step.started
model.requested
model.completed
validation.failed
step.retry_scheduled
budget.reserved
approval.requested
approval.decided
side_effect.completed
job.completed

事件包含 job_id、step_name、attempt、trace_id、版本、耗时和使用量；敏感正文单独存储并受访问控制。界面展示时间线，让运维人员看到任务卡在哪一步，而不是只给一个旋转图标。

12. 必测的故障场景 ​

上线前用故障注入验证：

  1. 模型调用完成后、保存结果前进程退出；
  2. 外部写入成功后、记录副作用前连接中断；
  3. 同一任务被重复投递两次；
  4. 审批等待期间部署新工作流；
  5. 预算只够完成一半步骤；
  6. 用户在模型调用中途取消；
  7. 上游输入在等待审批时发生变化；
  8. Worker 时钟偏差导致租约误判。

验收不是“最终能跑完”，而是没有重复副作用、状态可以解释、预算没有突破、失败后可以继续或安全结束。

13. 最小上线顺序 ​

  1. 单队列、单 Worker、固定工作流，先实现状态与检查点；
  2. 加幂等键和错误分类，再允许自动重试；
  3. 加预算预留和取消；
  4. 加人工审批；
  5. 做重放工具与故障注入；
  6. 最后才提高并发、增加动态分支或多 Agent。

可靠的 AI 长任务更像支付和订单系统，而不是聊天会话：每个状态可落库，每个副作用可识别，每个决定有证据，每次失败有恢复路径。

先找工作，再找工具。

AGI 未来之门