共计 2568 个字符，预计需要花费 7 分钟才能阅读完成。
什么是 Agent？
Agent（智能体）可以理解为一个能自主行动的软件实体。它有点像我们日常生活中的快递员——不用你一步步指挥，自己就知道取件、规划路线、送货。在编程领域，Agent 最接近 Actor 模型（Actor Model），但比简单的对象（Object）更独立。
- 自治性（Autonomy）：Agent 自己管理内部状态，比如快递员记得包裹要送哪里
- 反应性（Reactivity）：能对外部刺激做出响应，像快递员听到门铃会开门
- 主动性（Pro-activeness）：可以自主发起动作，比如快递员发现收件人不在会自动发起第二次派送
Agent vs 传统对象：架构差异
用寄快递的例子对比两种编程范式：
-
传统 OOP 写法
# 需要外部控制每一步 package = Package(address="北京") courier = Courier() courier.pick_up(package) # 手动调用取件 courier.navigate() # 手动触发导航 courier.deliver() # 手动执行送货 -
Agent 写法
# 只需发送消息，Agent 自主处理 courier = CourierAgent() courier.send({"type": "deliver", "address": "北京"}) # 发送任务消息后就不用管了
关键差异点：
- 状态管理 ：Agent 自己维护送货状态，外部无法直接修改
- 通信方式 ：通过消息（message）而非方法调用（method call）交互
- 执行控制 ：Agent 内部有事件循环处理消息队列
动手实现基础 Agent
下面用 Python 实现一个能接收订单的咖啡师 Agent：
import asyncio
from typing import Dict, Any
class BaristaAgent:
def __init__(self):
self._queue = asyncio.Queue()
self._is_running = False
self._current_order = None # 内部状态
async def start(self):
"""启动 Agent 的事件循环"""
self._is_running = True
while self._is_running:
try:
message = await self._queue.get()
await self._process_message(message)
except Exception as e:
print(f"处理消息失败: {e}")
async def stop(self):
"""安全停止 Agent"""
self._is_running = False
await self._queue.put(None) # 发送停止信号
async def send(self, message: Dict[str, Any]):
"""外部接口：发送消息到 Agent"""
await self._queue.put(message)
async def _process_message(self, message):
"""处理消息的核心逻辑"""
if message is None: # 停止信号
return
if message["type"] == "order":
self._current_order = message["items"]
print(f"开始制作: {self._current_order}")
await asyncio.sleep(2) # 模拟制作耗时
print(f"已完成: {self._current_order}")
self._current_order = None使用示例：
async def main():
barista = BaristaAgent()
# 启动 Agent 后台任务
task = asyncio.create_task(barista.start())
# 发送订单
await barista.send({"type": "order", "items": "拿铁"})
# 5 秒后停止
await asyncio.sleep(5)
await barista.stop()
await task
asyncio.run(main())生产环境实用技巧
线程安全方案
推荐使用 asyncio 的异步队列，避免多线程竞争。如果需要跨线程通信：
# 在另一个线程发送消息
def thread_send(agent, message):
asyncio.run_coroutine_threadsafe(agent.send(message),
loop=agent.event_loop # 需要传入 Agent 的事件循环
)状态持久化
用 Redis 保存 Agent 状态（需安装 redis 包）：
import redis
class PersistentAgent:
def __init__(self, redis_url):
self._redis = redis.Redis.from_url(redis_url)
async def save_state(self):
"""定时保存状态到 Redis"""
while True:
await asyncio.sleep(10) # 每 10 秒保存
self._redis.set(
"agent_state",
json.dumps(self._internal_state)
)新手常见问题
- 阻塞事件循环
- ❌ 错误做法：在消息处理中调用
time.sleep() -
✅ 正确方案：始终使用
await asyncio.sleep() -
消息丢失
- ❌ 直接丢弃处理失败的消息
-
✅ 添加重试机制或死信队列
async def _process_message(self, msg): try: # 业务逻辑 except Exception: await self._queue.put(msg) # 重新入队 -
状态不一致
- ❌ 允许外部直接修改
self._current_order - ✅ 所有状态变更通过消息触发
总结建议
初次开发 Agent 系统时，建议先用 asyncio 实现单进程版本，熟悉消息驱动模式。之后逐步扩展：
- 添加监控接口检查消息积压情况
- 实现
__eq__方法方便测试时比对 Agent 状态 - 用
pytest-asyncio编写异步测试用例
Agent 架构特别适合需要处理异步事件、维护复杂状态的场景，比如物联网设备控制、游戏 NPC 行为等。关键是保持每个 Agent 的职责单一，通过消息组合复杂行为。
正文完
