Agent 事件驱动架构：用消息驱动替代轮询等待

Agent 事件驱动架构用消息驱动替代轮询等待一、Agent 推理完一次就阻塞等待CPU 时间全浪费在空转上最早的 Agent 实现是这样while True: response llm.call(); if response.has_tool_call: result execute_tool(); continue; else: break。LLM 调用期间的 2-5 秒主线程完全阻塞等待。如果 Agent 在一次对话中多次调用 LLMReAct 模式下 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_factorylambda: uuid.uuid4().hex[:12]) event_type: EventType session_id: str turn_id: str timestamp: float field(default_factorylambda: datetime.now().timestamp()) payload: Dict[str, Any] field(default_factorydict) # 回调地址事件处理器收到事件后把结果发给这个地址 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 # 事件处理器注册表 # keyEventType, valuelist 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, encodingutf-8, decode_responsesTrue ) async def publish(self, event: AgentEvent): 发布事件到 Stream stream_key fagent: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 , }, maxlen10000, # 每个 Stream 最多保留 10000 条 ) async def subscribe(self, event_type: EventType, handler: Callable): 订阅事件类型并开始消费 self._handlers[event_type].append(handler) # 创建消费者组幂等 stream_key fagent:events:{event_type.value} try: await self.redis.xgroup_create( stream_key, self._consumer_group, id0, mkstreamTrue ) 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 fagent:events:{event_type.value} while True: try: # 批量读取一次处理多条事件 messages await self.redis.xreadgroup( groupnameself._consumer_group, consumernameconsumer_name, streams{stream_key: }, count10, block1000, # 阻塞 1s等待新消息 ) if not messages: continue for stream, msgs in messages: for msg_id, msg_data in msgs: # 重建事件对象 event AgentEvent( event_idmsg_data.get(event_id, ), event_typeEventType(msg_data.get(event_type, )), session_idmsg_data.get(session_id, ), turn_idmsg_data.get(turn_id, ), timestampfloat(msg_data.get(timestamp, 0)), payloadjson.loads(msg_data.get(payload, {})), reply_tomsg_data.get(reply_to, None), correlation_idmsg_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_typeEventType.AGENT_ERROR, session_idevent.session_id, turn_idevent.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_typeEventType.AGENT_START, session_idsession_id, turn_idturn_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_typeEventType.LLM_REQUEST, session_idsession_id, turn_idturn_id, payload{ messages: full_prompt, model: gpt-4, temperature: 0.3, }, # 告诉 LLM 服务完成后把结果发到这个事件类型 reply_toEventType.LLM_RESPONSE.value, correlation_idturn_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_typeEventType.TOOL_INVOKE, session_idevent.session_id, turn_idevent.turn_id, payload{ tool_name: tool_call[function][name], tool_args: json.loads(tool_call[function][arguments]), }, reply_toEventType.TOOL_RESULT.value, correlation_idevent.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_typeEventType.AGENT_COMPLETE, session_idevent.session_id, turn_idevent.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_typeEventType.LLM_REQUEST, session_idsession_id, turn_idturn_id, payload{ messages: updated_messages, model: gpt-4, }, reply_toEventType.LLM_RESPONSE.value, correlation_idturn_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 BusRedis Stream解耦了 Agent 推理、LLM 调用和工具执行Agent 线程只在处理事件的瞬间活跃。资源利用率从每个会话持有一个 LLM 连接变为按需获取100 个并发会话可能只需要 10 个 LLM 连接。代价是执行流程从线性变成了事件图排查和调试需要事件链追溯工具。