cfy666
3dc2015a91
- 架构设计：更新为 Eino Graph 声明式编排，增加三级存储架构说明 - 接口文档：AI 编排器章节重写为 Eino Graph，更新 LLM 服务接口 - 技术选型：新增 Eino 框架选型章节，修正 STT/LLM/TTS 默认方案 - 语音交互：Pipeline 描述改为 Eino Graph - 成本控制：模型引用修正为 qwen3-vl-plus - 技术名词解释：新增 Eino 框架相关术语 - README：增加 10/11/12 Eino 文档索引 - 10-Eino重构方案：状态更新为已实施 - CLAUDE.md：同步所有变更
811 lines
28 KiB
Markdown
811 lines
28 KiB
Markdown
# 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)
|
||
```
|