# 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. **实际系统**常采用混合架构,结合多种设计模式