diff --git a/docs/11-Eino框架技术文档.md b/docs/11-Eino框架技术文档.md new file mode 100644 index 0000000..0995b97 --- /dev/null +++ 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 为完整值(流式 → 非流式) + +这使得不同流式模式的节点可以无缝连接。 diff --git a/docs/12-Eino重构实施记录.md b/docs/12-Eino重构实施记录.md new file mode 100644 index 0000000..cafcdca --- /dev/null +++ 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 |