Agent 事件驱动架构：用消息队列解耦 Agent 内部模块

一、"Agent 推理链里加一个日志步骤，推理慢了 40%"

Agent 的推理流程是同步串联的：分析意图 → 规划步骤 → 调用工具 → 记录日志 → 返回结果。"记录日志"步骤本身 50ms，但因它阻塞在推理链中，导致整个 Agent 的响应被延迟。更糟的是，日志写入失败会导致整个 Agent 请求失败。

Agent 内部不需要所有操作都是同步的。记录日志、更新统计数据、触发后续通知——这些操作不应该阻塞用户等待 Agent 回复。事件驱动架构把 Agent 内部的模块从"同步调用"变为"异步事件"，核心推理链只保留真正必须等待的步骤。

二、Agent 的事件驱动架构

事件驱动后的效果：核心推理链只保留必须同步等待的操作（工具调用、模型推理），其他操作（日志、监控、通知、分析）全部异步化。Agent 响应延迟降低 30-50%，而且旁路模块的故障不影响主流程。

三、Go 实现：Agent 内部事件总线

package eventdriven

import (
    "context"
    "encoding/json"
    "fmt"
    "sync"
    "time"
)

// ========== Agent 事件定义 ==========

// AgentEvent Agent 事件
type AgentEvent struct {
    ID        string          `json:"id"`
    Type      string          `json:"type"`       // task.started, tool.called, task.completed
    TaskID    string          `json:"task_id"`
    TraceID   string          `json:"trace_id"`
    Timestamp time.Time       `json:"timestamp"`
    Payload   json.RawMessage `json:"payload"`
    Metadata  map[string]string `json:"metadata"`
}

// 事件类型常量
const (
    EventTaskStarted   = "task.started"
    EventToolCalled    = "tool.called"
    EventReasoningStep = "reasoning.step"
    EventTaskCompleted = "task.completed"
    EventTaskFailed    = "task.failed"
)

// TaskStartedPayload 任务开始事件载荷
type TaskStartedPayload struct {
    UserID  string `json:"user_id"`
    Message string `json:"message"`
}

// ToolCalledPayload 工具调用事件载荷
type ToolCalledPayload struct {
    ToolName string                 `json:"tool_name"`
    Params   map[string]interface{} `json:"params"`
    Latency  time.Duration          `json:"latency_ms"`
}

// ========== 事件发布器 ==========

// EventPublisher 事件发布接口
type EventPublisher interface {
    Publish(ctx context.Context, event *AgentEvent) error
}

// AsyncEventPublisher 异步事件发布器
type AsyncEventPublisher struct {
    handlers  map[string][]EventHandler // 事件类型 → 处理器列表
    eventCh   chan *AgentEvent
    ctx       context.Context
    cancel    context.CancelFunc
    mu        sync.RWMutex
    bufferSize int
}

type EventHandler func(ctx context.Context, event *AgentEvent) error

func NewAsyncEventPublisher(bufferSize int) *AsyncEventPublisher {
    ctx, cancel := context.WithCancel(context.Background())
    
    p := &AsyncEventPublisher{
        handlers:   make(map[string][]EventHandler),
        eventCh:    make(chan *AgentEvent, bufferSize),
        ctx:        ctx,
        cancel:     cancel,
        bufferSize: bufferSize,
    }
    
    // 启动事件处理协程池
    for i := 0; i < 4; i++ {
        go p.processEvents(i)
    }
    
    return p
}

// Subscribe 订阅事件
func (p *AsyncEventPublisher) Subscribe(eventType string, handler EventHandler) {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    p.handlers[eventType] = append(p.handlers[eventType], handler)
}

// Publish 发布事件（非阻塞）
func (p *AsyncEventPublisher) Publish(ctx context.Context, event *AgentEvent) error {
    select {
    case p.eventCh <- event:
        return nil
    default:
        // 缓冲区满 → 丢弃事件但不阻塞 Agent 主流程
        return fmt.Errorf("事件缓冲区已满，事件被丢弃: %s", event.Type)
    }
}

// processEvents 事件处理循环
func (p *AsyncEventPublisher) processEvents(workerID int) {
    for {
        select {
        case <-p.ctx.Done():
            return
        case event := <-p.eventCh:
            p.dispatch(event)
        }
    }
}

// dispatch 分发事件给所有订阅者
func (p *AsyncEventPublisher) dispatch(event *AgentEvent) {
    p.mu.RLock()
    handlers := p.handlers[event.Type]
    p.mu.RUnlock()
    
    for _, handler := range handlers {
        // 每个 handler 独立执行，互不影响
        go func(h EventHandler) {
            ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
            defer cancel()
            
            if err := h(ctx, event); err != nil {
                // handler 失败不影响主流程，只记录日志
                fmt.Printf("[EventBus] handler 失败: event=%s, err=%v\n", event.Type, err)
            }
        }(handler)
    }
}

// Close 关闭事件发布器
func (p *AsyncEventPublisher) Close() {
    p.cancel()
}

// ========== Agent 集成 ==========

// EventDrivenAgent 事件驱动的 Agent
type EventDrivenAgent struct {
    eventBus    *AsyncEventPublisher
    modelClient ModelClient
    toolRegistry ToolRegistry
}

func NewEventDrivenAgent(bus *AsyncEventPublisher) *EventDrivenAgent {
    agent := &EventDrivenAgent{
        eventBus: bus,
    }
    
    // 注册事件处理器（在 Agent 启动时）
    bus.Subscribe(EventTaskStarted, agent.handleTaskStarted)
    bus.Subscribe(EventToolCalled, agent.handleToolCalled)
    bus.Subscribe(EventTaskCompleted, agent.handleTaskCompleted)
    
    return agent
}

// RunTask 执行 Agent 任务（核心推理链）
func (a *EventDrivenAgent) RunTask(ctx context.Context, taskID, userID, message string) (string, error) {
    // 1. 发布任务开始事件（不等待）
    startedPayload, _ := json.Marshal(TaskStartedPayload{
        UserID: userID, Message: message,
    })
    a.eventBus.Publish(ctx, &AgentEvent{
        ID:        fmt.Sprintf("evt-%d", time.Now().UnixNano()),
        Type:      EventTaskStarted,
        TaskID:    taskID,
        Timestamp: time.Now(),
        Payload:   startedPayload,
    })
    
    // 2. 核心推理（同步等待）
    plan, err := a.planSteps(ctx, message)
    if err != nil {
        a.publishFailed(ctx, taskID, err)
        return "", err
    }
    
    // 3. 执行工具调用
    for _, step := range plan.Steps {
        // 发布推理步骤事件
        a.publishReasoningStep(ctx, taskID, step.Description)
        
        // 调用工具（同步等待）
        start := time.Now()
        result, err := a.toolRegistry.Call(ctx, step.ToolName, step.Params)
        latency := time.Since(start)
        
        // 发布工具调用事件（不等待）
        toolPayload, _ := json.Marshal(ToolCalledPayload{
            ToolName: step.ToolName,
            Params:   step.Params,
            Latency:  latency,
        })
        a.eventBus.Publish(ctx, &AgentEvent{
            ID:        fmt.Sprintf("evt-%d", time.Now().UnixNano()),
            Type:      EventToolCalled,
            TaskID:    taskID,
            Timestamp: time.Now(),
            Payload:   toolPayload,
        })
        
        if err != nil {
            a.publishFailed(ctx, taskID, err)
            return "", err
        }
        _ = result
    }
    
    // 4. 生成最终回答
    answer := "最终回答"
    
    // 5. 发布任务完成事件（不等待）
    completedPayload, _ := json.Marshal(map[string]string{"answer": answer})
    a.eventBus.Publish(ctx, &AgentEvent{
        ID:        fmt.Sprintf("evt-%d", time.Now().UnixNano()),
        Type:      EventTaskCompleted,
        TaskID:    taskID,
        Timestamp: time.Now(),
        Payload:   completedPayload,
    })
    
    return answer, nil
}

// ========== 事件处理器（旁路模块） ==========

func (a *EventDrivenAgent) handleTaskStarted(ctx context.Context, event *AgentEvent) error {
    var payload TaskStartedPayload
    json.Unmarshal(event.Payload, &payload)
    
    // 异步：更新用户活跃度统计
    // 异步：记录请求日志
    fmt.Printf("[Handler] 任务开始: task=%s user=%s\n", event.TaskID, payload.UserID)
    return nil
}

func (a *EventDrivenAgent) handleToolCalled(ctx context.Context, event *AgentEvent) error {
    var payload ToolCalledPayload
    json.Unmarshal(event.Payload, &payload)
    
    // 异步：记录工具调用频率
    // 异步：更新工具调用延迟指标
    fmt.Printf("[Handler] 工具调用: %s 耗时 %v\n", payload.ToolName, payload.Latency)
    return nil
}

func (a *EventDrivenAgent) handleTaskCompleted(ctx context.Context, event *AgentEvent) error {
    // 异步：发送通知
    // 异步：更新分析数据
    fmt.Printf("[Handler] 任务完成: %s\n", event.TaskID)
    return nil
}

func (a *EventDrivenAgent) publishFailed(ctx context.Context, taskID string, err error) {
    failedPayload, _ := json.Marshal(map[string]string{"error": err.Error()})
    a.eventBus.Publish(ctx, &AgentEvent{
        ID:        fmt.Sprintf("evt-%d", time.Now().UnixNano()),
        Type:      EventTaskFailed,
        TaskID:    taskID,
        Timestamp: time.Now(),
        Payload:   failedPayload,
    })
}

func (a *EventDrivenAgent) publishReasoningStep(ctx context.Context, taskID, desc string) {
    stepPayload, _ := json.Marshal(map[string]string{"description": desc})
    a.eventBus.Publish(ctx, &AgentEvent{
        ID:        fmt.Sprintf("evt-%d", time.Now().UnixNano()),
        Type:      EventReasoningStep,
        TaskID:    taskID,
        Timestamp: time.Now(),
        Payload:   stepPayload,
    })
}

// ========== 类型定义 ==========

type Plan struct {
    Steps []PlanStep
}

type PlanStep struct {
    Description string
    ToolName    string
    Params      map[string]interface{}
}

type ModelClient struct{}
type ToolRegistry struct{}

func (tr *ToolRegistry) Call(ctx context.Context, name string, params map[string]interface{}) (interface{}, error) {
    return nil, nil
}

func (a *EventDrivenAgent) planSteps(ctx context.Context, message string) (*Plan, error) {
    return &Plan{}, nil
}

四、事件驱动架构的边界与风险

事件顺序性。 异步事件不保证处理顺序。如果"任务完成"事件先于"工具调用"事件被处理，分析数据可能出错。需要依赖事件时间戳或序列号来保证分析的正确性。

缓冲区满时的丢弃策略。 事件通道缓冲区满了，是丢弃、阻塞还是告警？推荐丢弃 + 告警——Agent 主流程的稳定性优先于事件记录的完整性。

测试复杂度增加。 同步调用可以线性追踪，异步事件的处理发生在"别的地方"和"别的时间"。单元测试需要能捕获和断言发布的事件。

事件 Schema 的版本兼容。 事件结构会随版本演化。新增字段需要向后兼容（老消费者忽略未知字段），删除字段需要渐进式迁移（先标记废弃再删除）。

五、总结

Agent 事件驱动架构的核心原则：核心推理链只保留必须同步等待的操作，其他旁路操作全部异步化。事件总线 + 发布订阅模式实现模块解耦，每个旁路模块独立订阅、独立失败。实施路径：先把最明显的可异步操作（日志、监控指标）从主流程中拆出来，确认稳定后再逐步迁移其他模块。