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)
+```