# 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 |