当前位置：首页 > news >正文

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

news 2026/7/16 18:58:01

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 }