a90f7adfa1
将 Markdown 源文件移入 md/,LaTeX 工作目录保留在 latex/, Word 导出移入 word/;删除临时脚本、调试截图和空 stub。 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
30 KiB
30 KiB
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)
特点:直接将感知映射到行动,无内部状态
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)
特点:维护内部状态,跟踪世界的部分不可观察方面
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)
特点:显式表示目标,能够规划行动序列
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)
特点:使用效用函数量化目标状态的价值,处理冲突目标
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对比演示
"""
四种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,它结合了多种设计模式:
"""
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)采用了多层混合架构:
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)
反思与延伸
思考问题
-
Agent类型选择:你的空间分析项目适合哪种Agent类型?
-
状态管理:如何在部分可观察环境中维护准确的内部状态?
-
目标冲突:当生态保护与经济发展冲突时,如何量化权衡?
-
规划成本:复杂规划的计算成本何时超过了其收益?
延伸阅读
- "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"
关键要点
- Agent的核心特征是自主性、感知-行动循环和目标导向
- Reflex Agent最简单,适合完全可观察环境
- Model-based Agent通过内部状态处理部分可观察性
- Goal-based Agent能够规划多步行动达成目标
- Utility-based Agent通过效用函数处理多目标冲突
- 实际系统常采用混合架构,结合多种设计模式