共计 3226 个字符，预计需要花费 9 分钟才能阅读完成。
为什么需要 Agent 范式
当我们需要开发一个能自主决策的系统时，传统面向对象编程 (OOP) 会显得力不从心。比如要开发一个智能家居控制系统，用 OOP 写出来的代码往往是这样的：
class Thermostat:
def set_temperature(self, temp):
print(f"Temperature set to {temp}")
# 需要外部不断调用
thermostat = Thermostat()
while True:
if 室温过高:
thermostat.set_temperature(25)而 Agent 范式则是这样的存在：
class SmartThermostat(Agent):
def __init__(self):
self.comfort_range = (22, 26)
def on_perceive(self, env_data):
if env_data['temp'] > self.comfort_range[1]:
self.act({'type': 'cool', 'target': 25})两者的关键差异在于：
- 自主性：Agent 会主动感知环境并做出决策
- 反应性：能实时响应环境变化
- 目标导向：行为始终围绕预设目标展开
架构风格选型指南
在实现 Agent 时，我们通常面临两种架构选择：
基于事件的架构（Event-Driven）
class EventDrivenAgent:
def handle_event(self, event):
if event.type == 'TEMPERATURE_ALERT':
self.adjust_AC(event.value)优点：
– 代码结构清晰
– 容易集成现有消息系统（如 RabbitMQ）
缺点：
– 复杂决策逻辑会变成嵌套 if-else
– 难以实现长期目标跟踪
基于行为的架构（Behavior-Based）
class BehaviorAgent:
behaviors = [CoolingBehavior(), SafetyCheckBehavior()]
def decide_action(self):
for behavior in self.behaviors:
if behavior.should_activate(self.state):
return behavior.execute()优点：
– 行为模块可插拔
– 天然支持优先级机制
缺点：
– 行为间协调较复杂
– 调试难度略高
新手建议：从事件驱动入手，待业务复杂后再引入行为树。
手把手实现基础 Agent
下面我们实现一个符合 PEP8 的完整 Agent 基类：
from typing import Any, Dict, List
from queue import Queue
import threading
class BaseAgent:
"""Agent 基础实现，包含消息队列和简单行为树"""
def __init__(self):
self.message_queue = Queue()
self.running = False
self.behaviors: List[Behavior] = []
def perceive(self, env_data: Dict[str, Any]) -> None:
"""接收环境数据"""
self.message_queue.put(('percept', env_data))
def act(self, action: Dict[str, Any]) -> None:
"""执行动作"""
print(f"Executing action: {action}")
def start(self) -> None:
"""启动 Agent 主循环"""
self.running = True
worker = threading.Thread(target=self._run_loop)
worker.daemon = True
worker.start()
def _run_loop(self) -> None:
"""处理消息的内部循环"""
while self.running:
msg_type, content = self.message_queue.get()
if msg_type == 'percept':
self._process_percept(content)
def _process_percept(self, percept: Dict[str, Any]) -> None:
"""处理感知数据的模板方法"""
for behavior in self.behaviors:
if behavior.check_conditions(percept):
self.act(behavior.generate_action())使用示例：
class TemperatureBehavior:
def check_conditions(self, percept):
return percept.get('temp', 0) > 30
def generate_action(self):
return {'type': 'cool', 'level': 'high'}
agent = BaseAgent()
agent.behaviors.append(TemperatureBehavior())
agent.start()
# 模拟环境输入
agent.perceive({'temp': 35}) # 将触发降温动作新手避坑三大指南
在帮助 20+ 同学调试 Agent 代码后，我总结出这些高频问题：
- 消息积压导致响应延迟
# 错误示范
def on_message(msg):
time.sleep(5) # 同步阻塞处理
# 正确做法
def on_message(msg):
threading.Thread(target=long_running_task, args=(msg,)).start()- 状态同步出现竞态条件
# 危险代码
self.counter += 1
# 线程安全版本
with threading.Lock():
self.counter += 1- 行为树陷入死循环
# 错误行为树设计
def should_activate(self):
return not self.activated # 可能反复切换状态
# 应加入冷却机制
def should_activate(self):
return (not self.activated
and time.time() - self.last_activate > 5)进阶路线图
当基础 Agent 跑通后，可以尝试这两个方向：
集成 LLM 实现智能决策
class LLMBehavior(Behavior):
def generate_action(self, percept):
prompt = f"当前环境：{percept}\n 建议采取什么行动？"
response = openai.ChatCompletion.create(
model="gpt-4",
messages=[{"role": "user", "content": prompt}]
)
return parse_action(response.choices[0].message.content)多 Agent 协作系统
class Marketplace:
def __init__(self):
self.agents: Dict[str, BaseAgent] = {}
def broadcast(self, sender, message):
for agent_id, agent in self.agents.items():
if agent_id != sender:
agent.receive(message)
class SellerAgent(BaseAgent):
def handle_message(self, msg):
if msg.type == 'purchase_request':
self.check_inventory(msg.item)思考题
当 Agent 出现异常行为时，如何设计熔断机制？可以考虑：
- 心跳检测机制
- 异常行为频率统计
- 回退到安全状态
- 上级 Agent 监管策略
正文完
