AI Agent 的异步任务架构 - 文章 - 开发者社区

我正致力于两个项目的开发和升级，感兴趣的可以私信我，加入项目小组。

划水AI Node 全栈 AIGC 知识库，包括 AI 写作、多人协同编辑。复杂业务，真实上线。
智语 AI Agent 智能体项目。一个智能面试官，可以优化简历、模拟面试、解答题目等。

开始

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

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

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

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

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

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

整体架构

Frontend UI
Next.js

API Layer

Task Service

Postgres DB

Queue
BullMQ + Redis

Worker

Agent Runtime

LLM

Tools

External APIs

SSE / Event Stream

三条核心路径

1️⃣ 用户操作路径

arduino
 体验AI代码助手
 代码解读
复制代码
Frontend → API → Task Service → DB

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

👉 非阻塞

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

 体验AI代码助手
 代码解读
复制代码
Queue → Worker → Agent Runtime → Tools → LLM

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

👉 真正干活的地方

3️⃣ 实时反馈路径

 体验AI代码助手
 代码解读
复制代码
Worker → DB → SSE → Frontend

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

👉 用户可以看到：

erlang
 体验AI代码助手
 代码解读
复制代码
🧠 Thinking...
🔧 Calling API...
📄 Got result...

Agent Runtime 执行流程

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

是

否

启动 Agent Runtime

构建 Context

调用 LLM

解析输出

是否完成?

写入结果

结束任务

解析 Action

调用 Tool

获取结果

记录 Step

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

arduino
 体验AI代码助手
 代码解读
复制代码
while (没完成) {
  想 → 干 → 记录 → 再想
}

伪代码

js
 体验AI代码助手
 代码解读
复制代码
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 saveStep(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 saveStep(taskId, {
      type: 'observation',
      content: result
    })

    stepCount++
  }

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

Step 管理

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

step 的作用

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

step 数据定义

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

ts
 体验AI代码助手
 代码解读
复制代码
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 重新执行，这样就不用重复之前的步骤了。

任务中断

读取历史 Steps

构建 Context

重新进入 Agent Loop

继续执行

用户取消

更新 Task 状态为 cancelled

Worker 停止执行

定时任务流程

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

否

是

用户输入自然语言

LLM 解析意图

是否为定时任务

普通任务执行

生成 cron 表达式

创建 Task

定时触发

Worker 执行 Agent

输出结果

 体验AI代码助手
 代码解读
复制代码
自然语言 → 结构化 cron → 调度系统 → 定期触发 Agent

说一个具体的例子

js
 体验AI代码助手
 代码解读
复制代码
const result = await llm.invoke(`
每天早上9点帮我查 BTC 价格
`)

// LLM 识别到这是一个定时任务，输入 cron 表达方式
{
  type: "cron",
  cron: "0 9 * * *",
  task: "查询 BTC 价格"
}

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

js
 体验AI代码助手
 代码解读
复制代码
const task = await db.task.create({
  input: result.task,
  cron: result.cron,
  isCron: true
})

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

js
 体验AI代码助手
 代码解读
复制代码
await queue.add('agent-task', {
  taskId: task.id
}, {
  repeat: {
    cron: task.cron,
    tz: 'Asia/Singapore'
  }
})

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

js
 体验AI代码助手
 代码解读
复制代码
new Worker('agent-task', async (job) => {
  const { taskId } = job.data
  await runAgent(taskId)
})

总结

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