pengxiao
a90f7adfa1
将 Markdown 源文件移入 md/，LaTeX 工作目录保留在 latex/， Word 导出移入 word/；删除临时脚本、调试截图和空 stub。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1008 lines
30 KiB
Markdown
1008 lines
30 KiB
Markdown
# 03.2 Agent设计模式
|
||
|
||
## 核心问题
|
||
|
||
> 什么使一个系统成为"Agent"而非简单的程序？
|
||
> 不同类型的Agent有何区别，各适用于什么场景？
|
||
> 如何选择合适的Agent架构来解决空间分析问题？
|
||
|
||
---
|
||
|
||
## 概念讲解
|
||
|
||
### Agent的本质
|
||
|
||
**Agent** 是能够**感知环境**并**采取行动**以实现目标的实体：
|
||
|
||
```
|
||
┌─────────────────────────────────────────────────────────────┐
|
||
│ Agent的基本结构 │
|
||
├─────────────────────────────────────────────────────────────┤
|
||
│ │
|
||
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
|
||
│ │ Sensors │ ──→ │ Agent │ ──→ │Actuators│ │
|
||
│ │ (感知) │ │ (决策) │ │ (行动) │ │
|
||
│ └─────────┘ └────┬────┘ └─────────┘ │
|
||
│ │ │
|
||
│ ↓ │
|
||
│ ┌──────────┐ │
|
||
│ │Environment│ │
|
||
│ │ (环境) │ │
|
||
│ └──────────┘ │
|
||
│ │
|
||
│ 感知-决策-行动循环 (Perception-Decision-Action Loop) │
|
||
│ │
|
||
└─────────────────────────────────────────────────────────────┘
|
||
```
|
||
|
||
**Agent vs. 程序**：
|
||
|
||
| 特征 | 普通程序 | Agent |
|
||
|-----|---------|-------|
|
||
| 控制流 | 调用者驱动 | 自主驱动 |
|
||
| 状态 | 被动存储 | 主动维护世界模型 |
|
||
| 目标 | 无明确目标 | 有内在目标 |
|
||
| 环境 | 不感知 | 持续感知 |
|
||
| 适应性 | 固定行为 | 可学习适应 |
|
||
|
||
### 四种经典Agent类型
|
||
|
||
根据Russell & Norvig的AI教材，Agent有四种基本设计模式：
|
||
|
||
```
|
||
Agent类型演进
|
||
|
||
Reflex (反应式)
|
||
│
|
||
├──→ 无状态，直接映射
|
||
│ 感知→规则→行动
|
||
│
|
||
↓
|
||
Model-based (基于模型)
|
||
│
|
||
├──→ 有内部状态
|
||
│ 感知→状态更新→行动
|
||
│
|
||
↓
|
||
Goal-based (基于目标)
|
||
│
|
||
├──→ 有目标导向的规划
|
||
│ 状态+目标→规划→行动
|
||
│
|
||
↓
|
||
Utility-based (基于效用)
|
||
│
|
||
└──→ 有量化评估
|
||
状态+目标+效用→最优决策
|
||
```
|
||
|
||
---
|
||
|
||
## 设计原理
|
||
|
||
### 1. Reflex Agent（反应式Agent）
|
||
|
||
**特点**：直接将感知映射到行动，无内部状态
|
||
|
||
```python
|
||
class ReflexAgent:
|
||
"""
|
||
反应式Agent：最简单的Agent类型
|
||
|
||
适用场景：
|
||
- 环境完全可观察
|
||
- 当前行动只依赖当前感知
|
||
- 不需要历史信息
|
||
"""
|
||
|
||
def __init__(self, rules: dict):
|
||
"""
|
||
Args:
|
||
rules: {condition: action} 映射规则
|
||
"""
|
||
self.rules = rules
|
||
|
||
def act(self, percept: dict) -> str:
|
||
"""
|
||
根据当前感知选择行动
|
||
|
||
Args:
|
||
percept: 当前感知状态
|
||
|
||
Returns:
|
||
选择的行动
|
||
"""
|
||
for condition, action in self.rules.items():
|
||
if self._match_condition(condition, percept):
|
||
return action
|
||
|
||
return self.default_action()
|
||
|
||
def _match_condition(self, condition: dict, percept: dict) -> bool:
|
||
"""检查条件是否匹配"""
|
||
for key, value in condition.items():
|
||
if percept.get(key) != value:
|
||
return False
|
||
return True
|
||
|
||
def default_action(self) -> str:
|
||
"""默认行动"""
|
||
return "wait"
|
||
|
||
|
||
# 示例：简单的土地覆盖分类Agent
|
||
class LandCoverReflexAgent(ReflexAgent):
|
||
"""基于NDVI的土地覆盖分类Agent"""
|
||
|
||
def __init__(self):
|
||
rules = {
|
||
{'ndvi_high': True}: 'vegetation',
|
||
{'ndvi_low': True, 'nir_high': True}: 'water',
|
||
{'ndvi_low': True, 'temperature_high': True}: 'urban',
|
||
}
|
||
super().__init__(rules)
|
||
|
||
def classify(self, ndvi: float, nir: float, temperature: float) -> str:
|
||
"""分类土地覆盖类型"""
|
||
percept = {
|
||
'ndvi_high': ndvi > 0.4,
|
||
'ndvi_low': ndvi <= 0.4,
|
||
'nir_high': nir > 0.3,
|
||
'temperature_high': temperature > 25
|
||
}
|
||
return self.act(percept)
|
||
```
|
||
|
||
**优点**：
|
||
- 简单高效
|
||
- 响应快速
|
||
- 易于理解和调试
|
||
|
||
**缺点**：
|
||
- 无法处理部分可观察环境
|
||
- 无法规划未来行动
|
||
- 规则冲突时难以决策
|
||
|
||
### 2. Model-based Agent（基于模型的Agent）
|
||
|
||
**特点**：维护内部状态，跟踪世界的部分不可观察方面
|
||
|
||
```python
|
||
class ModelBasedAgent:
|
||
"""
|
||
基于模型的Agent：维护世界状态
|
||
|
||
适用场景：
|
||
- 环境部分可观察
|
||
- 需要跟踪历史信息
|
||
- 需要推断隐藏状态
|
||
"""
|
||
|
||
def __init__(self, transition_model, sensor_model):
|
||
"""
|
||
Args:
|
||
transition_model: 状态转移模型 P(s'|s,a)
|
||
sensor_model: 传感器模型 P(o|s)
|
||
"""
|
||
self.state = None
|
||
self.transition_model = transition_model
|
||
self.sensor_model = sensor_model
|
||
self.history = []
|
||
|
||
def update_state(self, action: str, percept: dict):
|
||
"""
|
||
更新内部状态
|
||
|
||
使用贝叶斯推断：
|
||
P(s'|o,a,s) ∝ P(o|s') * Σ P(s'|s,a) * P(s)
|
||
"""
|
||
if self.state is None:
|
||
# 初始化状态
|
||
self.state = self.sensor_model.estimate(percept)
|
||
else:
|
||
# 预测：基于转移模型
|
||
predicted = self.transition_model.predict(self.state, action)
|
||
|
||
# 更新：基于感知
|
||
self.state = self.sensor_model.update(predicted, percept)
|
||
|
||
self.history.append({
|
||
'action': action,
|
||
'percept': percept,
|
||
'state': self.state
|
||
})
|
||
|
||
def act(self, percept: dict) -> str:
|
||
"""选择行动"""
|
||
self.update_state(self.last_action, percept)
|
||
return self._choose_action()
|
||
|
||
def _choose_action(self) -> str:
|
||
"""基于当前状态选择行动"""
|
||
raise NotImplementedError
|
||
|
||
|
||
# 示例：生态变化检测Agent
|
||
class EcologicalChangeAgent(ModelBasedAgent):
|
||
"""检测生态系统变化的Agent"""
|
||
|
||
class TransitionModel:
|
||
"""状态转移模型"""
|
||
def predict(self, state, action):
|
||
# 简单的马尔可夫假设
|
||
new_state = state.copy()
|
||
if action == 'monitor':
|
||
# 状态可能自然变化
|
||
new_state['change_probability'] *= 0.95
|
||
return new_state
|
||
|
||
class SensorModel:
|
||
"""传感器模型"""
|
||
def estimate(self, percept):
|
||
return {
|
||
'baseline': percept['ndvi'],
|
||
'change_probability': 0.0,
|
||
'confidence': percept['quality']
|
||
}
|
||
|
||
def update(self, predicted, percept):
|
||
# 融合预测和观测
|
||
alpha = 0.7 # 预测权重
|
||
new_ndvi = alpha * predicted['baseline'] + (1-alpha) * percept['ndvi']
|
||
|
||
change_prob = predicted['change_probability']
|
||
if abs(new_ndvi - predicted['baseline']) > 0.1:
|
||
change_prob += 0.2
|
||
|
||
return {
|
||
'baseline': new_ndvi,
|
||
'change_probability': min(1.0, change_prob),
|
||
'confidence': predicted['confidence']
|
||
}
|
||
|
||
def __init__(self):
|
||
super().__init__(self.TransitionModel(), self.SensorModel())
|
||
self.last_action = None
|
||
|
||
def _choose_action(self) -> str:
|
||
"""基于变化概率选择行动"""
|
||
if self.state['change_probability'] > 0.6:
|
||
return 'alert'
|
||
elif self.state['change_probability'] > 0.3:
|
||
return 'investigate'
|
||
else:
|
||
return 'monitor'
|
||
```
|
||
|
||
### 3. Goal-based Agent（基于目标的Agent）
|
||
|
||
**特点**：显式表示目标，能够规划行动序列
|
||
|
||
```python
|
||
class GoalBasedAgent(ModelBasedAgent):
|
||
"""
|
||
基于目标的Agent：有明确的追求目标
|
||
|
||
适用场景：
|
||
- 需要规划多步行动
|
||
- 有明确的目标状态
|
||
- 需要考虑行动后果
|
||
"""
|
||
|
||
def __init__(self, transition_model, sensor_model, planner):
|
||
"""
|
||
Args:
|
||
planner: 规划器，搜索从当前状态到目标的路径
|
||
"""
|
||
super().__init__(transition_model, sensor_model)
|
||
self.planner = planner
|
||
self.current_goal = None
|
||
self.current_plan = []
|
||
|
||
def set_goal(self, goal: dict):
|
||
"""设置目标"""
|
||
self.current_goal = goal
|
||
self.current_plan = []
|
||
return self
|
||
|
||
def act(self, percept: dict) -> str:
|
||
"""选择行动"""
|
||
self.update_state(self.last_action, percept)
|
||
|
||
# 检查是否达到目标
|
||
if self._goal_achieved():
|
||
return 'goal_reached'
|
||
|
||
# 如果没有计划或计划过时，重新规划
|
||
if not self.current_plan or self._plan_stale():
|
||
self.current_plan = self.planner.plan(
|
||
self.state,
|
||
self.current_goal
|
||
)
|
||
|
||
# 执行计划的下一步
|
||
if self.current_plan:
|
||
action = self.current_plan.pop(0)
|
||
self.last_action = action
|
||
return action
|
||
|
||
return 'no_plan'
|
||
|
||
def _goal_achieved(self) -> bool:
|
||
"""检查目标是否达成"""
|
||
if not self.current_goal:
|
||
return False
|
||
return all(
|
||
self.state.get(k) == v
|
||
for k, v in self.current_goal.items()
|
||
)
|
||
|
||
def _plan_stale(self) -> bool:
|
||
"""检查计划是否需要更新"""
|
||
# 简化版本：检查最近的状态变化
|
||
if len(self.history) < 2:
|
||
return False
|
||
# 实际实现会更复杂
|
||
return False
|
||
|
||
|
||
# 示例：保护区选址Agent
|
||
class ReserveSiteAgent(GoalBasedAgent):
|
||
"""寻找最佳保护区选址的Agent"""
|
||
|
||
class Planner:
|
||
"""前向搜索规划器"""
|
||
def plan(self, current_state, goal):
|
||
"""
|
||
使用前向搜索规划
|
||
|
||
返回行动序列：[action1, action2, ...]
|
||
"""
|
||
plan = []
|
||
|
||
# 目标：找到至少3个候选地点
|
||
while current_state.get('n_candidates', 0) < goal.get('min_sites', 3):
|
||
# 下一步行动
|
||
if not current_state.get('searched_regions', []):
|
||
action = ('search_region', 0)
|
||
else:
|
||
next_region = max(current_state['searched_regions']) + 1
|
||
action = ('search_region', next_region)
|
||
|
||
plan.append(action)
|
||
# 模拟状态更新
|
||
current_state = self._simulate(current_state, action)
|
||
|
||
if next_region > 10: # 防止无限循环
|
||
break
|
||
|
||
return plan
|
||
|
||
def _simulate(self, state, action):
|
||
"""模拟行动后的状态"""
|
||
new_state = state.copy()
|
||
if action[0] == 'search_region':
|
||
searched = state.get('searched_regions', [])
|
||
searched.append(action[1])
|
||
new_state['searched_regions'] = searched
|
||
# 模拟可能发现候选点
|
||
if action[1] % 3 == 0: # 每3个区域有1个候选
|
||
new_state['n_candidates'] = state.get('n_candidates', 0) + 1
|
||
return new_state
|
||
|
||
def __init__(self):
|
||
super().__init__(
|
||
super().TransitionModel(),
|
||
super().SensorModel(),
|
||
self.Planner()
|
||
)
|
||
self.last_action = None
|
||
|
||
def find_reserve_sites(self, min_sites: int = 3, budget: float = 1000000):
|
||
"""寻找保护区选址"""
|
||
self.set_goal({
|
||
'min_sites': min_sites,
|
||
'budget': budget,
|
||
'status': 'found'
|
||
})
|
||
return self
|
||
```
|
||
|
||
### 4. Utility-based Agent（基于效用的Agent）
|
||
|
||
**特点**：使用效用函数量化目标状态的价值，处理冲突目标
|
||
|
||
```python
|
||
class UtilityBasedAgent(GoalBasedAgent):
|
||
"""
|
||
基于效用的Agent：量化目标价值
|
||
|
||
适用场景：
|
||
- 有多个冲突目标
|
||
- 目标有不同重要性
|
||
- 需要在不确定环境下决策
|
||
"""
|
||
|
||
def __init__(self, transition_model, sensor_model, planner, utility_fn):
|
||
"""
|
||
Args:
|
||
utility_fn: 效用函数 U(state) → 实数
|
||
"""
|
||
super().__init__(transition_model, sensor_model, planner)
|
||
self.utility_fn = utility_fn
|
||
|
||
def set_preferences(self, preferences: dict):
|
||
"""设置偏好（权重）"""
|
||
self.utility_fn.set_weights(preferences)
|
||
return self
|
||
|
||
def evaluate_plan(self, plan: list) -> float:
|
||
"""评估计划的期望效用"""
|
||
expected_state = self.state
|
||
total_utility = 0
|
||
|
||
for action in plan:
|
||
# 模拟行动
|
||
expected_state = self.transition_model.predict(
|
||
expected_state, action
|
||
)
|
||
# 累积效用
|
||
total_utility += self.utility_fn(expected_state)
|
||
|
||
return total_utility
|
||
|
||
def choose_best_action(self, available_actions: list) -> str:
|
||
"""选择效用最大的行动"""
|
||
best_action = None
|
||
best_utility = float('-inf')
|
||
|
||
for action in available_actions:
|
||
# 预测行动后的状态
|
||
predicted_state = self.transition_model.predict(
|
||
self.state, action
|
||
)
|
||
# 计算效用
|
||
utility = self.utility_fn(predicted_state)
|
||
|
||
if utility > best_utility:
|
||
best_utility = utility
|
||
best_action = action
|
||
|
||
return best_action
|
||
|
||
|
||
class UtilityFunction:
|
||
"""效用函数"""
|
||
|
||
def __init__(self, objectives: dict):
|
||
"""
|
||
Args:
|
||
objectives: {name: (weight, function)}
|
||
"""
|
||
self.objectives = objectives
|
||
|
||
def set_weights(self, weights: dict):
|
||
"""更新目标权重"""
|
||
for name, weight in weights.items():
|
||
if name in self.objectives:
|
||
old_weight, fn = self.objectives[name]
|
||
self.objectives[name] = (weight, fn)
|
||
|
||
def __call__(self, state: dict) -> float:
|
||
"""计算状态的总效用"""
|
||
total = 0
|
||
for (weight, fn) in self.objectives.values():
|
||
total += weight * fn(state)
|
||
return total
|
||
|
||
|
||
# 示例：土地利用规划Agent
|
||
class LandUsePlanningAgent(UtilityBasedAgent):
|
||
"""土地利用规划Agent，平衡多个目标"""
|
||
|
||
def __init__(self):
|
||
# 定义多个目标
|
||
objectives = {
|
||
'economic': (0.3, self._economic_value),
|
||
'ecological': (0.4, self._ecological_value),
|
||
'social': (0.3, self._social_value),
|
||
}
|
||
|
||
super().__init__(
|
||
super().TransitionModel(),
|
||
super().SensorModel(),
|
||
super().Planner(),
|
||
UtilityFunction(objectives)
|
||
)
|
||
|
||
@staticmethod
|
||
def _economic_value(state: dict) -> float:
|
||
"""经济价值：开发土地产生的收益"""
|
||
return state.get('developed_area', 0) * 1000
|
||
|
||
@staticmethod
|
||
def _ecological_value(state: dict) -> float:
|
||
"""生态价值：保护的自然栖息地"""
|
||
return -state.get('habitat_loss', 0) * 500
|
||
|
||
@staticmethod
|
||
def _social_value(state: dict) -> float:
|
||
"""社会价值：住房供应和公共空间"""
|
||
housing = state.get('housing_units', 0)
|
||
green_space = state.get('green_space_ratio', 0)
|
||
return housing * 100 + green_space * 2000
|
||
|
||
def plan_land_use(self, area: float, economic_weight: float = 0.3):
|
||
"""规划土地利用"""
|
||
self.set_preferences({
|
||
'economic': economic_weight,
|
||
'ecological': 1 - economic_weight - 0.3,
|
||
'social': 0.3
|
||
})
|
||
return self
|
||
```
|
||
|
||
---
|
||
|
||
## 代码示例
|
||
|
||
### 完整的四类Agent对比演示
|
||
|
||
```python
|
||
"""
|
||
四种Agent类型的完整对比演示
|
||
|
||
场景：生态监测站需要决定每日行动
|
||
"""
|
||
import numpy as np
|
||
from typing import Dict, List, Optional, Tuple
|
||
from dataclasses import dataclass
|
||
from enum import Enum
|
||
|
||
|
||
class SensorReading(Enum):
|
||
"""传感器读数类型"""
|
||
NORMAL = "normal"
|
||
ANOMALY_DETECTED = "anomaly"
|
||
CRITICAL = "critical"
|
||
|
||
|
||
@dataclass
|
||
class EnvironmentState:
|
||
"""环境状态"""
|
||
temperature: float
|
||
humidity: float
|
||
species_count: int
|
||
vegetation_health: float # 0-1
|
||
detected_anomaly: bool
|
||
time_step: int
|
||
|
||
|
||
class MonitoringStation:
|
||
"""模拟生态监测站环境"""
|
||
|
||
def __init__(self):
|
||
self.state = EnvironmentState(
|
||
temperature=25.0,
|
||
humidity=60.0,
|
||
species_count=15,
|
||
vegetation_health=0.8,
|
||
detected_anomaly=False,
|
||
time_step=0
|
||
)
|
||
self.anomaly_schedule = [5, 12, 18] # 预定异常发生时间
|
||
|
||
def step(self, action: str) -> Tuple[SensorReading, EnvironmentState]:
|
||
"""执行一步模拟"""
|
||
self.state.time_step += 1
|
||
|
||
# 环境动态变化
|
||
self.state.temperature += np.random.normal(0, 1)
|
||
self.state.humidity += np.random.normal(0, 2)
|
||
self.state.vegetation_health = max(0, min(1,
|
||
self.state.vegetation_health + np.random.normal(0, 0.05)
|
||
))
|
||
|
||
# 检查是否发生异常
|
||
if self.state.time_step in self.anomaly_schedule:
|
||
self.state.detected_anomaly = True
|
||
self.state.vegetation_health -= 0.2
|
||
|
||
# 行动影响
|
||
if action == "collect_sample":
|
||
self.state.species_count += np.random.randint(-1, 2)
|
||
elif action == "irrigate":
|
||
self.state.humidity = min(100, self.state.humidity + 10)
|
||
self.state.vegetation_health = min(1, self.state.vegetation_health + 0.05)
|
||
|
||
# 生成传感器读数
|
||
reading = self._get_sensor_reading()
|
||
|
||
return reading, self.state.copy()
|
||
|
||
def _get_sensor_reading(self) -> SensorReading:
|
||
"""生成传感器读数"""
|
||
if self.state.vegetation_health < 0.3:
|
||
return SensorReading.CRITICAL
|
||
elif self.state.detected_anomaly:
|
||
return SensorReading.ANOMALY_DETECTED
|
||
return SensorReading.NORMAL
|
||
|
||
def reset(self):
|
||
"""重置环境"""
|
||
self.__init__()
|
||
|
||
|
||
# ==================== 1. Reflex Agent ====================
|
||
|
||
class ReflexMonitoringAgent:
|
||
"""反应式监测Agent"""
|
||
|
||
def __init__(self):
|
||
self.rules = {
|
||
SensorReading.CRITICAL: "emergency_response",
|
||
SensorReading.ANOMALY_DETECTED: "investigate",
|
||
SensorReading.NORMAL: "routine_check"
|
||
}
|
||
|
||
def act(self, reading: SensorReading) -> str:
|
||
"""根据读数直接行动"""
|
||
return self.rules.get(reading, "routine_check")
|
||
|
||
|
||
# ==================== 2. Model-based Agent ====================
|
||
|
||
class ModelBasedMonitoringAgent:
|
||
"""基于模型的监测Agent"""
|
||
|
||
def __init__(self):
|
||
self.belief_state = {
|
||
'anomaly_active': False,
|
||
'anomaly_duration': 0,
|
||
'vegetation_trend': 'stable',
|
||
'last_reading': None
|
||
}
|
||
|
||
def act(self, reading: SensorReading) -> str:
|
||
"""更新信念状态并行动"""
|
||
# 更新内部状态
|
||
if reading == SensorReading.ANOMALY_DETECTED:
|
||
self.belief_state['anomaly_active'] = True
|
||
self.belief_state['anomaly_duration'] += 1
|
||
elif reading == SensorReading.NORMAL:
|
||
if self.belief_state['anomaly_active']:
|
||
self.belief_state['anomaly_duration'] -= 1
|
||
if self.belief_state['anomaly_duration'] <= 0:
|
||
self.belief_state['anomaly_active'] = False
|
||
|
||
# 基于信念状态决策
|
||
if self.belief_state['anomaly_active']:
|
||
if self.belief_state['anomaly_duration'] > 2:
|
||
return "intensive_monitoring"
|
||
return "investigate"
|
||
return "routine_check"
|
||
|
||
|
||
# ==================== 3. Goal-based Agent ====================
|
||
|
||
class GoalBasedMonitoringAgent:
|
||
"""基于目标的监测Agent"""
|
||
|
||
def __init__(self):
|
||
self.current_goal = None
|
||
self.belief_state = {
|
||
'data_coverage': 0.0,
|
||
'samples_collected': 0,
|
||
'anomalies_investigated': 0
|
||
}
|
||
|
||
def set_goal(self, goal: str):
|
||
"""设置当前目标"""
|
||
self.current_goal = goal
|
||
return self
|
||
|
||
def act(self, reading: SensorReading) -> str:
|
||
"""基于目标选择行动"""
|
||
# 目标导向的规划
|
||
if self.current_goal == "comprehensive_survey":
|
||
if self.belief_state['data_coverage'] < 1.0:
|
||
return "collect_sample"
|
||
elif self.belief_state['samples_collected'] < 10:
|
||
return "collect_sample"
|
||
else:
|
||
return "compile_report"
|
||
|
||
elif self.current_goal == "anomaly_investigation":
|
||
if reading == SensorReading.ANOMALY_DETECTED:
|
||
self.belief_state['anomalies_investigated'] += 1
|
||
return "investigate"
|
||
return "search_for_anomalies"
|
||
|
||
return "routine_check"
|
||
|
||
|
||
# ==================== 4. Utility-based Agent ====================
|
||
|
||
class UtilityMonitoringAgent:
|
||
"""基于效用的监测Agent"""
|
||
|
||
def __init__(self):
|
||
self.belief_state = {
|
||
'anomaly_active': False,
|
||
'data_coverage': 0.0,
|
||
'resource_remaining': 100,
|
||
'scientific_value': 0
|
||
}
|
||
self.weights = {
|
||
'safety': 0.5,
|
||
'science': 0.3,
|
||
'efficiency': 0.2
|
||
}
|
||
|
||
def utility(self, action: str, reading: SensorReading) -> float:
|
||
"""计算行动的效用"""
|
||
utility = 0
|
||
|
||
# 安全效用
|
||
if reading == SensorReading.CRITICAL:
|
||
if action == "emergency_response":
|
||
utility += 100 * self.weights['safety']
|
||
else:
|
||
utility -= 50 * self.weights['safety']
|
||
elif reading == SensorReading.ANOMALY_DETECTED:
|
||
if action == "investigate":
|
||
utility += 30 * self.weights['safety']
|
||
elif action == "routine_check":
|
||
utility -= 20 * self.weights['safety']
|
||
|
||
# 科学价值效用
|
||
if action == "collect_sample":
|
||
if self.belief_state['data_coverage'] < 0.8:
|
||
utility += 20 * self.weights['science']
|
||
else:
|
||
utility += 5 * self.weights['science']
|
||
|
||
# 效率效用
|
||
if self.belief_state['resource_remaining'] < 20:
|
||
if action == "routine_check":
|
||
utility += 10 * self.weights['efficiency']
|
||
elif action == "collect_sample":
|
||
utility -= 15 * self.weights['efficiency']
|
||
|
||
return utility
|
||
|
||
def act(self, reading: SensorReading) -> str:
|
||
"""选择效用最大的行动"""
|
||
actions = ["emergency_response", "investigate", "collect_sample",
|
||
"routine_check", "rest"]
|
||
|
||
best_action = "routine_check"
|
||
best_utility = float('-inf')
|
||
|
||
for action in actions:
|
||
u = self.utility(action, reading)
|
||
if u > best_utility:
|
||
best_utility = u
|
||
best_action = action
|
||
|
||
return best_action
|
||
|
||
|
||
# ==================== 演示对比 ====================
|
||
|
||
def compare_agents(n_steps: int = 20):
|
||
"""对比四种Agent的表现"""
|
||
print("=" * 60)
|
||
print("四种Agent类型在生态监测任务中的对比")
|
||
print("=" * 60)
|
||
|
||
agents = {
|
||
"Reflex": ReflexMonitoringAgent(),
|
||
"Model-based": ModelBasedMonitoringAgent(),
|
||
"Goal-based": GoalBasedMonitoringAgent().set_goal("comprehensive_survey"),
|
||
"Utility-based": UtilityMonitoringAgent()
|
||
}
|
||
|
||
results = {name: [] for name in agents.keys()}
|
||
|
||
for step in range(n_steps):
|
||
env = MonitoringStation()
|
||
|
||
for name, agent in agents.items():
|
||
reading, state = env.step("observe")
|
||
action = agent.act(reading)
|
||
results[name].append(action)
|
||
|
||
# 打印结果对比
|
||
print("\n行动序列对比:")
|
||
print("-" * 60)
|
||
print(f"{'时间':<6} {'Reflex':<20} {'Model-based':<20}")
|
||
print("-" * 60)
|
||
|
||
for i in range(n_steps):
|
||
print(f"{i:<6} {results['Reflex'][i]:<20} {results['Model-based'][i]:<20}")
|
||
|
||
print("-" * 60)
|
||
print(f"{'时间':<6} {'Goal-based':<20} {'Utility-based':<20}")
|
||
print("-" * 60)
|
||
|
||
for i in range(n_steps):
|
||
print(f"{i:<6} {results['Goal-based'][i]:<20} {results['Utility-based'][i]:<20}")
|
||
|
||
# 统计分析
|
||
print("\n行动统计:")
|
||
print("-" * 60)
|
||
for name, actions in results.items():
|
||
from collections import Counter
|
||
counts = Counter(actions)
|
||
print(f"\n{name}:")
|
||
for action, count in counts.most_common():
|
||
print(f" {action}: {count}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
compare_agents()
|
||
```
|
||
|
||
---
|
||
|
||
## 案例分析
|
||
|
||
### Claude Code的Agent架构
|
||
|
||
Claude Code是一个典型的**Utility-based Agent**，它结合了多种设计模式：
|
||
|
||
```python
|
||
"""
|
||
Claude Code的Agent架构简化示意
|
||
"""
|
||
|
||
class ClaudeCodeAgent:
|
||
"""
|
||
Claude Code Agent设计
|
||
|
||
特点：
|
||
- Model-based: 维护对话上下文状态
|
||
- Goal-based: 追求用户的任务目标
|
||
- Utility-based: 平衡正确性、效率、安全性
|
||
"""
|
||
|
||
def __init__(self):
|
||
# 内部状态
|
||
self.state = {
|
||
'conversation_history': [],
|
||
'workspace_state': {}, # 文件系统状态
|
||
'tool_results': [],
|
||
'user_goal': None
|
||
}
|
||
|
||
# 效用函数组件
|
||
self.utility_components = {
|
||
'task_completion': 0.5, # 完成任务
|
||
'correctness': 0.3, # 正确性
|
||
'safety': 0.2, # 安全性
|
||
}
|
||
|
||
def perceive(self, user_input: str, tool_outputs: list):
|
||
"""感知：更新内部状态"""
|
||
self.state['conversation_history'].append({
|
||
'role': 'user',
|
||
'content': user_input
|
||
})
|
||
self.state['tool_results'] = tool_outputs
|
||
|
||
def plan(self):
|
||
"""规划：生成行动计划"""
|
||
# 分析用户意图
|
||
intent = self._analyze_intent()
|
||
|
||
# 生成候选行动序列
|
||
candidates = self._generate_candidates(intent)
|
||
|
||
# 评估每个候选
|
||
best_plan = max(
|
||
candidates,
|
||
key=lambda p: self._evaluate_plan(p)
|
||
)
|
||
|
||
return best_plan
|
||
|
||
def act(self, plan: list):
|
||
"""执行：按计划调用工具"""
|
||
results = []
|
||
for action in plan:
|
||
result = self._execute_action(action)
|
||
results.append(result)
|
||
return results
|
||
|
||
def _evaluate_plan(self, plan: list) -> float:
|
||
"""评估计划的效用"""
|
||
utility = 0
|
||
for component, weight in self.utility_components.items():
|
||
if component == 'safety':
|
||
# 检查危险操作
|
||
if any(a.get('dangerous') for a in plan):
|
||
utility -= 100 * weight
|
||
# ... 其他评估
|
||
return utility
|
||
```
|
||
|
||
### ENAgent的混合设计
|
||
|
||
ENAgent（生态网络分析Agent）采用了**多层混合架构**：
|
||
|
||
```python
|
||
class ENAgent:
|
||
"""
|
||
ENAgent: 多层混合Agent架构
|
||
|
||
反应层：处理简单操作
|
||
规划层：处理复杂分析流程
|
||
效用层：优化分析参数
|
||
"""
|
||
|
||
def __init__(self):
|
||
# 反应式处理简单命令
|
||
self.reflex_layer = ReflexLayer({
|
||
'load_data': self._load_data,
|
||
'show_status': self._show_status,
|
||
})
|
||
|
||
# 规划层处理复杂流程
|
||
self.planning_layer = PlanningLayer()
|
||
self.planning_layer.set_goal('build_ecological_network')
|
||
|
||
# 效用层优化参数
|
||
self.utility_layer = UtilityLayer({
|
||
'accuracy': self._accuracy_fn,
|
||
'computation_time': self._time_fn,
|
||
'memory_usage': self._memory_fn,
|
||
})
|
||
|
||
def process(self, user_command: str):
|
||
"""处理用户命令"""
|
||
# 1. 反应层快速响应
|
||
if user_command in self.reflex_layer.handlers:
|
||
return self.reflex_layer.handle(user_command)
|
||
|
||
# 2. 规划层生成流程
|
||
plan = self.planning_layer.generate_plan(user_command)
|
||
|
||
# 3. 效用层优化参数
|
||
optimized_plan = self.utility_layer.optimize(plan)
|
||
|
||
# 4. 执行计划
|
||
return self._execute(optimized_plan)
|
||
```
|
||
|
||
---
|
||
|
||
## 反思与延伸
|
||
|
||
### 思考问题
|
||
|
||
1. **Agent类型选择**：你的空间分析项目适合哪种Agent类型？
|
||
|
||
2. **状态管理**：如何在部分可观察环境中维护准确的内部状态？
|
||
|
||
3. **目标冲突**：当生态保护与经济发展冲突时，如何量化权衡？
|
||
|
||
4. **规划成本**：复杂规划的计算成本何时超过了其收益？
|
||
|
||
### 延伸阅读
|
||
|
||
- **"Artificial Intelligence: A Modern Approach"** (Russell & Norvig) - Chapter 2: Intelligent Agents
|
||
- **"Agent-Based Modeling"** (Railsback & Grimm) - 基于Agent的建模
|
||
- **ReAct论文** - "ReAct: Synergizing Reasoning and Acting in Language Models"
|
||
|
||
---
|
||
|
||
## 关键要点
|
||
|
||
1. **Agent的核心特征**是自主性、感知-行动循环和目标导向
|
||
2. **Reflex Agent**最简单，适合完全可观察环境
|
||
3. **Model-based Agent**通过内部状态处理部分可观察性
|
||
4. **Goal-based Agent**能够规划多步行动达成目标
|
||
5. **Utility-based Agent**通过效用函数处理多目标冲突
|
||
6. **实际系统**常采用混合架构，结合多种设计模式
|