本文目录一、前言二、工具概览三、工具详解3.1 任务分解器3.2 计划生成器3.3 执行验证器3.4 计划管理器四、实战案例五、工具链整合六、总结一、前言1.1 工具化的规划有了理论基础我们来动手实现。规划能力是Agent从执行指令到主动规划的关键升级。把规划功能拆分为4个独立工具灵活组合使用可以单独使用任务分解器来拆解复杂问题可以用工具链完成分解-生成-验证-管理完整流程可以集成到Agent中让Agent自主规划任务1.2 你将学到什么✅ 实现任务分解工具✅ 实现计划生成工具✅ 实现执行验证工具✅ 实现计划管理工具二、工具概览2.1 4个工具工具功能任务分解器把复杂任务拆成子任务计划生成器为子任务制定详细计划执行验证器验证计划的可行性计划管理器管理和调整计划2.2 文件结构07_planning/ ├── shared_planning.py # 共享存储 ├── tool_1_task_decomposer.py # 工具1 ├── tool_2_plan_generator.py # 工具2 ├── tool_3_execution_validator.py # 工具3 ├── tool_4_plan_manager.py # 工具4 ├── planning_chain_demo.py # 工具链演示 ├── planning_agent_demo.py # Agent演示 └── sample_planning_tasks.json # 示例数据三、工具详解3.1 任务分解器功能说明把复杂任务智能拆分为多个子任务。支持两种分解模式简单分解平级子任务和分层分解多层子任务。3.2 完整代码from langchain.tools import tool import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from shared_planning import add_plan tool def task_decomposer(task: str, decomposition_type: str simple): 任务分解器 - 将复杂任务拆分为子任务 参数: task: 要分解的复杂任务 decomposition_type: 分解类型 (simple/hierarchical) 返回: 分解后的子任务列表 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 初始化LLM try: llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, modelqwen2.5:3b-instruct, temperature0.7 ) except Exception as e: return f[ERROR] LLM初始化失败: {e} # 不同分解策略的提示 if decomposition_type simple: prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务分解专家。请将给定的复杂任务分解为3-5个清晰的子任务。 请用以下格式输出 【任务理解】 简要说明这个任务的目标 【子任务列表】 1. [子任务1名称] - [子任务1描述] 2. [子任务2名称] - [子任务2描述] 3. [子任务3名称] - [子任务3描述] ... 【执行顺序】 说明子任务应该按照什么顺序执行), (user, 任务: {task}) ]) else: prompt ChatPromptTemplate.from_messages([ (system, 你是一个分层任务分解专家。请将给定的复杂任务分解为多个层次的子任务。 请用以下格式输出 【第一层主任务】 - 主任务名称 【第二层子任务】 1. [子任务1] - 子子任务1.1 - 子子任务1.2 2. [子任务2] - 子子任务2.1 - 子子任务2.2 【执行顺序】 说明执行顺序), (user, 任务: {task}) ]) chain prompt | llm result chain.invoke({task: task}) # 解析子任务简化版 subtasks [] lines result.content.split(\n) for i, line in enumerate(lines): if line.strip().startswith((1., 2., 3., 4., 5.)): subtasks.append({step: len(subtasks)1, description: line.strip()}) # 保存计划 plan_id fplan_{len(get_all_plans())} if get_all_plans in dir() else fplan_{hash(task)} add_plan(plan_id, task, subtasks, decomposition_type) output [] output.append( * 80) output.append(【任务分解器】) output.append( * 80) output.append(f\n原始任务: {task}) output.append(f分解类型: {decomposition_type}) output.append(\n * 80) output.append(【分解结果】) output.append( * 80) output.append(result.content) output.append(\n * 80) output.append([TIP] 说明将复杂任务分解为可管理的小任务) output.append( * 80) return \n.join(output)3.3 使用示例result task_decomposer.invoke({ task: 周末在家里举办一个小型派对, decomposition_type: simple }) print(result)3.2 计划生成器功能说明为子任务制定详细的执行计划包括时间安排、负责人、具体步骤等。如果没有提供子任务会先调用任务分解器。完整代码from langchain.tools import tool import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from shared_planning import add_plan tool def plan_generator(task: str, subtasks: list None): 计划生成器 - 为子任务生成详细的执行计划 参数: task: 原始任务 subtasks: 子任务列表可选如不提供会先调用任务分解 返回: 详细执行计划 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 初始化LLM try: llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, modelqwen2.5:3b-instruct, temperature0.7 ) except Exception as e: return f[ERROR] LLM初始化失败: {e} # 如果没有子任务先生成 if not subtasks: from tool_1_task_decomposer import task_decomposer subtasks_text task_decomposer.invoke({task: task, decomposition_type: simple}) else: subtasks_text \n.join([f{s[step]}. {s[description]} for s in subtasks]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个计划制定专家。请为给定的任务和子任务生成详细的执行计划。 请用以下格式输出 【总体目标】 明确任务的最终目标 【执行时间表】 1. [子任务1] - [预计时间] - [负责人/工具] - 具体步骤 * 步骤1 * 步骤2 2. [子任务2] - [预计时间] - [负责人/工具] - 具体步骤 ... 【注意事项】 列出执行过程中需要注意的关键点 【成功标准】 说明如何判断任务成功完成), (user, 任务: {task}\n\n子任务:\n{subtasks_text}) ]) chain prompt | llm result chain.invoke({task: task, subtasks_text: subtasks_text}) output [] output.append( * 80) output.append(【计划生成器】) output.append( * 80) output.append(f\n任务: {task}) output.append(\n * 80) output.append(【详细执行计划】) output.append( * 80) output.append(result.content) output.append(\n * 80) output.append([TIP] 说明好的计划让执行更有条理) output.append( * 80) return \n.join(output)4.3 使用示例result plan_generator.invoke({task: 周末在家里举办一个小型派对}) print(result)3.3 执行验证器功能说明验证计划的可行性发现潜在风险提供改进建议。在执行前进行验证可以大大减少实际执行时的问题。完整代码from langchain.tools import tool import sys import os sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from shared_planning import get_all_plans tool def execution_validator(plan: str, task: str None): 执行验证器 - 验证计划的可行性并提供改进建议 参数: plan: 要验证的计划 task: 原始任务可选 返回: 验证结果和改进建议 from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate # 初始化LLM try: llm ChatOpenAI( base_urlhttp://localhost:11434/v1, api_keyollama, modelqwen2.5:3b-instruct, temperature0.7 ) except Exception as e: return f[ERROR] LLM初始化失败: {e} prompt ChatPromptTemplate.from_messages([ (system, 你是一个计划验证专家。请仔细分析给定的计划评估其可行性并提供改进建议。 请用以下格式输出 【可行性评估】 - 总体评分: [1-10分] - 评估理由: [详细说明] 【潜在风险】 列出执行过程中可能遇到的问题 【改进建议】 提供具体的改进建议 【优化后的计划】 基于以上分析提供一个优化后的版本), (user, 任务: {task}\n\n计划:\n{plan}) ]) chain prompt | llm result chain.invoke({task: task or 未指定, plan: plan}) output [] output.append( * 80) output.append(【执行验证器】) output.append( * 80) output.append(\n * 80) output.append(【验证结果】) output.append( * 80) output.append(result.content) output.append(\n * 80) output.append([TIP] 说明验证可以提前发现问题减少执行风险) output.append( * 80) return \n.join(output)5.3 使用示例test_plan 【子任务列表】 1. 邀请客人 2. 准备食物 3. 布置场地 4. 组织活动 【执行顺序】 先邀请再准备然后布置最后组织 result execution_validator.invoke({plan: test_plan, task: 举办派对}) print(result)3.4 计划管理器功能说明管理计划的生命周期查看、获取详情、摘要、清空、导出、调整等。方便追踪和管理多个计划。完整代码from langchain.tools import tool import sys import os import json sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from shared_planning import ( get_all_plans, get_all_comparisons, clear_plan_store, format_plan_summary ) tool def plan_manager(command: str, data: dict None): 计划管理器 - 管理和调整计划 参数: command: 命令类型 (list/get/summary/clear/export/adjust) data: 相关数据用于adjust命令 返回: 操作结果 if command list: # 列出所有计划 plans get_all_plans() output [ * 80, 【计划列表】, * 80] for i, plan in enumerate(plans, 1): output.append(f\n计划 {i}:) output.append(f ID: {plan[id]}) output.append(f 任务: {plan[task][:50]}...) output.append(f 类型: {plan[type]}) output.append(f 子任务数: {len(plan[subtasks])}) output.append(\n * 80) return \n.join(output) elif command get and data and plan_id in data: # 获取特定计划 plans get_all_plans() plan next((p for p in plans if p[id] data[plan_id]), None) if plan: output [ * 80, 【计划详情】, * 80] output.append(fID: {plan[id]}) output.append(f任务: {plan[task]}) output.append(f类型: {plan[type]}) output.append(f\n子任务:) for subtask in plan[subtasks]: output.append(f {subtask[step]}. {subtask[description]}) output.append(\n * 80) return \n.join(output) else: return f[ERROR] 未找到计划: {data[plan_id]} elif command summary: # 获取摘要 return format_plan_summary() elif command clear: # 清空 clear_plan_store() return [OK] 已清空所有计划 elif command export: # 导出 data { plans: get_all_plans(), comparisons: get_all_comparisons() } output_file output_plan_traces.json with open(output_file, w, encodingutf-8) as f: json.dump(data, f, ensure_asciiFalse, indent2) return f[OK] 已导出到 {output_file} elif command adjust and data: # 调整计划模拟 return f[OK] 计划已根据反馈调整: {data.get(feedback, 无反馈)} else: return f[ERROR] 未知命令: {command}\n可用命令: list/get/summary/clear/export/adjust6.3 使用示例# 获取摘要 result plan_manager.invoke({command: summary}) print(result) # 列出所有计划 result plan_manager.invoke({command: list}) print(result)四、实战案例4.1 工具链演示运行完整的工具链演示cd 07_planning python planning_chain_demo.py4.2 工具功能对比表工具核心功能适用场景优势任务分解器将复杂任务拆分为子任务任务规划初期简化复杂问题计划生成器为子任务制定详细计划执行前准备步骤清晰执行验证器验证计划可行性执行前检查降低风险计划管理器管理计划生命周期计划跟踪便于管理4.3 实际应用案例案例1组织派对任务分解任务周末在家里举办一个小型派对分解结果【任务理解】 在周末举办一个小型家庭派对需要邀请朋友、准备食物、布置场地、组织活动。 【子任务列表】 1. 邀请嘉宾 - 通过电话或消息邀请朋友确认出席人数 2. 准备食物 - 根据人数准备菜单采购食材 3. 布置场地 - 装饰房间准备桌椅 4. 组织活动 - 准备游戏和娱乐项目 【执行顺序】 先邀请嘉宾再准备食物然后布置场地最后组织活动代码调用from tool_1_task_decomposer import task_decomposer result task_decomposer.invoke({ task: 周末在家里举办一个小型派对, decomposition_type: simple }) print(result)案例2项目计划生成任务开发一个个人博客网站计划结果【任务计划】 任务: 开发一个个人博客网站 【详细计划】 阶段1: 需求分析1天 - 确定功能需求 - 设计页面结构 - 选择技术栈 阶段2: 环境搭建1天 - 配置开发环境 - 创建项目结构 - 初始化Git仓库 阶段3: 核心功能开发3天 - 用户认证模块 - 文章管理模块 - 评论系统 阶段4: 测试部署1天 - 单元测试 - 部署上线代码调用from tool_2_plan_generator import plan_generator result plan_generator.invoke({ task: 开发一个个人博客网站 }) print(result)案例3计划可行性验证计划3天完成一个大型项目验证结果【可行性评估】 - 总体评分: 6/10 - 主要风险: 时间紧张 【风险分析】 1. 时间不足 - 大型项目通常需要更多时间 2. 需求不明确 - 可能导致返工 【改进建议】 1. 增加时间预算至5-7天 2. 先进行需求评审 3. 制定详细的每日计划 【调整建议】 建议将项目拆分为多个阶段逐步推进代码调用from tool_3_execution_validator import execution_validator test_plan 阶段1: 需求分析1天 阶段2: 设计开发1天 阶段3: 测试上线1天 result execution_validator.invoke({ plan: test_plan, task: 完成大型项目 }) print(result)4.4 完整工具链调用示例我们将4个工具组合成完整的规划流程from tool_1_task_decomposer import task_decomposer from tool_2_plan_generator import plan_generator from tool_3_execution_validator import execution_validator from tool_4_plan_manager import plan_manager # 步骤1分解任务 task 开发一个个人博客网站 decomposition_result task_decomposer.invoke({ task: task, decomposition_type: simple }) print(decomposition_result) # 步骤2生成计划 plan_result plan_generator.invoke({task: task}) print(plan_result) # 步骤3验证计划 validation_result execution_validator.invoke({ plan: plan_result, task: task }) print(validation_result) # 步骤4管理计划 summary plan_manager.invoke({command: summary}) print(summary)五、工具链整合5.1 规划工具链架构用户任务 ↓ [任务分解器] 将复杂任务拆分为子任务 ↓ [计划生成器] 为子任务制定详细计划 ↓ [执行验证器] 验证计划可行性发现风险 ↓ [计划管理器] 存储、管理和调整计划 ↓ 执行阶段5.2 工具协作流程步骤工具作用1任务分解器拆分任务为子任务2计划生成器生成详细执行计划3执行验证器验证并优化计划4计划管理器存储和管理计划六、总结6.1 本文要点工具核心能力价值✅ 任务分解器将复杂任务拆分为子任务简化问题✅ 计划生成器制定详细执行计划步骤清晰✅ 执行验证器验证计划可行性降低风险✅ 计划管理器管理计划生命周期便于跟踪6.2 学习路径建议阶段任务目标基础理解每个工具的功能掌握单个工具使用进阶组合使用多个工具构建简单工具链高级集成到完整Agent实现智能规划系统点赞 关注更新不迷路