掌握Agent设计模式：小白程序员轻松入门，收藏提升技能！

ai绘画-安安妮

496人浏览 · 2026-03-17 11:05:58

ai绘画-安安妮 · 2026-03-17 11:05:58 发布

本节目标

学完本课程后，你应该能够：

理解观察者模式在Agent中的应用
掌握策略模式在Agent中的实现
了解其他常用的Agent设计模式
实现多种设计模式的综合应用

理论讲解

设计模式在Agent系统中的重要性

设计模式是在软件设计中反复出现的问题的可重用解决方案。在Agent系统中，设计模式有助于：

提高代码的可维护性和可扩展性
促进组件间的松耦合
便于理解和复用代码
支持系统的灵活性和适应性

观察者模式（Observer Pattern）

观察者模式定义了对象之间的一对多依赖关系，当一个对象的状态发生改变时，所有依赖于它的对象都会得到通知并自动更新。

在Agent系统中，观察者模式可用于：

环境状态变化通知
Agent状态监控
事件驱动的响应机制

2.1 观察者模式结构

Subject（主题）：被观察的对象
Observer（观察者）：观察Subject的对象
ConcreteSubject（具体主题）：具体实现Subject的类
ConcreteObserver（具体观察者）：具体实现Observer的类

策略模式（Strategy Pattern）

策略模式定义了一系列算法，并将每种算法分别放入独立的类中，从而使算法的变化独立于使用它们的客户端。

在Agent系统中，策略模式可用于：

决策算法的切换
行为策略的动态调整
不同规划算法的选择

3.1 策略模式结构

Strategy（策略）：定义算法的接口
ConcreteStrategy（具体策略）：实现算法的类
Context（上下文）：使用策略的类

其他常用的Agent设计模式

4.1 状态模式（State Pattern）

允许对象在其内部状态改变时改变其行为。

4.2 模板方法模式（Template Method Pattern）

定义算法的骨架，将一些步骤延迟到子类中实现。

4.3 命令模式（Command Pattern）

将请求封装为对象，从而可以用不同的请求对客户进行参数化。

设计思路

在Agent系统中应用设计模式需要考虑：

模式的适用场景
模式间的组合使用
系统的性能影响
代码的可读性和维护性

代码实现

下面我们将实现多种Agent设计模式：

from abc import ABC, abstractmethod
from typing import List, Dict, Any, Optional
import time
import threading
from enum import Enum
# 观察者模式实现
class Observer(ABC):
"""
观察者抽象类
"""
@abstractmethod
def update(self, subject, event_type: str, data: Any):
"""
当被观察对象状态改变时被调用
:param subject: 被观察的主题对象
:param event_type: 事件类型
:param data: 事件数据
"""
pass
class Subject(ABC):
"""
主题抽象类
"""
def __init__(self):
self._observers: List[Observer] = []
def attach(self, observer: Observer):
"""添加观察者"""
if observer not in self._observers:
self._observers.append(observer)
def detach(self, observer: Observer):
"""移除观察者"""
if observer in self._observers:
self._observers.remove(observer)
def notify(self, event_type: str, data: Any = None):
"""通知所有观察者"""
for observer in self._observers:
observer.update(self, event_type, data)
class AgentMonitor(Subject):
"""
Agent监控器 - 被观察的主题
"""
def __init__(self, agent_id: str):
super().__init__()
self.agent_id = agent_id
self.state = "idle"
self.health_score = 100
self.performance_metrics = {}
def change_state(self, new_state: str):
"""改变Agent状态"""
old_state = self.state
self.state = new_state
print(f"Agent {self.agent_id} 状态从 '{old_state}' 变更为 '{new_state}'")
self.notify("state_changed", {"old_state": old_state, "new_state": new_state})
def update_health(self, health: int):
"""更新健康分数"""
old_health = self.health_score
self.health_score = health
print(f"Agent {self.agent_id} 健康分数从 {old_health} 更新为 {health}")
self.notify("health_changed", {"old_health": old_health, "new_health": health})
def log_metric(self, metric_name: str, value: Any):
"""记录性能指标"""
self.performance_metrics[metric_name] = value
self.notify("metric_updated", {"metric_name": metric_name, "value": value})
class Logger(Observer):
"""
日志观察者
"""
def __init__(self, name: str):
self.name = name
self.logs = []
def update(self, subject, event_type: str, data: Any):
log_entry = {
'timestamp': time.time(),
'event_type': event_type,
'data': data,
'subject_id': getattr(subject, 'agent_id', 'unknown')
}
self.logs.append(log_entry)
print(f"[{self.name}] LOG: Event '{event_type}' for agent {log_entry['subject_id']}, Data: {data}")
class AlertSystem(Observer):
"""
警报系统观察者
"""
def __init__(self, threshold: int = 30):
self.threshold = threshold
self.alerts = []
def update(self, subject, event_type: str, data: Any):
if event_type == "health_changed":
new_health = data.get('new_health', 0)
if new_health < self.threshold:
alert = {
'timestamp': time.time(),
'agent_id': getattr(subject, 'agent_id', 'unknown'),
'health': new_health,
'message': f'Health below threshold: {new_health} < {self.threshold}'
}
self.alerts.append(alert)
print(f"🚨 ALERT: {alert['message']}")
elif event_type == "state_changed":
new_state = data.get('new_state', '')
if new_state == "error":
alert = {
'timestamp': time.time(),
'agent_id': getattr(subject, 'agent_id', 'unknown'),
'state': new_state,
'message': f'Agent entered error state'
}
self.alerts.append(alert)
print(f"🚨 ALERT: Agent {alert['agent_id']} entered error state")
class PerformanceTracker(Observer):
"""
性能追踪观察者
"""
def __init__(self):
self.metrics_history = {}
def update(self, subject, event_type: str, data: Any):
if event_type == "metric_updated":
metric_name = data.get('metric_name')
value = data.get('value')
agent_id = getattr(subject, 'agent_id', 'unknown')
if agent_id not in self.metrics_history:
self.metrics_history[agent_id] = {}
if metric_name not in self.metrics_history[agent_id]:
self.metrics_history[agent_id][metric_name] = []
self.metrics_history[agent_id][metric_name].append({
'timestamp': time.time(),
'value': value
})
print(f"📊 Performance metric '{metric_name}' updated for agent {agent_id}: {value}")
# 策略模式实现
class PlanningStrategy(ABC):
"""
规划策略抽象类
"""
@abstractmethod
def plan(self, goals: List[str], current_state: Dict[str, Any]) -> List[str]:
"""
根据目标和当前状态制定计划
:param goals: 目标列表
:param current_state: 当前状态
:return: 行动计划
"""
pass
class SimplePlanningStrategy(PlanningStrategy):
"""
简单规划策略
"""
def plan(self, goals: List[str], current_state: Dict[str, Any]) -> List[str]:
plan = []
for goal in goals:
if goal == "navigate_to_target":
plan.extend(["move_forward", "turn_right", "move_forward"])
elif goal == "collect_resources":
plan.extend(["scan_area", "approach_resource", "collect"])
elif goal == "avoid_obstacle":
plan.extend(["stop", "analyze_obstacle", "find_alternative_path"])
elif goal == "communicate":
plan.extend(["establish_connection", "send_message", "wait_for_response"])
else:
plan.append("default_action")
return plan
class AdvancedPlanningStrategy(PlanningStrategy):
"""
高级规划策略 - 考虑当前状态和环境因素
"""
def plan(self, goals: List[str], current_state: Dict[str, Any]) -> List[str]:
plan = []
# 根据当前状态调整计划
battery_level = current_state.get('battery_level', 100)
obstacle_distance = current_state.get('obstacle_distance', float('inf'))
target_distance = current_state.get('target_distance', float('inf'))
for goal in goals:
if goal == "navigate_to_target":
if battery_level < 20:
plan.append("return_to_base")
continue
if obstacle_distance < 5:  # 障碍物在5米以内
plan.extend(["stop", "scan_surroundings", "plan_detour"])
else:
# 根据距离选择最优路径
if target_distance > 50:
plan.extend(["move_fast", "navigate_long_range"])
else:
plan.extend(["navigate_short_range", "fine_position_control"])
elif goal == "collect_resources":
if battery_level < 30:
plan.append("return_to_base")
continue
resource_density = current_state.get('resource_density', 0)
if resource_density > 0.8:
plan.extend(["approach_resource", "collect_efficiently"])
elif resource_density > 0.3:
plan.extend(["approach_resource", "collect_normally"])
else:
plan.extend(["scan_new_area", "find_better_location"])
elif goal == "avoid_obstacle":
obstacle_size = current_state.get('obstacle_size', 'small')
if obstacle_size == 'large':
plan.extend(["stop", "analyze_obstacle", "plan_large_detour"])
else:
plan.extend(["small_adjustment", "continue_navigation"])
elif goal == "communicate":
signal_strength = current_state.get('signal_strength', 0)
if signal_strength < 0.3:
plan.extend(["find_better_signal", "retry_connection"])
else:
plan.extend(["establish_connection", "send_message", "verify_delivery"])
return plan
class LearningPlanningStrategy(PlanningStrategy):
"""
学习型规划策略 - 基于历史经验优化规划
"""
def __init__(self):
self.experience_db = {}  # 存储历史经验
self.performance_history = []
def plan(self, goals: List[str], current_state: Dict[str, Any]) -> List[str]:
# 检查是否有类似的历史经验
state_hash = self._hash_state(current_state)
if state_hash in self.experience_db:
# 使用历史经验
best_plan = self.experience_db[state_hash]
print(f"Using learned plan for similar state: {best_plan}")
return best_plan
else:
# 使用默认规划策略
default_strategy = AdvancedPlanningStrategy()
plan = default_strategy.plan(goals, current_state)
# 记录新经验（这里简化，实际应用中需要评估执行效果）
self.experience_db[state_hash] = plan
print(f"Learned new plan for state: {plan}")
return plan
def _hash_state(self, state: Dict[str, Any]) -> str:
"""简单地哈希状态"""
import hashlib
state_str = str(sorted(state.items()))
return hashlib.md5(state_str.encode()).hexdigest()
def record_performance(self, state: Dict[str, Any], plan: List[str], performance: float):
"""记录执行性能"""
state_hash = self._hash_state(state)
self.performance_history.append({
'state_hash': state_hash,
'plan': plan,
'performance': performance,
'timestamp': time.time()
})
class AdaptiveAgent:
"""
自适应Agent - 使用策略模式
"""
def __init__(self, agent_id: str, initial_strategy: PlanningStrategy):
self.agent_id = agent_id
self.strategy = initial_strategy
self.goals: List[str] = []
self.current_state: Dict[str, Any] = {}
self.action_history: List[str] = []
self.performance_score = 1.0
def set_strategy(self, strategy: PlanningStrategy):
"""切换规划策略"""
old_strategy_type = type(self.strategy).__name__
self.strategy = strategy
new_strategy_type = type(strategy).__name__
print(f"Agent {self.agent_id} 策略从 {old_strategy_type} 切换到 {new_strategy_type}")
def set_goals(self, goals: List[str]):
"""设置目标"""
self.goals = goals
print(f"Agent {self.agent_id} 设置了目标: {goals}")
def update_state(self, state_update: Dict[str, Any]):
"""更新状态"""
self.current_state.update(state_update)
print(f"Agent {self.agent_id} 状态更新: {state_update}")
def plan_actions(self) -> List[str]:
"""制定行动计划"""
if not self.goals:
print(f"Agent {self.agent_id} 没有目标，无法制定计划")
return []
plan = self.strategy.plan(self.goals, self.current_state)
print(f"Agent {self.agent_id} 制定的计划: {plan}")
return plan
def execute_plan(self, plan: List[str]):
"""执行计划"""
for action in plan:
self.action_history.append(action)
print(f"Agent {self.agent_id} 执行动作: {action}")
# 模拟执行时间
time.sleep(0.1)
def adapt_strategy(self):
"""根据性能自适应调整策略"""
# 简单的自适应逻辑：如果性能低于阈值，切换到高级策略
if self.performance_score < 0.5:
if not isinstance(self.strategy, AdvancedPlanningStrategy):
print(f"Agent {self.agent_id} 性能较低，切换到高级策略")
self.set_strategy(AdvancedPlanningStrategy())
elif self.performance_score > 0.8:
if isinstance(self.strategy, SimplePlanningStrategy):
print(f"Agent {self.agent_id} 性能良好，保持当前策略")
def evaluate_performance(self) -> float:
"""评估性能（简化版本）"""
# 基于行动历史的简单评估
if not self.action_history:
return 0.5
# 计算效率：避免重复动作
unique_actions = len(set(self.action_history))
total_actions = len(self.action_history)
if total_actions == 0:
return 0.5
efficiency = unique_actions / total_actions
self.performance_score = min(1.0, efficiency + 0.2)  # 确保至少0.2的基准分数
return self.performance_score
# 状态模式实现
class AgentState(ABC):
"""
Agent状态抽象类
"""
@abstractmethod
def handle(self, agent: 'StateMachineAgent'):
"""处理Agent的行为"""
pass
@abstractmethod
def get_state_name(self) -> str:
"""获取状态名称"""
pass
class IdleState(AgentState):
"""空闲状态"""
def handle(self, agent: 'StateMachineAgent'):
print(f"Agent {agent.agent_id} 处于空闲状态，正在监听任务...")
# 检查是否有新任务
if agent.has_task():
agent.transition_to(BusyState())
def get_state_name(self) -> str:
return "idle"
class BusyState(AgentState):
"""忙碌状态"""
def handle(self, agent: 'StateMachineAgent'):
print(f"Agent {agent.agent_id} 正在执行任务...")
# 模拟任务执行
agent.execute_current_task()
# 检查任务是否完成
if agent.current_task_completed():
agent.transition_to(IdleState())
def get_state_name(self) -> str:
return "busy"
class ErrorState(AgentState):
"""错误状态"""
def handle(self, agent: 'StateMachineAgent'):
print(f"Agent {agent.agent_id} 处于错误状态，正在尝试恢复...")
# 尝试恢复
if agent.attempt_recovery():
agent.transition_to(IdleState())
else:
print(f"Agent {agent.agent_id} 恢复失败，保持错误状态")
def get_state_name(self) -> str:
return "error"
class StateMachineAgent:
"""
状态机Agent
"""
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.state = IdleState()
self.current_task = None
self.task_queue = []
self.error_count = 0
def transition_to(self, state: AgentState):
"""切换状态"""
old_state = self.state.get_state_name()
self.state = state
new_state = state.get_state_name()
print(f"Agent {self.agent_id} 状态从 '{old_state}' 切换到 '{new_state}'")
def handle(self):
"""处理当前状态"""
self.state.handle(self)
def assign_task(self, task: str):
"""分配任务"""
self.task_queue.append(task)
print(f"任务 '{task}' 已分配给 Agent {self.agent_id}")
def has_task(self) -> bool:
"""检查是否有任务"""
return len(self.task_queue) > 0
def execute_current_task(self):
"""执行当前任务"""
if self.task_queue:
self.current_task = self.task_queue.pop(0)
print(f"Agent {self.agent_id} 开始执行任务: {self.current_task}")
# 模拟任务执行
time.sleep(0.5)
print(f"Agent {self.agent_id} 完成任务: {self.current_task}")
self.current_task = None
def current_task_completed(self) -> bool:
"""检查当前任务是否完成"""
return self.current_task is None
def attempt_recovery(self) -> bool:
"""尝试恢复"""
print(f"Agent {self.agent_id} 尝试恢复...")
# 简单的恢复逻辑：重置错误计数
self.error_count = 0
return True  # 假设总是能恢复
# 命令模式实现
class Command(ABC):
"""
命令抽象类
"""
@abstractmethod
def execute(self):
"""执行命令"""
pass
@abstractmethod
def undo(self):
"""撤销命令"""
pass
class MoveCommand(Command):
"""移动命令"""
def __init__(self, agent: 'CommandAgent', direction: str, distance: float):
self.agent = agent
self.direction = direction
self.distance = distance
self.previous_position = agent.position
def execute(self):
print(f"Agent {self.agent.agent_id} 执行移动命令: {self.direction} {self.distance}m")
self.previous_position = self.agent.position.copy()
if self.direction == "forward":
self.agent.position[1] += self.distance
elif self.direction == "backward":
self.agent.position[1] -= self.distance
elif self.direction == "left":
self.agent.position[0] -= self.distance
elif self.direction == "right":
self.agent.position[0] += self.distance
print(f"Agent {self.agent.agent_id} 新位置: {self.agent.position}")
def undo(self):
print(f"撤销移动命令，恢复到位置: {self.previous_position}")
self.agent.position = self.previous_position.copy()
class RotateCommand(Command):
"""旋转命令"""
def __init__(self, agent: 'CommandAgent', angle: float):
self.agent = agent
self.angle = angle
self.previous_angle = agent.orientation
def execute(self):
print(f"Agent {self.agent.agent_id} 执行旋转命令: {self.angle}度")
self.previous_angle = self.agent.orientation
self.agent.orientation += self.angle
# 标准化角度到0-360度
self.agent.orientation %= 360
print(f"Agent {self.agent.agent_id} 新朝向: {self.agent.orientation}度")
def undo(self):
print(f"撤销旋转命令，恢复到朝向: {self.previous_angle}")
self.agent.orientation = self.previous_angle
class CommandAgent:
"""
支持命令模式的Agent
"""
def __init__(self, agent_id: str, initial_position: List[float] = None):
self.agent_id = agent_id
self.position = initial_position or [0.0, 0.0]
self.orientation = 0.0  # 朝向，单位度
self.command_history = []
def execute_command(self, command: Command):
"""执行命令"""
command.execute()
self.command_history.append(command)
def undo_last_command(self):
"""撤销最后一个命令"""
if self.command_history:
last_command = self.command_history.pop()
last_command.undo()
else:
print("没有可撤销的命令")
def get_position(self) -> List[float]:
"""获取当前位置"""
return self.position.copy()
def get_orientation(self) -> float:
"""获取当前朝向"""
return self.orientation
def demo_observer_pattern():
"""
演示观察者模式
"""
print("=== 观察者模式演示 ===")
# 创建Agent监控器
monitor = AgentMonitor("robot_001")
# 创建观察者
logger = Logger("SystemLogger")
alert_system = AlertSystem(threshold=50)
perf_tracker = PerformanceTracker()
# 注册观察者
monitor.attach(logger)
monitor.attach(alert_system)
monitor.attach(perf_tracker)
# 模拟状态变化
print("\n1. 状态变化:")
monitor.change_state("working")
monitor.change_state("maintenance")
print("\n2. 健康分数变化:")
monitor.update_health(80)
monitor.update_health(40)  # 触发警报
monitor.update_health(20)  # 触发警报
print("\n3. 性能指标更新:")
monitor.log_metric("response_time", 0.15)
monitor.log_metric("accuracy", 0.95)
# 移除观察者
monitor.detach(alert_system)
print("\n4. 移除警报系统后健康变化:")
monitor.update_health(10)  # 不会触发警报
# 显示日志
print(f"\n5. 日志观察者记录了 {len(logger.logs)} 条日志")
print(f"   警报系统记录了 {len(alert_system.alerts)} 条警报")
print(f"   性能追踪器记录了 {len(perf_tracker.metrics_history.get('robot_001', {}))} 种指标")
def demo_strategy_pattern():
"""
演示策略模式
"""
print("\n\n=== 策略模式演示 ===")
# 创建不同策略
simple_strategy = SimplePlanningStrategy()
advanced_strategy = AdvancedPlanningStrategy()
learning_strategy = LearningPlanningStrategy()
# 创建Agent
agent = AdaptiveAgent("navigator_001", simple_strategy)
# 设置目标
goals = ["navigate_to_target", "avoid_obstacle"]
agent.set_goals(goals)
print("\n1. 使用简单策略:")
current_state = {"battery_level": 80, "obstacle_distance": 10}
agent.update_state(current_state)
plan1 = agent.plan_actions()
agent.execute_plan(plan1)
print("\n2. 切换到高级策略:")
agent.set_strategy(advanced_strategy)
current_state = {"battery_level": 25, "obstacle_distance": 3, "target_distance": 60}
agent.update_state(current_state)
plan2 = agent.plan_actions()
agent.execute_plan(plan2)
print("\n3. 切换到学习策略:")
agent.set_strategy(learning_strategy)
current_state = {"battery_level": 90, "obstacle_distance": 15, "target_distance": 25}
agent.update_state(current_state)
plan3 = agent.plan_actions()
agent.execute_plan(plan3)
# 演示性能评估和自适应
print(f"\n4. Agent性能评估: {agent.evaluate_performance():.2f}")
agent.adapt_strategy()
def demo_state_pattern():
"""
演示状态模式
"""
print("\n\n=== 状态模式演示 ===")
# 创建状态机Agent
agent = StateMachineAgent("worker_001")
print(f"初始状态: {agent.state.get_state_name()}")
# 模拟一系列操作
print("\n1. 分配任务，状态应该改变:")
agent.assign_task("clean_room")
agent.handle()  # 检查状态
print(f"当前状态: {agent.state.get_state_name()}")
print("\n2. 处理忙碌状态:")
agent.handle()  # 执行任务
print(f"任务完成后状态: {agent.state.get_state_name()}")
print("\n3. 再次分配任务:")
agent.assign_task("deliver_package")
agent.handle()  # 检查状态
print(f"分配任务后状态: {agent.state.get_state_name()}")
agent.handle()  # 执行任务
print(f"任务完成后状态: {agent.state.get_state_name()}")
def demo_command_pattern():
"""
演示命令模式
"""
print("\n\n=== 命令模式演示 ===")
# 创建命令Agent
agent = CommandAgent("rover_001", [0, 0])
print(f"初始位置: {agent.get_position()}, 朝向: {agent.get_orientation()}度")
# 创建并执行命令
print("\n1. 执行移动命令:")
move_cmd1 = MoveCommand(agent, "forward", 5.0)
agent.execute_command(move_cmd1)
rotate_cmd = RotateCommand(agent, 90)
agent.execute_command(rotate_cmd)
move_cmd2 = MoveCommand(agent, "forward", 3.0)
agent.execute_command(move_cmd2)
print(f"当前位置: {agent.get_position()}, 朝向: {agent.get_orientation()}度")
print("\n2. 撤销命令:")
print("撤销前进3米:")
agent.undo_last_command()
print(f"位置: {agent.get_position()}")
print("撤销右转90度:")
agent.undo_last_command()
print(f"位置: {agent.get_position()}, 朝向: {agent.get_orientation()}度")
print("撤销前进5米:")
agent.undo_last_command()
print(f"最终位置: {agent.get_position()}")
def demo_combined_patterns():
"""
演示多种模式的组合使用
"""
print("\n\n=== 组合模式演示 ===")
# 创建一个综合Agent，结合多种设计模式
class ComprehensiveAgent:
def __init__(self, agent_id: str):
# 使用观察者模式监控状态
self.monitor = AgentMonitor(agent_id)
# 使用策略模式进行规划
self.adaptive_agent = AdaptiveAgent(agent_id, SimplePlanningStrategy())
# 使用状态模式管理状态
self.state_machine_agent = StateMachineAgent(agent_id)
# 使用命令模式执行动作
self.command_agent = CommandAgent(agent_id)
# 添加观察者
self.logger = Logger(f"{agent_id}_logger")
self.monitor.attach(self.logger)
def execute_task(self, task: str, environment_state: Dict[str, Any]):
"""执行任务的完整流程"""
print(f"\nComprehensiveAgent {self.monitor.agent_id} 开始执行任务: {task}")
# 更新监控状态
self.monitor.change_state("executing_task")
# 使用策略模式制定计划
self.adaptive_agent.set_goals([task])
self.adaptive_agent.update_state(environment_state)
plan = self.adaptive_agent.plan_actions()
# 使用状态模式管理执行状态
self.state_machine_agent.assign_task(task)
# 使用命令模式执行具体动作
for action in plan:
if action.startswith("move_"):
direction = action.split("_")[1] if "_" in action else "forward"
cmd = MoveCommand(self.command_agent, direction, 2.0)
self.command_agent.execute_command(cmd)
elif action.startswith("turn_"):
angle = 90 if "right" in action else -90
cmd = RotateCommand(self.command_agent, angle)
self.command_agent.execute_command(cmd)
# 更新监控状态
self.monitor.change_state("idle")
self.monitor.update_health(90)
# 演示综合Agent
comp_agent = ComprehensiveAgent("comp_001")
# 执行不同任务
tasks_with_states = [
("navigate_to_target", {"battery_level": 85, "target_distance": 20}),
("avoid_obstacle", {"obstacle_distance": 5, "battery_level": 70}),
("collect_resources", {"resource_density": 0.6, "battery_level": 50})
]
for task, state in tasks_with_states:
comp_agent.execute_task(task, state)
print(f"\n综合Agent执行了 {len(comp_agent.logger.logs)} 个监控事件")
if __name__ == "__main__":
demo_observer_pattern()
demo_strategy_pattern()
demo_state_pattern()
demo_command_pattern()
demo_combined_patterns()
print("\n=== Agent设计模式演示完成 ===")
print("本演示展示了以下设计模式在Agent系统中的应用:")
print("1. 观察者模式 - 用于状态监控和事件通知")
print("2. 策略模式 - 用于动态切换算法和行为策略")
print("3. 状态模式 - 用于管理Agent的不同行为状态")
print("4. 命令模式 - 用于封装和执行动作")
print("5. 多模式组合 - 展示如何结合多种模式构建复杂Agent")