diff --git a/docs/10-Eino重构方案.md b/docs/10-Eino重构方案.md new file mode 100644 index 0000000..f7f9a3e --- /dev/null +++ b/docs/10-Eino重构方案.md @@ -0,0 +1,810 @@ +# CamTalk 后端 AI 编排层 Eino 重构方案 + +> 创建日期：2026-06-19 +> 状态：草案 + +## 1. 背景与目标 + +### 1.1 现状问题 + +当前后端 AI 编排层（`internal/orchestrator/pipeline.go`）为手写 goroutine 管道： + +``` +STT → LLM(Stream) ──→ Splitter → TTS(Stream) → Sender + └→ Sender(LLMChunk) +``` + +存在以下问题： + +1. **编排逻辑硬编码**：STT→LLM→TTS 流程写死在 `ProcessQuery()` 中，扩展新流程（如视觉分析链路、多轮工具调用）需要重写 goroutine 调度 +2. **并发控制粗糙**：手动 `go func()` + `sync.WaitGroup`，缺乏结构化的流式数据传递 +3. **无回调/AOP 机制**：日志、指标、追踪散落在各处，无法统一注入 +4. **配置耦合**：模型名、TTS 参数等硬编码在 Pipeline 结构体，无法按请求动态切换 +5. **错误处理不一致**：TTS 错误被静默吞掉，STT/LLM 错误通过 Sender 发送，缺乏统一模式 + +### 1.2 重构目标 + +| 目标 | 说明 | +|------|------| +| 用 Eino Graph 替换手写 Pipeline | 声明式编排，类型安全，可组合 | +| 流式处理原生支持 | 利用 Eino 的 Transform/Stream 模式，替代手动 goroutine | +| 统一回调机制 | 通过 Eino Callback 实现日志、指标、追踪的 AOP | +| 按请求动态配置 | 利用 Eino Option 机制，支持每请求切换模型/参数 | +| 保持 API 兼容 | WebSocket 协议、REST API、Session 管理不变 | +| 渐进式迁移 | 可分阶段实施，新旧编排器并存 | + +## 2. Eino 编排模型选择 + +### 2.1 为什么选 Graph 而非 Chain 或 Workflow + +| 编排模式 | 适用场景 | CamTalk 适用性 | +|----------|----------|----------------| +| **Chain** | 线性流水线 | ❌ LLM 和 TTS 需要并行执行，非纯线性 | +| **Workflow** | DAG + 字段映射 | ⚠️ 不支持循环，未来 ReAct Agent 需要循环 | +| **Graph** | 任意有向图，支持分支/并行/循环 | ✅ 完美匹配，支持当前并行需求和未来扩展 | + +**选择 Graph**，理由： +- LLM Stream 输出需要同时分发给 TTS 和客户端（多下游分支） +- 未来需要支持 ReAct Agent 循环（Graph + Branch） +- 支持 Pregel 执行引擎，兼容未来有状态节点 + +### 2.2 Graph 拓扑设计 + +``` + ┌─────────────────────────────────────────┐ + │ CamTalk Pipeline Graph │ + │ │ + START │ │ END + │ │ ▲ + ▼ │ │ + ┌──────┴──────┐ │ + │ STT Node │ (Lambda: audio → text) │ + │ (可选跳过) │ │ + └──────┬──────┘ │ + │ text │ + ▼ │ + ┌──────────────┐ │ + │ History Node │ (Lambda: 组装对话历史) │ + └──────┬───────┘ │ + │ []*schema.Message │ + ▼ │ + ┌──────────────┐ ┌────────────────┐ │ + │ LLM Node │─────→│ Sentence Split │───┐ │ + │ (ChatModel) │stream│ Node (Lambda) │ │ │ + └──────┬───────┘ └────────────────┘ │ │ + │ stream │ │ + ▼ ▼ │ + ┌──────────────┐ ┌──────────────┐│ + │ Chunk Sender │ │ TTS Node ││ + │ Node (Lambda)│ │ (Lambda) ││ + └──────────────┘ └──────┬───────┘│ + │ │ + ▼ │ + ┌──────────────┐ │ + │Audio Sender │──┘ + │Node (Lambda) │ + └──────────────┘ +``` + +**关键设计决策：** + +- STT 作为起始 Lambda 节点（非 Eino 原生组件，需封装） +- LLM 使用 Eino 原生 ChatModel 组件（`eino-ext` 的 OpenAI 实现） +- LLM 输出通过 Graph 的多下游边分发：一条到 Chunk Sender（推文字），一条到 Sentence Split → TTS（推语音） +- TTS 封装为 Lambda 节点 +- 所有 Sender 操作封装为 Lambda 节点，注入 `Sender` 依赖 + +## 3. 详细设计 + +### 3.1 数据类型定义 + +```go +// internal/eino/types.go + +// Graph 统一输入 +type PipelineInput struct { + AudioData []byte // base64 解码后的音频（可选） + ImageData []byte // base64 解码后的图像（可选） + Text string // 直接文本输入（可选，跳过 STT） + SessionID string + RequestID string + Language string // zh / en + Scenario string // free_chat, interviewer, etc. +} + +// Graph 统一输出 +type PipelineOutput struct { + TranscribedText string // STT 结果 + FullResponse string // LLM 完整回复 +} + +// STT 节点输出 +type STTOutput struct { + Text string + Language string +} + +// LLM 节点输入（组装好的对话历史） +type LLMInput struct { + Messages []*schema.Message +} + +// 句子分割中间类型 +type SentenceChunk struct { + Sentence string + IsLast bool +} + +// TTS 节点输出 +type TTSAudioChunk struct { + AudioData []byte + Format string + Sentence string + IsLast bool +} +``` + +### 3.2 Eino Graph 构建 + +```go +// internal/eino/graph.go + +package eino + +import ( + "context" + "github.com/cloudwego/eino/components/model" + "github.com/cloudwego/eino/compose" + "github.com/cloudwego/eino/schema" +) + +// GraphOption 图级别配置 +type GraphOption struct { + ChatModel model.ToolCallingChatModel // Eino 原生 ChatModel + STTService stt.Service // 现有 STT 接口 + TTSService tts.Service // 现有 TTS 接口 + SessionMgr session.Manager // 会话管理 + Sender orchestrator.Sender // WS 消息推送 + PromptCfg *PromptConfig // 提示词配置 +} + +// NewPipelineGraph 构建编排图 +func NewPipelineGraph(ctx context.Context, opt *GraphOption) (compose.Runnable[PipelineInput, PipelineOutput], error) { + g := compose.NewGraph[PipelineInput, PipelineOutput]() + + // 1. STT 节点（Lambda） + sttNode := compose.InvokableLambda(sttLambda(opt.STTService)) + g.AddLambdaNode("stt", sttNode) + + // 2. 历史组装节点（Lambda） + historyNode := compose.InvokableLambda(historyLambda(opt.SessionMgr, opt.PromptCfg)) + g.AddLambdaNode("history", historyNode) + + // 3. LLM 节点（ChatModel，原生流式） + g.AddChatModelNode("llm", opt.ChatModel) + + // 4. 句子分割节点（Transform Lambda：stream → stream） + splitterNode := compose.TransformableLambda(splitterLambda()) + g.AddLambdaNode("splitter", splitterNode) + + // 5. LLM Chunk 推送节点（Transform Lambda） + chunkSenderNode := compose.TransformableLambda(chunkSenderLambda(opt.Sender)) + g.AddLambdaNode("chunk_sender", chunkSenderNode) + + // 6. TTS 节点（Collect Lambda：stream → non-stream） + ttsNode := compose.CollectableLambda(ttsLambda(opt.TTSService, opt.Sender)) + g.AddLambdaNode("tts", ttsNode) + + // 7. 完成通知节点（Invokable Lambda） + doneNode := compose.InvokableLambda(doneLambda(opt.Sender)) + g.AddLambdaNode("done", doneNode) + + // === 边连接 === + + // START → STT + g.AddEdge(compose.START, "stt") + // STT → History + g.AddEdge("stt", "history") + // History → LLM + g.AddEdge("history", "llm") + + // LLM 输出分发到两个下游（利用 Graph 多下游边） + // LLM → Chunk Sender（推送原始 token） + g.AddEdge("llm", "chunk_sender") + // LLM → Splitter → TTS（句子级语音合成） + g.AddEdge("llm", "splitter") + g.AddEdge("splitter", "tts") + + // Chunk Sender 和 TTS 都汇入 Done + g.AddEdge("chunk_sender", "done") + g.AddEdge("tts", "done") + + // Done → END + g.AddEdge("done", compose.END) + + // 编译 + return g.Compile(ctx, + compose.WithGraphName("camtalk_pipeline"), + compose.WithMaxRunSteps(50), + ) +} +``` + +### 3.3 节点实现 + +#### 3.3.1 STT Lambda + +```go +// internal/eino/nodes_stt.go + +func sttLambda(sttSvc stt.Service) func(ctx context.Context, input PipelineInput) (STTOutput, error) { + return func(ctx context.Context, input PipelineInput) (STTOutput, error) { + // 文本模式：跳过 STT + if input.Text != "" { + return STTOutput{Text: input.Text, Language: input.Language}, nil + } + + if len(input.AudioData) == 0 { + return STTOutput{}, fmt.Errorf("no audio data provided") + } + + // 调用现有 STT 服务 + result, err := sttSvc.Recognize(ctx, input.AudioData, stt.Options{ + Language: input.Language, + }) + if err != nil { + return STTOutput{}, fmt.Errorf("STT error: %w", err) + } + + return STTOutput{ + Text: result.Text, + Language: result.Language, + }, nil + } +} +``` + +#### 3.3.2 历史组装 Lambda + +```go +// internal/eino/nodes_history.go + +func historyLambda(sessionMgr session.Manager, promptCfg *PromptConfig) func(ctx context.Context, input STTOutput) ([]*schema.Message, error) { + return func(ctx context.Context, input STTOutput) ([]*schema.Message, error) { + sessionID := getSessionID(ctx) // 从 context 或 state 获取 + + history, err := sessionMgr.GetHistory(ctx, sessionID) + if err != nil { + return nil, fmt.Errorf("get history error: %w", err) + } + + // 构建系统提示词 + systemPrompt := promptCfg.BuildSystemPrompt(input.Language, getScenario(ctx)) + + messages := []*schema.Message{ + {Role: schema.System, Content: systemPrompt, + MultiContent: buildVisionContent(getImageData(ctx))}, + } + + // 追加历史消息 + for _, msg := range history { + messages = append(messages, &schema.Message{ + Role: schema.Role(msg.Role), + Content: msg.Content, + }) + } + + // 追加当前用户输入 + messages = append(messages, &schema.Message{ + Role: schema.User, + Content: input.Text, + }) + + // 保存用户消息到历史 + _ = sessionMgr.AppendMessage(ctx, sessionID, models.Message{ + Role: "user", + Content: input.Text, + }) + + return messages, nil + } +} +``` + +#### 3.3.3 句子分割 Transform Lambda + +```go +// internal/eino/nodes_splitter.go + +func splitterLambda() func(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (*schema.StreamReader[SentenceChunk], error) { + return func(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (*schema.StreamReader[SentenceChunk], error) { + sr, sw := schema.Pipe[SentenceChunk](8) + + go func() { + defer sw.Close() + var buffer []rune + + for { + chunk, err := stream.Recv() + if err != nil { + if err.Error() == "EOF" { + // 流结束，发送剩余缓冲 + if len(buffer) > 0 { + sw.Send(SentenceChunk{Sentence: string(buffer), IsLast: true}, nil) + } + return + } + sw.Send(SentenceChunk{}, err) + return + } + + for _, r := range chunk.Content { + buffer = append(buffer, r) + if isSentenceDelimiter(r) { + sw.Send(SentenceChunk{Sentence: string(buffer), IsLast: false}, nil) + buffer = buffer[:0] + } + } + } + }() + + return sr, nil + } +} +``` + +#### 3.3.4 TTS Collect Lambda + +```go +// internal/eino/nodes_tts.go + +func ttsLambda(ttsSvc tts.Service, sender orchestrator.Sender) func(ctx context.Context, stream *schema.StreamReader[SentenceChunk]) (struct{}, error) { + return func(ctx context.Context, stream *schema.StreamReader[SentenceChunk]) (struct{}, error) { + for { + chunk, err := stream.Recv() + if err != nil { + if err.Error() == "EOF" { + break + } + return struct{}{}, err + } + + if chunk.Sentence == "" { + continue + } + + // 调用 TTS 服务 + audioData, err := ttsSvc.Synthesize(ctx, chunk.Sentence, tts.Options{ + // 从 Option 或 Config 获取 + }) + if err != nil { + // TTS 失败不中断流程，仅记录日志 + log.Warn("TTS synthesis failed", zap.Error(err), + zap.String("sentence", chunk.Sentence)) + continue + } + + // 推送音频到客户端 + sender.SendTTSAudio(orchestrator.TTSAudioPayload{ + Audio: audioData, + Format: "mp3", + IsLast: chunk.IsLast, + }) + } + + return struct{}{}, nil + } +} +``` + +#### 3.3.5 Chunk Sender Transform Lambda + +```go +// internal/eino/nodes_sender.go + +func chunkSenderLambda(sender orchestrator.Sender) func(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (*schema.StreamReader[*schema.Message], error) { + return func(ctx context.Context, stream *schema.StreamReader[*schema.Message]) (*schema.StreamReader[*schema.Message], error) { + sr, sw := schema.Pipe[*schema.Message](8) + + go func() { + defer sw.Close() + for { + msg, err := stream.Recv() + if err != nil { + if err.Error() == "EOF" { + return + } + sw.Send(nil, err) + return + } + + // 推送 LLM 文本 chunk 到客户端 + sender.SendLLMChunk(orchestrator.LLMChunkPayload{ + Content: msg.Content, + }) + + // 透传给下游 + sw.Send(msg, nil) + } + }() + + return sr, nil + } +} +``` + +#### 3.3.6 Done Lambda + +```go +// internal/eino/nodes_done.go + +func doneLambda(sender orchestrator.Sender) func(ctx context.Context, input struct{}) (PipelineOutput, error) { + return func(ctx context.Context, input struct{}) (PipelineOutput, error) { + // 通知客户端 LLM 回复完成 + sender.SendLLMDone(orchestrator.LLMDonePayload{}) + + // 保存助手消息到历史 + // 注意：完整回复需要从某处收集，可通过 State 机制实现 + return PipelineOutput{}, nil + } +} +``` + +### 3.4 State 机制（收集完整回复） + +由于 LLM 输出被分发到两个下游，完整回复文本需要通过 Graph State 收集： + +```go +// internal/eino/state.go + +type PipelineState struct { + FullResponse strings.Builder + SessionID string + RequestID string +} + +func genLocalState(ctx context.Context) *PipelineState { + return &PipelineState{} +} + +// 在构建 Graph 时注册 State +func NewPipelineGraph(ctx context.Context, opt *GraphOption) (compose.Runnable[PipelineInput, PipelineOutput], error) { + g := compose.NewGraph[PipelineInput, PipelineOutput]( + compose.WithGenLocalState(genLocalState), + ) + + // ... 添加节点 ... + + // Chunk Sender 的 StatePostHandler 累积完整回复 + g.AddLambdaNode("chunk_sender", chunkSenderNode, + compose.WithStatePostHandler(func(ctx context.Context, output *schema.Message, state *PipelineState) *schema.Message { + state.FullResponse.WriteString(output.Content) + return output + }), + ) + + // Done 节点的 StatePreHandler 读取完整回复 + g.AddLambdaNode("done", doneNode, + compose.WithStatePreHandler(func(ctx context.Context, input struct{}, state *PipelineState) struct{} { + // 将完整回复存入 state 供 done 节点使用 + return input + }), + ) + + // ... +} +``` + +### 3.5 Callback 集成（日志/指标/追踪） + +```go +// internal/eino/callback.go + +type MetricsCallback struct { + logger *zap.Logger + metrics *MetricsCollector // Prometheus 等 +} + +func (m *MetricsCallback) OnStart(ctx context.Context, info *compose.RunInfo, input compose.CallbackInput) context.Context { + m.logger.Debug("node started", + zap.String("node", info.Name), + zap.String("graph", info.GraphName)) + return ctx +} + +func (m *MetricsCallback) OnEnd(ctx context.Context, info *compose.RunInfo, output compose.CallbackOutput) context.Context { + m.logger.Debug("node completed", + zap.String("node", info.Name)) + return ctx +} + +func (m *MetricsCallback) OnError(ctx context.Context, info *compose.RunInfo, err error) context.Context { + m.logger.Error("node failed", + zap.String("node", info.Name), + zap.Error(err)) + m.metrics.IncrementError(info.Name) + return ctx +} + +// 注册到 Graph +func NewPipelineGraph(ctx context.Context, opt *GraphOption) (compose.Runnable[PipelineInput, PipelineOutput], error) { + // ... + callback := &MetricsCallback{logger: opt.Logger, metrics: opt.Metrics} + + return g.Compile(ctx, + compose.WithCallbacks(callback), // 全局回调 + compose.WithCallbacks(llmCallback).DesignateNode("llm"), // LLM 专用回调 + ) +} +``` + +### 3.6 按请求动态配置 + +```go +// internal/eino/options.go + +// 运行时 Option：每请求可变 +func WithModelName(name string) compose.Option { + return compose.WithChatModelOption(model.WithModel(name)) +} + +func WithTemperature(temp float32) compose.Option { + return compose.WithChatModelOption(model.WithTemperature(temp)) +} + +func WithTTSVoice(voice string) compose.Option { + return compose.WithCallbacks(&ttsVoiceCallback{voice: voice}). + DesignateNode("tts") +} + +// WebSocket Handler 中的调用 +func (c *Client) handleQuery(req QueryRequest) { + opts := []compose.Option{} + + // 根据请求配置动态注入 + if req.Model != "" { + opts = append(opts, WithModelName(req.Model)) + } + if req.TTSVoice != "" { + opts = append(opts, WithTTSVoice(req.TTSVoice)) + } + + output, err := c.pipeline.Invoke(ctx, PipelineInput{...}, opts...) +} +``` + +### 3.7 ChatModel 适配（接入 eino-ext OpenAI） + +```go +// internal/eino/chatmodel.go + +import ( + openaiImpl "github.com/cloudwego/eino-ext/components/model/openai" +) + +func NewChatModel(cfg *config.AIConfig) (model.ToolCallingChatModel, error) { + return openaiImpl.NewChatModel(context.Background(), &openaiImpl.ChatModelConfig{ + APIKey: cfg.LLM.APIKey, + Model: cfg.LLM.Model, + BaseURL: cfg.LLM.BaseURL, + }) +} +``` + +## 4. 目录结构变更 + +``` +backend/internal/ +├── eino/ # 新增：Eino 编排层 +│ ├── graph.go # Graph 构建与编译 +│ ├── types.go # 数据类型定义 +│ ├── state.go # Graph State 定义 +│ ├── options.go # 运行时 Option +│ ├── callback.go # 回调实现（日志/指标） +│ ├── chatmodel.go # ChatModel 适配器 +│ ├── nodes_stt.go # STT Lambda 节点 +│ ├── nodes_history.go # 历史组装 Lambda 节点 +│ ├── nodes_splitter.go # 句子分割 Transform Lambda +│ ├── nodes_tts.go # TTS Collect Lambda 节点 +│ ├── nodes_sender.go # Chunk Sender Transform Lambda +│ ├── nodes_done.go # 完成通知 Lambda 节点 +│ └── graph_test.go # 集成测试 +├── orchestrator/ # 保留：兼容层（Phase 1） +│ ├── orchestrator.go # 接口定义（不变） +│ ├── pipeline.go # 旧实现（Phase 3 移除） +│ ├── splitter.go # 被 eino/nodes_splitter.go 替代 +│ ├── sender.go # Sender 接口（不变，被 eino 层引用） +│ └── eino_adapter.go # 新增：Eino 编排器适配为 Orchestrator 接口 +├── ai/ # 保留：AI 服务接口不变 +│ ├── llm/ # 保留接口，实现被 eino-ext 替代 +│ ├── stt/ # 完全保留 +│ └── tts/ # 完全保留 +└── ws/ # 保留：WebSocket Handler + └── handler.go # 切换到 Eino 编排器 +``` + +## 5. 分阶段实施计划 + +### Phase 1：基础设施（预计 2-3 天） + +| 任务 | 文件 | 说明 | +|------|------|------| +| 引入 Eino 依赖 | `go.mod` | `go get github.com/cloudwego/eino/...` | +| 引入 eino-ext OpenAI | `go.mod` | `go get github.com/cloudwego/eino-ext/...` | +| 定义数据类型 | `eino/types.go` | PipelineInput/Output、中间类型 | +| 定义 State | `eino/state.go` | PipelineState | +| 实现 ChatModel 适配器 | `eino/chatmodel.go` | 包装 eino-ext OpenAI | +| 编写 Callback 框架 | `eino/callback.go` | 日志 + 指标回调 | + +### Phase 2：节点实现与 Graph 构建（预计 3-4 天） + +| 任务 | 文件 | 说明 | +|------|------|------| +| STT Lambda | `eino/nodes_stt.go` | 包装现有 stt.Service | +| 历史组装 Lambda | `eino/nodes_history.go` | 对话历史 + 提示词 | +| 句子分割 Transform | `eino/nodes_splitter.go` | 重写 splitter.go 为 Eino Lambda | +| TTS Collect Lambda | `eino/nodes_tts.go` | 包装现有 tts.Service | +| Chunk Sender Transform | `eino/nodes_sender.go` | LLM token 推送 | +| Done Lambda | `eino/nodes_done.go` | 完成通知 | +| Graph 构建 | `eino/graph.go` | 组装所有节点 | +| 单元测试 | `eino/graph_test.go` | Mock 各节点测试图结构 | + +### Phase 3：集成与切换（预计 2-3 天） + +| 任务 | 文件 | 说明 | +|------|------|------| +| Eino 适配器 | `orchestrator/eino_adapter.go` | 将 Eino Graph 包装为现有 Orchestrator 接口 | +| WS Handler 切换 | `ws/handler.go` | 使用新的 Eino 编排器 | +| main.go 依赖注入 | `cmd/server/main.go` | 构建 ChatModel + Graph | +| 集成测试 | `eino/graph_test.go` | 端到端测试 | +| 性能对比 | - | 延迟、内存、CPU 对比 | + +### Phase 4：清理与增强（预计 1-2 天） + +| 任务 | 说明 | +|------|------| +| 移除旧 Pipeline | 删除 `orchestrator/pipeline.go`、`splitter.go` | +| 更新文档 | 更新架构文档、接口文档 | +| 启用 ReAct Agent（可选） | 基于 Graph Branch 实现工具调用循环 | +| 动态配置完善 | 按请求切换模型、TTS 参数 | + +## 6. 风险与缓解 + +| 风险 | 影响 | 缓解措施 | +|------|------|----------| +| Eino 框架不稳定（v0.x） | 生产故障 | 锁定版本，保留旧 Pipeline 可回退 | +| 流式处理延迟增加 | 用户体验下降 | 性能对比测试，必要时绕过 Eino 直接调用 | +| LLM 输出多下游分发丢失数据 | TTS 无输入 | 充分测试 Stream Copy 机制，添加监控 | +| 学习曲线 | 开发效率 | 先从简单 Chain 开始，逐步过渡到 Graph | +| eino-ext OpenAI 不兼容现有 API | 功能回退 | 验证 BaseURL 和参数映射，必要时自定义适配器 | + +## 7. 测试策略 + +### 7.1 单元测试 + +```go +// eino/graph_test.go + +func TestPipelineGraph_WithTextInput(t *testing.T) { + // Mock STT, LLM, TTS, Sender + mockLLM := &mockChatModel{responses: []string{"你好！"}} + mockSender := &mockSender{} + + graph, err := NewPipelineGraph(ctx, &GraphOption{ + ChatModel: mockLLM, + Sender: mockSender, + // ... + }) + require.NoError(t, err) + + output, err := graph.Invoke(ctx, PipelineInput{ + Text: "你好", + SessionID: "test-session", + }) + require.NoError(t, err) + assert.Equal(t, "你好！", output.FullResponse) + assert.True(t, mockSender.LLMDoneSent) +} + +func TestPipelineGraph_WithAudioInput(t *testing.T) { + mockSTT := &mockSTT{text: "你好"} + mockLLM := &mockChatModel{responses: []string{"你好！"}} + mockTTS := &mockTTS{audio: []byte("fake-audio")} + mockSender := &mockSender{} + + graph, _ := NewPipelineGraph(ctx, &GraphOption{ + ChatModel: mockLLM, + STTService: mockSTT, + TTSService: mockTTS, + Sender: mockSender, + }) + + output, err := graph.Invoke(ctx, PipelineInput{ + AudioData: []byte("fake-audio-data"), + SessionID: "test-session", + }) + require.NoError(t, err) + assert.True(t, mockSender.TTSAudioSent) +} +``` + +### 7.2 集成测试 + +- 启动真实 OpenAI API 调用（使用测试 key） +- 验证 WebSocket 消息序列：`stt_result` → `llm_chunk` × N → `llm_done` → `tts_audio` × N +- 验证 interrupt 取消功能 +- 验证多并发请求隔离 + +## 8. 依赖清单 + +```go +// go.mod 新增 +require ( + github.com/cloudwego/eino v0.4.x // 核心框架 + github.com/cloudwego/eino-ext v0.1.x // 组件实现 +) +``` + +## 9. 未来扩展路径 + +基于 Eino Graph 的重构完成后，可无缝扩展： + +1. **ReAct Agent**：Graph 添加 Branch 节点，实现 LLM → Tool → LLM 循环 +2. **多模态理解**：添加视觉分析 Lambda 节点（图像描述 → 上下文注入） +3. **Model Router**：Graph 前置分支节点，按场景/成本路由不同 LLM +4. **Rate Limiter**：通过 Callback 的 OnStart 实现令牌桶 +5. **Checkpoint/Resume**：利用 Eino 的 CheckpointStore 实现断点续传 +6. **Multi-Agent**：利用 ADK 的 Supervisor/SequentialAgent 编排复杂对话流程 + +--- + +## 附录 A：Eino vs 现有实现对比 + +| 维度 | 现有实现 | Eino 重构后 | +|------|----------|------------| +| 编排方式 | 手写 goroutine + channel | 声明式 Graph，类型安全 | +| 流式处理 | 手动 channel 传递 | StreamReader + Pipe，自动转换 | +| 错误处理 | 各节点独立处理 | 统一 Callback OnError | +| 日志/追踪 | 散落在各处 | AOP Callback 注入 | +| 配置灵活性 | Pipeline 创建时固定 | 每请求 Option 动态注入 | +| 可测试性 | 需要启动 goroutine | Graph.Invoke 直接测试 | +| 扩展性 | 修改 Pipeline 代码 | 添加节点 + 边，无需改已有逻辑 | +| 并发安全 | 手动 sync | State 自动加锁 | + +## 附录 B：关键 Eino API 参考 + +```go +// 构建 Graph +g := compose.NewGraph[I, O](opts...) +g.AddChatModelNode(key, chatModel) +g.AddLambdaNode(key, lambda, opts...) +g.AddEdge(from, to) +g.AddBranch(from, branchFunc, mapping) + +// 编译 +runnable, err := g.Compile(ctx, opts...) + +// 执行四种模式 +output, err := runnable.Invoke(ctx, input, opts...) +stream, err := runnable.Stream(ctx, input, opts...) +output, err := runnable.Collect(ctx, inputStream, opts...) +stream, err := runnable.Transform(ctx, inputStream, opts...) + +// Lambda 四种构造器 +lambda := compose.InvokableLambda(fn) // I → O +lambda := compose.StreamableLambda(fn) // I → StreamReader[O] +lambda := compose.CollectableLambda(fn) // StreamReader[I] → O +lambda := compose.TransformableLambda(fn) // StreamReader[I] → StreamReader[O] + +// Stream 操作 +sr, sw := schema.Pipe[T](bufSize) +sw.Send(chunk, err) +chunk, err := sr.Recv() +sw.Close() + +// Option +compose.WithCallbacks(handler) +compose.WithCallbacks(handler).DesignateNode("node_key") +compose.WithChatModelOption(model.WithTemperature(0.7)) +compose.WithGenLocalState(genFunc) +```