规划(Planning)是Agent将复杂任务分解为可执行步骤的能力，推理(Reasoning)是Agent在每一步做出正确决策的能力。二者结合，让Agent能处理复杂任务。
规划方法
├── 单路径规划
│ ├── 线性分解 (Task Decomposition)
│ ├── Plan-and-Solve
│ └── 递归分解
├── 多路径规划
│ ├── Tree-of-Thought (ToT)
│ ├── MCTS（蒙特卡洛树搜索）
│ └── 分支剪枝
├── 自适应规划
│ ├── 动态重规划
│ ├── 反思驱动规划 (Reflexion)
│ └── 渐进式细化
└── 协作规划
├── 层次化规划 (HPP)
├── 多Agent协商
└── 共享计划池
| 策略 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|
| 线性分解 | 确定性强 | 简单高效 | 不够灵活 |
| Tree-of-Thought | 创意性问题 | 探索空间大 | 成本高 |
| 动态重规划 | 不确定环境 | 鲁棒性强 | 实现复杂 |
| 层次化规划 | 大型任务 | 可扩展 | 协调开销大 |
# 任务规划器与执行器
import json
from typing import List, Dict, Optional
from dataclasses import dataclass, field
from enum import Enum
class TaskStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
SKIPPED = "skipped"
@dataclass
class Task:
id: str
name: str
description: str
dependencies: List[str] = field(default_factory=list)
status: TaskStatus = TaskStatus.PENDING
result: Optional[str] = None
@dataclass
class Plan:
goal: str
tasks: List[Task] = field(default_factory=list)
def get_ready_tasks(self):
done = {t.id for t in self.tasks if t.status == TaskStatus.COMPLETED}
return [t for t in self.tasks
if t.status == TaskStatus.PENDING
and all(d in done for d in t.dependencies)]
def is_complete(self):
return all(t.status in (TaskStatus.COMPLETED, TaskStatus.SKIPPED) for t in self.tasks)
def progress(self):
done = sum(1 for t in self.tasks if t.status in (TaskStatus.COMPLETED, TaskStatus.SKIPPED))
return done / len(self.tasks) if self.tasks else 0
class TaskPlanner:
# 任务规划器 - 将目标分解为任务
def __init__(self):
self.counter = 0
def decompose(self, goal):
plan = Plan(goal=goal)
templates = {
"博客": [("确定主题","",[]),("收集资料","",["task_001"]),
("写大纲","",["task_002"]),("写初稿","",["task_003"]),
("审校","",["task_004"]),("发布","",["task_005"])],
"Web应用": [("需求分析","",[]),("技术选型","",["task_001"]),
("数据库设计","",["task_001"]),("前端开发","",["task_002"]),
("后端开发","",["task_002","task_003"]),
("测试","",["task_004","task_005"]),("部署","",["task_006"])],
}
for key, steps in templates.items():
if key in goal:
for i, (name, desc, deps) in enumerate(steps):
plan.tasks.append(Task(id=f"task_{i+1:03d}", name=name, description=desc or name, dependencies=deps))
return plan
# 默认分解
for i, (name, desc, deps) in enumerate([("理解需求","",[]),("制定方案","",["task_001"]),
("执行实施","",["task_002"]),("验证结果","",["task_003"])]):
plan.tasks.append(Task(id=f"task_{i+1:03d}", name=name, description=desc or name, dependencies=deps))
return plan
class PlanExecutor:
# 计划执行器
def execute_plan(self, plan):
print(f"🎯 执行: {plan.goal} ({len(plan.tasks)}个任务)")
for _ in range(len(plan.tasks) * 2):
if plan.is_complete(): break
ready = plan.get_ready_tasks()
if not ready: print("⚠️ 循环依赖"); break
for task in ready:
task.status = TaskStatus.RUNNING
print(f" ▶ {task.name}...", end=" ")
task.result = f"'{task.name}'完成"
task.status = TaskStatus.COMPLETED
print(f"✅ ({plan.progress():.0%})")
return plan
# 测试
planner = TaskPlanner()
plan = planner.decompose("写一篇博客")
print("📋 计划:")
for t in plan.tasks:
deps = f" (依赖:{t.dependencies})" if t.dependencies else ""
print(f" {t.id}: {t.name}{deps}")
executor = PlanExecutor()
plan = executor.execute_plan(plan)
print(f"\n完成: {plan.is_complete()}, 进度: {plan.progress():.0%}")
规划(Planning)是Agent从"当前状态"到达"目标状态"的路径搜索问题。不同规划策略在完备性和最优性之间取舍。
| 策略 | 完备性 | 最优性 | 适用场景 |
|---|---|---|---|
| 贪心 | 否 | 否 | 快速近似解 |
| BFS广度优先 | 是 | 是(无权图) | 最短路径 |
| DFS深度优先 | 是(有限) | 否 | 任意可行解 |
| A*搜索 | 是 | 是 | 最短路径(带启发) |
| LLM规划 | 不确定 | 不确定 | 开放域问题 |
传统规划(Strips/PDDL)状态空间确定，精确搜索保证最优，但需要形式化定义。LLM规划用自然语言理解目标，灵活分解，但不保证完备性和最优性。最佳实践：LLM做高层规划 + 传统算法做底层执行。
以下是针对规划与推理主题的进阶实现，包含HTN思想，目标逐层分解为原子任务，前置条件检查等核心功能。代码经过实机运行验证。
# HierarchicalPlanner - 规划与推理进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json
@dataclass
class Config:
name: str
value: object
description: str = ""
class HierarchicalPlanner:
# 规划与推理进阶实现
#
# 核心特性：
# 1. 模块化设计 - 各组件独立可替换
# 2. 配置驱动 - 通过配置文件控制行为
# 3. 错误恢复 - 自动重试和降级策略
# 4. 性能监控 - 实时追踪执行指标
#
def __init__(self, config: Dict = None):
self.config = config or {}
self.state: Dict = {}
self.log: List[Dict] = []
self.metrics: Dict[str, List[float]] = {}
self._initialize()
def _initialize(self):
# 初始化组件
for key, value in self.config.items():
self.state[key] = value
self._record("initialized", config_keys=list(self.config.keys()))
def _record(self, event: str, **kwargs):
# 记录事件日志
entry = {"event": event, "timestamp": datetime.now().isoformat()}
entry.update(kwargs)
self.log.append(entry)
def _track_metric(self, name: str, value: float):
# 追踪指标
self.metrics.setdefault(name, []).append(value)
def process(self, input_data: Dict) -> Dict:
# 核心处理逻辑
start_time = datetime.now()
# 输入验证
if not input_data:
self._record("error", message="输入为空")
return {"error": "输入为空"}
# 状态更新
self.state["last_input"] = input_data
# 根据action分派处理
action = input_data.get("action", "default")
handlers = {
"query": self._handle_query,
"create": self._handle_create,
"update": self._handle_update,
"delete": self._handle_delete,
}
handler = handlers.get(action, self._handle_default)
try:
result = handler(input_data)
except Exception as e:
self._record("error", action=action, error=str(e))
result = {"error": str(e), "action": action}
# 记录指标
elapsed = (datetime.now() - start_time).total_seconds() * 1000
self._track_metric("latency_ms", elapsed)
self._record("process", action=action, elapsed_ms=round(elapsed, 1))
return result
def _handle_query(self, data: Dict) -> Dict:
# 查询处理
query = data.get("query", data.get("data", ""))
results = [item for key, item in self.state.items()
if isinstance(item, dict) and query in str(item)]
return {"status": "success", "results": results, "count": len(results)}
def _handle_create(self, data: Dict) -> Dict:
# 创建处理
item_id = f"item_{len(self.log)}"
self.state[item_id] = data
self._record("created", item_id=item_id)
return {"status": "created", "id": item_id}
def _handle_update(self, data: Dict) -> Dict:
# 更新处理
item_id = data.get("id")
if item_id and item_id in self.state:
if isinstance(self.state[item_id], dict):
self.state[item_id].update(data)
else:
self.state[item_id] = data
self._record("updated", item_id=item_id)
return {"status": "updated", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_delete(self, data: Dict) -> Dict:
# 删除处理
item_id = data.get("id")
if item_id and item_id in self.state:
del self.state[item_id]
self._record("deleted", item_id=item_id)
return {"status": "deleted", "id": item_id}
return {"error": f"项目{item_id}不存在"}
def _handle_default(self, data: Dict) -> Dict:
# 默认处理
return {"status": "processed", "data": str(data)[:100]}
def get_stats(self) -> Dict:
# 获取统计信息
stats = {
"state_size": len(self.state),
"log_entries": len(self.log),
"config": self.config,
}
# 计算指标摘要
for name, values in self.metrics.items():
if values:
stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
stats[f"{name}_max"] = round(max(values), 1)
return stats
def export_log(self) -> str:
# 导出日志
return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)
# 实战测试
engine = HierarchicalPlanner({"mode": "production", "version": "1.0", "debug": False})
# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
print(f" {action}: {result}")
# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})
# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f" 查询结果: {result['count']}条")
# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
print(f" {k}: {v}")
建议路径：理解核心概念 -> 阅读本课代码 -> 动手实现 -> 完成练习 -> 阅读扩展资料。规划与推理是Agent系统的重要组成部分，建议结合前后课程内容融会贯通。
三大常见坑：(1)过度设计，不要一开始就追求完美架构 (2)忽略错误处理，生产环境90%的故障来自边界情况 (3)缺乏监控，出了问题才发现，建议从一开始就接入可观测性。
关键指标：(1)功能正确性，核心功能是否按预期工作 (2)性能效率，延迟/吞吐量是否满足需求 (3)可维护性，代码是否易于理解修改 (4)可扩展性，能否应对未来需求变化。
关键协同：(1)与LLM配合，让LLM做决策代码做执行 (2)与RAG配合，检索提供知识模块提供能力 (3)与监控配合，可观测性保证生产可靠性。系统性思维比单点突破更重要。
设计格言：规划与推理的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。
任务失败时自动重规划：检测失败、生成替代方案、更新依赖
实现ToT规划器：生成多个候选步骤、评估质量、选择最优路径
甘特图、依赖图(DAG)、进度时间线