模块二，Agent规划模式的四个工具思考

📋 本文目录

• 一、前言

• 二、工具概览

• 三、工具详解

  • 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_url="http://localhost:11434/v1", api_key="ollama", model="qwen2.5:3b-instruct", temperature=0.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 = f"plan_{len(get_all_plans())}" if 'get_all_plans' in dir() else f"plan_{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_url="http://localhost:11434/v1", api_key="ollama", model="qwen2.5:3b-instruct", temperature=0.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_url="http://localhost:11434/v1", api_key="ollama", model="qwen2.5:3b-instruct", temperature=0.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(f"ID: {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", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) 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/adjust"

6.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.py

4.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 工具协作流程

六、总结

6.1 本文要点

6.2 学习路径建议

点赞 + 关注，更新不迷路！🚀