Ch23: EventManager 与 Plugin——源码中的解耦神经中枢

Part 8: WeKnora 源码走读 前置：Ch22 请求旅程 后续：Ch24 三路检索实现

结论先行：WeKnora 用 EventManager + Plugin 把「解析、分块、检索、融合」拆成可插拔工位。读源码时先找 Register( 和 Emit(，再顺订阅关系画 DAG——比从 main 逐行读快 10 倍。

一、Plugin 接口（读代码的锚点）

type Plugin interface {
    Name() string
    SubscribedEvents() []EventType
    Handle(ctx context.Context, event Event) (Event, error)
    Priority() int
}

二、事件类型（摄入 vs 检索）

摄入链：

RawDocumentReceived → DocumentParsed → ChunksCreated → ChunksIndexed

检索链：

QueryReceived → (BM25Done, VectorDone, GraphDone) → AllRetrievalsDone
              → FusionCompleted → RerankCompleted

维护链：

ScheduledLint → StaleDetected / ConflictDetected / GapDetected

读 eventmanager 包中的 EventType 常量定义，与上表对照——不一致处以源码为准。

三、并发：QueryReceived 为何快

Ch06 讲过：Emit(QueryReceived) 后 BM25 / Vector / Graph 用 goroutine 并行。

func (em *EventManager) dispatch(event Event) {
    plugins := em.subscribers[event.Type]
    var wg sync.WaitGroup
    for _, p := range plugins {
        wg.Add(1)
        go func(pl Plugin) {
            defer wg.Done()
            out, err := pl.Handle(context.Background(), event)
            if err != nil { em.emitError(pl, err); return }
            if out != nil { em.Emit(out) }
        }(p)
    }
    wg.Wait()
}

大师级理解：延迟 ≈ max(三路) + 融合 + Rerank，不是三路相加。改一路超时配置会影响整体 P99。

四、如何新增 Plugin（竞赛扩展点）

场景：为每次检索自动打 RAGAS 分（Ch20）。

1. 实现 EvalPlugin，订阅 RerankCompleted
2. 在 Handle 里异步写 eval 表（勿阻塞主路径）
3. 启动时 eventManager.Register(RerankCompleted, evalPlugin)
4. 加 feature flag 默认关

不要：在 ChatService 里硬编码 50 行 eval——破坏解耦。

五、读源码步骤（45 分钟）

rg "type EventManager" -g '*.go'
rg "Register\(" -g '*event*'
rg "QueryReceived" -g '*.go'

1. 打开 EventManager 结构体字段
2. 找 Register 调用栈（通常在 wire/init/main）
3. 列出订阅 QueryReceived 的 Plugin 名称
4. 选一个 Plugin（如 Vector）读完整 Handle

六、与 Dify 工作流对比

见 Ch16 Dify。

章末 · 大师自检

• 写出摄入链 4 个事件名
• 解释 goroutine 并行对延迟的意义
• 说明新增 EvalPlugin 应订阅哪个事件
• 在仓库定位 Register 至少一处

延伸阅读

• Ch06 § EventManager
• 精读 WeKnora § EventManager