开源 AI 工具链：轻量化 Agent 框架的事件驱动架构设计

一、痛点与场景：为什么 Agent 框架需要事件驱动

在 AI 应用落地的过程中，Agent 框架正在从简单的请求-响应模式，走向复杂的多步骤编排。传统的方式是使用线性调用链——Agent A 调用 Agent B，B 调用 Agent C，层层嵌套。这种模式在简单场景下可行，但面对生产环境的真实需求时，问题接踵而至。

首先是耦合度过高。当一个 Agent 需要同时触发多个下游任务时，线性调用链必须等待所有下游返回，整体延迟取决于最慢的那个。其次是扩展性差——新增一个 Agent 节点，需要修改上游调用方的代码。最后是容错能力弱，任何一个节点失败，整条链路都会中断。

事件驱动架构（EDA）天然解耦了生产者和消费者。Agent 不再直接调用其他 Agent，而是发布事件到消息总线，由感兴趣的消费方自行订阅。这种模式下，新增 Agent 只需订阅已有事件，无需修改任何上游代码。同时，事件可以异步处理，整体吞吐量显著提升。

flowchart TB
    subgraph 传统线性调用
        A1[用户请求] --> B1[路由 Agent]
        B1 --> C1[检索 Agent]
        C1 --> D1[生成 Agent]
        D1 --> E1[审核 Agent]
        E1 --> F1[响应]
    end

    subgraph 事件驱动架构
        A2[用户请求] --> B2[事件总线]
        B2 -->|request事件| C2[路由 Agent]
        C2 -->|route事件| B2
        B2 -->|search事件| D2[检索 Agent]
        B2 -->|generate事件| E2[生成 Agent]
        D2 -->|result事件| B2
        E2 -->|draft事件| B2
        B2 -->|review事件| F2[审核 Agent]
        F2 -->|approve事件| B2
        B2 -->|response事件| G2[响应]
    end

二、事件驱动 Agent 框架的核心机制

事件驱动架构的核心由三个组件构成：事件总线（Event Bus）、事件定义（Event Schema）和 Agent 注册表（Agent Registry）。事件总线负责消息的路由和分发，事件定义约束了事件的格式和语义，Agent 注册表管理所有 Agent 的订阅关系。

在轻量化设计中，我们选择基于 Python 的 asyncio 实现事件总线，避免引入 Kafka、RabbitMQ 等重量级中间件。对于单机或小集群部署的 AI 应用，进程内事件总线的延迟远低于网络消息队列，同时运维成本几乎为零。

sequenceDiagram
    participant User as 用户
    participant Bus as 事件总线
    participant Router as 路由Agent
    participant Search as 检索Agent
    participant Gen as 生成Agent
    participant Review as 审核Agent

    User->>Bus: 发布 user_request 事件
    Bus->>Router: 投递 user_request
    Router->>Bus: 发布 route_result 事件
    Bus->>Search: 投递 route_result
    Search->>Bus: 发布 search_result 事件
    Bus->>Gen: 投递 search_result
    Gen->>Bus: 发布 draft_complete 事件
    Bus->>Review: 投递 draft_complete
    Review->>Bus: 发布 review_pass 事件
    Bus->>User: 投递 final_response

三、生产级代码实现

3.1 事件总线核心实现

import asyncio
from typing import Any, Callable, Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
import uuid
import logging

logger = logging.getLogger(__name__)


@dataclass
class Event:
    """事件定义：所有 Agent 间通信的标准化消息格式"""
    event_type: str          # 事件类型，如 "user_request"、"search_result"
    payload: Dict[str, Any]  # 事件载荷，携带业务数据
    event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
    timestamp: str = field(default_factory=lambda: datetime.utcnow().isoformat())
    source: str = ""         # 事件来源 Agent 标识
    correlation_id: str = "" # 关联 ID，用于追踪同一请求的完整事件链


class EventBus:
    """轻量化异步事件总线

    设计考量：
    - 基于 asyncio 实现，无需外部消息队列依赖
    - 支持通配符订阅（如 "search.*" 匹配所有检索相关事件）
    - 内置背压控制：当消费者处理速度跟不上时，自动降级
    """

    def __init__(self, max_queue_size: int = 1000):
        self._subscribers: Dict[str, List[asyncio.Queue]] = {}
        self._max_queue_size = max_queue_size
        self._event_log: List[Event] = []  # 事件日志，用于调试和审计

    async def publish(self, event: Event) -> None:
        """发布事件到总线，投递给所有匹配的订阅者"""
        self._event_log.append(event)

        # 精确匹配
        queues = self._subscribers.get(event.event_type, [])

        # 通配符匹配：遍历所有订阅模式，检查是否匹配当前事件
        for pattern, pattern_queues in self._subscribers.items():
            if pattern.endswith(".*"):
                prefix = pattern[:-2]
                if event.event_type.startswith(prefix + "."):
                    queues.extend(pattern_queues)

        if not queues:
            logger.debug(f"事件 {event.event_type} 无订阅者，已丢弃")
            return

        for queue in queues:
            # 背压控制：队列满时丢弃最旧的事件，而非阻塞生产者
            if queue.full():
                try:
                    queue.get_nowait()  # 丢弃最旧事件
                    logger.warning(f"队列已满，丢弃旧事件以腾出空间")
                except asyncio.QueueEmpty:
                    pass
            await queue.put(event)

    def subscribe(self, event_type: str) -> asyncio.Queue:
        """订阅指定类型的事件，返回一个异步队列供消费"""
        if event_type not in self._subscribers:
            self._subscribers[event_type] = []
        queue = asyncio.Queue(maxsize=self._max_queue_size)
        self._subscribers[event_type].append(queue)
        return queue

    def unsubscribe(self, event_type: str, queue: asyncio.Queue) -> None:
        """取消订阅，Agent 下线时必须调用，防止内存泄漏"""
        if event_type in self._subscribers:
            self._subscribers[event_type] = [
                q for q in self._subscribers[event_type] if q is not queue
            ]

3.2 Agent 基类与注册机制

from abc import ABC, abstractmethod


class BaseAgent(ABC):
    """Agent 基类：定义事件驱动 Agent 的标准生命周期

    设计考量：
    - 每个 Agent 独立运行自己的事件循环，互不阻塞
    - 通过 subscribe/publish 与事件总线交互，而非直接调用
    - 支持优雅关闭：收到 shutdown 事件后完成当前任务再退出
    """

    def __init__(self, name: str, event_bus: EventBus):
        self.name = name
        self.event_bus = event_bus
        self._queues: Dict[str, asyncio.Queue] = {}
        self._running = False

    def listen(self, event_type: str) -> None:
        """声明关注的事件类型"""
        queue = self.event_bus.subscribe(event_type)
        self._queues[event_type] = queue

    async def start(self) -> None:
        """启动 Agent，开始监听事件"""
        self._running = True
        tasks = []
        for event_type, queue in self._queues.items():
            tasks.append(self._consume_loop(event_type, queue))
        await asyncio.gather(*tasks)

    async def _consume_loop(self, event_type: str, queue: asyncio.Queue) -> None:
        """事件消费循环：持续从队列中取出事件并处理"""
        while self._running:
            try:
                event = await asyncio.wait_for(queue.get(), timeout=1.0)
            except asyncio.TimeoutError:
                continue

            try:
                result = await self.handle(event)
                if result:
                    # 处理完成后自动发布结果事件
                    result.source = self.name
                    result.correlation_id = event.correlation_id or event.event_id
                    await self.event_bus.publish(result)
            except Exception as e:
                logger.error(f"Agent {self.name} 处理事件失败: {e}", exc_info=True)
                # 发布错误事件，供监控或降级 Agent 消费
                error_event = Event(
                    event_type=f"error.{event.event_type}",
                    payload={"error": str(e), "original_event_id": event.event_id},
                    source=self.name,
                    correlation_id=event.correlation_id,
                )
                await self.event_bus.publish(error_event)

    @abstractmethod
    async def handle(self, event: Event) -> Optional[Event]:
        """子类实现：处理事件并返回结果事件（或 None）"""
        pass

    async def stop(self) -> None:
        """优雅关闭"""
        self._running = False
        for event_type, queue in self._queues.items():
            self.event_bus.unsubscribe(event_type, queue)

3.3 具体业务 Agent 实现

class RouterAgent(BaseAgent):
    """路由 Agent：分析用户意图，决定后续调用哪些 Agent"""

    def __init__(self, event_bus: EventBus):
        super().__init__("router", event_bus)
        self.listen("user_request")

    async def handle(self, event: Event) -> Optional[Event]:
        query = event.payload.get("query", "")
        # 简化的意图分类逻辑，生产环境应接入 LLM
        intent = self._classify_intent(query)

        return Event(
            event_type="route_result",
            payload={"query": query, "intent": intent, "original_event_id": event.event_id},
        )

    def _classify_intent(self, query: str) -> str:
        """基于关键词的快速意图分类，低延迟优先"""
        if any(kw in query for kw in ["搜索", "查找", "检索"]):
            return "search"
        elif any(kw in query for kw in ["生成", "写", "创作"]):
            return "generate"
        elif any(kw in query for kw in ["分析", "对比", "评估"]):
            return "analyze"
        return "general"


class SearchAgent(BaseAgent):
    """检索 Agent：根据路由结果执行知识库检索"""

    def __init__(self, event_bus: EventBus):
        super().__init__("search", event_bus)
        self.listen("route_result")

    async def handle(self, event: Event) -> Optional[Event]:
        intent = event.payload.get("intent", "")
        query = event.payload.get("query", "")

        # 仅处理需要检索的意图
        if intent not in ("search", "analyze", "general"):
            return None

        # 模拟检索过程，生产环境接入向量数据库
        results = await self._search(query)

        return Event(
            event_type="search_result",
            payload={"query": query, "results": results},
        )

    async def _search(self, query: str) -> List[Dict]:
        """执行检索，生产环境替换为 Milvus/Pinecone 调用"""
        await asyncio.sleep(0.1)  # 模拟网络延迟
        return [{"content": f"与 '{query}' 相关的知识片段", "score": 0.92}]

四、架构权衡与边界分析

4.1 进程内事件总线的局限性

这套轻量化方案的核心取舍在于：放弃了分布式消息队列的持久化和跨进程能力，换取了极低的部署复杂度和亚毫秒级的事件投递延迟。

适用场景：单机部署、小集群（2-5 节点）、AI 应用原型验证阶段。在这些场景下，引入 Kafka 或 RabbitMQ 的运维成本远超其收益。

禁用场景：需要跨服务持久化事件的微服务架构、要求消息不丢失的金融级场景、Agent 数量超过 50 个且需要动态扩缩容的大规模部署。这些场景必须使用专业的分布式消息队列。

4.2 背压策略的权衡

当前实现采用"丢弃最旧事件"的背压策略。这意味着在极端高负载下，早期事件可能被丢弃。对于 AI Agent 场景，这个取舍是合理的——用户更关心最新请求的响应，而非排队等待的旧请求。但如果业务要求每条消息都必须处理，则需要改为阻塞式背压，代价是整体吞吐量下降。

4.3 事件顺序性

异步事件驱动架构不保证事件的严格顺序。如果业务要求 A 事件必须在 B 事件之前处理，需要在事件中携带序列号，由消费方自行排序。这增加了消费端的复杂度，但对于大多数 Agent 编排场景，宽松的顺序性已经足够。

五、总结

事件驱动架构为 Agent 框架带来了三个关键收益：解耦（Agent 之间无直接依赖）、弹性（新增 Agent 无需修改已有代码）、容错（单个 Agent 故障不影响事件总线的运行）。基于 asyncio 的进程内事件总线，在轻量化部署场景下提供了足够的能力，同时将运维复杂度降到最低。

落地路线建议：第一步，用本文的事件总线实现替换现有的线性调用链；第二步，为关键事件添加持久化日志，支持故障后的事件回放；第三步，当单机性能不足时，将事件总线替换为 Redis Streams 或 NATS，Agent 代码无需修改，只需更换事件总线的实现即可。