任务队列是 Agent 系统的"调度中枢"——它决定先做什么、后做什么、什么可以并行、失败了怎么办。如果说 Agent工作流 是宏观的流程编排，任务队列就是微观的执行调度。

一句话解释

任务队列管理 Agent 的任务分解、优先级排序、并发控制和错误恢复，是构建生产级 Agent 系统的基础设施。

任务分解策略

为什么需要任务分解

LLM 直接处理复杂任务时，容易出现"规划不足"或"遗漏步骤"的问题。将大任务分解为小任务，可以：

• 提高每步的成功率（小任务更容易完成）
• 支持并行执行（独立子任务可同时跑）
• 支持断点续跑（失败后只需重试一个子任务）

三种分解方式

分解方式	描述	优势	劣势	适用场景
LLM 分解	让 LLM 分析任务并输出子任务列表	灵活，适应任意任务	分解质量取决于 LLM 能力	开放式、创意性任务
规则分解	用预定义规则拆分（如按数据行数、按章节）	确定性强，可预测	不适应非结构化任务	批处理、数据处理
层级分解	先粗粒度分解，再对子任务递归分解	大小适中，可控制粒度	层级过深增加复杂度	大型复杂项目

LLM 分解的 Prompt 模式

你是一个任务规划器。请将以下任务分解为 3-7 个子任务。

任务：{user_request}

要求：
1. 每个子任务应该可以独立执行
2. 子任务之间标注依赖关系（哪些必须先完成）
3. 标注哪些子任务可以并行执行
4. 为每个子任务预估难度（低/中/高）

输出格式：
[
  {"id": "t1", "name": "...", "depends_on": [], "parallelizable": false, "difficulty": "low"},
  {"id": "t2", "name": "...", "depends_on": ["t1"], "parallelizable": true, "difficulty": "medium"},
  ...
]

动态任务图

生产级 Agent 需要支持运行时动态调整任务列表：

初始计划：[t1, t2, t3]
执行 t1 后发现新需求：
  → 新增 t1.5（t1 发现需要额外数据）
  → 取消 t3（t2 结果表明 t3 不再需要）
最终执行：[t1, t1.5, t2]

详见 Planner与Executor 中的重规划机制。

优先级调度

优先级模型

优先级 = f(紧急度, 重要性, 用户等级, SLA 倒计时)

高优先级：用户正在等待的交互请求
中优先级：有 SLA 承诺的异步任务
低优先级：后台批处理、数据更新

调度算法

算法	描述	适用场景
FIFO	先进先出，简单公平	同质化任务
优先级队列	按优先级排序执行	多等级任务混合
加权轮询	按权重分配执行时间	多用户公平性
最短任务优先	优先执行预估时间短的任务	减少平均等待时间
截止时间优先	优先执行即将超时的任务	SLA 保障

并发控制

并行 vs 串行 vs 条件执行

并发度控制

# 伪代码：并发度限制
max_concurrent = 5  # 最多同时执行 5 个子任务
semaphore = asyncio.Semaphore(max_concurrent)

async def execute_task(task):
    async with semaphore:
        result = await agent.run(task)
        return result

# 所有独立子任务并行执行，但不超过并发上限
results = await asyncio.gather(*[execute_task(t) for t in independent_tasks])

并发度设置的考量：

• LLM API 有速率限制（RPM/TPM）
• 工具 API 可能有并发限制
• 并发太高增加成本（同时消耗多个会话）
• 并发太低浪费时间（串行等待）

任务依赖图（DAG）

什么是任务 DAG

DAG 的优势

• 并行识别：自动发现可并行执行的独立任务
• 关键路径分析：找到决定总时长的关键路径
• 影响分析：某个任务失败时，快速识别受影响的下游任务
• 可视化：DAG 天然适合图形化展示执行进度

与工作流的关系

维度	任务队列（DAG）	工作流（Agent工作流）
层级	微观（子任务级）	宏观（流程级）
动态性	可运行时调整	通常预定义
依赖	显式 DAG 依赖	节点间顺序/条件
调度	优先级 + 并发控制	按图遍历执行

重试与错误恢复

重试策略

策略	描述	适用场景
固定间隔重试	每次等待固定时间后重试	瞬时错误
指数退避	重试间隔逐渐增加（1s→2s→4s→8s）	API 限流、网络问题
最大重试次数	超过 N 次后放弃	防止无限重试
降级策略	重试失败后切换到备选方案	有 fallback 的场景
死信队列	多次重试失败的任务进入死信队列	需要人工排查

错误分类与处理

错误类型              处理策略
──────────────────────────────────────
瞬时错误（网络超时）   → 指数退避重试
参数错误（输入格式）   → 不重试，返回错误
权限不足              → 不重试，请求人工授权
LLM 输出错误           → 重新生成（可调温度）
工具不可用             → 等待恢复或降级
计划错误               → 触发重规划

幂等性保证

重试时必须保证任务的幂等性——多次执行同一个任务不会产生副作用：

非幂等（危险）：
  "发送邮件通知用户"  → 重试会导致重复发送

幂等（安全）：
  "查询用户信息"      → 重试无副作用

保证幂等的方式：
  - 使用唯一 ID 避免重复操作
  - 先检查状态再执行（检查是否已发送）
  - 使用事务保证原子性

任务队列的可观测性

需要监控的指标

指标	描述	告警阈值
队列长度	待执行任务数	> 100
平均等待时间	任务从入队到开始执行的时间	> 30s
任务成功率	成功完成的任务比例	< 90%
平均执行时间	单个任务的执行耗时	因任务而异
重试率	需要重试的任务比例	> 10%
死信队列长度	重试失败的任务数	> 0

任务追踪

每个任务应有完整的执行记录：

• 任务 ID：唯一标识
• 创建时间：入队时间
• 开始时间：开始执行时间
• 结束时间：完成时间
• 状态：pending/running/success/failed/retrying
• 重试次数：已重试次数
• 输入/输出：任务参数和结果
• 错误信息：失败原因和堆栈

与 Planner 的协同

Planner与Executor 中的 Planner 负责任务分解和计划生成，任务队列负责执行调度：

• Planner 决定"做什么、什么顺序"
• 任务队列决定"何时执行、如何并发、失败了怎么办"
• 当任务失败时，任务队列通知 Planner 触发重规划

实战场景

场景 1：批量数据处理

任务：处理 1000 个文件
分解：每个文件一个子任务
调度：并发度 10（受 API 限流约束）
错误：单个文件失败重试 3 次，超过则跳过并记录
输出：成功 998 个，失败 2 个（人工排查）

场景 2：多步研究任务

任务：研究某公司的财务状况
分解：
  t1: 搜索公司基本信息
  t2: 查询财报数据（依赖 t1）
  t3: 搜索行业对比数据（与 t2 并行）
  t4: 分析数据（依赖 t2, t3）
  t5: 生成报告（依赖 t4）
执行：t1 → (t2 ∥ t3) → t4 → t5

常见误区

• 不设并发上限：所有任务同时执行，打爆 API 限流
• 重试无上限：任务一直失败一直重试，浪费资源和成本
• 忽视幂等性：重试导致重复操作（重复发邮件、重复扣款）
• 不追踪死信任务：失败任务被遗忘，无人排查
• DAG 过于复杂：分解太细，管理开销超过执行开销

可继续补充的方向

• 任务的优先级动态调整（基于运行时信息）
• 任务取消与回滚机制
• 分布式任务队列（跨多台机器调度）
• 任务预算控制（Token/成本上限）

关联笔记

• Agent状态管理 — 任务状态依赖状态管理持久化
• Agent工程架构 — 任务队列是架构执行层的核心
• Agent工作流 — 工作流是宏观编排，任务队列是微观调度
• Planner与Executor — Planner 生成任务，队列调度执行
• Agent失败模式 — 失败后的重试与恢复策略