长时任务 Agent 的持久化队列架构

AI Agent Durable Queue：阻止长时间运行的工作在任务中途失败

AI agent 不需要犯错就能让用户失望。有时候它已经完成了一半有用的工作，工作进程重启了，模型调用超时了，工具卡住了，整个任务就像从未存在过一样消失了。

这种失败在日志中看起来很小，但对客户来说是个大问题。他们让 agent 分析账户、起草迁移计划、处理文档、更新记录或准备报告。agent 看起来在忙。然后什么都没发生。

如果你在构建生产级 AI 工作流，解决方案不是一个更长的 prompt。你需要一个持久队列：一个将 agent 工作存储为可恢复状态的系统，而不是一个脆弱的循环存在于单个进程中。

本指南为独立开发者、微型 SaaS 创建者和想让长时间运行的 agent 完成工作并提供证据的 AI 产品团队介绍了一个实用的架构。

为什么 agent 工作在生产环境中失败

第一个 agent 原型通常是一个 while 循环：调用模型、追加工具结果、重复直到模型返回最终答案。这对演示来说没问题。对于实际的客户任务来说则很危险，因为所有重要的东西都存在于内存中：

• 对话状态
• 当前的工具调用
• 用户和租户上下文
• 步骤失败的原因

如果进程重启，堆栈就消失了。如果工具成功但响应写入失败，你可能会重复该操作。如果模型调用超时，你可能不知道是重试、暂停还是将工作标记为失败。

长时间运行的 AI 工作与支付、导入、计费工作和 webhooks 有相同的可靠性需求：持久状态、租赁、幂等性、重试和审计日志。

什么是 AI agent durable queue？

AI agent durable queue 是一个持久化支持的 agent 任务执行层。

不是在一个请求或一个工作进程堆栈中运行整个 agent 工作，而是把工作拆分成持久化的记录：

• 一个用于整个用户请求的运行记录
• 每个模型调用、工具调用、批准或验证的步骤记录
• 每个重要状态转换的事件记录
• 文件、摘要、报告或生成输出的 artifact 记录
• 解释发生了什么以及为什么的收据记录

一个普通的任务队列说"运行这个后台工作"。一个持久化的 agent 队列说"记住每一步、安全地恢复、避免重复的副作用、证明发生了什么、当需要人工审查时暂停"。

这个差异很重要，因为单个用户请求可能涉及检索、规划、工具调用、文件生成、数据库更新、批准和最终验证。

Durable queue 的心智模型

将 agent 想象为一个状态机，而不是一个聊天循环。

一个运行通过以下状态移动：

queued -> planning -> waiting_for_tool -> waiting_for_approval
       -> verifying -> completed
       -> failed
       -> paused
       -> cancelled

每个转换都在下一个风险操作之前被写入。这给了你一个恢复点。

如果工作进程在调用模型时崩溃，另一个工作进程可以检查最后提交的状态并继续。如果工具调用已经创建了一条记录，系统可以检测幂等性密钥并避免做两次。如果一个步骤需要批准，运行暂停而不是试图聪明。

目标不是使失败不可能。目标是使失败可见、可恢复和安全。

一个简单的数据库模式

你可以在转向复杂基础设施之前用 Postgres 构建这个。从小开始。

create table agent_runs (
  id uuid primary key,
  tenant_id uuid not null,
  user_id uuid not null,
  status text not null,
  task text not null,
  current_step_id uuid,
  priority int not null default 100,
  attempts int not null default 0,
  max_attempts int not null default 3,
  leased_until timestamptz,
  leased_by text,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now(),
  completed_at timestamptz
);

create table agent_steps (
  id uuid primary key,
  run_id uuid not null references agent_runs(id),
  type text not null, -- model_call, tool_call, approval, verify, final_answer
  status text not null,
  input_json jsonb not null default '{}',
  output_json jsonb,
  error_json jsonb,
  idempotency_key text,
  attempt int not null default 0,
  created_at timestamptz not null default now(),
  updated_at timestamptz not null default now()
);

create unique index agent_steps_idempotency_key_idx
on agent_steps(run_id, idempotency_key)
where idempotency_key is not null;

create table agent_events (
  id bigserial primary key,
  run_id uuid not null,
  step_id uuid,
  event_type text not null,
  event_json jsonb not null default '{}',
  created_at timestamptz not null default now()
);

这个模式不花哨，但它给了你基础：

• 重复的工具调用可以被阻止
• 失败有结构化的上下文
• 支持可以检查发生了什么
• 指标稍后可以被计算

对于小团队，无聊的表格胜过神奇的 agent 内存。

用租赁声明工作，而不是希望

一个常见的错误是让许多工作进程获取同一个工作。另一个是让一个死的工作进程永远持有一个工作。

使用租赁。一个工作进程在短时间内声明一个运行。如果工作进程继续取得进展，它会延长租赁。如果它死了，租赁过期，另一个工作进程可以恢复。

update agent_runs
set leased_by = $1,
    leased_until = now() + interval '60 seconds',
    status = 'running',
    updated_at = now()
where id = (
  select id
  from agent_runs
  where status in ('queued', 'running')
    and (leased_until is null or leased_until < now())
  order by priority asc, created_at asc
  for update skip locked
  limit 1
)
returning *;

skip locked 很有用，因为多个工作进程可以安全地查找工作而不会相互阻止。

保持租赁足够短以快速恢复，但足够长以避免嘈杂的接管。对于许多 agent 工作流，30 到 120 秒是一个合理的起点。

将 agent 循环拆分成可恢复的步骤

不要将整个 agent 运行视为一个不透明的工作。在每个有意义的操作发生前后存储它。

type StepType = "model_call" | "tool_call" | "approval" | "verify";

async function executeNextStep(runId: string) {
  const run = await db.getRun(runId);
  const context = await buildContextPacket(run);

  const modelStep = await db.createStep({
    runId,
    type: "model_call",
    status: "started",
    input_json: { contextVersion: context.version }
  });

  try {
    const response = await llm.chat(context.messages);

    await db.completeStep(modelStep.id, {
      output_json: {
        model: response.model,
        usage: response.usage,
        tool_calls: response.tool_calls ?? [],
        content_preview: response.content?.slice(0, 500)
      }
    });

    if (response.tool_calls?.length) {
      await enqueueToolSteps(runId, response.tool_calls);
      return;
    }

    await enqueueVerificationStep(runId, response.content);
  } catch (err) {
    await db.failStep(modelStep.id, normalizeError(err));
    await scheduleRetryOrPause(runId, err);
  }
}

执行风险操作。

决定下一步。

这个序列使工作流可检查。它还给了你一个干净的地方来添加预算、策略、批准和测试。

使每个工具调用都是幂等的

Agent 调用工具。工具创建副作用。副作用是可靠性 bug 变成客户痛苦的地方。

如果一个 agent 发送同一张发票两次、删除同一个文件两次或发布同一条消息两次，用户不会在乎模型多么聪明。

每个工具调用都需要一个幂等性密钥。密钥应该代表预期的效果，而不仅仅是一个随机的重试尝试。

function toolIdempotencyKey(runId: string, toolName: string, args: unknown) {
  return hashJson({
    runId,
    toolName,
    args,
    purpose: "agent-tool-effect-v1"
  });
}

在执行工具之前，检查相同的密钥是否已经完成。

async function runToolStep(step: AgentStep) {
  const previous = await db.findCompletedStepByKey(
    step.run_id,
    step.idempotency_key
  );

  if (previous) {
    await db.completeStep(step.id, {
      output_json: previous.output_json,
      reused_result: true
    });
    return;
  }

  const result = await tools.execute(step.input_json.name, step.input_json.args);

  await db.completeStep(step.id, {
    output_json: result
  });
}

对于外部 API，当 API 支持时也要传递幂等性密钥。对于内部写入，在创建的行上存储密钥。

幂等性不仅仅是为了支付。它适用于任何你可能重试的 agent 操作。

根据失败类型使用重试策略

一个持久队列不应该盲目地重试所有内容。

为不同的失败使用不同的策略：

这是一个简单的重试决策函数：

function retryDecision(error: AgentError, attempt: number) {
  if (error.kind === "rate_limit") {
    return { retry: true, delayMs: error.retryAfterMs ?? 60_000 };
  }

  if (error.kind === "timeout" && attempt < 3) {
    return { retry: true, delayMs: Math.pow(2, attempt) * 1000 };
  }

  if (error.kind === "invalid_tool_args") {
    return { retry: false, next: "revise_plan" };
  }

  if (error.kind === "permission_denied") {
    return { retry: false, next: "pause_for_admin" };
  }

  return { retry: false, next: "fail_with_receipt" };
}

重点是避免将重试变成隐藏的成本泄漏。每个重试都应该有一个原因、一个限制和一个追踪。

将批准暂停添加为一级队列状态

长时间运行的 agent 通常到达它们不应该单独继续的时刻：

• 更改计费数据
• 更新面向客户的设置
• 运行高成本操作
• 使用敏感集成

不要用一个 prompt 来处理这个，比如"在需要时要求批准"。将批准存储为一个持久步骤。

{
  "type": "approval",
  "status": "waiting",
  "input_json": {
    "risk": "high",
    "action": "send_customer_email",
    "summary": "Send a renewal reminder to 84 trial users",
    "sample": "Hi {{first_name}}, your trial ends...",
    "required_role": "workspace_admin"
  }
}

运行保持暂停，直到正确的人类批准、拒绝或编辑操作。

这给了你干净的产品行为：

• 工作进程不会在等待时空转
• 用户可以稍后恢复
• 审查者看到确切的提议操作
• 审计日志显示谁批准了什么
• agent 不能悄悄绕过这个门

批准门当它们保护信任时不是摩擦。它们是工作流契约的一部分。

分别存储 artifact

一个长时间运行的 agent 生成的不仅仅是消息。它可能创建文件、报告、差异、SQL 查询、图表、摘要或提取的数据。

不要把这些 artifact 埋在一个巨大的 prompt 文本中。将它们存储为单独的记录，然后只将相关的 artifact 摘要传回模型。这保持上下文更小，并使最终输出更容易验证。

例如，一个数据清理 agent 可能存储：

• 上传的文件快照
• 生成的 Python 脚本

模型不需要每次调用时都有完整的文件。它需要正确的摘要、正确的链接和正确的约束。

在完成运行前验证

一个持久队列不应该仅仅因为模型写了一个自信的最终答案就将工作标记为完成。

添加一个验证步骤。

对于确定性任务，验证可以是代码：

async function verifyReport(runId: string) {
  const artifacts = await db.listArtifacts(runId);
  const report = artifacts.find(a => a.artifact_type === "final_report");
  const sources = artifacts.filter(a => a.artifact_type === "source_snapshot");

  return {
    hasReport: Boolean(report),
    hasSources: sources.length > 0,
    hasEmptySections: report?.content_text?.includes("TODO") ?? false
  };
}

对于模糊任务，使用混合方法：

• 必需的部分检查
• LLM-as-judge 用于狭隘的评分标准
• 对于风险输出的人类审查

一个有用的完成规则：agent 完成只有当队列能够解释什么改变了、什么证据支持它、什么仍然未解决。

从第一天开始跟踪的指标

你不需要一个大规模的可观测性堆栈来开始。跟踪显示队列是否使 agent 工作更可靠的指标。

• 中位数和 p95 运行持续时间
• 按工具名的工具失败
• 按提供者的模型失败
• 阻止的重复工具调用
• 在租赁过期后恢复的运行
• 每个完成的运行的成本
• 用户可见的失败率

最重要的指标是成功结果的成本。一个便宜的失败运行仍然是浪费。

实际的构建顺序

如果你从头开始构建这个，不要试图一下子做完。从运行、步骤、事件、租赁、幂等工具调用、类型化重试、批准暂停、artifact、验证和客户可见状态开始。

对于独立构建者，Postgres 加一个工作进程足以开始。更大的团队稍后可以将调度移到专用队列或工作流引擎，同时保持持久化运行状态和收据在产品数据库中。

一个长时间运行的 AI agent 不仅仅是一个更聪明的 prompt。它是一个分布式工作流，有不可靠的模型、脆弱的 API、人类批准、可变成本和真实的副作用。

核心模式很简单：持久化运行、持久化每个步骤、使用租赁、使工具幂等、谨慎重试、暂停批准、验证完成并留下收据。

FAQ

什么是 AI agent durable queue？

AI agent durable queue 是一个持久化支持的执行系统，存储 agent 运行、步骤、事件、artifact、重试、批准和结果，以便长时间运行的 AI 工作可以在崩溃或超时后恢复。

一个普通的后台工作队列对 AI agent 来说足够吗？

一个普通的队列是一个很好的开始，但 AI agent 通常需要更多的状态。你需要步骤历史、工具幂等性、批准暂停、模型使用、artifact 和验证收据。一个基本的队列可以运行工作进程，但持久化的 agent 状态应该存在于你的应用数据库或工作流存储中。

我如何阻止 agent 重复相同的工具操作？

为每个有副作用的工具调用使用幂等性密钥。用工具步骤存储密钥，当可能时，将其传递给外部 API。在重试时，在执行操作前检查相同的密钥是否已经完成。

当一个 agent 工作进程崩溃时应该发生什么？

工作进程租赁应该过期。另一个工作进程应该声明运行、检查最后完成的步骤并从下一个安全状态继续。如果前一步可能造成了副作用，新的工作进程在重试前应该使用幂等性记录。

什么时候 AI agent 应该请求人工批准？

对于发送消息、花费金钱、删除或修改重要数据、外部发布、访问敏感集成或超过成本和风险阈值的操作，需要批准。批准应该存储为一个持久步骤，而不是仅留给 prompt 指令。