AI Agent异步任务架构设计与实现解析

在构建Agent应用时，一个核心区别在于任务执行模型。Chat App（像 ChatGPT 或 Claude Desktop）一直用的是同步交互模型——你发起一次请求，等 LLM 推理完，才能继续下一轮。说白了，一次请求就是一次完整的推理，阻塞的。

而 Agent App（比如 Manus、OpenClaw）完全不一样：任务可以在后台执行，用户可以继续发起新请求，任务完成后自动返回结果，甚至能设置 cron 定时执行。这背后的本质是：异步任务系统 + Agent 执行引擎。

开始

Chat App（如 ChatGPT / Claude Desktop）通常是同步交互模型：

• 发起一次请求
• 等待 LLM 推理完成
• 才能继续下一轮对话

本质：一次请求 = 一次完整推理（阻塞）

而 Agent App（如 Manus / OpenClaw）则完全不同：

• 任务可以在后台执行
• 用户可以继续发起新的请求
• 任务完成后自动返回结果
• 甚至可以设置 cron 定时执行

本质：异步任务系统 + Agent 执行引擎

整体架构

flowchart LR
A[Frontend UI
Next.js] --> B[API Layer]
B --> C[Task Service]
C --> D[(Postgres DB)]
C --> E[Queue
BullMQ + Redis]
E --> F[Worker]
F --> G[Agent Runtime]
G --> H[LLM]
G --> I[Tools]
I --> J[External APIs]
F --> D
D --> K[SSE / Event Stream]
K --> A

三条核心路径

1️⃣ 用户操作路径

Frontend → API → Task Service → DB

• 用户提交任务
• 系统创建 Task
• 不等待执行，立即返回 taskId

非阻塞

2️⃣ 后台执行路径（核心）

Queue → Worker → Agent Runtime → Tools → LLM

• Worker 异步执行任务
• Agent Runtime 控制执行流程
• 调用 LLM + Tools 完成任务

真正干活的地方

3️⃣ 实时反馈路径

Worker → DB → SSE → Frontend

• 执行过程持续写入 Step
• 前端通过 SSE 实时订阅

用户可以看到：

? Thinking...
? Calling API...
? Got result...

Agent Runtime 执行流程

这是 Agent 最核心的部分，更加详细的流程如下：

flowchart TD
A[启动 Agent Runtime] --> B[构建 Context]
B --> C[调用 LLM]
C --> D[解析输出]
D --> E{是否完成?}
E -- 是 --> F[写入结果]
F --> G[结束任务]
E -- 否 --> H[解析 Action]
H --> I[调用 Tool]
I --> J[获取结果]
J --> K[记录 Step]
K --> B

它的核心思想如下，这就是 Agent 和 Chat 的本质区别

while (没完成) {
  想 → 干 → 记录 → 再想
}

伪代码

async function runAgent(taskId: string) {
  let stepCount = 0
  const MAX_STEPS = 10

  while (stepCount < MAX_STEPS) {
    // 1. 构建上下文
    const context = await buildContext(taskId)

    // 2. LLM 推理
    const output = await llm.invoke(context)

    // 3. 记录 thought
    await sa veStep(taskId, {
      type: 'thought',
      content: output.thought
    })

    // 4. 判断是否结束
    if (output.type === 'finish') {
      await finishTask(taskId, output.result)
      return
    }

    // 5. 调用工具
    const result = await runTool(output.action)

    // 6. 记录 observation
    await sa veStep(taskId, {
      type: 'observation',
      content: result
    })

    stepCount++
  }

  await failTask(taskId, 'max steps reached')
}

Step 管理

上面可以看到，一个任务需要一个 while 循环分多步骤执行，每一个步骤就是一个 step ，要单独记录下来。

step 的作用

• 记录执行过程（可视化）
• 构建上下文（context）
• 支持恢复（resume）

step 数据定义

step 分多种类型，如思考中、调用 tool 执行中、执行结果观察中。所以它数据结构也要有不同的定义。

type Step =
| {
    id: string
    taskId: string
    type: 'thought'
    content: string
    createdAt: Date
  }
| {
    id: string
    taskId: string
    type: 'action'
    tool: string
    input: any
    createdAt: Date
  }
| {
    id: string
    taskId: string
    type: 'observation'
    content: string
    createdAt: Date
  }

取消/恢复 step

异步任务可以在后台执行，用户也可以随时操作取消任务、恢复任务。恢复任务的时候就需要读取历史 steps ，重新构建 Context 重新执行，这样就不用重复之前的步骤了。

flowchart TD
A[任务中断] --> B[读取历史 Steps]
B --> C[构建 Context]
C --> D[重新进入 Agent Loop]
D --> E[继续执行]

A --> F[用户取消]
F --> G[更新 Task 状态为 cancelled]
G --> H[Worker 停止执行]

定时任务流程

有了以上的异步的架构，才可以在此基础之上执行定时任务。

flowchart TD
A[用户输入自然语言] --> B[LLM 解析意图]
B --> C{是否为定时任务}
C -- 否 --> D[普通任务执行]
C -- 是 --> E[生成 cron 表达式]
E --> F[创建 Task]
F --> G[注册 BullMQ repeat job]
G --> H[定时触发]
H --> I[Worker 执行 Agent]
I --> J[输出结果]

核心思路就是：

自然语言 → 结构化 cron → 调度系统 → 定期触发 Agent

说一个具体的例子

const result = await llm.invoke(`每天早上9点帮我查 BTC 价格`)
// LLM 识别到这是一个定时任务，输入 cron 表达方式
{
  type: "cron",
  cron: "0 9 * * *",
  task: "查询 BTC 价格"
}

然后，在数据库创建一个 task 记录

const task = await db.task.create({
  input: result.task,
  cron: result.cron,
  isCron: true
})

然后，把定时任务插入到 queue 任务队列

await queue.add('agent-task', { taskId: task.id }, {
  repeat: {
    cron: task.cron,
    tz: 'Asia/Singapore'
  }
})

最后，使用 queue worker 启动定时任务，就开始定时执行了。

new Worker('agent-task', async (job) => {
  const { taskId } = job.data
  await runAgent(taskId)
})

总结

Agent 本质是一个可持续执行的 LLM 状态机，这就有点接近于黄仁勋说的“Agent 是全新的操作系统”这个概念了。