CamTalk/docs/10-Eino重构方案.md

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 数据类型定义

// 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 构建

// 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

// 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

// 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

// 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

// 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

// 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

// 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 收集：

// 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 集成（日志/指标/追踪）

// 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 按请求动态配置

// 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）

// 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 单元测试

// 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.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 参考

// 构建 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)