多步骤任务规划

要点

  • 任务规划是 Agent 在执行前先把目标拆解成有顺序、有依赖的步骤列表，再按计划逐步执行
  • 前面的 Agent Loop 每一轮都由 LLM 临时决定下一步，没有全局计划，复杂任务上容易跳步或重复
  • ReAct 策略交替进行推理和行动，适合步骤数不确定的任务；Plan-and-Execute 先一次性生成计划再执行，适合步骤可提前确定的任务
  • 任务拆解需要给每个步骤标注工具名、参数来源、输出变量和依赖关系
  • 拓扑排序把步骤按依赖关系转成合法的执行顺序，无依赖的步骤可以并行执行
  • 状态跟踪让 Agent 区分步骤的完成状态，失败时只回滚受影响的部分
  • 用 TypeScript + Hono 实现，核心是 Planner 生成计划、Executor 按序执行、Tracker 管理状态

1. 什么是任务规划

前面几篇的 Agent，工作模式是这样的：收到请求 → 调 LLM → LLM 决定调什么工具 → 执行工具 → 拿到结果再调 LLM → 直到 LLM 给出最终回答。

这个循环能处理不少场景，但有一个隐含假设：每一步的决策都是临时做出的。LLM 在每一轮只看当前对话历史，凭直觉选下一步。对于简单的线性任务（查天气 → 发提醒），这没有问题。

任务稍微复杂一点，「走一步看一步」就会出问题。用户说：「分析上月销售数据，找出前五名商品，整理报告发邮件给运营团队。」

这个任务涉及多个有依赖关系的步骤：查数据 → 排序 → 取前五名 → 查商品详情 → 生成报告 → 发邮件。Agent Loop 里 LLM 在每一轮都需要重新推理「下一步做什么」，可能重复查询、跳步或者漏掉环节。

任务规划要解决的问题：执行之前先制定计划，把目标拆解成有顺序、有依赖的步骤，然后按计划执行。

2. 两种规划策略

ReAct：交替推理和行动

ReAct（Reasoning + Acting）不预先生成完整计划，而是每一步先「思考」当前状态，决定下一步行动，执行后观察结果再思考下一步。

Thought: 需要先查询上月销售数据。
Action: query_sales_data(month="上月")
Observation: 返回 500 条销售记录
 
Thought: 数据已拿到，按商品汇总取前五名。
Action: aggregate_and_sort(data=..., limit=5)
Observation: 返回前五名商品列表
 
Thought: 查这五个商品的详情，然后生成报告发邮件。
...

ReAct 的优势是灵活——每一步都能根据最新观察调整方向。弱点是每一步都需要一次 LLM 调用，步骤多的任务消耗大。

Plan-and-Execute：先规划后执行

Plan-and-Execute 把过程分成两个阶段：LLM 一次性生成完整步骤列表，然后执行引擎按顺序执行，执行阶段不需要 LLM 参与。

// 阶段一：规划
步骤 1：query_sales_data(month="上月") → sales_data
步骤 2：aggregate_and_sort(data=$steps.1, limit=5) → top_5
步骤 3：batch_get_product_info(ids=$steps.2) → details
步骤 4：generate_report(top=$steps.2, details=$steps.3) → report
步骤 5：send_email(to="[email protected]", content=$steps.4) → result
 
// 阶段二：按序执行，不需要 LLM

Plan-and-Execute 的延迟和成本更低，但计划生成后固定，遇到意外需要重新规划。

怎么选

维度ReActPlan-and-Execute
规划时机每步都规划开始前一次性规划
LLM 调用次数每步一次规划一次
灵活性高低，需要重规划机制补充
适合场景步骤数不确定、中间结果影响路径步骤可提前确定、追求低延迟

两种策略可以组合：Plan-and-Execute 生成初始计划，执行时遇到意外切换到 ReAct 做局部调整。这篇的实现以 Plan-and-Execute 为主线。

3. 任务拆解与依赖关系

步骤的类型定义

每个步骤需要三个信息：做什么（工具名）、输入从哪来（参数）、输出存到哪（变量名）。

// src/planner/types.ts
 
export type TaskStep = {
  id: number
  description: string
  tool: string
  args: Record<string, StepArgValue>
  output: string
  dependsOn: number[]
  status: 'pending' | 'running' | 'completed' | 'failed'
}
 
// 参数值：固定值或对其他步骤输出的引用
export type StepArgValue =
  | { type: 'literal'; value: string | number | boolean }
  | { type: 'reference'; path: string }
 
export type TaskPlan = {
  goal: string
  steps: TaskStep[]
  status: 'planning' | 'executing' | 'completed' | 'failed'
}
 
export type ExecutionContext = {
  outputs: Record<string, unknown>
}

用 LLM 生成计划

Planner 调一次 LLM，把用户请求转成步骤列表。

// src/planner/planner.ts
import type { TaskStep, TaskPlan } from './types'
 
const AVAILABLE_TOOLS = [
  { name: 'query_sales_data', description: '查询指定月份的销售数据' },
  { name: 'aggregate_and_sort', description: '按指定字段分组汇总并排序' },
  { name: 'batch_get_product_info', description: '根据商品 ID 列表批量查询详情' },
  { name: 'generate_report', description: '根据数据生成文字报告' },
  { name: 'send_email', description: '发送邮件' },
]
 
export async function generatePlan(
  goal: string,
  callLLM: (prompt: string) => Promise<string>,
): Promise<TaskPlan> {
  const prompt = `根据用户目标，把任务拆解成执行步骤。
 
可用工具：
${AVAILABLE_TOOLS.map((t) => `- ${t.name}: ${t.description}`).join('\n')}
 
输出 JSON 数组，每个元素包含 id、description、tool、args、output、dependsOn。
args 中固定值用 { "type": "literal", "value": "xxx" }，
引用其他步骤输出用 { "type": "reference", "path": "steps.{id}.output" }。
只输出 JSON。
 
用户目标：${goal}`
 
  const response = await callLLM(prompt)
  const steps: TaskStep[] = JSON.parse(response)
 
  return {
    goal,
    steps: steps.map((s) => ({ ...s, status: 'pending' as const })),
    status: 'planning',
  }
}

LLM 返回的计划示例：

// planner-output.json
[
  { "id": 1, "tool": "query_sales_data",
    "args": { "month": { "type": "literal", "value": "上月" } },
    "output": "sales_data", "dependsOn": [] },
  { "id": 2, "tool": "aggregate_and_sort",
    "args": { "data": { "type": "reference", "path": "steps.1.output" } },
    "output": "top_5", "dependsOn": [1] },
  { "id": 3, "tool": "batch_get_product_info",
    "args": { "ids": { "type": "reference", "path": "steps.2.output" } },
    "output": "details", "dependsOn": [2] }
]

4. 拓扑排序与并行执行

步骤之间的依赖关系构成有向无环图。拓扑排序把步骤转成合法的执行顺序。

// src/planner/topological-sort.ts
import type { TaskStep } from './types'
 
export function topologicalSort(steps: TaskStep[]): TaskStep[] {
  const sorted: TaskStep[] = []
  const visited = new Set<number>()
  const stepMap = new Map(steps.map((s) => [s.id, s]))
 
  function visit(step: TaskStep) {
    if (visited.has(step.id)) return
    visited.add(step.id)
    for (const depId of step.dependsOn) {
      const dep = stepMap.get(depId)
      if (!dep) throw new Error(`步骤 ${step.id} 依赖了不存在的步骤 ${depId}`)
      visit(dep)
    }
    sorted.push(step)
  }
 
  for (const step of steps) visit(step)
 
  if (sorted.length !== steps.length) {
    throw new Error('检测到循环依赖')
  }
  return sorted
}

依赖关系不紧密的步骤可以并行执行。getReadySteps 找出当前所有依赖都已完成的步骤：

// src/planner/scheduler.ts
import type { TaskStep } from './types'
 
export function getReadySteps(steps: TaskStep[]): TaskStep[] {
  const completedIds = new Set(
    steps.filter((s) => s.status === 'completed').map((s) => s.id),
  )
  return steps.filter(
    (s) => s.status === 'pending' && s.dependsOn.every((id) => completedIds.has(id)),
  )
}

5. 执行引擎与状态跟踪

执行引擎按计划顺序执行步骤，把结果存入上下文，更新步骤状态。

// src/planner/executor.ts
import type { TaskPlan, TaskStep, ExecutionContext, StepArgValue } from './types'
import { topologicalSort } from './topological-sort'
import { getReadySteps } from './scheduler'
 
type ToolExecutor = (tool: string, args: Record<string, unknown>) => Promise<unknown>
 
export async function executePlan(
  plan: TaskPlan,
  executeTool: ToolExecutor,
): Promise<{ outputs: Record<string, unknown>; failedStep?: TaskStep }> {
  const context: ExecutionContext = { outputs: {} }
  const sorted = topologicalSort(plan.steps)
  plan.status = 'executing'
 
  while (true) {
    const readySteps = getReadySteps(sorted)
    if (readySteps.length === 0) {
      const failed = sorted.find((s) => s.status === 'failed')
      plan.status = failed ? 'failed' : 'completed'
      return { outputs: context.outputs, failedStep: failed }
    }
 
    // 并行执行当前批次的所有就绪步骤
    await Promise.allSettled(
      readySteps.map(async (step) => {
        step.status = 'running'
        const resolvedArgs = resolveArgs(step.args, context)
        try {
          context.outputs[step.output] = await executeTool(step.tool, resolvedArgs)
          step.status = 'completed'
        } catch (error) {
          step.status = 'failed'
          context.outputs[step.output] = { error: (error as Error).message }
        }
      }),
    )
  }
}
 
function resolveArgs(
  args: Record<string, StepArgValue>,
  context: ExecutionContext,
): Record<string, unknown> {
  const resolved: Record<string, unknown> = {}
  for (const [key, arg] of Object.entries(args)) {
    if (arg.type === 'literal') resolved[key] = arg.value
    else if (arg.type === 'reference') resolved[key] = context.outputs[arg.path]
  }
  return resolved
}

Promise.allSettled 保证单步失败不会导致整个批次崩溃。status 字段让调度逻辑能区分已完成、正在执行和等待中的步骤。

6. 失败处理与回滚

某一步失败后，前面已完成的步骤要不要撤销，取决于业务需求。「创建订单」和「扣减库存」是关键步骤，失败了需要回滚；「发送确认邮件」是非关键步骤，失败了跳过即可。

// src/planner/rollback.ts
import type { TaskStep, StepArgValue } from './types'
 
type RollbackStep = TaskStep & {
  rollback?: (output: unknown) => Promise<void>
  critical: boolean
}
 
export async function executeWithRollback(
  steps: RollbackStep[],
  executeTool: (tool: string, args: Record<string, unknown>) => Promise<unknown>,
): Promise<{ outputs: Record<string, unknown>; rolledBack: boolean }> {
  const outputs: Record<string, unknown> = {}
  const completed: RollbackStep[] = []
 
  for (const step of steps) {
    step.status = 'running'
    try {
      const resolvedArgs = resolveArgs(step.args, { outputs })
      outputs[step.output] = await executeTool(step.tool, resolvedArgs)
      step.status = 'completed'
      completed.push(step)
    } catch (error) {
      step.status = 'failed'
      if (!step.critical) {
        outputs[step.output] = { error: (error as Error).message }
        continue
      }
      // 关键步骤失败：按逆序回滚
      for (const prev of completed.reverse()) {
        if (prev.rollback) {
          try { await prev.rollback(outputs[prev.output]) } catch { /* 回滚失败只能记录 */ }
        }
      }
      return { outputs, rolledBack: true }
    }
  }
  return { outputs, rolledBack: false }
}
 
function resolveArgs(args: Record<string, StepArgValue>, ctx: { outputs: Record<string, unknown> }) {
  const resolved: Record<string, unknown> = {}
  for (const [k, v] of Object.entries(args)) {
    resolved[k] = v.type === 'literal' ? v.value : ctx.outputs[v.path]
  }
  return resolved
}

critical: true 的步骤失败时整个计划中断并回滚；critical: false 的步骤失败时跳过继续。

7. 在 Hono 中集成

// src/routes/task-planner.ts
import { Hono } from 'hono'
import { generatePlan } from '../planner/planner'
import { executePlan } from '../planner/executor'
import type { AppEnv } from '../types'
 
const taskApp = new Hono<AppEnv>()
 
async function executeTool(tool: string, args: Record<string, unknown>): Promise<unknown> {
  switch (tool) {
    case 'query_sales_data': return querySalesData(args.month as string)
    case 'aggregate_and_sort': return aggregateAndSort(args.data as unknown[], args.limit as number)
    case 'batch_get_product_info': return batchGetProductInfo(args.ids as string[])
    case 'generate_report': return generateReport(args.top, args.details)
    case 'send_email': return sendEmail(args.to as string, args.content as string)
    default: throw new Error(`未知工具: ${tool}`)
  }
}
 
taskApp.post('/execute', async (c) => {
  const { goal } = await c.req.json<{ goal: string }>()
  if (!goal) return c.json({ error: 'goal 不能为空' }, 400)
 
  const callLLM = async (prompt: string) => {
    const res = await c.env.AI.run('@cf/meta/llama-3.1-8b-instruct', {
      messages: [{ role: 'user', content: prompt }],
    })
    return typeof res === 'string' ? res : JSON.stringify(res)
  }
 
  // 阶段一：LLM 生成计划
  const plan = await generatePlan(goal, callLLM)
  // 阶段二：执行引擎按序执行
  const { outputs, failedStep } = await executePlan(plan, executeTool)
 
  if (failedStep) {
    return c.json({ status: 'partial', failedStep: failedStep.id, outputs })
  }
  return c.json({ status: 'completed', outputs })
})
 
export default taskApp

前端发送 &#123; "goal": "分析上月销售数据，找出前五名商品，整理报告发邮件给 [email protected]" &#125; 到 POST /api/task/execute，Planner 生成步骤列表，Executor 按依赖顺序执行，最后返回结果。

总结

回顾一下这篇的要点：

  • 任务规划让 Agent 在执行前先生成步骤列表，避免「走一步看一步」导致的跳步和重复
  • ReAct 每步都推理，灵活但 LLM 调用多；Plan-and-Execute 一次规划逐步执行，延迟低但需要重规划补充灵活性
  • 每个步骤标注工具名、参数来源、输出变量和依赖关系，执行引擎才能正确调度
  • 拓扑排序确定执行顺序，无依赖的步骤可以并行执行减少总耗时
  • 执行引擎用 status 跟踪步骤状态，Promise.allSettled 保证单步失败不崩全局
  • 回滚策略通过 critical 字段区分：关键步骤失败则回滚，非关键步骤失败则跳过

下一篇讨论 Agent 记忆系统，解决 Agent 如何跨会话保持上下文和积累经验。