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

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