1. 从轮询到事件驱动 传统 Agent 采用轮询模式：定期检查是否有新任务或状态变化。这种模式简单但低效——要么延迟高（轮询间隔长），要么浪费资源（轮询间隔短）。
事件驱动架构（EDA）将 Agent 从"主动检查"转变为"被动响应"：当有意义的事件发生时，系统主动通知 Agent 处理。核心优势：
低延迟：事件发生即触发，无需等待轮询周期 低资源消耗：无事件时 Agent 处于休眠状态 天然解耦：事件生产者和消费者完全分离 弹性扩展：事件积压时可动态增加消费者 2. 事件驱动 Agent 整体架构 ┌──────────┐ ┌──────────┐ ┌──────────┐ │ GitHub │ │ Slack │ │ Sensor │ 事件源 │ Webhook │ │ Events │ │ Data │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │ │ ▼ ▼ ▼ ┌────────────────────────────────────────┐ │ Event Ingestion Layer │ │ (API Gateway / Message Queue) │ └───────────────────┬────────────────────┘ │ ▼ ┌────────────────────────────────────────┐ │ Event Router / Filter │ └───────────────────┬────────────────────┘ │ ┌───────────┼───────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Agent A │ │ Agent B │ │ Agent C │ 事件消费者 │(Coder) │ │(Reviewer)│ │(Notifier)│ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────────────────────────────────┐ │ Action Executor │ │ (API calls, DB writes, etc.) │ └─────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────┐ │ Event Publisher (新事件) │ └─────────────────────────────────────┘ 3. 事件模型设计 3.1 事件 Schema from pydantic import BaseModel, Field from datetime import datetime from enum import Enum from typing import Any, Optional import uuid class EventSeverity(str, Enum): INFO = "info" WARNING = "warning" ERROR = "error" CRITICAL = "critical" class AgentEvent(BaseModel): """统一事件格式 - CloudEvents 兼容""" event_id: str = Field(default_factory=lambda: str(uuid.uuid4())) event_type: str # "github.pr.opened", "agent.task.completed" source: str # 事件来源标识 subject: str = "" # 事件主题（如 PR #123） data: dict[str, Any] # 事件负载 time: datetime = Field(default_factory=datetime.now) severity: EventSeverity = EventSeverity.INFO trace_id: str = "" # 分布式追踪 ID correlation_id: str = "" # 关联 ID（同一会话） # 事件版本控制 spec_version: str = "1.0" data_content_type: str = "application/json" class TaskCompletedEvent(AgentEvent): event_type: str = "agent.task.completed" data: dict[str, Any] = Field(description="包含 task_id, result, duration 等") class ErrorEvent(AgentEvent): event_type: str = "agent.error" severity: EventSeverity = EventSeverity.ERROR 3.2 事件注册表 class EventRegistry: """事件类型注册表 - 管理 Schema 和路由规则""" def __init__(self): self._schemas: dict[str, type[BaseModel]] = {} self._handlers: dict[str, list[callable]] = {} def register(self, event_type: str, schema: type[BaseModel]): self._schemas[event_type] = schema def subscribe(self, event_type: str, handler: callable): self._handlers.setdefault(event_type, []).append(handler) def get_handlers(self, event_type: str) -> list[callable]: # 支持通配符匹配: "github.pr.*" 匹配 "github.pr.opened" handlers = self._handlers.get(event_type, []) for pattern, h_list in self._handlers.items(): if "*" in pattern: prefix = pattern.replace("*", "") if event_type.startswith(prefix): handlers.extend(h_list) return handlers def validate(self, event: AgentEvent) -> bool: schema = self._schemas.get(event.event_type) if schema: schema(**event.model_dump()) return True 4. 事件总线实现 4.1 内存事件总线（单机） import asyncio from collections import defaultdict from concurrent.futures import ThreadPoolExecutor class InMemoryEventBus: def __init__(self, max_workers: int = 10): self._subscribers: dict[str, list] = defaultdict(list) self._executor = ThreadPoolExecutor(max_workers=max_workers) self._dead_letter_queue: list[tuple[AgentEvent, Exception]] = [] self._middleware: list[callable] = [] def use(self, middleware: callable): """注册中间件：日志、认证、限流等""" self._middleware.append(middleware) async def subscribe(self, event_type: str, handler: callable): self._subscribers[event_type].append(handler) async def publish(self, event: AgentEvent): # 执行中间件链 for mw in self._middleware: event = await mw(event) if event is None: return # 被中间件拦截 handlers = self._subscribers.get(event.event_type, []) # 匹配通配符订阅 for pattern, h_list in self._subscribers.items(): if "*" in pattern: prefix = pattern.replace("*", "") if event.event_type.startswith(prefix): handlers.extend(h_list) # 并行执行所有处理器 tasks = [self._safe_execute(h, event) for h in handlers] await asyncio.gather(*tasks, return_exceptions=True) async def _safe_execute(self, handler: callable, event: AgentEvent): try: if asyncio.iscoroutinefunction(handler): await handler(event) else: await asyncio.get_event_loop().run_in_executor( self._executor, handler, event ) except Exception as e: self._dead_letter_queue.append((event, e)) # 发布错误事件 error_event = ErrorEvent( source="event_bus", data={"original_event": event.event_id, "error": str(e)}, trace_id=event.trace_id, ) await self.publish(error_event) 4.2 Kafka 事件总线（分布式） from aiokafka import AIOKafkaProducer, AIOKafkaConsumer import json class KafkaEventBus: def __init__(self, bootstrap_servers: str = "localhost:9092"): self.bootstrap_servers = bootstrap_servers self._producer: AIOKafkaProducer = None self._consumers: list[AIOKafkaConsumer] = [] self._handlers: dict[str, callable] = {} async def start(self): self._producer = AIOKafkaProducer( bootstrap_servers=self.bootstrap_servers, value_serializer=lambda v: json.dumps(v.model_dump(mode="json")).encode(), key_serializer=lambda k: k.encode() if k else None, acks="all", # 等待所有副本确认 enable_idempotence=True, # 幂等生产者 compression_type="lz4", # 压缩 max_in_flight_requests_per_connection=5, ) await self._producer.start() async def publish(self, event: AgentEvent, topic: str = None): topic = topic or event.event_type.split(".")[0] # 按事件类型分 Topic partition_key = event.correlation_id or event.source await self._producer.send_and_wait( topic, event, key=partition_key ) async def subscribe(self, event_type: str, handler: callable, group_id: str = "agent-group"): topic = event_type.split(".")[0] consumer = AIOKafkaConsumer( topic, bootstrap_servers=self.bootstrap_servers, group_id=group_id, value_deserializer=lambda v: AgentEvent(**json.loads(v.decode())), auto_offset_reset="latest", enable_auto_commit=False, ) await consumer.start() self._consumers.append(consumer) self._handlers[event_type] = handler async for msg in consumer: event = msg.value if event.event_type == event_type or event_type.endswith("*"): try: await handler(event) await consumer.commit() except Exception as e: # 重试或进入死信队列 await self._handle_error(event, e) async def stop(self): await self._producer.stop() for c in self._consumers: await c.stop() 5. Agent 状态机 事件驱动 Agent 的核心是一个状态机：不同事件触发不同状态转换。
...
