docs: 新增 Eino 框架技术文档和重构实施记录

- docs/11-Eino框架技术文档.md: 框架简介、技术选型对比、核心概念（Lambda/Graph/ChatModel/StreamReader/Callback/State）、CamTalk Graph 设计、目录结构、注意事项 - docs/12-Eino重构实施记录.md: 重构背景、架构变更、四阶段实施详情、代码统计、遗留事项 Co-Authored-By: Claude <noreply@anthropic.com>
2026-06-19 22:09:07 +08:00
parent 4ffd84510e
commit 9576884619
2 changed files with 450 additions and 0 deletions
--- a/docs/11-Eino框架技术文档.md
+++ b/docs/11-Eino框架技术文档.md
@@ -0,0 +1,246 @@
+# CamTalk Eino 框架技术文档
+
+> 创建日期：2026-06-19
+> 状态：已实施
+
+## 1. 框架简介
+
+[CloudWeGo Eino](https://github.com/cloudwego/eino) 是字节跳动 CloudWeGo 团队开源的 AI 应用开发框架，提供基于图（Graph）的编排能力、组件抽象和流式处理支持。
+
+CamTalk 使用 Eino 替代原有的手写 goroutine 管道，实现 STT → LLM → TTS 的声明式编排。
+
+## 2. 技术选型
+
+### 2.1 为什么选 Eino
+
+| 维度 | 手写 goroutine（旧方案） | Eino Graph（新方案） |
+|------|------------------------|---------------------|
+| 编排方式 | 手动 `go func()` + `sync.WaitGroup` | 声明式 DAG，类型安全 |
+| 流式处理 | 自定义 `chan` 传递 | `StreamReader` + `Pipe`，自动转换 |
+| 错误处理 | 各节点独立处理，不一致 | Graph 级别统一错误传播 |
+| 回调/AOP | 日志散落各处 | `callbacks.Handler` 统一注入 |
+| 配置灵活性 | Pipeline 创建时固定 | 每请求 `Option` 动态注入 |
+| 可测试性 | 需启动 goroutine | `Graph.Invoke()` 直接测试 |
+| 扩展性 | 修改 Pipeline 代码 | 添加节点 + 边，无侵入 |
+| 并发安全 | 手动 `sync` | State 自动加锁 |
+
+### 2.2 Eino vs 其他编排框架
+
+| 框架 | 特点 | CamTalk 适用性 |
+|------|------|---------------|
+| **Eino** | Go 原生、类型安全、流式原生 | ✅ 完美匹配 |
+| LangChain Go | 生态丰富但较重 | ❌ 过度抽象 |
+| 自研编排 | 完全可控 | ❌ 维护成本高 |
+
+**选择 Eino 的核心理由**：
+1. Go 原生，泛型支持，编译时类型检查
+2. 原生流式处理（`StreamReader`），适合 LLM token 级推送
+3. Graph 支持分支、并行、循环，满足当前和未来需求
+4. Callback 机制实现 AOP（日志、指标、消息推送）
+5. eino-ext 提供 OpenAI ChatModel 实现，直接对接 DashScope
+
+### 2.3 核心依赖版本
+
+```
+github.com/cloudwego/eino v0.9.9
+github.com/cloudwego/eino-ext/components/model/openai v0.1.13
+```
+
+## 3. Eino 核心概念
+
+### 3.1 Lambda
+
+Lambda 是 Graph 中的可组合函数单元，支持四种模式：
+
+| 模式 | 函数签名 | 构造方法 | 说明 |
+|------|---------|---------|------|
+| Invoke | `I → O` | `compose.InvokableLambda()` | 同步调用 |
+| Stream | `I → StreamReader[O]` | `compose.StreamableLambda()` | 流式输出 |
+| Collect | `StreamReader[I] → O` | `compose.CollectableLambda()` | 流式输入 |
+| Transform | `StreamReader[I] → StreamReader[O]` | `compose.TransformableLambda()` | 双向流式 |
+
+**返回类型**：所有 Lambda 构造函数返回 `*compose.Lambda`。
+
+### 3.2 Graph
+
+Graph 是有向无环图（DAG）编排器，支持：
+- **节点**：Lambda、ChatModel、ToolsNode 等
+- **边**：`g.AddEdge(from, to)` 定义数据流向
+- **分支**：`g.AddBranch()` 条件路由
+- **State**：`compose.WithGenLocalState()` 跨节点共享状态
+
+```go
+g := compose.NewGraph[PipelineInput, PipelineOutput]()
+g.AddLambdaNode("stt", sttLambda)
+g.AddChatModelNode("llm", chatModel)
+g.AddEdge(compose.START, "stt")
+g.AddEdge("stt", "llm")
+g.AddEdge("llm", compose.END)
+
+runnable, err := g.Compile(ctx)
+output, err := runnable.Invoke(ctx, input)   // 同步调用
+stream, err := runnable.Stream(ctx, input)    // 流式调用
+```
+
+### 3.3 ChatModel
+
+ChatModel 是 LLM 组件抽象，接口定义：
+
+```go
+type BaseChatModel interface {
+    Generate(ctx, []*schema.Message, ...Option) (*schema.Message, error)
+    Stream(ctx, []*schema.Message, ...Option) (*schema.StreamReader[*schema.Message], error)
+}
+```
+
+CamTalk 使用 `eino-ext/components/model/openai` 实现，通过 `BaseURL` 对接 DashScope：
+
+```go
+chatModel, _ := openai.NewChatModel(ctx, &openai.ChatModelConfig{
+    APIKey:  cfg.AI.LLM.APIKey,
+    Model:   cfg.AI.LLM.Model,
+    BaseURL: cfg.AI.LLM.Endpoint, // "https://dashscope.aliyuncs.com/compatible-mode/v1"
+})
+```
+
+### 3.4 StreamReader
+
+`schema.StreamReader[T]` 是 Eino 的流式数据抽象：
+- `sr.Recv()` 读取一帧，`io.EOF` 表示流结束
+- `schema.Pipe[T](bufSize)` 创建 `StreamReader` + `StreamWriter` 对
+- 框架自动处理 `T ↔ StreamReader[T]` 的转换（装箱/concat）
+
+### 3.5 Callback
+
+Callback 是 Eino 的 AOP 机制，支持节点生命周期钩子：
+
+```go
+type Handler interface {
+    OnStart(ctx, *RunInfo, CallbackInput) context.Context
+    OnEnd(ctx, *RunInfo, CallbackOutput) context.Context
+    OnError(ctx, *RunInfo, error) context.Context
+    OnStartWithStreamInput(ctx, *RunInfo, *StreamReader[CallbackInput]) context.Context
+    OnEndWithStreamOutput(ctx, *RunInfo, *StreamReader[CallbackOutput]) context.Context
+}
+```
+
+CamTalk 使用 `utils/callbacks.NewHandlerHelper()` 构建 typed handler：
+- `ModelCallbackHandler.OnEndWithStreamOutput`：逐 token 推送 `llm_chunk`
+
+### 3.6 State
+
+Graph 全局状态，通过 `WithGenLocalState` 注册：
+
+```go
+type PipelineState struct {
+    FullResponse    strings.Builder
+    TranscribedText string
+    TokenUsage      *TokenUsage
+}
+
+g := compose.NewGraph[I, O](compose.WithGenLocalState(func(ctx context.Context) *PipelineState {
+    return &PipelineState{}
+}))
+```
+
+节点通过 `compose.ProcessState` 读写 State。
+
+## 4. CamTalk Graph 设计
+
+### 4.1 拓扑
+
+```
+START → STT → History → ChatModel → Splitter → TTS → Done → END
+```
+
+| 节点 | 类型 | 输入 → 输出 | 职责 |
+|------|------|------------|------|
+| STT | InvokableLambda | `PipelineInput → STTOutput` | 语音识别，写入 State |
+| History | InvokableLambda | `STTOutput → []*schema.Message` | 组装提示词和历史 |
+| ChatModel | ChatModel（原生） | `[]*schema.Message → StreamReader[*Message]` | LLM 流式推理 |
+| Splitter | TransformableLambda | `StreamReader[string] → StreamReader[[]string]` | 句子切分 |
+| TTS | InvokableLambda | `[]string → struct{}` | 语音合成，推送音频 |
+| Done | InvokableLambda | `struct{} → PipelineOutput` | 发送 llm_done |
+
+### 4.2 流式模式
+
+Graph 使用 **Stream 模式**调用：
+- 内部所有节点以 Transform 模式运行
+- ChatModel 的 `Stream()` 方法实现真正的 token 级流式
+- 适配器消费 `StreamReader[PipelineOutput]` 触发整条链路
+
+### 4.3 消息推送机制
+
+| 消息 | 推送方式 | 时机 |
+|------|---------|------|
+| `stt_result` | Lambda 内部直接调用 Sender | STT 完成后 |
+| `llm_chunk` | Callback `OnEndWithStreamOutput` | ChatModel 逐 token |
+| `tts_audio` | Lambda 内部直接调用 Sender | TTS 逐句合成 |
+| `llm_done` | Lambda 内部直接调用 Sender | Done 节点执行时 |
+
+**Context 注入**：Sender、RequestID、SessionID、PipelineState 通过 `context.WithValue` 传递。
+
+### 4.4 多模态支持
+
+History 节点将图片构建为 `schema.Message.UserInputMultiContent`：
+
+```go
+systemMsg.UserInputMultiContent = []schema.MessageInputPart{
+    {
+        Type: schema.ChatMessagePartTypeImageURL,
+        Image: &schema.MessageInputImage{
+            MessagePartCommon: schema.MessagePartCommon{
+                Base64Data: &base64Str,
+                MIMEType:   "image/jpeg",
+            },
+            Detail: schema.ImageURLDetailAuto,
+        },
+    },
+}
+```
+
+## 5. 目录结构
+
+```
+backend/internal/eino/
+├── types.go           # PipelineInput/Output、STTOutput、TokenUsage
+├── state.go           # PipelineState（跨节点状态）
+├── callback.go        # Callback handler（LLM token 推送）
+├── graph.go           # Graph 构建与编译
+├── adapter.go         # EinoOrchestrator（Orchestrator 接口适配器）
+├── nodes_stt.go       # STT Lambda
+├── nodes_history.go   # 历史组装 Lambda
+├── nodes_splitter.go  # 句子分割 Transform Lambda
+├── nodes_tts.go       # TTS Lambda
+├── nodes_done.go      # Done Lambda
+└── graph_test.go      # 单元测试
+```
+
+## 6. 注意事项
+
+### 6.1 值类型 vs 指针类型
+
+Graph 泛型参数必须使用值类型（`PipelineInput`/`PipelineOutput`），所有 Lambda 的输入输出也使用值类型。框架在 Transform 模式下会自动处理 `T` 和 `StreamReader[T]` 的转换。
+
+### 6.2 Callback 运行时传入
+
+Callback 通过 `Stream()` 的 option 传入，不在 `Compile()` 时注册：
+
+```go
+streamReader, err := runnable.Stream(ctx, input, compose.WithCallbacks(handler))
+```
+
+### 6.3 eino-ext 与 DashScope 兼容性
+
+eino-ext OpenAI ChatModel 通过 `BaseURL` 对接 DashScope 兼容接口。需注意：
+- 多模态图片使用 `Base64Data` + `MIMEType` 格式
+- `Timeout` 控制单次请求超时
+- 流式输出通过 `Stream()` 方法获取 `StreamReader[*schema.Message]`
+
+### 6.4 框架自动类型转换
+
+Eino 框架在编排场景中自动处理以下转换：
+- **T → StreamReader[T]**：将完整值装箱为单帧流（非流式 → 假流式）
+- **StreamReader[T] → T**：将流 concat 为完整值（流式 → 非流式）
+
+这使得不同流式模式的节点可以无缝连接。
--- a/docs/12-Eino重构实施记录.md
+++ b/docs/12-Eino重构实施记录.md
@@ -0,0 +1,204 @@
+# 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 |