CamTalk/docs/12-Eino重构实施记录.md

CamTalk Eino 重构实施记录

创建日期：2026-06-19 状态：已完成

1. 重构背景

CamTalk 原 AI 编排层（internal/orchestrator/pipeline.go）使用手写 goroutine + WaitGroup + channel 实现 STT → LLM → TTS 流式管道，存在以下问题：

1. 编排逻辑硬编码：流程写死在 ProcessQuery() 中，扩展需重写 goroutine 调度
2. 并发控制粗糙：手动 go func() + sync.WaitGroup，缺乏结构化流式传递
3. 无回调/AOP 机制：日志、指标、追踪散落各处
4. 配置耦合：模型名、TTS 参数硬编码在 Pipeline 结构体
5. 错误处理不一致：TTS 错误被静默吞掉，缺乏统一模式

重构目标：使用 Eino Graph 替换手写 Pipeline，实现声明式编排、统一回调、按请求动态配置，保持 WebSocket 协议和 REST API 不变。

2. 整体架构变更

2.1 重构前

WS Handler → Orchestrator.Pipeline.ProcessQuery()
              ├→ goroutine: STT.Recognize()
              ├→ goroutine: LLM.ChatStream() ──→ chan chunk ──→ Sender
              └→ goroutine: Splitter → TTS.SynthesizeStream() ──→ chan audio ──→ Sender
              WaitGroup.Wait()
              Sender.SendLLMDone()

2.2 重构后

WS Handler → EinoOrchestrator.ProcessQuery()
              ├→ Graph.Stream(ctx, input)
              │   ├→ STT Lambda ─→ History Lambda ─→ ChatModel ─→ Splitter ─→ TTS ─→ Done
              │   │                  (State 写入)    (Callback    (Transform)  (Invoke)  (Invoke)
              │   │                                    流式推送)
              │   └→ 消费 StreamReader（触发整条链路惰性执行）
              └→ 追加助手消息到历史

2.3 关键设计决策

决策	选择	理由
Graph 调用模式	Stream	ChatModel 需要真正的 token 级流式输出
LLM 组件	eino-ext ChatModel	原生 Eino 组件，直接对接 DashScope
消息推送	Callback（LLM）+ Sender（其他）	LLM token 流式推送需要 Callback
值类型 vs 指针	值类型统一	避免框架类型转换不匹配
历史追加	适配器负责	Done 节点只负责发送 llm_done

3. 分阶段实施

Phase 1：基础设施（提交 fd5c771）

目标：引入 Eino 依赖，创建基础类型和 Callback。

任务清单：

任务	文件	说明
引入 Eino 依赖	go.mod	eino v0.9.9 + eino-ext/components/model/openai v0.1.13
数据类型定义	eino/types.go	PipelineInput、PipelineOutput、STTOutput、TokenUsage
State 定义	eino/state.go	PipelineState 含 sync.Mutex 并发保护
消息推送 Callback	eino/callback.go	BuildCallbackHandler() 使用 callbacks.NewHandlerHelper()

关键实现：

• PipelineState 使用 strings.Builder + sync.Mutex 累积 LLM 完整回复
• Callback 通过 ModelCallbackHandler.OnEndWithStreamOutput 逐 token 推送 llm_chunk
• Sender/RequestID/PipelineState 通过 context.WithValue 注入

验证：go build ./cmd/server ✓

---

Phase 2：节点实现（提交 fd5c771）

目标：实现 Graph 中的 5 个 Lambda 节点。

任务清单：

任务	文件	Lambda 类型	输入 → 输出
STT Lambda	eino/nodes_stt.go	InvokableLambda	PipelineInput → STTOutput
历史组装 Lambda	eino/nodes_history.go	InvokableLambda	STTOutput → []*schema.Message
句子分割 Lambda	eino/nodes_splitter.go	TransformableLambda	StreamReader[string] → StreamReader[[]string]
TTS Lambda	eino/nodes_tts.go	InvokableLambda	[]string → struct{}
Done Lambda	eino/nodes_done.go	InvokableLambda	struct{} → PipelineOutput

关键实现：

• STT 节点将输入元数据写入 State，供下游节点读取
• History 节点从 State 读取 SessionID/Scenario/ImageData，构建系统提示词 + 多模态消息
• Splitter 使用 TransformableLambda 按句子分隔符切分，逐句输出给 TTS
• TTS 节点调用 ttsService.SynthesizeStream()，逐 chunk 推送 tts_audio
• Done 节点从 State 读取完整回复，发送 llm_done
• 所有 Lambda 使用值类型（非指针），返回 *compose.Lambda

验证：go build ./internal/eino/... ✓

---

Phase 3：Graph 构建与适配器（提交 4b731b5）

目标：构建 Graph、实现适配器、切换 main.go。

任务清单：

任务	文件	说明
Graph 构建	eino/graph.go	NewPipelineGraph() 组装 6 个节点 + 边 + 编译
适配器	eino/adapter.go	EinoOrchestrator 实现 orchestrator.Orchestrator 接口
main.go 切换	cmd/server/main.go	移除旧 LLM + orchestrator，替换为 Eino

Graph 拓扑：

START → STT → History → ChatModel → Splitter → TTS → Done → END

适配器职责：

1. 解码 base64 音频/图片
2. 获取会话配置
3. 注入 Sender/RequestID/SessionID/StartTime/State 到 context
4. 追加用户消息到历史
5. 调用 graph.Stream(ctx, input, callbacks) 触发惰性执行
6. 消费 StreamReader[PipelineOutput]
7. 追加助手消息到历史

关键实现：

• eino-ext ChatModel 配置：BaseURL 对接 DashScope，Timeout 控制请求超时
• Callback 在运行时通过 compose.WithCallbacks() 传入，不在编译时注册
• 元数据（SessionID/Scenario 等）通过 State 跨节点传递，不通过 Graph 边传递

变更文件：

• 修改 state.go：新增 SessionID/RequestID/ImageData 等字段
• 修改 nodes_stt.go：写入元数据到 State
• 修改 nodes_history.go：从 State 读取元数据（移除 HistoryInput 依赖）
• 修改 nodes_done.go：移除历史追加（由适配器负责）

验证：go build ./cmd/server ✓，go vet ./... ✓

---

Phase 4：清理与测试（提交 4ffd845）

目标：删除旧代码，编写单元测试。

删除的文件：

文件	说明
orchestrator/pipeline.go	旧 STT→LLM→TTS 手写 goroutine 管道（-547 行）
orchestrator/splitter.go	旧句子切分器（-114 行）
orchestrator/pipeline_test.go	旧 Pipeline 测试（-309 行）
ai/llm/openai.go	旧 LLM OpenAI 实现（-548 行）
ai/llm/openai_test.go	旧 LLM 测试（-143 行）

保留的文件：

文件	保留原因
orchestrator/orchestrator.go	Orchestrator 接口（ws/handler 依赖）
orchestrator/sender.go	Sender 接口（eino/callback 依赖）
ai/llm/llm.go	Request/Chunk/TokenUsage 类型定义
ai/llm/prompt.go	BuildSystemPrompt（eino/nodes_history 依赖）
ai/llm/scenarios.go	GetScenarioPrompt（eino/nodes_history 依赖）

新增测试：eino/graph_test.go（13 个测试）

测试	覆盖内容
TestDetectImageMimeType	JPEG/PNG/GIF/WebP/未知格式检测
TestBuildPipelineInput	文本输入构建
TestBuildPipelineInput_WithAudioData	音频+图片输入构建
TestPipelineState_AppendAndGet	State 文本追加和读取
TestPipelineState_ConcurrentAccess	State 并发安全（100 goroutine）
TestContextInjection	Sender/RequestID/State 注入和提取
TestLatencyFromCtx	延迟计算
TestEinoOrchestrator_ImplementsInterface	接口实现检查
TestNew*Lambda_ReturnsNonNil	5 个 Lambda 构造函数非空检查

验证：go build ./... ✓，go vet ./... ✓，go test ./... ✓

4. 代码变更统计

阶段	提交	新增	删除	净变化
Phase 1 + 2	fd5c771	+946	-24	+922
Phase 3	4b731b5	+395	-98	+297
Phase 4	4ffd845	+235	-1661	-1426
合计		+1576	-1783	-207

重构后代码量净减少 207 行，同时获得了更好的可维护性、可测试性和可扩展性。

5. 遗留事项

事项	优先级	说明
eino-ext ChatModel DashScope 兼容性端到端验证	高	需要真实 API Key 验证流式输出和多模态
LLM 超时控制	中	eino-ext ChatModel 的 Timeout 配置需验证
TTS 流式优化	中	当前 TTS 是 InvokableLambda，可改为 StreamableLambda
ReAct Agent 扩展	低	基于 Graph Branch 实现工具调用循环
Model Router	低	按场景/成本路由不同 LLM
指标监控	低	通过 Callback 接入 Prometheus