refactor(officefile): 按 md/latex/word 三层结构重组文档目录

将 Markdown 源文件移入 md/,LaTeX 工作目录保留在 latex/,
Word 导出移入 word/;删除临时脚本、调试截图和空 stub。

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-29 14:25:21 +08:00
parent de7a47db9d
commit a90f7adfa1
64 changed files with 7 additions and 58 deletions
@@ -0,0 +1,562 @@
# 01.1 智能的模块化视角
## 核心问题
> 为什么智能系统需要模块化设计?
> 技能(Skill)的本质是什么?如何设计可复用的智能组件?
> 函数式组合思想如何应用于AI系统?
---
## 概念讲解
### 模块化的必要性
随着系统复杂度增加,模块化变得必不可少:
```
复杂度与模块化的关系
低复杂度 ──→ [单体脚本] ──→ 可维护
简单直接
中复杂度 ──→ [函数库] ──→ 需要组织
按功能分类
高复杂度 ──→ [模块化系统] ──→ 必须模块化
清晰边界,可组合
```
**模块化的收益**
| 收益类型 | 说明 | 例子 |
|---------|------|------|
| 可理解性 | 每个模块可独立理解 | 理解缓冲区分析不需要理解投影变换 |
| 可测试性 | 模块可单独测试 | 测试空间索引不需要完整工作流 |
| 可复用性 | 模块可在不同场景使用 | 缓冲区算法用于多个项目 |
| 可替换性 | 模块可用等价实现替换 | QGIS ↔ ArcGIS 同一功能 |
| 可维护性 | 修改局限在模块内 | 修复bug不影响其他模块 |
### 函数式组合思想
函数式编程的核心:**组合小函数构建复杂行为**
```
简单函数 ──┬─── buffer(geom, distance)
├─── intersect(a, b)
├─── centroid(geom)
└─── distance(a, b)
↓ 组合
complex_operation = pipe(
load_data,
clean_geometry,
buffer(100),
intersect(study_area),
calculate_area,
format_output
)
```
**关键特性**
1. **纯函数**:相同输入→相同输出,无副作用
2. **高阶函数**:函数可以作为参数和返回值
3. **不可变数据**:数据不修改,而是创建新版本
### 技能即能力封装
在Claude Code中,技能(Skill)是智能的模块化单元:
```yaml
# 技能的结构
---
name: skill-name # 技能名称
description: 技能描述 # 何时使用
parameters: # 输入参数
- param1: type
- param2: type
returns: # 输出
- result: type
---
## 技能逻辑
具体的执行步骤...
```
**技能设计的三个层次**
```
Level 1: 原子技能
└── 单一功能,不可再分
例如:buffer, intersect, dissolve
Level 2: 组合技能
└── 由原子技能组合而成
例如:site_selection = buffer + intersect + rank
Level 3: 工作流技能
└── 完整的决策流程
例如:ecological_network_analysis
```
---
## 设计原理
### 模块化的设计原则
**1. 单一职责原则 (SRP)**
每个模块只做一件事,做好一件事:
```python
# 好的设计:每个函数职责单一
def calculate_distance(geom1, geom2):
"""只计算距离"""
return geom1.distance(geom2)
def format_distance(distance, unit='m'):
"""只格式化输出"""
if distance > 1000:
return f"{distance/1000:.2f} km"
return f"{distance:.0f} m"
# 使用
dist = calculate_distance(point_a, point_b)
formatted = format_distance(dist)
# 不好的设计:混合了计算和格式化
def calculate_and_format_distance(geom1, geom2):
distance = geom1.distance(geom2)
# 格式化逻辑混在一起
if distance > 1000:
return f"{distance/1000:.2f} km"
return f"{distance:.0f} m"
```
**2. 开闭原则 (OCP)**
对扩展开放,对修改关闭:
```python
# 使用抽象基类实现扩展性
from abc import ABC, abstractmethod
class SpatialOperation(ABC):
"""空间操作的抽象基类"""
@abstractmethod
def execute(self, data):
pass
class BufferOperation(SpatialOperation):
"""缓冲区操作"""
def __init__(self, distance):
self.distance = distance
def execute(self, data):
return data.buffer(self.distance)
class IntersectOperation(SpatialOperation):
"""相交操作"""
def __init__(self, other_data):
self.other_data = other_data
def execute(self, data):
return data.intersection(self.other_data)
# 可以添加新操作而不修改现有代码
class UnionOperation(SpatialOperation):
"""合并操作"""
def execute(self, data):
return data.union(self.other_data)
```
**3. 依赖倒置原则 (DIP)**
依赖抽象而非具体实现:
```python
# 好的设计:依赖抽象
class WorkflowProcessor:
def __init__(self, operation: SpatialOperation):
self.operation = operation # 依赖抽象
def process(self, data):
return self.operation.execute(data)
# 可以轻松替换具体实现
processor = WorkflowProcessor(BufferOperation(100))
# 不好的设计:依赖具体实现
class WorkflowProcessor:
def __init__(self, buffer_distance):
self.buffer_distance = buffer_distance
def process(self, data):
# 硬编码了具体操作
return data.buffer(self.buffer_distance)
```
### 技能接口设计
良好的技能接口设计:
```python
from typing import Protocol, TypeVar, Generic
T = TypeVar('T')
class SkillInput(Protocol[T]):
"""技能输入协议"""
def validate(self) -> bool:
"""验证输入有效性"""
...
class SkillOutput(Protocol[T]):
"""技能输出协议"""
def to_dict(self) -> dict:
"""转换为可序列化格式"""
...
class Skill(Generic[T]):
"""技能基类"""
name: str
description: str
def can_handle(self, input_data: T) -> bool:
"""判断是否能处理此输入"""
pass
def execute(self, input_data: T) -> SkillOutput[T]:
"""执行技能"""
pass
def estimate_cost(self, input_data: T) -> float:
"""估算执行成本(时间/资源)"""
pass
```
---
## 代码示例
### 模块化的空间分析系统
```python
"""
模块化空间分析系统示例
展示如何用函数式组合构建复杂分析
"""
from typing import Callable, List, Any, TypeVar
from functools import reduce
import geopandas as gpd
T = TypeVar('T')
class SpatialPipeline:
"""空间分析流水线"""
def __init__(self):
self.steps: List[Callable] = []
def add_step(self, step: Callable, name: str = None):
"""添加处理步骤"""
step.name = name or step.__name__
self.steps.append(step)
return self
def execute(self, initial_data):
"""执行流水线"""
result = initial_data
for step in self.steps:
print(f"执行步骤: {getattr(step, 'name', step.__name__)}")
result = step(result)
return result
def pipe(*functions):
"""函数式组合工具"""
return reduce(lambda f, g: lambda x: g(f(x)), functions)
# === 原子操作 ===
def load_data(path: str) -> gpd.GeoDataFrame:
"""加载数据"""
print(f"加载: {path}")
return gpd.read_file(path)
def clean_geometry(gdf: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""清理几何"""
print("清理几何")
# 修复无效几何
gdf['geometry'] = gdf.geometry.buffer(0)
return gdf[gdf.geometry.is_valid]
def reproject(gdf: gpd.GeoDataFrame, target_crs: str = 'EPSG:3857') -> gpd.GeoDataFrame:
"""重投影"""
print(f"重投影到: {target_crs}")
return gdf.to_crs(target_crs)
def buffer(gdf: gpd.GeoDataFrame, distance: float) -> gpd.GeoDataFrame:
"""缓冲区分析"""
print(f"缓冲距离: {distance}")
return gdf.buffer(distance)
def intersect(gdf: gpd.GeoDataFrame, other: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""相交分析"""
print("相交分析")
return gdf.overlay(other, how='intersection')
def calculate_area(gdf: gpd.GeoDataFrame) -> float:
"""计算面积"""
area = gdf.geometry.area.sum()
print(f"总面积: {area:.2f} 平方米")
return area
# === 高阶操作 ===
def make_buffer(distance: float) -> Callable:
"""缓冲操作工厂函数"""
return lambda gdf: buffer(gdf, distance)
def make_reproject(crs: str) -> Callable:
"""重投影工厂函数"""
return lambda gdf: reproject(gdf, crs)
def make_intersect(other_data: gpd.GeoDataFrame) -> Callable:
"""相交工厂函数"""
return lambda gdf: intersect(gdf, other_data)
# === 使用示例 ===
def example_pipeline_usage():
"""流水线使用示例"""
# 方式1:使用Pipeline类
pipeline = SpatialPipeline()
pipeline.add_step(load_data, "加载数据")
pipeline.add_step(clean_geometry, "清理几何")
pipeline.add_step(lambda gdf: reproject(gdf, 'EPSG:3857'), "重投影")
pipeline.add_step(lambda gdf: buffer(gdf, 100), "缓冲")
pipeline.add_step(calculate_area, "计算面积")
# result = pipeline.execute("data.geojson")
# 方式2:使用函数式组合
analysis_pipeline = pipe(
load_data,
clean_geometry,
lambda gdf: reproject(gdf, 'EPSG:3857'),
lambda gdf: buffer(gdf, 100),
calculate_area
)
# result = analysis_pipeline("data.geojson")
return pipeline
# === 技能封装 ===
class BufferSkill:
"""缓冲区技能"""
name = "buffer_analysis"
description = "执行缓冲区分析"
def __init__(self, distance: float, unit: str = 'm'):
self.distance = distance
self.unit = unit
def execute(self, data: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""执行技能"""
# 确保在合适的坐标系中
if data.crs and data.crs.is_geographic:
data = data.to_crs('EPSG:3857')
result = data.buffer(self.distance)
return gpd.GeoDataFrame(
geometry=result,
crs=data.crs
)
def __repr__(self):
return f"BufferSkill(distance={self.distance}{self.unit})"
class SiteSelectionSkill:
"""选址技能:组合多个原子操作"""
name = "site_selection"
description = "基于多准则的选址分析"
def __init__(self,
distance_from_road: float,
distance_from_water: float,
min_area: float):
self.road_distance = distance_from_road
self.water_distance = distance_from_water
self.min_area = min_area
def execute(self,
sites: gpd.GeoDataFrame,
roads: gpd.GeoDataFrame,
water: gpd.GeoDataFrame) -> gpd.GeoDataFrame:
"""
执行选址分析
组合操作:
1. 找到距离道路指定范围内的区域
2. 排除距离水体太近的区域
3. 筛选面积满足要求的区域
"""
# 1. 道路缓冲
road_buffer = roads.buffer(self.road_distance)
# 2. 水体缓冲(排除区)
water_buffer = water.buffer(self.water_distance)
# 3. 找到满足条件的site
suitable = sites[
sites.geometry.intersects(road_buffer.union_all()) &
~sites.geometry.intersects(water_buffer.union_all())
]
# 4. 面积筛选
suitable = suitable[suitable.geometry.area >= self.min_area]
return suitable
if __name__ == "__main__":
# 示例:构建一个选址分析流水线
print("=== 模块化空间分析系统 ===\n")
# 创建技能
buffer_skill = BufferSkill(distance=500, unit='m')
print(f"创建技能: {buffer_skill}")
# 技能可以独立测试
print("\n技能的核心优势:")
print("1. 可理解性 - 每个技能职责单一")
print("2. 可测试性 - 独立测试每个技能")
print("3. 可复用性 - 在不同场景中使用")
print("4. 可组合性 - 小技能组合成大技能")
```
---
## 案例分析
### QGIS插件架构分析
QGIS的模块化设计是学习的好例子:
```
QGIS架构
├── Core (核心库)
│ ├── QgsGeometry - 几何操作
│ ├── QgsVectorLayer - 矢量图层
│ ├── QgsRasterLayer - 栅格图层
│ └── QgsProcessing - 处理框架
├── Providers (数据提供者)
│ ├── OGR Provider - 矢量数据
│ ├── GDAL Provider - 栅格数据
│ └── PostGIS Provider - 数据库
├── Plugins (插件)
│ ├── 每个插件独立模块
│ ├── 通过接口访问核心功能
│ └── 可单独安装/卸载
└── Processing Algorithms (处理算法)
├── 算法库(600+算法)
├── 可组合使用
└── 模型构建器
```
**关键设计模式**
1. **Provider模式**:数据访问抽象
2. **Plugin模式**:功能扩展
3. **Algorithm模式**:处理步骤封装
**AI系统的启发**
```python
# 类似QGIS的AI技能架构
class AISkillRegistry:
"""AI技能注册表"""
def __init__(self):
self.skills = {}
def register(self, skill):
"""注册技能"""
self.skills[skill.name] = skill
def get(self, name: str):
"""获取技能"""
return self.skills.get(name)
def list_by_category(self, category: str):
"""按类别列出技能"""
return [s for s in self.skills.values()
if s.category == category]
# 使用
registry = AISkillRegistry()
registry.register(BufferSkill(distance=100))
registry.register(SiteSelectionSkill(...))
# 查找和使用技能
buffer = registry.get("buffer_analysis")
result = buffer.execute(data)
```
---
## 反思与延伸
### 思考问题
1. **边界划分**:如何确定一个模块的边界?太小的模块和太大的模块各有什么问题?
2. **接口设计**:设计一个技能接口时,应该考虑哪些因素?
3. **复用性**:什么代码值得复用?什么不值得?
4. **组合爆炸**:当模块数量很大时,如何管理模块之间的依赖?
### 实践练习
1. **重构练习**:找一个你写的复杂函数,将其拆分为多个小函数
2. **接口设计**:为你熟悉的空间分析操作设计技能接口
3. **组合挑战**:用5个以下的基本操作组合出10个不同的分析流程
### 延伸阅读
- **"Refactoring"** (Martin Fowler) - 重构与模块化
- **"The Art of Unix Programming"** - 模块化哲学
- QGIS Plugin开发指南
---
## 关键要点
1. **模块化是管理复杂性的核心方法**
2. **函数式组合让小函数构建大功能**
3. **技能是智能的封装单元**
4. **良好的接口设计是模块化的关键**
5. **QGIS的架构是学习的优秀范例**
@@ -0,0 +1,719 @@
# 01.2 状态与状态机
## 核心问题
> 在AI系统中,"状态"到底是什么?
> 为什么状态管理是Agent系统的核心?
> LangGraph是如何用状态机设计工作流的?
---
## 概念讲解
### 什么是状态
**状态**是系统在某一时刻的快照,包含所有影响未来行为的信息:
```
系统状态 = 所有相关的变量值
例如:生态分析工作流的状态
{
"input_data": {...}, # 输入数据
"current_step": "buffer", # 当前步骤
"intermediate_results": {...}, # 中间结果
"user_preferences": {...}, # 用户偏好
"error_count": 0, # 错误计数
"checkpoint_reached": False # 检查点状态
}
```
**状态的类型**
| 类型 | 说明 | 例子 |
|-----|------|------|
| 静态状态 | 初始输入,不变化 | 输入文件路径、参数 |
| 动态状态 | 运行中变化 | 当前步骤、累积结果 |
| 控制状态 | 影响流程走向 | 分支条件、错误标志 |
| 会话状态 | 跨请求持久化 | 用户偏好、历史记录 |
### 为什么状态管理很重要
**1. 断点续传**
```python
# 没有状态管理:出错后必须从头开始
def analysis_without_state():
step1()
step2() # 如果这里出错
step3() # 这些都要重做
# 有状态管理:可以从断点继续
class AnalysisWithState:
def __init__(self):
self.state = {"current_step": 0}
def run(self):
if self.state["current_step"] < 1:
step1()
self.state["current_step"] = 1
if self.state["current_step"] < 2:
try:
step2()
self.state["current_step"] = 2
except Exception:
# 保存状态,下次可以从这里继续
save_state(self.state)
raise
if self.state["current_step"] < 3:
step3()
```
**2. 人机协同**
```python
# HITL需要状态来知道在哪里需要人类介入
class HITLWorkflow:
def __init__(self):
self.state = {
"step": "identify_sources",
"pending_review": True,
"sources": None,
"human_feedback": None
}
def next_action(self):
if self.state["pending_review"]:
return "request_human_review"
elif self.state["human_feedback"]:
return "incorporate_feedback"
else:
return "proceed_to_next_step"
```
**3. 调试和可解释性**
```python
# 状态历史记录了整个决策过程
class StatefulAgent:
def __init__(self):
self.state_history = []
def decide(self, context):
# 记录状态
self.state_history.append({
"timestamp": now(),
"state": self.state.copy(),
"context": context,
"decision": None
})
# 做决策
decision = self._make_decision(context)
self.state_history[-1]["decision"] = decision
return decision
def explain(self):
"""回溯决策过程"""
return self.state_history
```
### 状态机
**状态机**是描述系统状态转换的模型:
```
┌─────────┐
│ 初始 │
│ state │
└────┬────┘
│ event: start
┌─────────┐
│ 加载数据 │
└────┬────┘
│ success
┌─────────┐ error ┌─────────┐
│ 分析处理 │ ─────────────→│ 错误 │
└────┬────┘ └─────────┘
│ success │ retry
↓ │
┌─────────┐ │
│ 人类 │ │
│ 审查 │ │
└────┬────┘ │
│ approve │
↓ │
┌─────────┐ │
│ 完成 │←───────────────────────┘
└─────────┘
```
**状态机的要素**
1. **状态 (State)**:系统可能处于的情况
2. **事件 (Event)**:触发状态转换的条件
3. **转换 (Transition)**:从一个状态到另一个状态
4. **动作 (Action)**:状态转换时执行的操作
---
## 设计原理
### LangGraph的状态设计哲学
LangGraph是构建有状态Agent的框架,其核心思想:
```python
from typing import TypedDict
# 定义状态类型
class AnalysisState(TypedDict):
"""生态网络分析状态"""
# 输入数据
input_path: str
parameters: dict
# 处理过程
current_step: str
intermediate_results: dict
# 人机交互
review_requested: bool
human_feedback: str
# 输出
final_result: dict
errors: list
# 状态图定义
workflow = StateGraph(AnalysisState)
# 添加节点(处理步骤)
workflow.add_node("load_data", load_data_node)
workflow.add_node("identify_sources", identify_sources_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("extract_corridors", extract_corridors_node)
# 添加边(状态转换)
workflow.add_edge("load_data", "identify_sources")
workflow.add_conditional_edge(
"identify_sources",
should_review, # 条件函数
{
"review": "human_review",
"continue": "extract_corridors"
}
)
# 编译为可执行图
app = workflow.compile()
```
**核心概念**
1. **状态即消息**:状态在节点间传递
2. **图即流程**:有向图描述工作流
3. **条件分支**:基于状态的动态路由
### 工作流状态机实现
```python
from enum import Enum
from typing import Dict, Any, Callable, Optional
from dataclasses import dataclass, field
class WorkflowState(Enum):
"""工作流状态枚举"""
IDLE = "idle"
LOADING = "loading"
PROCESSING = "processing"
REVIEWING = "reviewing"
COMPLETED = "completed"
ERROR = "error"
@dataclass
class WorkflowContext:
"""工作流上下文(状态数据)"""
data: Dict[str, Any] = field(default_factory=dict)
current_step: int = 0
errors: list = field(default_factory=list)
metadata: Dict[str, Any] = field(default_factory=dict)
class StateMachine:
"""通用状态机"""
def __init__(self, initial_state: WorkflowState):
self.state = initial_state
self.context = WorkflowContext()
self.transitions: Dict[WorkflowState, Dict[str, WorkflowState]] = {}
self.actions: Dict[tuple[WorkflowState, WorkflowState], Callable] = {}
def add_transition(self,
from_state: WorkflowState,
event: str,
to_state: WorkflowState,
action: Callable = None):
"""添加状态转换"""
if from_state not in self.transitions:
self.transitions[from_state] = {}
self.transitions[from_state][event] = to_state
if action:
self.actions[(from_state, to_state)] = action
def trigger(self, event: str, **kwargs) -> bool:
"""触发事件"""
if self.state not in self.transitions:
raise ValueError(f"没有从状态 {self.state} 的转换")
if event not in self.transitions[self.state]:
print(f"事件 {event} 在状态 {self.state} 下无效")
return False
# 获取目标状态
new_state = self.transitions[self.state][event]
old_state = self.state
# 执行转换动作
action = self.actions.get((old_state, new_state))
if action:
result = action(self.context, **kwargs)
if result is False: # 动作失败,不转换
return False
# 更新状态
self.state = new_state
print(f"状态转换: {old_state}{new_state}")
return True
# 生态分析工作流状态机
class EcologicalAnalysisWorkflow:
"""生态网络分析工作流"""
def __init__(self):
# 创建状态机
self.sm = StateMachine(WorkflowState.IDLE)
# 定义转换
self.sm.add_transition(WorkflowState.IDLE, "start", WorkflowState.LOADING)
self.sm.add_transition(WorkflowState.LOADING, "loaded", WorkflowState.PROCESSING)
self.sm.add_transition(WorkflowState.LOADING, "error", WorkflowState.ERROR)
self.sm.add_transition(WorkflowState.PROCESSING, "complete", WorkflowState.REVIEWING)
self.sm.add_transition(WorkflowState.PROCESSING, "error", WorkflowState.ERROR)
self.sm.add_transition(WorkflowState.REVIEWING, "approved", WorkflowState.COMPLETED)
self.sm.add_transition(WorkflowState.REVIEWING, "rejected", WorkflowState.PROCESSING)
self.sm.add_transition(WorkflowState.ERROR, "retry", WorkflowState.LOADING)
def run(self, data_path: str):
"""执行工作流"""
# 启动
self.sm.trigger("start", data_path=data_path)
# 模拟加载
print("加载数据...")
self.sm.trigger("loaded")
# 模拟处理
print("处理数据...")
self.sm.trigger("complete")
# 审查
print("等待审查...")
# 这里会等待人类输入
# 假设批准
self.sm.trigger("approved")
print(f"工作流完成,最终状态: {self.sm.state}")
```
---
## 代码示例
### 完整的状态机工作流
```python
"""
完整的状态机工作流示例
"""
import json
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field, asdict
from enum import Enum
import time
class State(Enum):
"""状态枚举"""
IDLE = "idle"
LOAD_DATA = "load_data"
IDENTIFY_SOURCES = "identify_sources"
BUILD_RESISTANCE = "build_resistance"
REVIEW_SOURCES = "review_sources"
REVIEW_RESISTANCE = "review_resistance"
EXTRACT_CORRIDORS = "extract_corridors"
COMPLETED = "completed"
ERROR = "error"
@dataclass
class WorkflowState:
"""工作流状态数据"""
current: State = State.IDLE
step_number: int = 0
data_path: Optional[str] = None
sources: Optional[List[Dict]] = None
resistance_weights: Optional[Dict] = None
corridors: Optional[List[Dict]] = None
errors: List[str] = field(default_factory=list)
history: List[Dict] = field(default_factory=list)
def transition_to(self, new_state: State, action: str = ""):
"""状态转换"""
old_state = self.current
self.current = new_state
self.step_number += 1
# 记录历史
self.history.append({
"step": self.step_number,
"from": old_state.value,
"to": new_state.value,
"action": action,
"timestamp": time.time()
})
def to_dict(self) -> Dict:
"""序列化"""
return {
"current": self.current.value,
"step_number": self.step_number,
"data_path": self.data_path,
"sources": self.sources,
"resistance_weights": self.resistance_weights,
"corridors": self.corridors,
"errors": self.errors,
"history": self.history
}
def save(self, path: str):
"""保存状态到文件"""
with open(path, 'w') as f:
json.dump(self.to_dict(), f, indent=2)
@classmethod
def load(cls, path: str) -> 'WorkflowState':
"""从文件加载状态"""
with open(path, 'r') as f:
data = json.load(f)
# 转换State枚举
data["current"] = State(data["current"])
return cls(**{k: v for k, v in data.items() if k != "history"})
class EcologicalAnalysisAgent:
"""生态分析智能体(有状态)"""
def __init__(self):
self.state = WorkflowState()
self.review_callbacks = {
State.REVIEW_SOURCES: self._review_sources,
State.REVIEW_RESISTANCE: self._review_resistance
}
def start(self, data_path: str):
"""启动分析"""
self.state.data_path = data_path
self.state.transition_to(State.LOAD_DATA, "开始加载数据")
self._execute_current_step()
def _execute_current_step(self):
"""执行当前状态对应的操作"""
handlers = {
State.LOAD_DATA: self._handle_load_data,
State.IDENTIFY_SOURCES: self._handle_identify_sources,
State.BUILD_RESISTANCE: self._handle_build_resistance,
State.REVIEW_SOURCES: self._handle_review,
State.REVIEW_RESISTANCE: self._handle_review,
State.EXTRACT_CORRIDORS: self._handle_extract_corridors,
State.COMPLETED: self._handle_completed,
State.ERROR: self._handle_error
}
handler = handlers.get(self.state.current)
if handler:
handler()
def _handle_load_data(self):
"""处理数据加载"""
print(f"\n[状态: {self.state.current.value}] 加载数据: {self.state.data_path}")
# 模拟加载
try:
# 这里实际会读取文件
time.sleep(0.5)
print("数据加载成功")
self.state.transition_to(State.IDENTIFY_SOURCES, "数据加载完成")
self._execute_current_step()
except Exception as e:
self.state.errors.append(str(e))
self.state.transition_to(State.ERROR, f"加载失败: {e}")
self._execute_current_step()
def _handle_identify_sources(self):
"""处理源地识别"""
print(f"\n[状态: {self.state.current.value}] 识别生态源地...")
# 模拟识别
self.state.sources = [
{"id": 1, "area": 1500, "type": "forest"},
{"id": 2, "area": 800, "type": "wetland"}
]
print(f"识别到 {len(self.state.sources)} 个源地")
self.state.transition_to(State.REVIEW_SOURCES, "源地识别完成,等待审查")
self._execute_current_step()
def _handle_build_resistance(self):
"""处理阻力面构建"""
print(f"\n[状态: {self.state.current.value}] 构建阻力面...")
# 模拟构建
self.state.resistance_weights = {
"forest": 1,
"grassland": 10,
"urban": 100,
"water": 50
}
print("阻力面构建完成")
self.state.transition_to(State.REVIEW_RESISTANCE, "阻力面构建完成,等待审查")
self._execute_current_step()
def _handle_review(self):
"""处理审查状态"""
print(f"\n[状态: {self.state.current.value}] 等待人类审查...")
callback = self.review_callbacks.get(self.state.current)
if callback:
result = callback()
if result == "approve":
if self.state.current == State.REVIEW_SOURCES:
self.state.transition_to(State.BUILD_RESISTANCE, "审查通过")
elif self.state.current == State.REVIEW_RESISTANCE:
self.state.transition_to(State.EXTRACT_CORRIDORS, "审查通过")
self._execute_current_step()
else:
# 拒绝,返回上一状态
print("审查未通过,重新执行...")
# 简化处理:直接继续
def _review_sources(self) -> str:
"""审查源地"""
print("\n=== 源地审查 ===")
print(f"识别到 {len(self.state.sources)} 个源地:")
for s in self.state.sources:
print(f" - ID {s['id']}: {s['type']}, 面积 {s['area']}")
# 实际实现中这里会等待人类输入
# 这里模拟自动批准
print("\n[模拟] 审查: 批准")
return "approve"
def _review_resistance(self) -> str:
"""审查阻力面"""
print("\n=== 阻力面审查 ===")
print("阻力权重:")
for land_type, weight in self.state.resistance_weights.items():
print(f" - {land_type}: {weight}")
print("\n[模拟] 审查: 批准")
return "approve"
def _handle_extract_corridors(self):
"""处理廊道提取"""
print(f"\n[状态: {self.state.current.value}] 提取生态廊道...")
# 模拟提取
self.state.corridors = [
{"from": 1, "to": 2, "length": 3500}
]
print(f"提取到 {len(self.state.corridors)} 条廊道")
self.state.transition_to(State.COMPLETED, "分析完成")
self._execute_current_step()
def _handle_completed(self):
"""处理完成状态"""
print(f"\n[状态: {self.state.current.value}] 工作流完成!")
print(f"\n=== 结果摘要 ===")
print(f"源地数量: {len(self.state.sources) if self.state.sources else 0}")
print(f"廊道数量: {len(self.state.corridors) if self.state.corridors else 0}")
print(f"执行步骤: {self.state.step_number}")
def _handle_error(self):
"""处理错误状态"""
print(f"\n[状态: {self.state.current.value}] 发生错误")
for error in self.state.errors:
print(f" - {error}")
def save_state(self, path: str):
"""保存当前状态"""
self.state.save(path)
print(f"状态已保存到: {path}")
def resume_from(self, path: str):
"""从保存的状态恢复"""
self.state = WorkflowState.load(path)
print(f"从状态恢复: {self.state.current.value}")
print(f"历史步骤: {self.state.step_number}")
self._execute_current_step()
# 使用示例
if __name__ == "__main__":
print("=== 生态分析状态机工作流 ===\n")
agent = EcologicalAnalysisAgent()
# 执行工作流
agent.start("data.geojson")
# 可以保存状态
# agent.save_state("workflow_state.json")
# 可以从状态恢复
# new_agent = EcologicalAnalysisAgent()
# new_agent.resume_from("workflow_state.json")
```
---
## 案例分析
### LangGraph在空间分析中的应用
```python
"""
LangGraph风格的生态网络分析工作流
"""
from typing import TypedDict, Annotated, Literal
from operator import add
class EcologicalState(TypedDict):
"""生态分析状态类型"""
messages: Annotated[list, add] # 消息历史
input_data: dict
sources: list
resistance: dict
corridors: list
next_step: str
human_feedback: str
# 节点函数
def load_data_node(state: EcologicalState) -> EcologicalState:
"""加载数据节点"""
print("执行: load_data")
state["sources"] = [{"id": 1, "area": 1000}]
state["next_step"] = "identify"
return state
def identify_sources_node(state: EcologicalState) -> EcologicalState:
"""识别源地节点"""
print("执行: identify_sources")
state["sources"] = [{"id": i, "area": i * 100} for i in range(1, 6)]
state["next_step"] = "review"
return state
def human_review_node(state: EcologicalState) -> EcologicalState:
"""人类审查节点"""
print("执行: human_review")
print(f"待审查: {state['sources']}")
# 在实际实现中,这里会等待人类输入
state["human_feedback"] = "approved"
state["next_step"] = "build_resistance"
return state
def build_resistance_node(state: EcologicalState) -> EcologicalState:
"""构建阻力面节点"""
print("执行: build_resistance")
state["resistance"] = {"forest": 1, "urban": 100}
state["next_step"] = "complete"
return state
# 路由函数
def should_review(state: EcologicalState) -> Literal["review", "skip"]:
"""决定是否需要审查"""
if len(state.get("sources", [])) > 3:
return "review"
return "skip"
# 条件边
def route_after_identify(state: EcologicalState) -> str:
"""识别源地后的路由"""
if state.get("human_feedback") == "approved":
return "build_resistance"
return "identify" # 重新识别
print("""
┌──────────────┐
│ load_data │
└──────┬───────┘
┌──────────────┐
│identify_sources│
└──────┬───────┘
├────→ [review?] ──→ human_review ──┐
│ No │
↓ ↓
┌──────────────┐ ┌──────────────┐
│build_resistance│◀──────────────────│ approved │
└──────────────┘ └──────────────┘
""")
```
---
## 反思与延伸
### 思考问题
1. **状态粒度**:状态应该有多细?太细会怎样,太粗会怎样?
2. **持久化策略**:哪些状态需要持久化?什么时候保存状态?
3. **并发处理**:如果多个Agent协同工作,如何管理共享状态?
4. **调试**:当状态机出错时,如何调试?
### 实践练习
1. **状态审计**:添加状态转换日志,分析工作流执行路径
2. **状态压缩**:实现状态序列化/反序列化,支持断点续传
3. **条件路由**:实现一个带多个分支的状态机
### 延伸阅读
- **"Designing Data-Intensive Applications"** (Kleppmann) - 状态管理理论
- **LangGraph文档** - 实际框架使用
- **"State Machine Design Patterns"** - 状态机设计模式
---
## 关键要点
1. **状态是系统在某一时刻的完整快照**
2. **状态机描述系统如何随事件转换状态**
3. **LangGraph用图结构表达有状态的工作流**
4. **良好的状态管理支持断点续传和HITL**
5. **状态历史是调试和可解释性的关键**
@@ -0,0 +1,677 @@
# 01.3 概率与不确定性
## 核心问题
> 空间分析中的不确定性从何而来?
> AI系统如何表示和处理不确定性?
> 如何在不确定性下做出稳健的决策?
---
## 概念讲解
### 不确定性的来源
在空间分析和AI系统中,不确定性无处不在:
```
空间分析中的不确定性来源
┌─────────────────────────────────────────────────────────────┐
│ │
│ 1. 数据不确定性 │
│ - 测量误差 │
│ - 空间采样不完整 │
│ - 分类错误 │
│ - 时间延迟 │
│ │
│ 2. 参数不确定性 │
│ - 阻力权重不确定 │
│ - 阈值选择主观 │
│ - 模型参数拟合误差 │
│ │
│ 3. 结构不确定性 │
│ - 模型选择 │
│ - 变量关系假设 │
│ - 尺度效应 │
│ │
│ 4. 语义不确定性 │
│ - 概念模糊("生态质量"是什么?) │
│ - 分类边界不清 │
│ - 专家意见分歧 │
│ │
└─────────────────────────────────────────────────────────────┘
```
### 不确定性的类型
| 类型 | 说明 | 例子 |
|-----|------|------|
| **偶然不确定性** (Aleatoric) | 系统固有的随机性,无法通过更多数据消除 | 降雨量的随机波动 |
| **认知不确定性** (Epistemic) | 知识不足导致的不确定性,可通过更多数据减少 | 未调查区域的物种分布 |
| **模糊性** (Ambiguity) | 概念或分类的不明确 | "高生态价值"的定义 |
| **冲突** (Conflict) | 不同信息源的不一致 | 两个专家给出相反意见 |
### AI如何处理不确定性
**传统GIS vs 概率AI**
```
传统GIS: 确定性输出
输入 → [处理] → 单一结果
例如:这个区域是/不是生态源地
概率AI: 概率输出
输入 → [处理] → (结果, 置信度)
例如:这个区域是生态源地的概率是 0.78 ± 0.12
```
**置信度的表示**
```python
# 方式1: 点估计 + 置信区间
estimate = 0.75
confidence_interval = (0.65, 0.85)
# 方式2: 概率分布
from scipy.stats import beta
distribution = beta(a=8, b=3) # 基于共8次成功,3次失败
# 方式3: 分类概率
class_probabilities = {
"high_suitability": 0.65,
"medium_suitability": 0.25,
"low_suitability": 0.10
}
# 方式4: 模糊隶属度
fuzzy_membership = {
"is_source": 0.72,
"is_not_source": 0.28
}
```
---
## 设计原理
### 不确定性传播
当多个步骤串联时,不确定性会累积:
```python
"""
不确定性传播示例
"""
import numpy as np
from scipy.stats import norm
class UncertainValue:
"""带不确定性的值"""
def __init__(self, mean, std):
self.mean = mean
self.std = std
def __add__(self, other):
"""加法:方差相加"""
return UncertainValue(
self.mean + other.mean,
np.sqrt(self.std**2 + other.std**2)
)
def __mul__(self, scalar):
"""乘以标量:标准差也乘"""
return UncertainValue(
self.mean * scalar,
self.std * abs(scalar)
)
def __repr__(self):
return f"{self.mean:.2f} ± {self.std:.2f}"
# 示例:源地适宜性评估中的不确定性传播
def assess_suitability_with_uncertainty():
"""
每个指标都有测量不确定性,
最终的适宜性评分会累积这些不确定性
"""
# 各项指标(均值 ± 标准差)
vegetation_quality = UncertainValue(0.75, 0.10)
connectivity = UncertainValue(0.60, 0.15)
distance_to_threat = UncertainValue(0.80, 0.08)
# 加权组合(权重也有不确定性)
weights = {
"vegetation": 0.4,
"connectivity": 0.3,
"distance": 0.3
}
# 计算总分(简化传播)
total = (vegetation_quality * weights["vegetation"] +
connectivity * weights["connectivity"] +
distance_to_threat * weights["distance"])
print("各指标不确定性:")
print(f" 植被质量: {vegetation_quality}")
print(f" 连通性: {connectivity}")
print(f" 威胁距离: {distance_to_threat}")
print(f"\n总分: {total}")
print(f" 置信区间95%: [{total.mean - 1.96*total.std:.2f}, "
f"{total.mean + 1.96*total.std:.2f}]")
return total
if __name__ == "__main__":
assess_suitability_with_uncertainty()
```
### 敏感性分析
了解哪些参数对结果影响最大:
```python
"""
敏感性分析:识别关键参数
"""
import numpy as np
from typing import Dict, List, Tuple
def sensitivity_analysis(model_fn, param_ranges: Dict[str, Tuple[float, float]],
n_samples=1000) -> Dict[str, float]:
"""
使用蒙特卡洛方法进行敏感性分析
Args:
model_fn: 模型函数,接受参数字典,返回结果
param_ranges: 参数范围 {param_name: (min, max)}
n_samples: 采样次数
Returns:
各参数的敏感性系数
"""
results = {param: [] for param in param_ranges}
model_outputs = []
# 蒙特卡洛采样
for _ in range(n_samples):
# 随机采样参数
sample = {k: np.random.uniform(v[0], v[1])
for k, v in param_ranges.items()}
# 记录参数值
for param, value in sample.items():
results[param].append(value)
# 计算模型输出
output = model_fn(sample)
model_outputs.append(output)
# 计算相关性作为敏感性指标
sensitivities = {}
for param in param_ranges:
correlation = np.corrcoef(results[param], model_outputs)[0, 1]
sensitivities[param] = abs(correlation)
return sensitivities
# 示例:生态阻力面构建的敏感性分析
def resistance_model(params):
"""简化的阻力面模型"""
# 参数:各土地类型的阻力权重
forest_weight = params["forest"]
grass_weight = params["grass"]
urban_weight = params["urban"]
# 简化:计算平均阻力
# 实际应用中会是空间计算
landscape_composition = {
"forest": 0.4,
"grass": 0.3,
"urban": 0.3
}
total_resistance = (
forest_weight * landscape_composition["forest"] +
grass_weight * landscape_composition["grass"] +
urban_weight * landscape_composition["urban"]
)
return total_resistance
def run_sensitivity_example():
"""运行敏感性分析示例"""
print("=== 阻力面参数敏感性分析 ===\n")
# 定义参数范围
param_ranges = {
"forest": (1, 10),
"grass": (10, 50),
"urban": (50, 200)
}
# 运行敏感性分析
sensitivities = sensitivity_analysis(
resistance_model,
param_ranges,
n_samples=5000
)
# 排序并输出
sorted_sens = sorted(sensitivities.items(),
key=lambda x: x[1], reverse=True)
print("参数敏感性排序:")
for param, sensitivity in sorted_sens:
bar = "" * int(sensitivity * 30)
print(f" {param}: {sensitivity:.3f} {bar}")
print("\n解释:")
print(f" 最敏感的参数是 {sorted_sens[0][0]}")
print(f" 应当优先精确确定该参数的值")
if __name__ == "__main__":
run_sensitivity_example()
```
### 鲁棒决策
当不确定性无法消除时,做鲁棒的决策:
```python
"""
鲁棒决策:在不确定性下的稳健决策
"""
from typing import List, Callable
import numpy as np
def robust_decision_scenarios():
"""
鲁棒决策的几种策略
"""
# 策略1: 最大最小 (Maximin) - 最坏情况最优
def maximin(payoff_matrix):
"""
选择在最坏情况下表现最好的选项
payoff_matrix: 选项 × 场景 的收益矩阵
"""
worst_case_outcomes = payoff_matrix.min(axis=1)
best_option = worst_case_outcomes.argmax()
return best_option, worst_case_outcomes
# 策略2: 最大平均 (Maximum Expected Value)
def max_expected(payoff_matrix, probabilities=None):
"""选择期望收益最大的选项"""
if probabilities is None:
probabilities = np.ones(payoff_matrix.shape[1]) / payoff_matrix.shape[1]
expected_values = payoff_matrix @ probabilities
best_option = expected_values.argmax()
return best_option, expected_values
# 策略3: 最小后悔 (Minimax Regret)
def minimax_regret(payoff_matrix):
"""选择最小化最大后悔的选项"""
# 每个场景的最佳收益
best_per_scenario = payoff_matrix.max(axis=0)
# 后悔矩阵:每个选项在每个场景与最佳的差距
regret_matrix = best_per_scenario - payoff_matrix
# 每个选项的最大后悔
max_regret = regret_matrix.max(axis=1)
# 选择最大后悔最小的选项
best_option = max_regret.argmin()
return best_option, max_regret
# 示例:生态廊道选址决策
# 选项:3个候选廊道路线
# 场景:不同的未来土地变化情景
payoff_matrix = np.array([
# 情景1 情景2 情景3 情景4
[80, 60, 40, 70], # 选项A:穿过森林
[50, 90, 70, 50], # 选项B:沿河流
[60, 70, 90, 60], # 选项C:绕行城市边缘
])
print("=== 生态廊道选址:鲁棒决策分析 ===\n")
print("收益矩阵(廊道质量评分):")
print(" 情景1 情景2 情景3 情景4")
for i, row in enumerate(payoff_matrix, ord('A')):
print(f"选项{i}: {row}")
print("\n--- 策略1: 最大最小 (最坏情况最优) ---")
option, worst = maximin(payoff_matrix)
print(f"推荐: 选项{chr(ord('A') + option)}")
print(f"各选项最坏情况: {worst}")
print(" 原理: 选择在最坏情景下表现最好的")
print("\n--- 策略2: 最大期望 (平均收益最大) ---")
option, expected = max_expected(payoff_matrix)
print(f"推荐: 选项{chr(ord('A') + option)}")
print(f"各选项期望收益: {expected}")
print(" 原理: 选择平均表现最好的")
print("\n--- 策略3: 最小最大后悔 ---")
option, regret = minimax_regret(payoff_matrix)
print(f"推荐: 选项{chr(ord('A') + option)}")
print(f"各选项最大后悔: {regret}")
print(" 原理: 选择让'选错'的后悔最小的")
return payoff_matrix
if __name__ == "__main__":
robust_decision_scenarios()
```
---
## 代码示例
### 概率源地识别
```python
"""
带不确定性的生态源地识别
"""
import numpy as np
from scipy.stats import beta
from typing import Dict, List, Tuple
class ProbabilisticSource:
"""概率源地:带置信度的源地"""
def __init__(self, source_id: int, geometry,
probability: float, confidence: float):
self.id = source_id
self.geometry = geometry
self.probability = probability # 是源地的概率
self.confidence = confidence # 概率估计的置信度
def __repr__(self):
return (f"Source({self.id}, P={self.probability:.2f}±{self.confidence:.2f})")
class ProbabilisticSourceIdentifier:
"""概率源地识别器"""
def __init__(self, threshold=0.5):
self.threshold = threshold
def identify(self, landscape_data) -> List[ProbabilisticSource]:
"""
识别源地,返回概率源地
返回的不是"是/否"的判断,而是"是源地的概率"
"""
sources = []
# 模拟:对每个斑块计算是源地的概率
for i, patch in enumerate(landscape_data):
# 基于多个指标计算概率
probability = self._calculate_source_probability(patch)
# 估计置信度(基于数据质量)
confidence = self._estimate_confidence(patch)
if probability >= self.threshold:
source = ProbabilisticSource(
source_id=i,
geometry=patch['geometry'],
probability=probability,
confidence=confidence
)
sources.append(source)
return sources
def _calculate_source_probability(self, patch) -> float:
"""计算斑块是源地的概率"""
# 使用贝叶斯推理
# P(是源地|数据) ∝ P(数据|是源地) × P(是源地)
# 各指标的似然
area_likelihood = self._area_likelihood(patch['area'])
veg_likelihood = self._vegetation_likelihood(patch['vegetation'])
shape_likelihood = self._shape_likelihood(patch['shape_index'])
# 先验概率
prior = 0.3 # 假设30%的斑块可能是源地
# 后验概率(简化)
probability = (area_likelihood * veg_likelihood *
shape_likelihood * prior)
probability = min(probability, 1.0) # 限制在[0,1]
return probability
def _area_likelihood(self, area: float) -> float:
"""面积似然:大面积更像源地"""
if area > 1000:
return 1.0
elif area > 500:
return 0.7
else:
return 0.3
def _vegetation_likelihood(self, veg_quality: float) -> float:
"""植被质量似然"""
return veg_quality # 假设已归一化到[0,1]
def _shape_likelihood(self, shape_index: float) -> float:
"""形状指数似然:紧凑形状更好"""
return 1.0 - min(abs(shape_index - 1.0), 0.5)
def _estimate_confidence(self, patch) -> float:
"""估计概率的置信度"""
# 基于数据质量、分辨率等因素
data_quality = patch.get('data_quality', 0.8)
resolution_factor = patch.get('resolution', 30) / 30 # 归一化
return data_quality * min(resolution_factor, 1.0)
def uncertainty_propagation_example():
"""不确定性传播示例"""
print("=== 不确定性传播示例 ===\n")
# 创建一些模拟斑块
patches = [
{'id': 1, 'area': 1200, 'vegetation': 0.85, 'shape_index': 1.2,
'geometry': 'POLYGON(...)', 'data_quality': 0.9},
{'id': 2, 'area': 800, 'vegetation': 0.75, 'shape_index': 1.5,
'geometry': 'POLYGON(...)', 'data_quality': 0.7},
{'id': 3, 'area': 400, 'vegetation': 0.65, 'shape_index': 1.8,
'geometry': 'POLYGON(...)', 'data_quality': 0.6},
]
identifier = ProbabilisticSourceIdentifier(threshold=0.4)
sources = identifier.identify(patches)
print("识别到的概率源地:")
for source in sources:
print(f" {source}")
# 计算整体不确定性
if sources:
avg_prob = np.mean([s.probability for s in sources])
avg_conf = np.mean([s.confidence for s in sources])
print(f"\n总体置信度: {avg_conf:.2f}")
print(f"平均源地概率: {avg_prob:.2f}")
# 置信区间
margin_of_error = (1 - avg_conf) * 0.2 # 简化计算
print(f"源地数量估计: {len(sources)} ± {margin_of_error * len(sources):.1f}")
if __name__ == "__main__":
uncertainty_propagation_example()
```
---
## 案例分析
### ENAgent中的不确定性处理
在ENAgent项目中,不确定性处理体现在:
**1. 源地识别的不确定性**
```python
class ENAgentSourceIdentifier:
"""ENAgent的源地识别模块"""
def identify_with_uncertainty(self, landcover, species_params):
"""
识别源地,同时估计不确定性
Returns:
sources: 源地列表
uncertainty_map: 不确定性空间分布
"""
sources = []
uncertainty_map = np.zeros_like(landcover)
# 对每个候选斑块
for patch in self._candidate_patches(landcover):
# 计算适宜性(考虑物种参数)
suitability = self._calculate_suitability(patch, species_params)
# 估计不确定性(来自多个来源)
uncertainty = self._estimate_uncertainty(
patch,
data_quality=landcover.metadata['quality'],
species_uncertainty=species_params['uncertainty']
)
# 记录
if suitability > self.threshold:
sources.append({
'geometry': patch,
'suitability': suitability,
'uncertainty': uncertainty
})
# 更新不确定性地图
self._add_to_uncertainty_map(uncertainty_map, patch, uncertainty)
return sources, uncertainty_map
def _estimate_uncertainty(self, patch, data_quality, species_uncertainty):
"""
估计源地识别的不确定性
来源:
1. 数据质量 (data_quality)
2. 物种参数的不确定性 (species_uncertainty)
3. 分类误差 (classification_error)
4. 边界效应 (edge_effect)
"""
# 组合各种不确定性源
uncertainty = np.sqrt(
(1 - data_quality)**2 +
species_uncertainty**2 +
0.1**2 + # 分类误差
self._edge_uncertainty(patch)**2
)
return min(uncertainty, 1.0)
```
**2. 阻力面的敏感性分析**
```python
class ResistanceSurfaceSensitivity:
"""阻力面敏感性分析"""
def analyze(self, base_weights, variation_ranges, n_simulations=1000):
"""
分析阻力面权重对结果的敏感性
Args:
base_weights: 基础权重 {land_type: weight}
variation_ranges: 权重变化范围 {land_type: (min, max)}
n_simulations: 蒙特卡洛模拟次数
Returns:
sensitivity_results: 敏感性分析结果
"""
results = []
for _ in range(n_simulations):
# 随机采样权重
sample_weights = {}
for land_type, (min_w, max_w) in variation_ranges.items():
sample_weights[land_type] = np.random.uniform(min_w, max_w)
# 计算对应的阻力面
resistance = self._compute_resistance(sample_weights)
# 评估结果(例如:平均连通性)
connectivity = self._assess_connectivity(resistance)
results.append({
'weights': sample_weights,
'connectivity': connectivity
})
# 分析敏感性
sensitivity = self._compute_sensitivity(results, base_weights)
return sensitivity
def _compute_sensitivity(self, results, base_weights):
"""计算各土地类型的敏感性"""
# 计算每个权重变化与连通性变化的相关性
sensitivities = {}
for land_type in base_weights:
weight_values = [r['weights'][land_type] for r in results]
connectivity_values = [r['connectivity'] for r in results]
correlation = np.corrcoef(weight_values, connectivity_values)[0, 1]
sensitivities[land_type] = abs(correlation)
return sensitivities
```
---
## 反思与延伸
### 思考问题
1. **不确定性识别**:在你的项目中,不确定性来自哪些方面?哪些是可减少的,哪些是固有的?
2. **表示选择**:你应该用标准差、置信区间,还是概率分布?各有什么优劣?
3. **决策权衡**:当不确定性很高时,你应该继续分析还是寻求更多数据?
4. **沟通问题**:如何向非专家解释不确定性?
### 实践练习
1. **不确定性审计**:对一个分析流程,识别所有不确定性来源并分类
2. **敏感性分析**:对你熟悉的空间模型进行敏感性分析
3. **鲁棒决策**:为你的项目设计一个鲁棒决策框架
### 延伸阅读
- **"Uncertainty Quantification in Predictive Modeling"** - 不确定性量化的理论基础
- **"Flaw of Averages"** (Sam Savage) - 为什么平均值会误导
- **空间数据质量标准** - 空间不确定性的行业实践
---
## 关键要点
1. **不确定性在空间分析中普遍存在**,有多个来源
2. **偶然不确定性无法消除**,认知不确定性可以通过更多数据减少
3. **不确定性会传播**,多步骤分析需要考虑累积效应
4. **敏感性分析识别关键参数**,优先减少高敏感参数的不确定性
5. **鲁棒决策在不确定性下做稳健选择**,而非追求最优
@@ -0,0 +1,718 @@
# 01.4 反馈与学习
## 核心问题
> 系统如何从经验中改进?
> 强化学习的基本直觉是什么?
> 如何设计一个好的奖励函数?
---
## 概念讲解
### 反馈循环
**反馈**是系统学习的基础机制:
```
┌─────────────────────────────────────────────────┐
│ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐│
│ │ Action │ ───→ │ Effect │ ───→ │Reward ││
│ └─────────┘ └─────────┘ └─────────┘│
│ │ │ │
│ │ ┌────────────┐ │ │
│ └───────────→│ Update │←──────┘ │
│ ↑ │
│ │ │
│ ┌──────┴──────┐ │
│ │ Policy │ │
│ │ Improvement│ │
│ └─────────────┘ │
│ │
└─────────────────────────────────────────────────┘
```
**反馈的类型**
| 类型 | 说明 | 例子 |
|-----|------|------|
| **正反馈** | 强化正确行为 | 生态廊道有效,增加类似策略 |
| **负反馈** | 抑制错误行为 | 阻力面不合理,调整权重 |
| **延迟反馈** | 效果滞后 | 生态工程几年后才见效 |
| **隐式反馈** | 未明确标注 | 用户不使用某功能 = 不好用 |
### 强化学习的直觉
强化学习(RL)是关于"如何通过试错学习":
```
强化学习核心概念
智能体 ──→ 动作 ──→ 环境 ──→ 奖励
↑ │
│ │
└─────────────── 观察状态 ←──────────────┘
更新策略
```
**关键要素**
1. **状态 (State)**:智能体看到的当前情况
2. **动作 (Action)**:智能体能做的事情
3. **奖励 (Reward)**:动作好坏的即时反馈
4. **策略 (Policy)**:状态到动作的映射规则
5. **价值函数 (Value)**:对长期收益的估计
```python
# RL的数学直觉
# 策略:在状态s采取动作a的概率
π(a|s) = P(action=a | state=s)
# 价值函数:从状态s开始的期望累积奖励
V(s) = E[Σ γ^t * r_t | s_0 = s]
# γ是折扣因子,平衡即时和长期奖励
# 动作价值函数:在状态s采取动作a后的期望累积奖励
Q(s,a) = E[Σ γ^t * r_t | s_0 = s, a_0 = a]
# 目标:找到最优策略,最大化累积奖励
π* = argmax_π V^π(s)
```
### 探索与利用的权衡
RL中经典的困境:
```
探索 (Explore) vs 利用 (Exploit)
利用 探索
↓ ↓
选择已知最好的动作 尝试新动作
获得稳定奖励 可能发现更好动作
可能错过最优 可能浪费资源
```
**策略**
| 策略 | 方法 | 适用场景 |
|-----|------|---------|
| **ε-greedy** | 以ε概率随机探索 | 通用,简单 |
| **Boltzmann** | 按价值概率选择 | 需要细粒度控制 |
| **UCB** | 上置信界选择 | 需要理论保证 |
| **Thompson Sampling** | 采样后验概率 | 贝叶斯框架 |
```python
def epsilon_greedy_action(q_values, epsilon, n_actions):
"""
ε-greedy策略
Args:
q_values: 各动作的估计价值
epsilon: 探索概率
n_actions: 动作数量
Returns:
选择的动作
"""
if np.random.random() < epsilon:
# 探索:随机选择
return np.random.randint(n_actions)
else:
# 利用:选择价值最高的
return np.argmax(q_values)
# ε的衰减策略
def epsilon_schedule(initial_epsilon, final_epsilon, total_steps, current_step):
"""线性衰减ε"""
decay = (initial_epsilon - final_epsilon) / total_steps
return max(final_epsilon, initial_epsilon - decay * current_step)
```
---
## 设计原理
### 奖励函数设计
奖励函数定义了"什么是好的行为":
```python
"""
奖励函数设计原则
"""
# 原则1: 清晰明确
# 好的奖励
def good_reward(ecological_quality):
"""生态质量越高,奖励越高"""
return ecological_quality
# 不好的奖励(有歧义)
def bad_reward(ecological_quality, cost):
"""混合多个目标,可能冲突"""
return ecological_quality - cost * 0.001
# 原则2: 适度塑形 (Reward Shaping)
# 不要过度引导,让智能体自己探索
def shaped_reward(base_reward, intermediate_metric):
"""
基础奖励 + 形状奖励
基础奖励:定义最终目标
形状奖励:引导到达目标(权重较小)
"""
return base_reward + 0.1 * intermediate_metric
# 原则3: 避免奖励黑客 (Reward Hacking)
# 防止智能体找到"作弊"方法
def safe_reward_with_constraints(action_result):
"""
带约束的奖励
"""
base_reward = action_result['quality']
# 如果违反约束,给予惩罚
if action_result['violates_constraint']:
base_reward -= 100 # 大惩罚
# 如果使用"作弊"方法,给予惩罚
if action_result['uses_exploit']:
base_reward -= 50
return base_reward
# 原则4: 多目标平衡
def multi_objective_reward(ecological, economic, social, weights):
"""
多目标加权
Args:
ecological: 生态效益
economic: 经济效益
social: 社会效益
weights: 各目标权重
Returns:
综合奖励
"""
# 归一化到[0,1]
normalized = {
'eco': min(ecological / 100, 1.0),
'eco': min(economic / 1000, 1.0),
'soc': min(social / 100, 1.0)
}
total = (weights['eco'] * normalized['eco'] +
weights['eco'] * normalized['eco'] +
weights['soc'] * normalized['soc'])
return total
```
### 在空间分析中的应用
```python
class SpatialOptimizerRL:
"""
用强化学习优化空间布局
场景:给定区域内选择最优生态廊道路线
"""
def __init__(self, landscape, constraints):
self.landscape = landscape
self.constraints = constraints
# 状态空间:当前的廊道路线
# 动作空间:下一步走向哪个像元
# 奖励:连通性、距离、穿越地类的综合
def state_representation(self):
"""将当前空间格局转换为状态表示"""
return {
'current_position': self.current_position,
'visited_cells': self.visited_cells,
'local_context': self._get_local_context()
}
def available_actions(self):
"""获取可用的动作"""
# 可以向8个方向移动
directions = [
(0, 1), (1, 0), (0, -1), (-1, 0), # 上下左右
(1, 1), (1, -1), (-1, 1), (-1, -1) # 对角
]
actions = []
for dx, dy in directions:
new_x = self.current_position[0] + dx
new_y = self.current_position[1] + dy
if self._is_valid_move(new_x, new_y):
actions.append((new_x, new_y))
return actions
def reward_function(self, action, new_state):
"""
定义奖励函数
考虑:
1. 穿越的土地类型(林地奖励,城市惩罚)
2. 距离目标的远近(越近越好)
3. 是否到达目标(大奖励)
"""
x, y = new_state['position']
# 1. 土地类型奖励/惩罚
land_type = self.landscape[y, x]
land_rewards = {
'forest': 10,
'grassland': 5,
'wetland': 8,
'agriculture': 0,
'urban': -50,
'water': -20
}
land_reward = land_rewards.get(land_type, -10)
# 2. 距离奖励(离目标越近越好)
dist_to_goal = self._distance_to_goal(new_state['position'])
distance_reward = -dist_to_goal * 0.1
# 3. 目标到达奖励
goal_reward = 0
if new_state['position'] == self.goal_position:
goal_reward = 1000
# 4. 约束惩罚
constraint_penalty = 0
if self._violates_constraint(new_state):
constraint_penalty = -100
# 总奖励
total_reward = (land_reward + distance_reward +
goal_reward + constraint_penalty)
return total_reward
def train(self, n_episodes=1000):
"""
训练智能体
使用Q-learning
"""
q_table = {} # Q值表
for episode in range(n_episodes):
state = self._reset()
epsilon = self._epsilon_schedule(episode)
done = False
while not done:
# ε-greedy选择动作
if np.random.random() < epsilon:
action = np.random.choice(self.available_actions())
else:
# 选择Q值最高的动作
q_values = [q_table.get((state, a), 0)
for a in self.available_actions()]
action = self.available_actions()[np.argmax(q_values)]
# 执行动作
new_state, reward, done = self._step(action)
# 更新Q值
old_q = q_table.get((state, action), 0)
max_next_q = max([q_table.get((new_state, a), 0)
for a in self.available_actions()] + [0])
# Q-learning更新公式
q_table[(state, action)] = old_q + 0.1 * (
reward + 0.99 * max_next_q - old_q
)
state = new_state
return q_table
```
---
## 代码示例
### 简化的生态网络优化RL
```python
"""
简化版:用Q-learning优化生态源地选择
"""
import numpy as np
from typing import List, Dict, Tuple
import random
class EcologicalNetworkOptimizer:
"""
生态网络优化器(RL简化版)
问题:从候选源地中选择最优组合
- 最大化总生态价值
- 满足连通性要求
- 预算约束
"""
def __init__(self, candidate_sites: List[Dict], budget: float):
self.candidate_sites = candidate_sites
self.budget = budget
# 动作:选择或不选择某个源地
self.n_actions = len(candidate_sites)
# 状态:已选源地列表
# 简化:用位掩码表示状态
self.n_states = 2 ** self.n_actions
# Q表
self.q_table = np.zeros((self.n_states, self.n_actions))
def state_to_mask(self, state: int) -> List[bool]:
"""状态索引转位掩码"""
return [(state >> i) & 1 for i in range(self.n_actions)]
def mask_to_state(self, mask: List[bool]) -> int:
"""位掩码转状态索引"""
state = 0
for i, bit in enumerate(mask):
if bit:
state |= (1 << i)
return state
def available_actions(self, state_mask: List[bool]) -> List[int]:
"""获取可用动作(未选的源地)"""
return [i for i, selected in enumerate(state_mask) if not selected]
def reward_function(self, state_mask: List[bool]) -> float:
"""
计算当前选择的奖励
考虑:
1. 总生态价值
2. 连通性
3. 预算约束
"""
# 选中源地
selected_sites = [self.candidate_sites[i]
for i, selected in enumerate(state_mask) if selected]
if not selected_sites:
return 0
# 1. 总生态价值
total_value = sum(site['value'] for site in selected_sites)
# 2. 连通性(简化:已选源地之间的平均距离)
if len(selected_sites) > 1:
positions = [(site['x'], site['y']) for site in selected_sites]
distances = []
for i in range(len(positions)):
for j in range(i + 1, len(positions)):
dist = np.sqrt((positions[i][0] - positions[j][0])**2 +
(positions[i][1] - positions[j][1])**2)
distances.append(dist)
avg_distance = np.mean(distances)
connectivity_reward = -0.1 * avg_distance # 距离越近越好
else:
connectivity_reward = 0
# 3. 预算惩罚
total_cost = sum(site['cost'] for site in selected_sites)
budget_penalty = 0
if total_cost > self.budget:
budget_penalty = -100 * (total_cost - self.budget) / self.budget
# 总奖励
total_reward = total_value + connectivity_reward + budget_penalty
return total_reward
def step(self, state: int, action: int) -> Tuple[int, float, bool]:
"""
执行一步
Returns:
next_state: 下一个状态
reward: 奖励
done: 是否结束
"""
state_mask = self.state_to_mask(state)
# 执行动作(选择一个源地)
if state_mask[action]: # 已经选过了
return state, -100, True # 惩罚并结束
new_mask = state_mask.copy()
new_mask[action] = True
# 计算奖励
reward = self.reward_function(new_mask)
# 检查是否结束(预算用完或所有源地都选了)
total_cost = sum(self.candidate_sites[i]['cost']
for i, selected in enumerate(new_mask) if selected)
done = (total_cost >= self.budget) or (sum(new_mask) == len(new_mask))
next_state = self.mask_to_state(new_mask)
return next_state, reward, done
def train(self, n_episodes=1000, alpha=0.1, gamma=0.99,
epsilon_start=1.0, epsilon_end=0.01):
"""
Q-learning训练
Args:
n_episodes: 训练回合数
alpha: 学习率
gamma: 折扣因子
epsilon_start: 初始探索率
epsilon_end: 最终探索率
"""
for episode in range(n_episodes):
# 线性衰减ε
epsilon = epsilon_start - (epsilon_start - epsilon_end) * episode / n_episodes
state = 0 # 初始状态(空)
done = False
while not done:
state_mask = self.state_to_mask(state)
available = self.available_actions(state_mask)
if not available:
break
# ε-greedy
if np.random.random() < epsilon:
action = random.choice(available)
else:
q_values = [self.q_table[state, a] for a in available]
action = available[np.argmax(q_values)]
# 执行动作
next_state, reward, done = self.step(state, action)
# Q-learning更新
old_q = self.q_table[state, action]
next_max = np.max(self.q_table[next_state])
self.q_table[state, action] = old_q + alpha * (
reward + gamma * next_max - old_q
)
state = next_state
# 定期报告
if episode % 100 == 0:
current_epsilon = epsilon_start - (epsilon_start - epsilon_end) * episode / n_episodes
print(f"Episode {episode}, ε={current_epsilon:.3f}, "
f"Best Q: {np.max(self.q_table[0]):.2f}")
return self.q_table
def get_solution(self) -> List[Dict]:
"""获取学习到的最优解"""
state = 0
state_mask = self.state_to_mask(state)
solution = []
while True:
available = self.available_actions(state_mask)
if not available:
break
# 选择Q值最高的动作
q_values = [self.q_table[state, a] for a in available]
action = available[np.argmax(q_values)]
solution.append(self.candidate_sites[action])
state, _, done = self.step(state, action)
if done:
break
return solution
# 示例使用
def example_usage():
"""示例使用"""
print("=== 生态网络优化:Q-learning ===\n")
# 创建候选源地
np.random.seed(42)
n_candidates = 10
candidates = []
for i in range(n_candidates):
candidates.append({
'id': i,
'x': np.random.randint(0, 100),
'y': np.random.randint(0, 100),
'value': np.random.randint(50, 150),
'cost': np.random.randint(20, 80)
})
budget = 200
print(f"候选源地数: {n_candidates}")
print(f"预算: {budget}\n")
# 创建优化器并训练
optimizer = EcologicalNetworkOptimizer(candidates, budget)
optimizer.train(n_episodes=500)
# 获取解
solution = optimizer.get_solution()
print("\n=== 最优解 ===")
print(f"选择源地数: {len(solution)}")
total_value = sum(s['value'] for s in solution)
total_cost = sum(s['cost'] for s in solution)
print(f"总价值: {total_value}")
print(f"总成本: {total_cost}")
print("\n选择的源地:")
for s in solution:
print(f" 源地 {s['id']}: 价值={s['value']}, 成本={s['cost']}")
if __name__ == "__main__":
example_usage()
```
---
## 案例分析
### ENAgent中的反馈机制
```python
class ENAgentFeedback:
"""
ENAgent的反馈机制
场景:生态网络迭代的改进
"""
def __init__(self):
self.iteration_history = []
self.performance_metrics = []
def collect_feedback(self, iteration, result, human_feedback):
"""
收集每轮的反馈
Args:
iteration: 迭代次数
result: 本轮结果
human_feedback: 人类专家的反馈
"""
feedback_record = {
'iteration': iteration,
'result': result,
'human_feedback': human_feedback,
'timestamp': time.time()
}
self.iteration_history.append(feedback_record)
def analyze_feedback(self) -> Dict:
"""
分析反馈,提取改进建议
Returns:
改进建议
"""
if not self.iteration_history:
return {}
# 分析模式
suggestions = {}
# 1. 常见问题
problem_counts = {}
for record in self.iteration_history:
for problem in record['human_feedback'].get('problems', []):
problem_counts[problem] = problem_counts.get(problem, 0) + 1
if problem_counts:
common_problems = sorted(problem_counts.items(),
key=lambda x: x[1], reverse=True)
suggestions['common_problems'] = common_problems
# 2. 趋势分析
if len(self.iteration_history) > 1:
recent_quality = self.iteration_history[-1]['result']['quality']
previous_quality = self.iteration_history[-2]['result']['quality']
if recent_quality > previous_quality:
suggestions['trend'] = 'improving'
else:
suggestions['trend'] = 'stagnant_or_degrading'
# 3. 参数调整建议
suggestions['parameter_adjustments'] = self._suggest_adjustments()
return suggestions
def _suggest_adjustments(self) -> Dict:
"""建议参数调整"""
# 基于反馈历史,建议如何调整参数
# 这是一个简化示例
return {
'resistance_weights': 'consider adjusting urban weight',
'source_threshold': 'might be too high/low'
}
```
---
## 反思与延伸
### 思考问题
1. **延迟奖励**:生态工程的效果多年后才显现,如何设计奖励函数?
2. **稀疏奖励**:当大多数步骤没有明确反馈时,如何学习?
3. **多目标冲突**:生态目标和经济目标冲突时,奖励函数如何平衡?
4. **人类反馈**:如何整合人类专家的定性反馈?
### 实践练习
1. **奖励设计**:为一个你熟悉的任务设计奖励函数
2. **调试RL**:观察Q表的变化,理解学习过程
3. **探索策略**:比较不同ε衰减策略的效果
### 延伸阅读
- **"Reinforcement Learning: An Introduction"** (Sutton & Barto) - RL圣经
- **"Algorithms for Decision Making"** (Mykel Kochenderfer) - 决策与RL
- **"Reward Shaping"**论文 - 奖励塑形理论
---
## 关键要点
1. **反馈是学习的基础机制**,正反馈强化正确行为,负反馈纠正错误
2. **强化学习核心**:状态、动作、奖励、策略、价值函数
3. **探索vs利用**:经典困境,需要平衡策略
4. **奖励函数设计**是RL的关键,定义了"什么是好的行为"
5. **在空间分析中**:RL可用于优化布局、路径选择、参数调整
@@ -0,0 +1,761 @@
# 01.5 人机协同的原理
## 核心问题
> 人类和AI各自的优势是什么?如何互补?
> 何时需要人类介入?如何设计审查点?
> 如何建立和维护对AI系统的信任?
---
## 概念讲解
### 人类与AI的能力对比
```
┌─────────────────────────────────────────────────────────────┐
│ 人类 vs AI 能力对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 能力维度 │ 人类 │ AI │
│ ───────────── │ ──────────── │ ───────────────── │
│ │
│ 模式识别 │ 不擅长大量 │ 非常擅长 │
│ │ 数据的模式 │ 大规模模式识别 │
│ │
│ 语义理解 │ 深度理解 │ 表层理解 │
│ │ 上下文关联 │ 统计关联 │
│ │
│ 创造力 │ 原创性强 │ 组合创新 │
│ │ 跳跃思维 │ 已有模式重组 │
│ │
│ 伦理判断 │ 天然具备 │ 需要显式编码 │
│ │ 直觉道德 │ 规则约束 │
│ │
│ 不确定性处理 │ 直觉判断 │ 概率计算 │
│ │ 启发式 │ 量化评估 │
│ │
│ 知识获取 │ 慢,深度 │ 快,广度 │
│ │ 需要学习 │ 即时查询 │
│ │
│ 注意力控制 │ 有限,易疲劳 │ 不知疲倦 │
│ │ 可自主转移 │ 需要任务定义 │
│ │
│ 可解释性 │ 可事后解释 │ 需要专门设计 │
│ │ 理由可能模糊 │ 逻辑清晰 │
│ │
└─────────────────────────────────────────────────────────────┘
```
### HITL的理论基础
**Human-in-the-Loop (HITL)** 不仅仅是"让人检查结果",而是有理论基础的系统设计方法:
```
HITL的理论支撑
┌─────────────────────────────────────────────────────────────┐
│ │
│ 1. 互补性原理 │
│ 人类和AI有互补优势,结合优于单独使用 │
│ │
│ 2. 控制论原理 │
│ 人类作为反馈回路的一部分,可以校正系统偏差 │
│ │
│ 3. 信任校准 │
│ 通过参与建立对AI能力的准确认知 │
│ │
│ 4. 价值对齐 │
│ 人类介入确保AI行为与人类价值观一致 │
│ │
│ 5. 责任归属 │
│ 人类在关键决策点参与,明确责任边界 │
│ │
└─────────────────────────────────────────────────────────────┘
```
### 信任建立的动态
```
信任建立过程
时间
│ ┌────────────┐
│ │ 初始信任 │ 基于声誉、宣传等
│ └─────┬──────┘
│ │ 第一次使用
│ ↓
│ ┌────────────┐
│ │ 体验信任 │ 基于实际交互
│ └─────┬──────┘
│ │
│ ┌──────┴──────┐
│ ↓ ↓
│ 成功 失败
│ │ │
│ ↓ ↓
│ ┌─────┐ ┌─────┐
│ │信任 │ │不信任│
│ │增强 │ │/怀疑 │
│ └──┬──┘ └──┬──┘
│ │ │
│ └──────┬──────┘
│ │ 解释/透明度
│ ↓
│ ┌────────────┐
│ │ 校准信任 │ 与能力匹配的信任水平
│ └────────────┘
└──────────────────────────→
```
---
## 设计原理
### 何时需要人类介入
**决策框架**:根据任务的特性决定介入程度
```python
def human_intervention_necessity(task_characteristics: Dict) -> str:
"""
评估任务需要人类介入的程度
Args:
task_characteristics: 任务特性描述
Returns:
介入程度: 'full', 'selective', 'minimal', 'none'
"""
scores = {
'consequence': 0, # 后果严重性
'uncertainty': 0, # 不确定性
'ethical': 0, # 伦理敏感性
'complexity': 0, # 复杂度
'novelty': 0 # 新颖性
}
# 评估后果严重性
if task_characteristics.get('life_critical', False):
scores['consequence'] = 3
elif task_characteristics.get('economic_impact', 0) > 1000000:
scores['consequence'] = 2
elif task_characteristics.get('economic_impact', 0) > 100000:
scores['consequence'] = 1
# 评估不确定性
uncertainty = task_characteristics.get('uncertainty_level', 'low')
scores['uncertainty'] = {'low': 0, 'medium': 1, 'high': 2}[uncertainty]
# 评估伦理敏感性
if task_characteristics.get('ethical_concerns', False):
scores['ethical'] = 3
# 评估复杂度
complexity = task_characteristics.get('complexity', 'low')
scores['complexity'] = {'low': 0, 'medium': 1, 'high': 2}[complexity]
# 评估新颖性
if task_characteristics.get('novel_situation', False):
scores['novelty'] = 2
# 总分
total_score = sum(scores.values())
# 决定介入程度
if total_score >= 10:
return 'full' # 完全由人类主导
elif total_score >= 6:
return 'selective' # 关键点介入
elif total_score >= 3:
return 'minimal' # 异常时介入
else:
return 'none' # AI自主执行
```
**ENAgent的三个审查点设计依据**
| 审查点 | 任务特性 | 介入理由 |
|-------|---------|---------|
| 源地识别 | 高不确定性 + 本地知识需求 | 遥感分类可能错误,地面实况重要 |
| 阻力权重 | 高价值判断 + 物种特异性 | 不同物种权重差异大,专家知识关键 |
| 廊道优化 | 多目标权衡 + 社会影响 | 生态与经济/社会的平衡,人类决策 |
### 信任校准机制
```python
class TrustCalibration:
"""
信任校准系统
目标:让用户的信任水平与AI的实际能力匹配
"""
def __init__(self):
self.declared_confidence = [] # AI声明的置信度
self.actual_performance = [] # 实际表现
self.user_trust_level = 0.5 # 用户信任水平
def record_outcome(self, ai_confidence: float,
actual_correct: bool,
user_trusted: bool):
"""
记录一次AI决策的结果
Args:
ai_confidence: AI声明的置信度 [0, 1]
actual_correct: 实际是否正确
user_trusted: 用户是否信任并采用了AI建议
"""
self.declared_confidence.append(ai_confidence)
self.actual_performance.append(1.0 if actual_correct else 0.0)
def assess_calibration(self) -> Dict:
"""
评估AI的校准程度
Returns:
校准报告
"""
if not self.declared_confidence:
return {'status': 'insufficient_data'}
# 按置信度分组统计
confidence_bins = {
'high': [], # > 0.8
'medium': [], # 0.5-0.8
'low': [] # < 0.5
}
for conf, perf in zip(self.declared_confidence, self.actual_performance):
if conf > 0.8:
confidence_bins['high'].append(perf)
elif conf > 0.5:
confidence_bins['medium'].append(perf)
else:
confidence_bins['low'].append(perf)
# 计算各组的平均实际表现
calibration_report = {}
for bin_name, performances in confidence_bins.items():
if performances:
avg_performance = sum(performances) / len(performances)
calibration_report[bin_name] = {
'ai_declared_range': self._get_bin_range(bin_name),
'actual_accuracy': avg_performance,
'calibration_gap': avg_performance - self._get_bin_expected(bin_name)
}
return calibration_report
def _get_bin_range(self, bin_name: str) -> str:
ranges = {
'high': '> 0.8',
'medium': '0.5-0.8',
'low': '< 0.5'
}
return ranges[bin_name]
def _get_bin_expected(self, bin_name: str) -> float:
"""该置信度组的期望表现"""
expected = {
'high': 0.9,
'medium': 0.65,
'low': 0.25
}
return expected[bin_name]
def recommend_trust_adjustment(self) -> str:
"""
基于校准结果,建议信任调整
Returns:
调整建议
"""
calibration = self.assess_calibration()
if calibration.get('status') == 'insufficient_data':
return "需要更多数据来评估"
overconfident = any(
v['calibration_gap'] < -0.1
for v in calibration.values()
if isinstance(v, dict)
)
underconfident = any(
v['calibration_gap'] > 0.1
for v in calibration.values()
if isinstance(v, dict)
)
if overconfident:
return "AI倾向于过度自信,建议降低信任度,增加审查"
elif underconfident:
return "AI实际表现优于声明,可以增加信任"
else:
return "AI校准良好,当前信任水平适当"
```
### 审查点设计模式
```python
class CheckpointDesign:
"""
审查点设计框架
"""
@staticmethod
def design_checkpoint(task_info: Dict) -> Dict:
"""
为任务设计审查点
Args:
task_info: 任务信息
Returns:
审查点设计
"""
checkpoint = {
'name': task_info['name'],
'trigger_condition': None,
'information_provided': [],
'decision_options': [],
'default_action': None,
'timeout_handling': None
}
# 1. 触发条件设计
checkpoint['trigger_condition'] = CheckpointDesign._design_trigger(task_info)
# 2. 信息提供设计
checkpoint['information_provided'] = CheckpointDesign._design_info_display(task_info)
# 3. 决策选项设计
checkpoint['decision_options'] = CheckpointDesign._design_options(task_info)
# 4. 默认行为
checkpoint['default_action'] = CheckpointDesign._design_default(task_info)
return checkpoint
@staticmethod
def _design_trigger(task_info: Dict) -> Dict:
"""设计触发条件"""
return {
'type': 'conditional', # always, conditional, on_error
'conditions': [
'confidence_below_threshold',
'conflicting_alternatives',
'ethical_concern_detected'
],
'threshold': task_info.get('confidence_threshold', 0.7)
}
@staticmethod
def _design_info_display(task_info: Dict) -> List[str]:
"""设计展示给人类的信息"""
base_info = [
'ai_proposal',
'confidence_level',
'reasoning_trace'
]
# 根据任务类型添加额外信息
if task_info.get('high_stakes', False):
base_info.extend([
'consequence_analysis',
'alternative_options'
])
if task_info.get('uncertain', False):
base_info.append('uncertainty_quantification')
return base_info
@staticmethod
def _design_options(task_info: Dict) -> List[str]:
"""设计人类决策选项"""
base_options = ['approve', 'reject', 'modify']
if task_info.get('allow_delegation', False):
base_options.append('delegate_to_ai')
return base_options
@staticmethod
def _design_default(task_info: Dict) -> str:
"""设计默认行为(人类不响应时)"""
if task_info.get('high_stakes', False):
return 'wait_for_human' # 等待人类
else:
return 'proceed_with_caution' # 谨慎继续
```
---
## 代码示例
### 完整的HITL工作流实现
```python
"""
完整的人机协同工作流实现
"""
import time
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass
from enum import Enum
class HumanDecision(Enum):
"""人类决策类型"""
APPROVE = "approve"
REJECT = "reject"
MODIFY = "modify"
DEFER = "defer"
REQUEST_INFO = "request_info"
@dataclass
class CheckpointResult:
"""审查点结果"""
checkpoint_name: str
decision: HumanDecision
modifications: Optional[Dict] = None
additional_input: Optional[Dict] = None
timestamp: float = None
class HITLWorkflow:
"""
人机协同工作流
"""
def __init__(self, name: str):
self.name = name
self.checkpoints: Dict[str, Dict] = {}
self.state = {}
self.history: List[CheckpointResult] = []
def add_checkpoint(self,
name: str,
trigger: Callable,
info_formatter: Callable = None,
critical: bool = False):
"""
添加审查点
Args:
name: 审查点名称
trigger: 触发条件函数,返回True时需要审查
info_formatter: 信息格式化函数
critical: 是否为关键审查点
"""
self.checkpoints[name] = {
'trigger': trigger,
'info_formatter': info_formatter or (lambda x: x),
'critical': critical,
'activated': False
}
def execute_step(self,
step_name: str,
step_function: Callable,
**kwargs) -> Dict:
"""
执行工作流步骤
Args:
step_name: 步骤名称
step_function: 执行函数
**kwargs: 传递给函数的参数
Returns:
执行结果
"""
print(f"\n{'='*50}")
print(f"执行步骤: {step_name}")
print('='*50)
# 检查是否有审查点
checkpoint = self.checkpoints.get(step_name)
if checkpoint:
# 执行步骤
result = step_function(self.state, **kwargs)
# 格式化信息
info = checkpoint['info_formatter'](result)
# 检查是否需要触发审查
if checkpoint['trigger'](result, self.state):
print(f"\n[审查点触发: {step_name}]")
checkpoint['activated'] = True
# 获取人类决策
decision = self._get_human_decision(info, step_name)
# 记录决策
self.history.append(CheckpointResult(
checkpoint_name=step_name,
decision=decision['type'],
modifications=decision.get('modifications'),
timestamp=time.time()
))
# 根据决策处理
if decision['type'] == HumanDecision.APPROVE:
print("✓ 人类批准,继续执行")
self.state[step_name] = result
elif decision['type'] == HumanDecision.REJECT:
print("✗ 人类拒绝,回退")
return {'status': 'rejected', 'checkpoint': step_name}
elif decision['type'] == HumanDecision.MODIFY:
print("✎ 人类修改结果")
result = self._apply_modifications(result, decision['modifications'])
self.state[step_name] = result
elif decision['type'] == HumanDecision.DEFER:
print("⏸ 暂停,等待更多信息")
return {'status': 'deferred', 'checkpoint': step_name}
else:
print(f"审查点未触发(条件不满足),自动继续")
self.state[step_name] = result
else:
# 没有审查点,直接执行
result = step_function(self.state, **kwargs)
self.state[step_name] = result
return result
def _get_human_decision(self, info: Dict, checkpoint_name: str) -> Dict:
"""
获取人类决策
实际实现中可能是GUI、CLI或其他交互方式
"""
print("\n" + "-"*40)
print("信息摘要:")
for key, value in info.items():
print(f" {key}: {value}")
print("\n可用决策:")
print(" 1. 批准 (approve)")
print(" 2. 拒绝 (reject)")
print(" 3. 修改 (modify)")
# 模拟人类输入
# 实际实现中等待真实输入
choice = "1" # 默认批准
decisions = {
"1": HumanDecision.APPROVE,
"2": HumanDecision.REJECT,
"3": HumanDecision.MODIFY
}
return {'type': decisions[choice]}
def _apply_modifications(self, original: Dict, modifications: Dict) -> Dict:
"""应用人类修改"""
if modifications:
original.update(modifications)
return original
def get_checkpoint_summary(self) -> Dict:
"""获取审查点摘要"""
return {
'total_checkpoints': len(self.checkpoints),
'activated_checkpoints': sum(1 for c in self.checkpoints.values() if c['activated']),
'human_decisions': [
{
'checkpoint': r.checkpoint_name,
'decision': r.decision.value,
'timestamp': r.timestamp
}
for r in self.history
]
}
# 示例:生态网络分析的HITL工作流
def ecological_hitl_example():
"""生态网络分析HITL示例"""
workflow = HITLWorkflow("ecological_network_analysis")
# 步骤1:加载数据(无审查)
def load_data(state):
print("加载土地利用数据...")
return {'data_loaded': True, 'n_pixels': 10000}
# 步骤2:识别源地(有审查)
def identify_sources(state):
print("识别生态源地...")
sources = [
{'id': 1, 'area': 1500, 'confidence': 0.85},
{'id': 2, 'area': 800, 'confidence': 0.65},
{'id': 3, 'area': 2000, 'confidence': 0.92}
]
return {'sources': sources, 'n_sources': len(sources)}
# 审查条件:有低置信度源地时触发
def source_trigger(result, state):
return any(s['confidence'] < 0.7 for s in result['sources'])
# 信息格式化
def format_source_info(result):
return {
'识别源地数': result['n_sources'],
'平均置信度': sum(s['confidence'] for s in result['sources']) / result['n_sources'],
'低置信度源地': [s['id'] for s in result['sources'] if s['confidence'] < 0.7]
}
workflow.add_checkpoint(
'identify_sources',
trigger=source_trigger,
info_formatter=format_source_info,
critical=True
)
# 步骤3:构建阻力面(有审查)
def build_resistance(state):
print("构建阻力面...")
return {'weights': {'forest': 1, 'urban': 100}, 'built': True}
# 审查条件:总是触发
def resistance_trigger(result, state):
return True # 权重设置总是需要人类审查
def format_resistance_info(result):
return result['weights']
workflow.add_checkpoint(
'build_resistance',
trigger=resistance_trigger,
info_formatter=format_resistance_info
)
# 执行工作流
print("=== 开始执行HITL工作流 ===")
workflow.execute_step('load_data', load_data)
workflow.execute_step('identify_sources', identify_sources)
workflow.execute_step('build_resistance', build_resistance)
# 摘要
summary = workflow.get_checkpoint_summary()
print("\n=== 工作流摘要 ===")
print(f"总审查点: {summary['total_checkpoints']}")
print(f"激活审查点: {summary['activated_checkpoints']}")
print("人类决策:")
for decision in summary['human_decisions']:
print(f" {decision['checkpoint']}: {decision['decision']}")
if __name__ == "__main__":
ecological_hitl_example()
```
---
## 案例分析
### ENAgent的审查点实现
```python
class ENAgentHITL:
"""
ENAgent的人机协同实现
"""
def __init__(self):
self.review_points = {
'source_identification': SourceReview(),
'resistance_surface': ResistanceReview(),
'corridor_extraction': CorridorReview()
}
class SourceReview:
"""源地识别审查"""
def trigger_condition(self, sources):
"""触发条件"""
# 条件1:有低置信度源地
low_confidence = any(s['confidence'] < 0.7 for s in sources)
# 条件2:源地数量异常
abnormal_count = len(sources) < 3 or len(sources) > 20
# 条件3:源地分布极不均匀
if len(sources) >= 2:
areas = [s['area'] for s in sources]
area_range = max(areas) - min(areas)
uneven = area_range > 10 * sum(areas) / len(areas)
else:
uneven = False
return low_confidence or abnormal_count or uneven
def format_for_review(self, sources):
"""格式化信息供审查"""
return {
'n_sources': len(sources),
'sources_by_confidence': sorted(sources,
key=lambda x: x['confidence']),
'spatial_distribution': self._analyze_distribution(sources),
'potential_issues': self._detect_issues(sources)
}
def _detect_issues(self, sources):
"""检测潜在问题"""
issues = []
if len(sources) < 3:
issues.append("源地数量偏少,可能遗漏重要栖息地")
low_conf = [s for s in sources if s['confidence'] < 0.7]
if low_conf:
issues.append(f"{len(low_conf)}个源地置信度低于0.7")
return issues
```
---
## 反思与延伸
### 思考问题
1. **责任边界**:当HITL系统出错时,责任应该如何划分?
2. **审查疲劳**:如果审查点太多,人类会产生疲劳,如何平衡?
3. **信任过度**:如何防止人类过度信任AI而减少必要的审查?
4. **可解释性**AI应该如何向人类解释其推理过程?
### 实践练习
1. **审查点设计**:为你熟悉的流程设计审查点
2. **信任评估**:记录你使用AI工具的经历,评估信任变化
3. **HITL实现**:实现一个简单的HITL工作流
### 延伸阅读
- **"Human-in-the-Loop Machine Learning"** - HITL系统设计
- **"Human-Centered AI"** (Ben Shneiderman) - 以人为本的AI
- **"Explainable AI"**论文集 - 可解释AI研究
---
## 关键要点
1. **人类和AI有互补优势**,结合优于单独使用
2. **HITL不是妥协**,而是有理论基础的系统设计方法
3. **审查点选择关键**:在需要人类独特能力的决策点介入
4. **信任需要校准**:让信任水平与实际能力匹配
5. **责任必须明确**:关键决策点的人类参与确保责任归属
@@ -0,0 +1,181 @@
# 第二部分:基础原理
## 本部分目标
理解现代AI系统的核心设计原理,超越具体工具:
- 智能系统的模块化设计思想
- 状态与状态机的设计哲学
- 概率思维与不确定性处理
- 反馈机制与学习原理
- 人机协同的理论基础
---
## 章节导航
| 章节 | 文件 | 核心问题 | 实践 |
|-----|------|---------|------|
| 01.1 | [智能的模块化视角](./01.1-modular-intelligence.md) | 为什么要模块化?技能如何封装? | QGIS技能架构分析 |
| 01.2 | [状态与状态机](./01.2-state-and-state-machines.md) | 状态是什么?为何重要? | 简单工作流状态机 |
| 01.3 | [概率与不确定性](./01.3-probability-and-uncertainty.md) | AI如何处理未知? | 生态源地识别不确定性 |
| 01.4 | [反馈与学习](./01.4-feedback-and-learning.md) | 系统如何改进? | 生态网络优化示例 |
| 01.5 | [人机协同的原理](./01.5-human-ai-collaboration.md) | 何时需要人类介入? | ENAgent审查点设计 |
---
## 学习路径
```
┌─────────────────┐
│ 01-foundations │
└────────┬────────┘
┌────────────────────┼────────────────────┐
│ │ │
↓ ↓ ↓
┌──────────┐ ┌──────────┐ ┌──────────┐
│设计思维 │ │系统思维 │ │协作思维 │
│01.1, 01.2│ │01.3, 01.4│ │ 01.5 │
└──────────┘ └──────────┘ └──────────┘
│ │ │
└────────────────────┼────────────────────┘
┌─────────────────┐
│ 综合理解 │
│ AI系统设计 │
└─────────────────┘
```
---
## 核心概念图谱
```
┌─────────────────────────────────────┐
│ AI系统设计核心 │
└─────────────────────────────────────┘
┌──────────────────────────────┼──────────────────────────────┐
│ │ │
↓ ↓ ↓
┌───────────────┐ ┌───────────────┐ ┌───────────────┐
│ 模块化 │ │ 状态机 │ │ 反馈循环 │
│ ────────── │ │ ────────── │ │ ────────── │
│ 技能封装 │ │ 工作流编排 │ │ 学习优化 │
│ 接口设计 │ │ 条件分支 │ │ 奖励信号 │
│ 组合模式 │ │ 错误处理 │ │ 探索利用 │
└───────────────┘ └───────────────┘ └───────────────┘
│ │ │
└──────────────────────────────┼──────────────────────────────┘
┌─────────────────────────────────────────────────────────────┐
│ 人机协同层 │
│ ───────────────────────────────────────────────── │
│ 何时介入 │ 如何信任 │ 责任边界 │ 互补优势 │
└─────────────────────────────────────────────────────────────┘
```
---
## 前置知识
**必需**
- Python面向对象编程基础
- 函数式编程概念(高阶函数、map/reduce)
- 基本的数据结构(图、树、字典)
**有助理解**
- 设计模式基础
- 状态机概念
- 概率论基础
---
## 预计学习时间
| 阅读类型 | 时间估计 |
|---------|---------|
| 快速浏览 | 3-4小时 |
| 理解性阅读 | 10-15小时 |
| 完成所有实践 | 20-25小时 |
---
## 章节亮点
### 01.1 智能的模块化视角
- 从QGIS插件架构理解模块化
- 函数式组合思想
- 技能即能力封装的设计理念
### 01.2 状态与状态机
- 为什么状态管理是核心
- LangGraph的状态设计哲学
- 工作流的状态机实现
### 01.3 概率与不确定性
- 空间分析中的不确定性来源
- 置信度的表示和传播
- 鲁棒决策的方法
### 01.4 反馈与学习
- 强化学习的直觉理解
- 奖励函数设计原则
- 探索与利用的权衡
### 01.5 人机协同的原理
- HITL的理论基础
- 信任校准机制
- 责任边界划分
---
## 实践案例01:用LangGraph构建空间决策工作流
详见 [practice/langgraph-workflow](./practice/langgraph-workflow/)
### 实践目标
1. 理解状态驱动的Agent设计
2. 实现一个简单的空间决策工作流
3. 添加Human-in-the-Loop审查点
4. 处理工作流中的错误和重试
---
## 思考框架
在学习每章时,问自己:
1. **概念理解**:这个概念解决了什么问题?
2. **设计权衡**:为什么这样设计?有哪些替代方案?
3. **实际应用**:这个原理在ENAgent中如何体现?
4. **迁移思考**:这个原理可以应用到我的工作中吗?
---
## 延伸资源
### 经典阅读
- **"Design Patterns"** (GoF) - 设计模式基础
- **"Introduction to Automata Theory"** - 状态机理论
- **"Reinforcement Learning: An Introduction"** - RL基础
### 在线资源
- LangGraph官方文档
- LangChain状态管理指南
- Human-in-the-Loop机器学习论文集
---
## 关键要点预览
1. **模块化是管理复杂性的核心方法**
2. **状态机是工作流编排的基础抽象**
3. **概率思维让AI能处理不确定性**
4. **反馈循环是学习和改进的机制**
5. **人机协同需要明确的责任边界**
> "原理是知识的骨架,工具是知识的血肉。骨架不变,血肉可生。"
@@ -0,0 +1,370 @@
# 实践案例01:用LangGraph构建空间决策工作流
## 目标
通过本实践,你将:
1. 理解状态驱动的Agent设计
2. 实现一个简单的空间决策工作流
3. 添加Human-in-the-Loop审查点
4. 处理工作流中的错误和重试
---
## 背景知识
### 什么是LangGraph
LangGraph是构建**有状态**的多Agent应用的框架:
```
核心概念:
1. State(状态): 在节点间传递的数据
2. Node(节点): 处理状态的函数
3. Edge(边): 节点之间的连接
4. Graph(图): 节点和边组成的完整工作流
```
### 为什么用LangGraph
- **状态管理**: 自动管理工作流状态
- **可视化**: 可以绘制和查看工作流图
- **持久化**: 支持中断和恢复
- **条件路由**: 基于状态动态选择路径
---
## 实践步骤
### 步骤1:安装依赖
```bash
pip install langgraph langchain-core langchain-anthropic
```
### 步骤2:定义状态
```python
from typing import TypedDict, Annotated, List, Optional
from operator import add
from typing_extensions import TypedDict
class EcologicalAnalysisState(TypedDict):
"""生态网络分析状态"""
# 输入
input_path: str
parameters: dict
# 处理过程
current_step: str
intermediate_results: dict
# 人机交互
review_requested: bool
human_feedback: Optional[str]
# 输出
final_result: Optional[dict]
errors: Annotated[List[str], add]
```
### 步骤3:定义节点
```python
def load_data_node(state: EcologicalAnalysisState) -> EcologicalAnalysisState:
"""加载数据节点"""
print("执行: load_data")
# 实际实现中读取文件
return {
**state,
"current_step": "data_loaded",
"intermediate_results": {"data": "loaded"}
}
def identify_sources_node(state: EcologicalAnalysisState) -> EcologicalAnalysisState:
"""识别源地节点"""
print("执行: identify_sources")
# 实际实现中运行源地识别算法
sources = [{"id": 1, "area": 1000}, {"id": 2, "area": 800}]
return {
**state,
"current_step": "sources_identified",
"intermediate_results": {**state["intermediate_results"], "sources": sources}
}
def human_review_node(state: EcologicalAnalysisState) -> EcologicalAnalysisState:
"""人类审查节点"""
print("执行: human_review")
print(f"待审查: {state['intermediate_results']}")
# 实际实现中等待人类输入
return {
**state,
"review_requested": False,
"human_feedback": "approved"
}
def build_resistance_node(state: EcologicalAnalysisState) -> EcologicalAnalysisState:
"""构建阻力面节点"""
print("执行: build_resistance")
return {
**state,
"current_step": "resistance_built"
}
```
### 步骤4:定义路由
```python
def should_review(state: EcologicalAnalysisState) -> str:
"""决定是否需要审查"""
sources = state["intermediate_results"].get("sources", [])
if len(sources) > 2: # 源地数量多时需要审查
return "review"
return "continue"
```
### 步骤5:构建图
```python
from langgraph.graph import StateGraph, END
def build_workflow():
"""构建工作流图"""
# 创建图
workflow = StateGraph(EcologicalAnalysisState)
# 添加节点
workflow.add_node("load_data", load_data_node)
workflow.add_node("identify_sources", identify_sources_node)
workflow.add_node("human_review", human_review_node)
workflow.add_node("build_resistance", build_resistance_node)
# 设置入口
workflow.set_entry_point("load_data")
# 添加边
workflow.add_edge("load_data", "identify_sources")
# 添加条件边
workflow.add_conditional_edges(
"identify_sources",
should_review,
{
"review": "human_review",
"continue": "build_resistance"
}
)
workflow.add_edge("human_review", "build_resistance")
workflow.add_edge("build_resistance", END)
# 编译
return workflow.compile()
```
### 步骤6:运行工作流
```python
def run_workflow():
"""运行工作流"""
# 初始状态
initial_state = {
"input_path": "data.geojson",
"parameters": {},
"current_step": "start",
"intermediate_results": {},
"review_requested": False,
"human_feedback": None,
"final_result": None,
"errors": []
}
# 构建并运行
app = build_workflow()
result = app.invoke(initial_state)
print("\n=== 最终结果 ===")
print(result)
```
---
## 扩展练习
### 1. 添加错误处理
```python
def with_error_handling(node_func):
"""装饰器:添加错误处理"""
def wrapper(state):
try:
return node_func(state)
except Exception as e:
return {
**state,
"errors": [str(e)]
}
return wrapper
# 使用
@with_error_handling
def risky_node(state):
# 可能出错的节点
...
```
### 2. 添加检查点(持久化)
```python
from langgraph.checkpoint.memory import MemorySaver
# 创建检查点保存器
memory = MemorySaver()
# 编译时添加检查点
app = workflow.compile(checkpointer=memory, interrupt_before=["human_review"])
# 运行时可以指定thread_id
config = {"configurable": {"thread_id": "conversation-1"}}
result = app.invoke(initial_state, config=config)
```
### 3. 可视化工作流
```python
from IPython.display import Image, display
# 生成图
app = build_workflow()
display(Image(app.get_graph().draw_mermaid_png()))
```
---
## 完整代码示例
```python
"""
完整的LangGraph空间决策工作流示例
"""
from typing import TypedDict, Annotated, List, Optional, Literal
from operator import add
from langgraph.graph import StateGraph, END
class State(TypedDict):
"""工作流状态"""
step: int
data: Optional[dict]
sources: Optional[list]
reviewed: bool
result: Optional[str]
errors: Annotated[List[str], add]
# 节点函数
def load_node(state: State) -> State:
"""加载数据"""
print(f"[节点: load] 步骤 {state['step']}")
return {**state, "step": state["step"] + 1, "data": {"loaded": True}}
def analyze_node(state: State) -> State:
"""分析数据"""
print(f"[节点: analyze] 步骤 {state['step']}")
return {
**state,
"step": state["step"] + 1,
"sources": [{"id": 1, "value": 100}]
}
def review_node(state: State) -> State:
"""人类审查"""
print(f"[节点: review] 步骤 {state['step']}")
print("等待人类审查...")
# 实际实现中等待输入
return {**state, "step": state["step"] + 1, "reviewed": True}
def finalize_node(state: State) -> State:
"""完成"""
print(f"[节点: finalize] 步骤 {state['step']}")
return {**state, "result": "completed"}
# 路由函数
def route_after_analyze(state: State) -> Literal["review", "finalize"]:
"""分析后的路由"""
if state.get("sources") and len(state["sources"]) > 0:
return "review"
return "finalize"
# 构建图
def build_graph():
"""构建工作流图"""
graph = StateGraph(State)
# 添加节点
graph.add_node("load", load_node)
graph.add_node("analyze", analyze_node)
graph.add_node("review", review_node)
graph.add_node("finalize", finalize_node)
# 添加边
graph.set_entry_point("load")
graph.add_edge("load", "analyze")
# 条件边
graph.add_conditional_edges(
"analyze",
route_after_analyze,
{"review": "review", "finalize": "finalize"}
)
graph.add_edge("review", "finalize")
graph.add_edge("finalize", END)
return graph.compile()
# 运行
if __name__ == "__main__":
print("=== LangGraph 空间决策工作流 ===\n")
app = build_graph()
initial_state: State = {
"step": 1,
"data": None,
"sources": None,
"reviewed": False,
"result": None,
"errors": []
}
result = app.invoke(initial_state)
print(f"\n最终状态: {result['step']}")
print(f"结果: {result['result']}")
```
---
## 反思问题
1. **状态设计**:你的状态中哪些信息是必需的?哪些可以省略?
2. **节点粒度**:节点应该多大?如何平衡?
3. **错误处理**:当节点失败时,工作流应该如何处理?
4. **审查点**:你的工作流中哪些地方需要人类介入?
---
## 下一步
完成这个实践后,你已经:
- ✅ 理解了状态驱动的Agent设计
- ✅ 实现了一个简单的LangGraph工作流
- ✅ 掌握了条件路由的基本方法
- ✅ 了解了如何添加HITL审查点
准备好进入下一章:**02-spatial-intelligence(空间智能)**