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 的核心是一个状态机：不同事件触发不同状态转换。

5.1 状态机实现

from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Callable

class AgentState(str, Enum):
    IDLE = "idle"
    PLANNING = "planning"
    EXECUTING = "executing"
    WAITING_INPUT = "waiting_input"
    REVIEWING = "reviewing"
    ERROR = "error"
    DONE = "done"

@dataclass
class StateTransition:
    from_state: AgentState
    to_state: AgentState
    trigger_event: str           # 触发事件类型
    guard: Callable = None       # 守卫条件
    action: Callable = None      # 转换时执行的动作

class AgentStateMachine:
    def __init__(self, agent_id: str, initial_state: AgentState = AgentState.IDLE):
        self.agent_id = agent_id
        self.state = initial_state
        self.transitions: list[StateTransition] = []
        self.history: list[tuple[AgentState, AgentState, str, datetime]] = []
        self._state_handlers: dict[AgentState, Callable] = {}

    def add_transition(self, transition: StateTransition):
        self.transitions.append(transition)

    def on_state(self, state: AgentState, handler: Callable):
        """注册状态进入时的回调"""
        self._state_handlers[state] = handler

    async def handle_event(self, event: AgentEvent) -> bool:
        """处理事件，尝试状态转换"""
        for t in self.transitions:
            if t.from_state == self.state and t.trigger_event == event.event_type:
                if t.guard and not t.guard(event):
                    continue
                old_state = self.state
                self.state = t.to_state
                self.history.append((old_state, self.state, event.event_type, datetime.now()))

                # 执行转换动作
                if t.action:
                    await t.action(event)

                # 执行新状态的处理函数
                handler = self._state_handlers.get(self.state)
                if handler:
                    await handler(event)

                return True
        return False

class CodeAgentStateMachine(AgentStateMachine):
    def __init__(self, agent_id: str):
        super().__init__(agent_id)
        self.setup_transitions()

    def setup_transitions(self):
        self.add_transition(StateTransition(
            AgentState.IDLE, AgentState.PLANNING, "task.assigned"
        ))
        self.add_transition(StateTransition(
            AgentState.PLANNING, AgentState.EXECUTING, "plan.ready",
            action=self._start_execution
        ))
        self.add_transition(StateTransition(
            AgentState.EXECUTING, AgentState.REVIEWING, "code.committed",
            action=self._trigger_review
        ))
        self.add_transition(StateTransition(
            AgentState.REVIEWING, AgentState.DONE, "review.passed"
        ))
        self.add_transition(StateTransition(
            AgentState.REVIEWING, AgentState.EXECUTING, "review.changes_requested",
            action=self._apply_feedback
        ))
        self.add_transition(StateTransition(
            AgentState.EXECUTING, AgentState.ERROR, "execution.failed",
            action=self._handle_error
        ))
        self.add_transition(StateTransition(
            AgentState.ERROR, AgentState.EXECUTING, "retry.triggered"
        ))

    async def _start_execution(self, event: AgentEvent):
        print(f"Agent {self.agent_id} 开始执行任务: {event.data.get('task_id')}")

    async def _trigger_review(self, event: AgentEvent):
        print(f"Agent {self.agent_id} 提交代码审查: {event.data.get('commit_sha')}")

    async def _apply_feedback(self, event: AgentEvent):
        print(f"Agent {self.agent_id} 根据反馈修改: {event.data.get('comments')}")

    async def _handle_error(self, event: AgentEvent):
        print(f"Agent {self.agent_id} 执行失败: {event.data.get('error')}")

6. Webhook 集成层

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

class WebhookIntegration:
    def __init__(self, app: FastAPI, event_bus: InMemoryEventBus):
        self.app = app
        self.bus = event_bus
        self._secrets: dict[str, str] = {}  # source -> secret

    def register_webhook(self, source: str, path: str, secret: str = None):
        self._secrets[source] = secret
        @self.app.post(path)
        async def handle_webhook(request: Request):
            body = await request.body()
            # 验证签名
            if secret:
                signature = request.headers.get("X-Hub-Signature-256", "")
                if not self._verify_signature(body, secret, signature):
                    raise HTTPException(status_code=401, detail="Invalid signature")

            payload = await request.json()
            event_type = self._map_to_event_type(source, payload, request.headers)

            event = AgentEvent(
                event_type=event_type,
                source=source,
                data=payload,
                trace_id=request.headers.get("X-Trace-Id", str(uuid.uuid4())),
            )
            await self.bus.publish(event)
            return {"status": "accepted"}

    def _verify_signature(self, body: bytes, secret: str, signature: str) -> bool:
        expected = "sha256=" + hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        return hmac.compare_digest(expected, signature)

    def _map_to_event_type(self, source: str, payload: dict, headers) -> str:
        if source == "github":
            event_name = headers.get("X-GitHub-Event", "push")
            action = payload.get("action", "")
            return f"github.{event_name}.{action}" if action else f"github.{event_name}"
        elif source == "slack":
            return f"slack.{payload.get('event', {}).get('type', 'unknown')}"
        elif source == "custom":
            return payload.get("event_type", "custom.event")
        return f"{source}.event"

7. 事件溯源与回放

class EventStore:
    """事件存储 - 支持回放和审计"""
    def __init__(self, storage_backend="sqlite"):
        self.backend = storage_backend
        self._events: list[AgentEvent] = []  # 生产环境用数据库

    async def append(self, event: AgentEvent):
        self._events.append(event)

    async def query(self, source: str = None, event_type: str = None,
                    since: datetime = None, limit: int = 100) -> list[AgentEvent]:
        results = self._events
        if source:
            results = [e for e in results if e.source == source]
        if event_type:
            results = [e for e in results if e.event_type == event_type]
        if since:
            results = [e for e in results if e.time > since]
        return results[-limit:]

    async def replay(self, agent: AgentStateMachine, since: datetime = None):
        """从指定时间点重放事件，重建 Agent 状态"""
        events = await self.query(since=since)
        for event in events:
            await agent.handle_event(event)

    async def snapshot(self, agent: AgentStateMachine) -> dict:
        """定期快照，减少回放量"""
        return {
            "agent_id": agent.agent_id,
            "state": agent.state,
            "history_length": len(agent.history),
            "timestamp": datetime.now().isoformat()
        }

8. 背压与流控

class BackpressureHandler:
    """背压处理：当消费者跟不上生产速度时"""
    def __init__(self, max_queue_size: int = 1000, strategy: str = "drop_oldest"):
        self.max_queue_size = max_queue_size
        self.strategy = strategy
        self.queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size)
        self.dropped_count = 0
        self.total_processed = 0

    async def submit(self, event: AgentEvent):
        if self.queue.full():
            if self.strategy == "drop_oldest":
                self.queue.get_nowait()
                self.dropped_count += 1
            elif self.strategy == "drop_newest":
                self.dropped_count += 1
                return False
            elif self.strategy == "block":
                await self.queue.put(event)
                return True
        await self.queue.put(event)
        return True

    async def consume(self, handler: callable):
        while True:
            event = await self.queue.get()
            try:
                await handler(event)
                self.total_processed += 1
            except Exception:
                pass  # 错误已在总线层处理
            self.queue.task_done()

    def health(self) -> dict:
        return {
            "queue_size": self.queue.qsize(),
            "max_size": self.max_queue_size,
            "utilization": self.queue.qsize() / self.max_queue_size,
            "dropped": self.dropped_count,
            "processed": self.total_processed,
        }

9. 模式对比

维度轮询模式事件驱动
延迟取决于轮询间隔毫秒级
资源消耗持续高空闲时近零
扩展性线性弹性
复杂度低中高
调试难度简单需分布式追踪
可靠性单点重试事件持久化+回放

10. 总结

事件驱动 Agent 架构是构建实时响应系统的核心模式。设计要点：

  1. 统一事件模型：CloudEvents 兼容格式，便于跨系统互通
  2. 事件总线选型：单机用内存队列，分布式用 Kafka/NATS
  3. 状态机驱动：每个 Agent 是一个有限状态机，事件触发转换
  4. Webhook 集成：统一入口 + 签名验证 + 事件类型映射
  5. 背压控制：防止事件洪峰压垮系统
  6. 事件溯源：持久化所有事件，支持回放和审计

推荐技术栈：FastAPI（Webhook 接入）+ Kafka/NATS（事件总线）+ Redis（状态缓存）+ OpenTelemetry（分布式追踪）+ SQLite/PostgreSQL（事件存储）。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。