2026_DesignAI/officefile/supplements/03-autonomous-design/03.5-planning-and-execution.md

# 03.5 规划与执行

## 核心问题

> Agent如何将复杂目标分解为可执行步骤？
> 当环境变化时，如何动态调整计划？
> 如何平衡规划深度与执行效率？

---

## 概念讲解

### 规划的基本概念

**规划 (Planning)** 是寻找从初始状态到目标状态的行动序列的过程：

```
规划问题的基本要素

    状态空间 (State Space)
         │
         ├─── 初始状态: S₀
         ├─── 目标状态: S_goal
         └─── 中间状态: S₁, S₂, ..., Sₙ
                      │
                      ↓
    行动 (Actions)
         │
         ├─── 前置条件: preconditions
         ├─── 效果: effects
         └─── 代价: cost
                      │
                      ↓
    规划 = 行动序列 [a₁, a₂, ..., aₙ]
            使得: S₀ ─a₁→ S₁ ─a₂→ S₂ ... ─aₙ→ S_goal
```

### 规划与执行的循环

```
    ┌─────────────────────────────────────────────────────┐
    │                  规划-执行循环                       │
    │                  (Plan-Execute Loop)                 │
    ├─────────────────────────────────────────────────────┤
    │                                                      │
    │    ┌─────────┐                                       │
    │    │  目标   │                                       │
    │    └────┬────┘                                       │
    │         │                                            │
    │         ↓                                            │
    │    ┌─────────┐      ┌─────────────┐                 │
    │    │  规划   │ ───→ │  执行步骤   │                 │
    │    │  Planner│      │  Executor   │                 │
    │    └────┬────┘      └──────┬──────┘                 │
    │         │                  │                         │
    │         │                  ↓                         │
    │         │         ┌─────────────┐                    │
    │         │         │  观察结果   │                    │
    │         │         └──────┬──────┘                    │
    │         │                  │                         │
    │         │                  ↓                         │
    │         │         ┌─────────────┐                    │
    │         │         │  监控状态   │                    │
    │         │         └──────┬──────┘                    │
    │         │                  │                         │
    │         ↓                  ↓                         │
    │    ┌─────────────────────────────┐                   │
    │    │      是否需要重新规划？      │                   │
    │    └─────────────┬───────────────┘                   │
    │          是             否                            │
    │          │              │                            │
    │          └──────────────┘                            │
    │                 │                                     │
    │                 ↓                                     │
    │         ┌─────────────┐                              │
    │         │  继续执行   │                              │
    │         └─────────────┘                              │
    │                                                      │
    └─────────────────────────────────────────────────────┘
```

### 规划方法的分类

```
规划方法谱系

    ┌────────────────────────────────────────────────────────┐
    │                                                        │
    │  前向搜索 (Forward Search)                             │
    │  ────────────────────────                              │
    │  从初始状态向目标状态搜索                               │
    │  适合：目标明确，分支因子较小                           │
    │                                                        │
    │           ↓                                            │
    │                                                        │
    │  后向搜索 (Backward Search)                            │
    │  ───────────────────────                               │
    │  从目标状态向初始状态搜索                               │
    │  适合：目标状态较少，起始状态较多                       │
    │                                                        │
    │           ↓                                            │
    │                                                        │
    │  双向搜索 (Bidirectional Search)                       │
    │  ──────────────────────────────                        │
    │  同时从两端搜索，在中间汇合                             │
    │  适合：状态空间大，双向都可搜索                         │
    │                                                        │
    │           ↓                                            │
    │                                                        │
    │  分层规划 (Hierarchical Planning)                      │
    │  ─────────────────────────────                         │
    │  先规划高层抽象，再细化具体步骤                         │
    │  适合：复杂任务，有多层抽象                             │
    │                                                        │
    └────────────────────────────────────────────────────────┘
```

---

## 设计原理

### 前向搜索规划

```python
from typing import Callable, List, Dict, Any, Optional, Tuple
from dataclasses import dataclass
from enum import Enum
import heapq
from collections import deque


class PlanningStatus(Enum):
    """规划状态"""
    SUCCESS = "success"
    FAILURE = "failure"
    IN_PROGRESS = "in_progress"
    NO_PLAN = "no_plan"


@dataclass
class Action:
    """行动定义"""
    name: str
    preconditions: Callable[[Any], bool]  # 前置条件检查
    effects: Callable[[Any], Any]  # 状态转换
    cost: float = 1.0  # 行动代价


@dataclass
class Plan:
    """计划"""
    actions: List[Action]
    expected_final_state: Any
    total_cost: float
    steps: List[str]


class ForwardSearchPlanner:
    """
    前向搜索规划器

    从初始状态开始，逐步应用行动直到达到目标
    """

    def __init__(self,
                 actions: List[Action],
                 goal_test: Callable[[Any], bool],
                 max_depth: int = 100,
                 heuristic: Callable[[Any], float] = None):
        """
        Args:
            actions: 可用行动列表
            goal_test: 目标测试函数
            max_depth: 最大搜索深度
            heuristic: 启式函数（估计到目标的距离）
        """
        self.actions = actions
        self.goal_test = goal_test
        self.max_depth = max_depth
        self.heuristic = heuristic or (lambda s: 0)

        # 搜索统计
        self.nodes_expanded = 0
        self.nodes_visited = 0

    def plan(self, initial_state: Any) -> Optional[Plan]:
        """
        执行前向搜索规划

        支持的搜索算法：
        - BFS（无启发式）
        - UCS（uniform cost search，无启发式但有代价）
        - A*（有启发式）
        """
        # 搜索节点：(f_score, g_score, state, action_sequence)
        initial_node = (
            self.heuristic(initial_state),
            0,
            initial_state,
            []
        )

        open_set = [initial_node]
        closed_set = set()

        while open_set:
            # 获取最优节点
            f, g, current_state, action_sequence = heapq.heappop(open_set)

            # 检查是否已访问
            state_hash = self._hash_state(current_state)
            if state_hash in closed_set:
                continue
            closed_set.add(state_hash)
            self.nodes_visited += 1

            # 检查是否达到目标
            if self.goal_test(current_state):
                return Plan(
                    actions=action_sequence,
                    expected_final_state=current_state,
                    total_cost=g,
                    steps=[a.name for a in action_sequence]
                )

            # 深度限制
            if len(action_sequence) >= self.max_depth:
                continue

            # 扩展节点
            self.nodes_expanded += 1
            for action in self.actions:
                # 检查前置条件
                if action.preconditions(current_state):
                    # 应用行动
                    new_state = action.effects(current_state)

                    # 创建新节点
                    new_g = g + action.cost
                    new_f = new_g + self.heuristic(new_state)
                    new_sequence = action_sequence + [action]

                    heapq.heappush(open_set, (
                        new_f,
                        new_g,
                        new_state,
                        new_sequence
                    ))

        return None

    def _hash_state(self, state: Any) -> int:
        """状态哈希（用于去重）"""
        return hash(str(state))


class BackwardSearchPlanner:
    """
    后向搜索规划器

    从目标状态开始，反向应用行动直到回到初始状态
    """

    def __init__(self,
                 actions: List[Action],
                 initial_state: Any,
                 max_depth: int = 100):
        """
        Args:
            actions: 可用行动列表
            initial_state: 初始状态
            max_depth: 最大搜索深度
        """
        self.actions = actions
        self.initial_state = initial_state
        self.max_depth = max_depth

    def plan(self, goal_state: Any) -> Optional[Plan]:
        """
        执行后向搜索规划

        反向行动需要能够"撤销"原行动的效果
        """
        # 构建反向行动
        reverse_actions = self._build_reverse_actions()

        # 从目标状态搜索到初始状态
        planner = ForwardSearchPlanner(
            actions=reverse_actions,
            goal_test=lambda s: s == self.initial_state,
            max_depth=self.max_depth
        )

        result = planner.plan(goal_state)

        if result:
            # 反转行动序列
            result.actions = list(reversed(result.actions))
            result.steps = list(reversed(result.steps))

        return result

    def _build_reverse_actions(self) -> List[Action]:
        """构建反向行动（简化实现）"""
        # 实际实现需要更复杂的逻辑
        # 这里假设行动是可逆的
        return self.actions
```

### 分层规划

```python
class HierarchicalTask:
    """分层任务"""

    def __init__(self, name: str,
                 is_primitive: bool = False,
                 subtasks: List['HierarchicalTask'] = None,
                 implementation: Callable = None):
        self.name = name
        self.is_primitive = is_primitive
        self.subtasks = subtasks or []
        self.implementation = implementation


class HierarchicalPlanner:
    """
    分层规划器 (HTN - Hierarchical Task Network)

    将复杂任务分解为可执行的原子任务
    """

    def __init__(self, root_task: HierarchicalTask):
        self.root_task = root_task
        self.task_hierarchy = self._build_hierarchy(root_task)

    def _build_hierarchy(self, task: HierarchicalTask,
                         level: int = 0) -> Dict:
        """构建任务层次结构"""
        return {
            'task': task,
            'level': level,
            'children': [
                self._build_hierarchy(t, level + 1)
                for t in task.subtasks
            ]
        }

    def plan(self, initial_state: Dict) -> List[Action]:
        """
        执行分层规划

        1. 从根任务开始
        2. 递归分解非原子任务
        3. 收集所有原子任务
        """
        execution_plan = []
        self._decompose_task(self.root_task, initial_state, execution_plan)
        return execution_plan

    def _decompose_task(self, task: HierarchicalTask,
                        state: Dict, plan: List) -> bool:
        """分解任务"""
        if task.is_primitive:
            # 原子任务，直接执行
            if task.implementation:
                result = task.implementation(state)
                plan.append(Action(
                    name=task.name,
                    preconditions=lambda s: True,
                    effects=lambda s: result,
                    cost=1.0
                ))
                return True
            return False

        # 非原子任务，递归分解子任务
        for subtask in task.subtasks:
            if not self._decompose_task(subtask, state, plan):
                return False

        return True

    def visualize_hierarchy(self) -> str:
        """可视化任务层次"""
        lines = []

        def print_node(node, prefix="", is_last=True):
            connector = "└── " if is_last else "├── "
            lines.append(f"{prefix}{connector}{node['task'].name}")

            children = node['children']
            for i, child in enumerate(children):
                is_last_child = (i == len(children) - 1)
                extension = "    " if is_last else "│   "
                print_node(child, prefix + extension, is_last_child)

        print_node(self.task_hierarchy)
        return "\n".join(lines)
```

### 动态重规划

```python
class ReplanningAgent:
    """
    支持动态重规划的Agent

    特点：
    1. 持续监控执行状态
    2. 检测计划失效
    3. 触发重新规划
    """

    def __init__(self, planner, monitor_interval: float = 1.0):
        self.planner = planner
        self.monitor_interval = monitor_interval

        self.current_plan: Optional[Plan] = None
        self.executed_steps: List[str] = []
        self.plan_status = PlanningStatus.IN_PROGRESS

    def execute_with_monitoring(self,
                                 initial_state: Any,
                                 environment) -> Any:
        """
        带监控的执行

        Args:
            initial_state: 初始状态
            environment: 环境接口（支持 step() 和 get_state()）
        """
        # 初始规划
        self.current_plan = self.planner.plan(initial_state)

        if not self.current_plan:
            self.plan_status = PlanningStatus.NO_PLAN
            return None

        current_state = initial_state

        # 执行-监控循环
        for action in self.current_plan.actions:
            print(f"执行: {action.name}")

            # 执行行动
            try:
                result = environment.execute(action.name, current_state)
                self.executed_steps.append(action.name)

                # 更新状态
                current_state = environment.get_state()

                # 检查是否需要重新规划
                if self._should_replan(current_state, action, result):
                    print("检测到计划失效，重新规划...")
                    self._replan(current_state)

                # 检查计划是否完成
                if self._plan_complete(current_state):
                    self.plan_status = PlanningStatus.SUCCESS
                    break

            except Exception as e:
                print(f"执行失败: {e}")
                self._replan(current_state)

        return current_state

    def _should_replan(self, state: Any,
                       last_action: Action,
                       result: Any) -> bool:
        """判断是否需要重新规划"""
        # 检查1：行动结果是否符合预期
        if not result.get('success', True):
            return True

        # 检查2：状态是否发生意外变化
        if result.get('unexpected_change', False):
            return True

        # 检查3：目标是否已改变
        if result.get('goal_changed', False):
            return True

        return False

    def _replan(self, current_state: Any) -> None:
        """执行重新规划"""
        # 保存已执行的步骤
        executed = self.executed_steps.copy()

        # 重新规划
        new_plan = self.planner.plan(current_state)

        if new_plan:
            self.current_plan = new_plan
            print(f"新计划: {' → '.join(new_plan.steps)}")
        else:
            print("无法找到新计划")
            self.plan_status = PlanningStatus.FAILURE

    def _plan_complete(self, state: Any) -> bool:
        """检查计划是否完成"""
        return self.planner.goal_test(state)
```

---

## 代码示例

### 空间分析任务规划器

```python
"""
空间分析任务规划器

演示如何为GIS分析任务创建分层规划系统
"""
from typing import List, Dict, Any, Optional, Callable
from dataclasses import dataclass, field
from enum import Enum
import json


class TaskType(Enum):
    """任务类型"""
    DATA_PREPARATION = "data_preparation"
    ANALYSIS = "analysis"
    VISUALIZATION = "visualization"
    EXPORT = "export"


class TaskStatus(Enum):
    """任务状态"""
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    COMPLETED = "completed"
    FAILED = "failed"
    SKIPPED = "skipped"


@dataclass
class Task:
    """任务定义"""
    id: str
    name: str
    type: TaskType
    description: str = ""
    depends_on: List[str] = field(default_factory=list)
    parameters: Dict[str, Any] = field(default_factory=dict)
    status: TaskStatus = TaskStatus.PENDING
    result: Any = None
    error: Optional[str] = None

    def is_ready(self, completed_tasks: set) -> bool:
        """检查任务是否准备就绪（依赖已完成）"""
        return all(dep in completed_tasks for dep in self.depends_on)

    def to_dict(self) -> Dict:
        """转换为字典"""
        return {
            'id': self.id,
            'name': self.name,
            'type': self.type.value,
            'description': self.description,
            'depends_on': self.depends_on,
            'parameters': self.parameters,
            'status': self.status.value,
            'error': self.error
        }


class SpatialAnalysisPlanner:
    """
    空间分析任务规划器

    功能：
    1. 定义任务依赖关系
    2. 生成执行计划
    3. 执行任务序列
    4. 处理失败和重试
    """

    def __init__(self, name: str = "Spatial Analysis"):
        self.name = name
        self.tasks: Dict[str, Task] = {}
        self.execution_history: List[Dict] = []

    def add_task(self,
                 task_id: str,
                 name: str,
                 task_type: TaskType,
                 description: str = "",
                 depends_on: List[str] = None,
                 parameters: Dict = None) -> 'SpatialAnalysisPlanner':
        """添加任务"""
        self.tasks[task_id] = Task(
            id=task_id,
            name=name,
            type=task_type,
            description=description,
            depends_on=depends_on or [],
            parameters=parameters or {}
        )
        return self

    def get_execution_plan(self) -> List[List[str]]:
        """
        获取执行计划（分层级）

        返回每层可并行执行的任务ID列表
        """
        plan = []
        completed = set()
        remaining = set(self.tasks.keys())

        while remaining:
            # 找出所有准备就绪的任务
            ready = [
                task_id for task_id in remaining
                if self.tasks[task_id].is_ready(completed)
            ]

            if not ready:
                # 循环依赖
                raise ValueError("检测到循环依赖或无法满足的依赖")

            plan.append(ready)
            completed.update(ready)
            remaining -= set(ready)

        return plan

    def execute(self,
                executor: Callable[[Task], Any],
                max_retries: int = 1) -> Dict[str, Any]:
        """
        执行计划

        Args:
            executor: 任务执行器函数
            max_retries: 最大重试次数

        Returns:
            执行结果摘要
        """
        plan = self.get_execution_plan()
        results = {}
        completed = set()

        for level, task_ids in enumerate(plan):
            print(f"\n=== 执行层级 {level + 1}/{len(plan)} ===")
            print(f"任务: {', '.join(task_ids)}")

            # 可以并行执行（这里简化为顺序）
            for task_id in task_ids:
                task = self.tasks[task_id]

                for attempt in range(max_retries + 1):
                    try:
                        print(f"  执行: {task.name} (尝试 {attempt + 1})")
                        task.status = TaskStatus.IN_PROGRESS

                        # 执行任务
                        result = executor(task)
                        task.result = result
                        task.status = TaskStatus.COMPLETED
                        completed.add(task_id)
                        results[task_id] = result

                        # 记录历史
                        self.execution_history.append({
                            'task_id': task_id,
                            'status': 'completed',
                            'attempt': attempt + 1
                        })

                        break

                    except Exception as e:
                        error_msg = str(e)
                        task.error = error_msg

                        if attempt < max_retries:
                            print(f"    失败，重试: {error_msg}")
                        else:
                            task.status = TaskStatus.FAILED
                            print(f"    最终失败: {error_msg}")

                            self.execution_history.append({
                                'task_id': task_id,
                                'status': 'failed',
                                'error': error_msg,
                                'attempts': attempt + 1
                            })

                            # 决定是否继续
                            if task_id in self._get_critical_tasks():
                                print("关键任务失败，终止执行")
                                return results

        return results

    def visualize_plan(self) -> str:
        """可视化执行计划（DAG）"""
        plan = self.get_execution_plan()

        lines = [f"\n{self.name} - 执行计划"]
        lines.append("=" * 50)

        for level, task_ids in enumerate(plan):
            lines.append(f"\n层级 {level + 1}:")
            for task_id in task_ids:
                task = self.tasks[task_id]
                deps = f" (依赖: {', '.join(task.depends_on)})" if task.depends_on else ""
                lines.append(f"  - {task.name}{deps}")

        return "\n".join(lines)

    def _get_critical_tasks(self) -> set:
        """获取关键任务（失败会终止整个流程）"""
        # 简化实现：所有数据准备任务是关键的
        return {
            t.id for t in self.tasks.values()
            if t.type == TaskType.DATA_PREPARATION
        }


# ==================== 预定义分析工作流 ====================

class CommonAnalysisWorkflows:
    """常见空间分析工作流模板"""

    @staticmethod
    def ecological_network_analysis() -> SpatialAnalysisPlanner:
        """
        生态网络分析工作流

        六阶段：
        1. 数据准备
        2. 源地识别
        3. 阻力面构建
        4. MCR分析
        5. 廊道提取
        6. 结果输出
        """
        planner = SpatialAnalysisPlanner("生态网络分析")

        # 数据准备阶段
        planner.add_task("load_landcover", "加载土地覆盖数据",
                        TaskType.DATA_PREPARATION,
                        "加载研究区土地覆盖栅格数据")
        planner.add_task("load_elevation", "加载高程数据",
                        TaskType.DATA_PREPARATION,
                        "加载DEM高程数据")

        # 分析阶段
        planner.add_task("identify_sources", "识别生态源地",
                        TaskType.ANALYSIS,
                        "基于形态空间格局分析识别核心生境斑块",
                        depends_on=["load_landcover"])

        planner.add_task("build_resistance", "构建生态阻力面",
                        TaskType.ANALYSIS,
                        "基于土地覆盖类型赋值构建阻力面",
                        depends_on=["load_landcover", "load_elevation"])

        planner.add_task("mcr_analysis", "最小累积阻力分析",
                        TaskType.ANALYSIS,
                        "计算从各源地到空间各点的最小累积阻力",
                        depends_on=["identify_sources", "build_resistance"])

        planner.add_task("extract_corridors", "提取生态廊道",
                        TaskType.ANALYSIS,
                        "基于MCR结果提取潜在生态廊道",
                        depends_on=["mcr_analysis"])

        # 输出阶段
        planner.add_task("visualize", "结果可视化",
                        TaskType.VISUALIZATION,
                        "生成源地、阻力面、廊道的可视化地图",
                        depends_on=["extract_corridors"])

        planner.add_task("export_results", "导出分析结果",
                        TaskType.EXPORT,
                        "导出矢量数据和统计报告",
                        depends_on=["extract_corridors"])

        return planner

    @staticmethod
    def site_selection_analysis() -> SpatialAnalysisPlanner:
        """
        选址分析工作流

        1. 加载约束数据
        2. 加载候选地块
        3. 叠加分析
        4. 适宜性评价
        5. 最优选址
        6. 结果导出
        """
        planner = SpatialAnalysisPlanner("设施选址分析")

        planner.add_task("load_constraints", "加载约束数据",
                        TaskType.DATA_PREPARATION,
                        "加载坡度、保护区、道路等约束数据")

        planner.add_task("load_candidates", "加载候选地块",
                        TaskType.DATA_PREPARATION,
                        "加载候选地块矢量数据")

        planner.add_task("overlay_analysis", "叠加分析",
                        TaskType.ANALYSIS,
                        "将约束数据叠加到候选地块",
                        depends_on=["load_constraints", "load_candidates"])

        planner.add_task("suitability", "适宜性评价",
                        TaskType.ANALYSIS,
                        "基于多准则评价计算适宜性得分",
                        depends_on=["overlay_analysis"])

        planner.add_task("select_optimal", "最优选址",
                        TaskType.ANALYSIS,
                        "选择得分最高的地块",
                        depends_on=["suitability"])

        planner.add_task("export", "导出选址方案",
                        TaskType.EXPORT,
                        "导出选址结果和评价报告",
                        depends_on=["select_optimal"])

        return planner


# ==================== 演示程序 ====================

def demonstrate_spatial_planning():
    """演示空间分析规划系统"""
    print("=" * 70)
    print("空间分析任务规划器演示")
    print("=" * 70)

    # 1. 创建生态网络分析工作流
    print("\n1. 创建生态网络分析工作流...")
    planner = CommonAnalysisWorkflows.ecological_network_analysis()

    # 2. 可视化执行计划
    print("\n2. 执行计划DAG:")
    print(planner.visualize_plan())

    # 3. 获取执行计划
    print("\n3. 执行层级:")
    execution_plan = planner.get_execution_plan()
    for i, level in enumerate(execution_plan, 1):
        task_names = [planner.tasks[t].name for t in level]
        print(f"  层级 {i}: {', '.join(task_names)}")

    # 4. 模拟执行
    print("\n4. 模拟执行:")

    def mock_executor(task: Task) -> Dict:
        """模拟任务执行"""
        import time
        import random

        time.sleep(0.1)  # 模拟执行时间

        # 模拟偶尔失败
        if random.random() < 0.1:
            raise Exception("模拟随机失败")

        return {
            'task': task.name,
            'status': 'success',
            'output': f"{task.name}_output"
        }

    results = planner.execute(mock_executor, max_retries=2)

    # 5. 统计结果
    print("\n5. 执行统计:")
    status_count = {}
    for task in planner.tasks.values():
        status = task.status.value
        status_count[status] = status_count.get(status, 0) + 1

    for status, count in status_count.items():
        print(f"  {status}: {count}")

    # 6. 创建自定义工作流
    print("\n6. 创建自定义工作流:")

    custom_planner = SpatialAnalysisPlanner("自定义分析")
    custom_planner.add_task("t1", "数据加载", TaskType.DATA_PREPARATION)
    custom_planner.add_task("t2", "数据清洗", TaskType.ANALYSIS, depends_on=["t1"])
    custom_planner.add_task("t3", "统计分析", TaskType.ANALYSIS, depends_on=["t2"])
    custom_planner.add_task("t4", "并行分析A", TaskType.ANALYSIS, depends_on=["t2"])
    custom_planner.add_task("t5", "并行分析B", TaskType.ANALYSIS, depends_on=["t2"])
    custom_planner.add_task("t6", "结果汇总", TaskType.ANALYSIS,
                           depends_on=["t3", "t4", "t5"])

    print(custom_planner.visualize_plan())


if __name__ == "__main__":
    demonstrate_spatial_planning()
```

### RL风格的规划执行

```python
class ReinforcementLearningPlanner:
    """
    基于强化学习思想的规划执行

    特点：
    1. 从执行中学习
    2. 评估行动效果
    3. 更新策略
    """

    def __init__(self, actions: List[Action],
                 reward_fn: Callable[[Any, Any, Any], float]):
        """
        Args:
            actions: 可用行动
            reward_fn: 奖励函数 (state, action, next_state) -> reward
        """
        self.actions = actions
        self.reward_fn = reward_fn

        # Q值表：state -> {action: q_value}
        self.q_table: Dict[str, Dict[str, float]] = {}

        # 学习参数
        self.learning_rate = 0.1
        self.discount_factor = 0.9
        self.epsilon = 0.1  # 探索率

    def select_action(self, state: Any, training: bool = True) -> Action:
        """
        选择行动（epsilon-greedy策略）

        Args:
            state: 当前状态
            training: 是否在训练模式
        """
        state_key = self._state_key(state)

        # 初始化Q值
        if state_key not in self.q_table:
            self.q_table[state_key] = {a.name: 0.0 for a in self.actions}

        # epsilon-greedy
        import random
        if training and random.random() < self.epsilon:
            # 探索：随机选择
            return random.choice(self.actions)
        else:
            # 利用：选择Q值最大的
            q_values = self.q_table[state_key]
            best_action_name = max(q_values, key=q_values.get)
            return next(a for a in self.actions if a.name == best_action_name)

    def update_q_value(self, state: Any, action: Action,
                       reward: float, next_state: Any) -> None:
        """
        更新Q值（Q-learning更新规则）

        Q(s,a) ← Q(s,a) + α[r + γ max Q(s',a') - Q(s,a)]
        """
        state_key = self._state_key(state)
        next_state_key = self._state_key(next_state)

        # 确保初始化
        if next_state_key not in self.q_table:
            self.q_table[next_state_key] = {
                a.name: 0.0 for a in self.actions
            }

        # 计算新的Q值
        current_q = self.q_table[state_key][action.name]
        max_next_q = max(self.q_table[next_state_key].values())

        new_q = current_q + self.learning_rate * (
            reward + self.discount_factor * max_next_q - current_q
        )

        self.q_table[state_key][action.name] = new_q

    def learn_episode(self, environment,
                      max_steps: int = 100) -> float:
        """
        执行一个学习episode

        Returns:
            总奖励
        """
        state = environment.reset()
        total_reward = 0

        for _ in range(max_steps):
            # 选择行动
            action = self.select_action(state, training=True)

            # 执行行动
            next_state, done = environment.step(action)

            # 计算奖励
            reward = self.reward_fn(state, action, next_state)
            total_reward += reward

            # 更新Q值
            self.update_q_value(state, action, reward, next_state)

            state = next_state

            if done:
                break

        return total_reward

    def _state_key(self, state: Any) -> str:
        """生成状态键（简化）"""
        return str(hash(str(state)))
```

---

## 案例分析

### LangGraph的规划执行

```python
"""
LangGraph风格的规划执行

LangGraph使用状态图来表示和执行复杂的Agent工作流
"""

from typing import TypedDict, Annotated, Sequence
import operator


class AgentState(TypedDict):
    """Agent状态定义"""
    messages: Annotated[Sequence[str], operator.add]
    current_plan: list[str]
    executed_steps: list[str]
    requires_replan: bool


def should_continue(state: AgentState) -> str:
    """
    条件边：决定继续执行还是结束

    类似LangGraph的条件边
    """
    if not state["current_plan"]:
        return "end"
    if state["requires_replan"]:
        return "replan"
    return "continue"


class LangGraphStylePlanner:
    """LangGraph风格的规划器"""

    def __init__(self):
        self.state = AgentState(
            messages=[],
            current_plan=[],
            executed_steps=[],
            requires_replan=False
        )
        self.nodes = {
            "plan": self._plan_node,
            "execute": self._execute_node,
            "observe": self._observe_node,
            "replan": self._replan_node
        }

    def build_graph(self) -> Dict:
        """构建执行图"""
        return {
            "nodes": self.nodes,
            "edges": {
                "plan": "execute",
                "execute": "observe",
                "observe": should_continue,
                "continue": "execute",
                "replan": "execute"
            }
        }

    def _plan_node(self, state: AgentState) -> AgentState:
        """规划节点：生成初始计划"""
        state["current_plan"] = ["step1", "step2", "step3"]
        state["messages"].append("计划已生成")
        return state

    def _execute_node(self, state: AgentState) -> AgentState:
        """执行节点：执行下一步"""
        if state["current_plan"]:
            step = state["current_plan"].pop(0)
            state["executed_steps"].append(step)
            state["messages"].append(f"已执行: {step}")
        return state

    def _observe_node(self, state: AgentState) -> AgentState:
        """观察节点：检查是否需要重新规划"""
        # 检查执行结果
        state["requires_replan"] = False  # 简化
        return state

    def _replan_node(self, state: AgentState) -> AgentState:
        """重规划节点：调整计划"""
        state["current_plan"] = ["new_step"] + state["current_plan"]
        state["requires_replan"] = False
        state["messages"].append("计划已更新")
        return state

    def run(self, initial_goal: str) -> AgentState:
        """运行整个图"""
        self.state["messages"].append(f"目标: {initial_goal}")

        graph = self.build_graph()
        current_node = "plan"

        while current_node != "end":
            # 执行节点
            self.state = self.nodes[current_node](self.state)

            # 获取下一个节点
            next_node = graph["edges"].get(current_node)
            if callable(next_node):
                next_node = next_node(self.state)
            current_node = next_node

        return self.state
```

---

## 反思与延伸

### 思考问题

1. **规划深度**：多深的规划是合适的？过度规划会有什么问题？

2. **不确定性**：如何在不确定环境中进行规划？

3. **多Agent协调**：多个Agent如何协调各自的规划？

4. **规划评估**：如何评估一个规划的质量？

### 延伸阅读

- **"Planning Algorithms"** (LaValle) - 规划算法权威教材
- **"Hierarchical Planning"** - 分层规划专题
- **"Reinforcement Learning"** (Sutton & Barto) - RL中的规划与学习

---

## 关键要点

1. **规划**是寻找从初始状态到目标的行动序列
2. **前向搜索**从初始状态向目标搜索，适合目标明确的场景
3. **后向搜索**从目标向初始状态搜索，适合目标较少的场景
4. **分层规划**将复杂任务分解，适合复杂多阶段任务
5. **重规划**是应对环境变化的关键机制
6. **执行-监控-调整**循环是实际系统的核心模式
7. **任务依赖管理**使用DAG表达和并行化执行