Agent 事件驱动架构：用消息驱动替代轮询等待
一、Agent 推理完一次就阻塞等待，CPU 时间全浪费在空转上
最早的 Agent 实现是这样：while True: response = llm.call(); if response.has_tool_call: result = execute_tool(); continue; else: break。LLM 调用期间的 2-5 秒，主线程完全阻塞等待。如果 Agent 在一次对话中多次调用 LLM（ReAct 模式下 3-5 次很正常），累计阻塞时间轻松突破 10 秒。
更严重的问题是资源利用率。一个 Agent 进程在等待 LLM 响应时，CPU 和内存被占用但无事可做。如果有 100 个并发的 Agent 会话，就需要 100 个进程/协程同时持有 LLM 连接——连接池被撑爆，后面来的请求排队超时。
事件驱动架构的解法：不阻塞等待，而是订阅事件。LLM 响应是一个事件，工具调用结果是事件，收到事件后触发下一步动作。
二、Agent 事件驱动模型的设计
sequenceDiagram participant User as 用户 participant Agent as Agent Core participant Bus as Event Bus participant LLM as LLM 服务 participant Tool as 工具服务 participant Store as 状态存储 User->>Agent: 发送消息 Agent->>Bus: 发布 AgentStartEvent Agent->>LLM: 异步调用 LLM Note over Agent: 释放线程/协程<br/>不阻塞等待 LLM-->>Bus: 发布 LLMResponseEvent Bus->>Agent: 投递事件 Agent->>Agent: 解析是否需要工具调用 alt 需要工具 Agent->>Bus: 发布 ToolInvokeEvent Bus->>Tool: 投递事件 Tool-->>Bus: 发布 ToolResultEvent Bus->>Agent: 投递事件 Agent->>LLM: 继续推理 Note over Agent: 再次释放，不阻塞 else 不需要工具 Agent->>Store: 保存对话结果 Agent->>Bus: 发布 AgentCompleteEvent Bus->>User: 投递结果 end核心转换：从"请求-响应"的同步模型，变成"发布-订阅"的异步事件模型。Agent 不再持有 LLM 连接等待响应，而是向 Event Bus 注册一个事件处理器。当 LLM 响应到达时，Event Bus 回调处理器继续推理。整个过程 Agent 的线程/协程只在"处理事件"时活跃，其余时间完全释放。
三、生产级事件驱动 Agent 实现
""" 事件驱动的 Agent 执行引擎 使用消息队列 + 事件处理器替代阻塞等待 架构：所有组件通过 Event Bus 解耦 - Agent 发布工具调用事件 - 工具服务订阅并处理 - 结果通过事件回传 """ import asyncio import json import uuid from typing import Dict, Any, Optional, Callable, Awaitable from dataclasses import dataclass, field from datetime import datetime from enum import Enum import aioredis class EventType(str, Enum): """Agent 事件类型""" AGENT_START = "agent.start" AGENT_THINK = "agent.think" LLM_REQUEST = "llm.request" LLM_RESPONSE = "llm.response" TOOL_INVOKE = "tool.invoke" TOOL_RESULT = "tool.result" AGENT_COMPLETE = "agent.complete" AGENT_ERROR = "agent.error" @dataclass class AgentEvent: """Agent 事件""" event_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12]) event_type: EventType session_id: str turn_id: str timestamp: float = field(default_factory=lambda: datetime.now().timestamp()) payload: Dict[str, Any] = field(default_factory=dict) # 回调地址：事件处理器收到事件后，把结果发给这个地址 reply_to: Optional[str] = None # 关联的事件 ID（如 TOOL_RESULT 关联到 TOOL_INVOKE） correlation_id: Optional[str] = None class EventBus: """ 基于 Redis Stream 的轻量级事件总线 为什么用 Redis Stream 而非 RabbitMQ/Kafka： 1. Agent 系统通常已经依赖 Redis 做缓存/状态存储 2. Stream 的消费者组支持同组竞争消费（负载均衡） 3. Stream 提供消息追溯（XREAD），方便排查 """ def __init__(self, redis_url: str): self.redis: Optional[aioredis.Redis] = None self.redis_url = redis_url # 事件处理器注册表 # key=EventType, value=list of handlers self._handlers: Dict[EventType, list] = { etype: [] for etype in EventType } # 消费者组名 self._consumer_group = "agent-workers" async def connect(self): """建立 Redis 连接""" self.redis = await aioredis.from_url( self.redis_url, encoding="utf-8", decode_responses=True ) async def publish(self, event: AgentEvent): """发布事件到 Stream""" stream_key = f"agent:events:{event.event_type.value}" await self.redis.xadd( stream_key, { "event_id": event.event_id, "event_type": event.event_type.value, "session_id": event.session_id, "turn_id": event.turn_id, "timestamp": str(event.timestamp), "payload": json.dumps(event.payload), "reply_to": event.reply_to or "", "correlation_id": event.correlation_id or "", }, maxlen=10000, # 每个 Stream 最多保留 10000 条 ) async def subscribe(self, event_type: EventType, handler: Callable): """订阅事件类型并开始消费""" self._handlers[event_type].append(handler) # 创建消费者组（幂等） stream_key = f"agent:events:{event_type.value}" try: await self.redis.xgroup_create( stream_key, self._consumer_group, id="0", mkstream=True ) except aioredis.ResponseError: # 消费者组已存在 pass # 启动后台消费循环 consumer_name = f"{event_type.value}-{uuid.uuid4().hex[:6]}" asyncio.create_task(self._consume_loop(event_type, consumer_name)) async def _consume_loop(self, event_type: EventType, consumer_name: str): """消费循环：持续拉取事件并分发到处理器""" stream_key = f"agent:events:{event_type.value}" while True: try: # 批量读取，一次处理多条事件 messages = await self.redis.xreadgroup( groupname=self._consumer_group, consumername=consumer_name, streams={stream_key: ">"}, count=10, block=1000, # 阻塞 1s，等待新消息 ) if not messages: continue for stream, msgs in messages: for msg_id, msg_data in msgs: # 重建事件对象 event = AgentEvent( event_id=msg_data.get("event_id", ""), event_type=EventType(msg_data.get("event_type", "")), session_id=msg_data.get("session_id", ""), turn_id=msg_data.get("turn_id", ""), timestamp=float(msg_data.get("timestamp", 0)), payload=json.loads(msg_data.get("payload", "{}")), reply_to=msg_data.get("reply_to", None), correlation_id=msg_data.get("correlation_id", None), ) # 分发给所有注册的处理器 for handler in self._handlers[event_type]: try: await handler(event) except Exception as e: # 单个处理器失败不应影响其他处理器 await self.publish(AgentEvent( event_type=EventType.AGENT_ERROR, session_id=event.session_id, turn_id=event.turn_id, payload={ "error": str(e), "handler_event": event.event_type.value, }, )) # 确认消息 await self.redis.xack(stream_key, self._consumer_group, msg_id) except asyncio.CancelledError: break except Exception as e: # 消费循环异常，记录后继续 print(f"消费循环异常 ({event_type.value}): {e}") await asyncio.sleep(1) class EventDrivenAgent: """ 事件驱动的 Agent 执行引擎 与传统 Agent 的区别： 1. 不持有 LLM 连接——通过事件触发推理 2. 不等待工具结果——订阅 TOOL_RESULT 事件 3. 状态通过 Event Sourcing 方式恢复 """ def __init__(self, event_bus: EventBus, state_store): self.bus = event_bus self.state_store = state_store # 注册事件处理器 self._register_handlers() def _register_handlers(self): """注册所有事件处理器""" # LLM 响应到达时，处理推理结果 self.bus.subscribe(EventType.LLM_RESPONSE, self._on_llm_response) # 工具结果到达时，注入上下文并继续推理 self.bus.subscribe(EventType.TOOL_RESULT, self._on_tool_result) async def handle_user_message(self, session_id: str, user_input: str): """ 处理用户消息的入口 不直接调 LLM，而是发布事件： 1. AgentStart → 表示新一轮对话开始 2. LLMRequest → 触发 LLM 调用 """ turn_id = str(uuid.uuid4()) # 发布 AgentStart 事件 await self.bus.publish(AgentEvent( event_type=EventType.AGENT_START, session_id=session_id, turn_id=turn_id, payload={"user_input": user_input}, )) # 构建 Prompt（同步操作，很快） system_prompt = await self.state_store.get_system_prompt(session_id) history = await self.state_store.get_history(session_id) full_prompt = self._build_messages(system_prompt, history, user_input) # 发布 LLMRequest 事件，让 LLM 服务处理 await self.bus.publish(AgentEvent( event_type=EventType.LLM_REQUEST, session_id=session_id, turn_id=turn_id, payload={ "messages": full_prompt, "model": "gpt-4", "temperature": 0.3, }, # 告诉 LLM 服务：完成后把结果发到这个事件类型 reply_to=EventType.LLM_RESPONSE.value, correlation_id=turn_id, )) # 此处不阻塞！Agent 线程/协程已释放 async def _on_llm_response(self, event: AgentEvent): """处理 LLM 响应事件""" llm_output = event.payload.get("output", {}) # 检查是否有工具调用 tool_calls = llm_output.get("tool_calls", []) if tool_calls: # 有工具调用：发布工具执行事件 for tool_call in tool_calls: await self.bus.publish(AgentEvent( event_type=EventType.TOOL_INVOKE, session_id=event.session_id, turn_id=event.turn_id, payload={ "tool_name": tool_call["function"]["name"], "tool_args": json.loads(tool_call["function"]["arguments"]), }, reply_to=EventType.TOOL_RESULT.value, correlation_id=event.correlation_id, )) else: # 无工具调用：推理完成 final_response = llm_output.get("content", "") # 保存到状态存储 await self.state_store.save_turn( event.session_id, event.turn_id, final_response ) # 发布完成事件 await self.bus.publish(AgentEvent( event_type=EventType.AGENT_COMPLETE, session_id=event.session_id, turn_id=event.turn_id, payload={"response": final_response}, )) async def _on_tool_result(self, event: AgentEvent): """处理工具结果事件""" # 将工具结果注入上下文，再次调用 LLM session_id = event.session_id turn_id = event.turn_id history = await self.state_store.get_history(session_id) tool_results = event.payload.get("results", {}) # 构建带工具结果的 Prompt updated_messages = self._augment_with_tool_results( history, tool_results ) # 再次发布 LLM 请求 await self.bus.publish(AgentEvent( event_type=EventType.LLM_REQUEST, session_id=session_id, turn_id=turn_id, payload={ "messages": updated_messages, "model": "gpt-4", }, reply_to=EventType.LLM_RESPONSE.value, correlation_id=turn_id, )) def _build_messages(self, system: str, history: list, user_input: str) -> list: """构建 LLM 消息列表""" pass def _augment_with_tool_results(self, history: list, tool_results: dict) -> list: """将工具结果注入消息列表""" pass四、事件驱动模式的边界
缺点：
- 调试复杂度显著上升：同步模式可以直接打断点跟踪，事件驱动模式下执行流散布在多个事件处理器中，排查问题时需要追溯整个事件链。
- 最终一致性挑战：状态存储的写入和事件发布不是原子的。如果 AgentComplete 事件发布成功但状态存储写入失败，状态和事件不一致。需要引入 Outbox 模式。
- 事件风暴风险：一个 ReAct 循环中如果 Agent 频繁调用工具，会产生大量 TOOL_INVOKE→TOOL_RESULT 事件对。事件总线的吞吐需要匹配峰值。
禁用场景：
- 单轮问答 Agent（无工具调用）：事件驱动的 overhead 超过收益，同步模式更简单。
- 需要强事务性的场景（如金融交易 Agent）：事件最终一致性无法满足 ACID 要求。
五、总结
事件驱动 Agent 的核心转换，是把"阻塞等待 LLM/工具响应"替换为"发布事件 + 异步回调"。Event Bus（Redis Stream）解耦了 Agent 推理、LLM 调用和工具执行，Agent 线程只在处理事件的瞬间活跃。资源利用率从"每个会话持有一个 LLM 连接"变为"按需获取"，100 个并发会话可能只需要 10 个 LLM 连接。代价是执行流程从线性变成了事件图，排查和调试需要事件链追溯工具。
)
![[C++]内存管理：串顺序存储的内存回收](http://pic.xiahunao.cn/yaotu/[C++]内存管理：串顺序存储的内存回收)
