Agent02：任务规划
理解 Agent 的任务规划能力，包括目标拆解、步骤生成、执行反馈和计划修正，让 Agent 能够处理复杂的多步骤任务。
任务规划是 Agent 的核心能力之一，它让 Agent 能够将复杂的目标分解为可执行的步骤，并根据执行反馈动态调整计划。理解任务规划的机制，才能构建能够处理复杂任务的 Agent 系统。
任务规划的基本原理
什么是任务规划
任务规划是指 Agent 根据用户的目标，自动生成执行计划并动态调整的能力：
| 组件 | 说明 | 示例 |
|---|---|---|
| 目标 | 用户想要达成的结果 | ”帮我订一张去北京的机票” |
| 计划 | 达成目标的步骤序列 | 查航班 → 选座位 → 支付 → 出票 |
| 执行 | 按计划执行每个步骤 | 调用航班查询 API |
| 反馈 | 执行结果的信息 | API 返回的航班列表 |
| 修正 | 根据反馈调整计划 | 航班售罄，换其他航班 |
任务规划的流程
用户目标 → 目标分析 → 计划生成 → 步骤执行 → 结果反馈 → 计划修正 → 完成目标任务规划的价值
| 场景 | 说明 | 价值 |
|---|---|---|
| 复杂任务 | 需要多步骤完成的任务 | 自动拆解，无需人工干预 |
| 动态环境 | 执行过程中条件变化 | 自适应调整计划 |
| 长时任务 | 需要长时间执行的任务 | 保持任务状态，逐步推进 |
| 多工具任务 | 需要调用多个工具的任务 | 合理编排工具调用顺序 |
目标拆解
目标拆解的原则
| 原则 | 说明 | 示例 |
|---|---|---|
| 可执行性 | 每个子目标都可以独立执行 | ”查询航班”是可执行的 |
| 完整性 | 所有子目标组合起来能达成总目标 | 包含查询、选择、支付、出票 |
| 独立性 | 子目标之间尽量独立 | 查询航班不依赖选座位 |
| 顺序性 | 子目标之间有合理的顺序 | 先查询再选择 |
目标拆解方法
def decompose_goal(goal):
"""目标拆解"""
# 使用模型进行目标拆解
prompt = f"""
将以下目标拆解为可执行的子目标：
目标：{goal}
请输出子目标列表，每个子目标一行，格式为：序号. 子目标描述
"""
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
# 解析子目标
sub_goals = []
for line in response.choices[0].message.content.split('\n'):
line = line.strip()
if line and '.' in line:
parts = line.split('.', 1)
if len(parts) == 2:
sub_goals.append(parts[1].strip())
return sub_goals目标拆解示例
| 原始目标 | 拆解后的子目标 |
|---|---|
| 帮我订一张去北京的机票 | 1. 查询去北京的航班 2. 选择合适的航班 3. 填写乘客信息 4. 完成支付 5. 获取电子机票 |
| 生成一份销售周报 | 1. 查询本周销售数据 2. 计算销售总额 3. 分析销售趋势 4. 生成周报图表 5. 撰写周报文案 |
| 部署一个 Web 应用 | 1. 打包应用代码 2. 配置服务器环境 3. 上传应用包 4. 启动应用服务 5. 验证服务可用性 |
步骤生成
步骤生成的要素
| 要素 | 说明 | 示例 |
|---|---|---|
| 步骤描述 | 明确的执行动作 | ”调用航班查询 API” |
| 前置条件 | 执行该步骤需要满足的条件 | ”已获取用户出发城市” |
| 工具选择 | 需要调用的工具 | ”search_flights” |
| 参数准备 | 工具调用所需的参数 | {“from_city”: “上海”, “to_city”: “北京”} |
| 预期结果 | 期望的执行结果 | 返回航班列表 |
步骤生成实现
class StepGenerator:
def __init__(self, tools):
self.tools = tools
def generate_steps(self, sub_goals):
"""根据子目标生成执行步骤"""
steps = []
for i, sub_goal in enumerate(sub_goals):
prompt = f"""
为以下子目标生成执行步骤：
子目标：{sub_goal}
可用工具：{json.dumps(self.tools, indent=2)}
请输出步骤详情，格式为 JSON：
{{
"step": "{i+1}",
"description": "步骤描述",
"tool": "工具名称或None",
"arguments": {{工具参数}}
}}
"""
response = openai.chat.completions.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
try:
step = json.loads(response.choices[0].message.content)
steps.append(step)
except json.JSONDecodeError:
# 如果解析失败，创建一个默认步骤
steps.append({
"step": i+1,
"description": sub_goal,
"tool": None,
"arguments": {}
})
return steps步骤生成策略
| 策略 | 说明 | 适用场景 |
|---|---|---|
| 贪心策略 | 每次选择最优的下一步 | 简单任务 |
| 深度优先 | 完整生成所有步骤再执行 | 确定性任务 |
| 动态规划 | 考虑多种可能性并选择最优路径 | 复杂任务 |
| 层次规划 | 先制定高层计划，再细化执行 | 大型任务 |
执行反馈
执行反馈的类型
| 类型 | 说明 | 示例 |
|---|---|---|
| 成功 | 步骤执行成功 | API 返回正确结果 |
| 失败 | 步骤执行失败 | API 返回错误 |
| 部分成功 | 步骤部分完成 | 返回部分数据 |
| 超时 | 步骤执行超时 | API 响应超时 |
| 异常 | 发生异常情况 | 网络错误 |
执行反馈处理
class ExecutionFeedback:
def __init__(self):
self.feedback_history = []
def record_feedback(self, step, result, status):
"""记录执行反馈"""
feedback = {
"step": step,
"timestamp": datetime.now().isoformat(),
"result": result,
"status": status
}
self.feedback_history.append(feedback)
return feedback
def analyze_feedback(self):
"""分析执行反馈"""
success_count = sum(1 for f in self.feedback_history if f["status"] == "success")
fail_count = sum(1 for f in self.feedback_history if f["status"] == "fail")
return {
"total_steps": len(self.feedback_history),
"success_rate": success_count / len(self.feedback_history) if self.feedback_history else 0,
"fail_count": fail_count,
"last_feedback": self.feedback_history[-1] if self.feedback_history else None
}反馈驱动的学习
| 学习方式 | 说明 | 实现方式 |
|---|---|---|
| 即时修正 | 根据当前反馈调整下一步 | 动态规划 |
| 历史学习 | 从历史执行中学习经验 | 记录成功/失败案例 |
| 模型微调 | 使用反馈数据微调模型 | 监督学习 |
| 规则优化 | 根据反馈优化规则 | 调整参数和阈值 |
计划修正
计划修正的触发条件
| 条件 | 说明 | 示例 |
|---|---|---|
| 执行失败 | 步骤执行失败 | API 返回错误 |
| 结果不符合预期 | 执行结果与预期不符 | 查询结果为空 |
| 环境变化 | 外部环境发生变化 | 航班价格变动 |
| 用户干预 | 用户修改了目标 | 用户改变出行日期 |
| 超时 | 执行时间过长 | 超过预期时间 |
计划修正策略
class PlanCorrector:
def __init__(self):
self.strategies = {
"retry": self.retry_strategy,
"skip": self.skip_strategy,
"replace": self.replace_strategy,
"backtrack": self.backtrack_strategy,
"abort": self.abort_strategy
}
def correct_plan(self, current_plan, feedback):
"""根据反馈修正计划"""
status = feedback["status"]
if status == "success":
return current_plan, "继续执行"
# 根据状态选择修正策略
if status == "fail":
return self.strategies["retry"](current_plan, feedback)
elif status == "timeout":
return self.strategies["replace"](current_plan, feedback)
elif status == "partial":
return self.strategies["skip"](current_plan, feedback)
return current_plan, "无法修正"
def retry_strategy(self, plan, feedback):
"""重试策略"""
current_step = feedback["step"]
plan["steps"][current_step-1]["retry_count"] = plan["steps"][current_step-1].get("retry_count", 0) + 1
if plan["steps"][current_step-1]["retry_count"] < 3:
return plan, "重试当前步骤"
else:
return self.replace_strategy(plan, feedback)
def replace_strategy(self, plan, feedback):
"""替换策略"""
current_step = feedback["step"]
# 生成替代步骤
alternative_step = self.generate_alternative_step(plan["steps"][current_step-1])
plan["steps"][current_step-1] = alternative_step
return plan, "替换为替代步骤"
def backtrack_strategy(self, plan, feedback):
"""回溯策略"""
current_step = feedback["step"]
# 回溯到上一个步骤
plan["current_step"] = max(1, current_step - 1)
return plan, "回溯到上一步骤"计划修正模式
| 模式 | 说明 | 适用场景 |
|---|---|---|
| 重试 | 重新执行当前步骤 | 临时错误 |
| 跳过 | 跳过当前步骤 | 非关键步骤失败 |
| 替换 | 用替代步骤替换 | 当前步骤无法执行 |
| 回溯 | 返回上一步重新规划 | 路径选择错误 |
| 中止 | 终止任务 | 无法完成目标 |
任务规划的常见问题
问题1：目标拆解不完整
表现：遗漏了关键步骤
解决方案：
- 提供目标拆解的示例
- 使用更强大的模型进行拆解
- 添加人工审核环节
问题2：步骤顺序不合理
表现：步骤之间顺序错误
解决方案：
- 优化步骤生成的提示词
- 添加步骤依赖关系检查
- 使用规划算法优化顺序
问题3：计划过于僵化
表现：无法适应执行过程中的变化
解决方案：
- 实现动态规划能力
- 添加计划修正机制
- 设计灵活的执行策略
问题4：执行反馈缺失
表现：无法获取执行结果的反馈
解决方案：
- 完善工具的返回值设计
- 添加执行日志记录
- 实现反馈分析机制
任务规划的最佳实践
任务规划流程
1. 目标理解
- 分析用户目标
- 识别关键约束
- 确定成功标准
2. 目标拆解
- 将目标分解为子目标
- 检查子目标的可执行性
- 确定子目标顺序
3. 步骤生成
- 为每个子目标生成执行步骤
- 选择合适的工具
- 准备工具参数
4. 计划执行
- 按步骤执行
- 记录执行反馈
- 处理异常情况
5. 计划修正
- 分析执行反馈
- 选择修正策略
- 更新执行计划
6. 结果总结
- 汇总执行结果
- 生成最终报告
- 提供完成确认评估指标
| 指标 | 定义 | 目标值 |
|---|---|---|
| 目标达成率 | 成功完成目标的比例 | > 85% |
| 计划执行率 | 按计划执行的步骤比例 | > 90% |
| 修正次数 | 平均每次任务的修正次数 | < 2 |
| 执行效率 | 完成任务的平均时间 | 根据任务类型 |
工具选择建议
| 工具 | 说明 | 适用场景 |
|---|---|---|
| LangChain | 提供链式调用和计划能力 | 通用 Agent 开发 |
| LlamaIndex | 提供索引和查询能力 | RAG + Agent |
| AutoGPT | 提供自主规划和执行能力 | 复杂任务自动化 |
| CrewAI | 提供多 Agent 协作能力 | 多 Agent 场景 |
项目判断清单
- 任务需要多步骤 → 实现目标拆解
- 需要动态调整计划 → 实现计划修正机制
- 需要调用多个工具 → 设计步骤生成策略
- 需要处理异常情况 → 添加执行反馈和重试
- 需要长期任务支持 → 实现状态持久化
- 需要复杂决策 → 使用规划算法
- 需要用户干预 → 添加人工确认节点
- 需要监控 → 记录任务执行日志和指标