120 lines
3.5 KiB
Go
120 lines
3.5 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
"time"
|
||
|
||
openaiImpl "github.com/cloudwego/eino-ext/components/model/openai"
|
||
"github.com/cloudwego/eino/compose"
|
||
|
||
"github.com/hhs/camtalk/internal/ai/stt"
|
||
"github.com/hhs/camtalk/internal/ai/tts"
|
||
"github.com/hhs/camtalk/internal/config"
|
||
"github.com/hhs/camtalk/internal/logger"
|
||
"github.com/hhs/camtalk/internal/models"
|
||
"github.com/hhs/camtalk/internal/session"
|
||
"github.com/hhs/camtalk/internal/store"
|
||
)
|
||
|
||
const (
|
||
nodeSTT = "stt"
|
||
nodeHistory = "history"
|
||
nodeLLM = "llm"
|
||
nodeMessageToString = "msg2str"
|
||
nodeSplitter = "splitter"
|
||
nodeTTS = "tts"
|
||
nodeDone = "done"
|
||
)
|
||
|
||
// PipelineGraph 封装编译后的 Eino Graph。
|
||
type PipelineGraph struct {
|
||
Runnable compose.Runnable[PipelineInput, PipelineOutput]
|
||
}
|
||
|
||
// NewPipelineGraph 构建 CamTalk AI 编排 Graph。
|
||
//
|
||
// 拓扑：START → STT → History → ChatModel → Splitter → TTS → Done → END
|
||
//
|
||
// Graph 使用 Stream 模式调用，ChatModel 实现真正的 token 级流式输出。
|
||
// LLM token 通过 Callback 的 OnEndWithStreamOutput 实时推送到客户端。
|
||
func NewPipelineGraph(
|
||
ctx context.Context,
|
||
cfg *config.Config,
|
||
sttService stt.Service,
|
||
ttsService tts.Service,
|
||
sessionMgr session.Manager,
|
||
scenarioRepo store.UserScenarioRepository,
|
||
) (*PipelineGraph, error) {
|
||
log := logger.Log
|
||
|
||
// 1. 创建 eino-ext ChatModel（对接 DashScope OpenAI 兼容接口）
|
||
chatModel, err := openaiImpl.NewChatModel(ctx, &openaiImpl.ChatModelConfig{
|
||
APIKey: cfg.AI.LLM.APIKey,
|
||
Model: cfg.AI.LLM.Model,
|
||
BaseURL: cfg.AI.LLM.Endpoint,
|
||
Timeout: time.Duration(cfg.AI.LLM.Timeout) * time.Second,
|
||
})
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
log.Infow("Eino ChatModel 初始化成功",
|
||
"model", cfg.AI.LLM.Model,
|
||
"endpoint", cfg.AI.LLM.Endpoint)
|
||
|
||
// 2. 构建 Graph（值类型，非指针）
|
||
g := compose.NewGraph[PipelineInput, PipelineOutput](
|
||
compose.WithGenLocalState(genLocalState),
|
||
)
|
||
|
||
// 3. 添加节点
|
||
maxHistory := cfg.Session.MaxHistory
|
||
|
||
_ = g.AddLambdaNode(nodeSTT, NewSTTLambda(sttService))
|
||
_ = g.AddLambdaNode(nodeHistory, NewHistoryLambda(sessionMgr.GetHistory, scenarioRepo, maxHistory))
|
||
_ = g.AddChatModelNode(nodeLLM, chatModel)
|
||
_ = g.AddLambdaNode(nodeMessageToString, NewMessageToStringLambda())
|
||
_ = g.AddLambdaNode(nodeSplitter, NewSplitterLambda())
|
||
_ = g.AddLambdaNode(nodeTTS, NewTTSLambda(
|
||
ttsService,
|
||
cfg.AI.TTS.Voice,
|
||
cfg.AI.TTS.Speed,
|
||
cfg.AI.TTS.OutputFormat,
|
||
cfg.AI.TTS.SampleRate,
|
||
))
|
||
_ = g.AddLambdaNode(nodeDone, NewDoneLambda(cfg.AI.LLM.Model))
|
||
|
||
// 4. 连接边
|
||
_ = g.AddEdge(compose.START, nodeSTT)
|
||
_ = g.AddEdge(nodeSTT, nodeHistory)
|
||
_ = g.AddEdge(nodeHistory, nodeLLM)
|
||
_ = g.AddEdge(nodeLLM, nodeMessageToString)
|
||
_ = g.AddEdge(nodeMessageToString, nodeSplitter)
|
||
_ = g.AddEdge(nodeSplitter, nodeTTS)
|
||
_ = g.AddEdge(nodeTTS, nodeDone)
|
||
_ = g.AddEdge(nodeDone, compose.END)
|
||
|
||
// 5. 编译（回调在运行时通过 Stream option 传入）
|
||
runnable, err := g.Compile(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
log.Infow("Eino Graph 编译成功", "nodes", 7)
|
||
return &PipelineGraph{Runnable: runnable}, nil
|
||
}
|
||
|
||
// buildPipelineInput 从 WebSocket 请求和会话配置构建 Graph 输入。
|
||
func buildPipelineInput(req models.WsQuery, sessionID string, sess *models.Session, audioData, imageData []byte) PipelineInput {
|
||
return PipelineInput{
|
||
AudioData: audioData,
|
||
ImageData: imageData,
|
||
Text: req.Text,
|
||
SessionID: sessionID,
|
||
RequestID: req.RequestID,
|
||
Language: sess.Config.Language,
|
||
Scenario: sess.Config.Scenario,
|
||
TTSEnabled: sess.Config.TTSEnabled,
|
||
UserID: sess.UserID,
|
||
}
|
||
}
|