# CamTalk 后端 AI 编排层 Eino 重构方案

> 创建日期：2026-06-19
> 状态：已实施（实施记录见 [12-Eino重构实施记录](12-Eino重构实施记录.md)）

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