CamTalk/docs/11-Eino框架技术文档.md

# 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 为完整值（流式 → 非流式）

这使得不同流式模式的节点可以无缝连接。