Agent设计模式入门指南：从原理到实战避坑

1次阅读

共计 2291 个字符，预计需要花费 6 分钟才能阅读完成。

假设我们有个电商订单处理系统，用传统回调方式实现时是这样的：

# 回调地狱示例
class OrderSystem:
    def create_order(self, callback):
        db.insert(order, lambda: callback.on_success())

    def pay_order(self, order_id, callback):
        payment.charge(lambda: inventory.check(callback.on_paid))

问题 1：箭头型代码：嵌套回调形成金字塔结构
问题 2：状态分散：订单状态可能散落在多个回调闭包中
问题 3：错误处理困难：每个异步操作都需要单独 try-catch

Agent 通过三个核心机制解决上述问题：

消息队列：所有操作通过消息触发（松耦合）
有限状态机（FSM）：集中管理业务状态
隔离上下文：每个 Agent 实例拥有独立存储

特性	Agent 模式	Actor 模型
通信方式	异步消息	异步消息
状态管理	显式 FSM	行为隐含状态
分布式支持	需额外实现	原生支持
适用规模	单机 / 轻量分布式	大规模分布式

from dataclasses import dataclass
from enum import Enum, auto
import asyncio

class OrderState(Enum):
    CREATED = auto()
    PAID = auto()
    SHIPPED = auto()

@dataclass
class OrderMessage:
    msg_type: str
    payload: dict

class OrderAgent:
    def __init__(self, order_id):
        self.order_id = order_id
        self.state = OrderState.CREATED
        self.mailbox = asyncio.Queue()  # 消息队列
        self._task = asyncio.create_task(self._process())

    async def _process(self):
        while True:
            try:
                msg = await self.mailbox.get()
                await self._handle_message(msg)
            except Exception as e:
                print(f"Agent {self.order_id} error: {e}")

    async def _handle_message(self, msg: OrderMessage):
        if msg.msg_type == "pay" and self.state == OrderState.CREATED:
            await self._process_payment(msg.payload)
            self.state = OrderState.PAID
        # 其他状态转换...

关键实现说明：

@dataclass：自动生成消息类的 __init__ 方法
asyncio.Queue：线程安全的异步消息队列
状态检查：处理消息前验证当前状态合法性

from concurrent.futures import ThreadPoolExecutor

class AgentPool:
    def __init__(self):
        self.executor = ThreadPoolExecutor(
            max_workers=8,  # 根据 CPU 核心数调整
            thread_name_prefix="agent_worker"
        )
        self.loop = asyncio.get_event_loop()

    async def dispatch(self, agent, msg):
        await self.loop.run_in_executor(
            self.executor,
            agent.mailbox.put_nowait,
            msg
        )

import uvloop
uvloop.install()  # 替换默认事件循环，性能提升 2 - 4 倍

背压 (Backpressure) 策略：
监控队列长度agent.mailbox.qsize()
达到阈值时返回 HTTP 503
客户端采用指数退避重试

消息去重方案：
为每条消息附加唯一 ID
使用 Redis 记录已处理 ID
采用 SETNX 原子操作

# Redis 幂等检查示例
async def is_duplicate(msg_id):
    return await redis_client.setnx(f"msg:{msg_id}", "1") == 0

尝试将 Web 请求转化为 Agent 消息：

@app.post("/orders")
async def create_order():
    order_id = generate_id()
    agent = OrderAgent(order_id)
    await agent.mailbox.put(OrderMessage("create", {...}))
    return {"id": order_id}