From c9c697fe645731ee763d75bb56d6b4c9748c0a1b Mon Sep 17 00:00:00 2001 From: hhs <386998068@qq.com> Date: Sat, 13 Jun 2026 16:05:18 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=20AI=20Orchestrator?= =?UTF-8?q?=EF=BC=88=E6=A0=B8=E5=BF=83=E7=BC=96=E6=8E=92=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 实现 Orchestrator 接口（4.1） - 实现 Sender 接口（4.2） - 实现 STT → LLM → TTS 流式并行管道（4.3） - 实现句子切分器（4.4） - 实现错误降级处理（4.5） - 实现 Interrupt 支持（4.6） - 编写完整的单元测试（4.7） --- backend/go.mod | 4 + backend/go.sum | 2 + backend/internal/orchestrator/orchestrator.go | 33 + backend/internal/orchestrator/pipeline.go | 325 +++++++++ .../internal/orchestrator/pipeline_test.go | 661 ++++++++++++++++++ backend/internal/orchestrator/sender.go | 22 + backend/internal/orchestrator/splitter.go | 55 ++ 7 files changed, 1102 insertions(+) create mode 100644 backend/internal/orchestrator/orchestrator.go create mode 100644 backend/internal/orchestrator/pipeline.go create mode 100644 backend/internal/orchestrator/pipeline_test.go create mode 100644 backend/internal/orchestrator/sender.go create mode 100644 backend/internal/orchestrator/splitter.go diff --git a/backend/go.mod b/backend/go.mod index 8728265..8e60b15 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -8,6 +8,7 @@ require ( github.com/gorilla/websocket v1.5.3 github.com/redis/go-redis/v9 v9.20.1 github.com/spf13/viper v1.21.0 + github.com/stretchr/testify v1.11.1 go.uber.org/zap v1.28.0 ) @@ -17,6 +18,7 @@ require ( github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/cloudwego/base64x v0.1.4 // indirect github.com/cloudwego/iasm v0.2.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/gabriel-vasile/mimetype v1.4.3 // indirect github.com/gin-contrib/sse v0.1.0 // indirect @@ -32,11 +34,13 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.2 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/sagikazarmark/locafero v0.11.0 // indirect github.com/sourcegraph/conc v0.3.1-0.20240121214520-5f936abd7ae8 // indirect github.com/spf13/afero v1.15.0 // indirect github.com/spf13/cast v1.10.0 // indirect github.com/spf13/pflag v1.0.10 // indirect + github.com/stretchr/objx v0.5.2 // indirect github.com/subosito/gotenv v1.6.0 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/ugorji/go/codec v1.2.12 // indirect diff --git a/backend/go.sum b/backend/go.sum index 5a8d338..01eeeb8 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -86,6 +86,8 @@ github.com/spf13/viper v1.21.0/go.mod h1:P0lhsswPGWD/1lZJ9ny3fYnVqxiegrlNrEmgLjb github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= diff --git a/backend/internal/orchestrator/orchestrator.go b/backend/internal/orchestrator/orchestrator.go new file mode 100644 index 0000000..d038603 --- /dev/null +++ b/backend/internal/orchestrator/orchestrator.go @@ -0,0 +1,33 @@ +// Package orchestrator 实现 STT → LLM → TTS 流式并行管道。 +package orchestrator + +import ( + "context" + + "github.com/hhs/camtalk/internal/models" +) + +// Orchestrator AI 编排器接口。 +// 接收查询并执行完整的 STT → LLM → TTS 管道。 +type Orchestrator interface { + // ProcessQuery 处理一次用户查询。 + // ctx 用于整体超时和中断控制。 + // sessionID 用于会话管理和历史获取。 + // req 包含图像和音频数据。 + // history 是最近的对话历史。 + // sender 用于向客户端推送消息。 + ProcessQuery( + ctx context.Context, + sessionID string, + req models.WsQuery, + history []models.Message, + sender Sender, + ) error +} + +// QueryRequest 查询请求（内部使用）。 +type QueryRequest struct { + Image []byte // JPEG 图片（已从 Base64 解码） + Audio []byte // 音频数据（已从 Base64 解码） + Language string // 语言，如 "zh-CN" +} diff --git a/backend/internal/orchestrator/pipeline.go b/backend/internal/orchestrator/pipeline.go new file mode 100644 index 0000000..9285e51 --- /dev/null +++ b/backend/internal/orchestrator/pipeline.go @@ -0,0 +1,325 @@ +package orchestrator + +import ( + "context" + "encoding/base64" + "strings" + "sync" + "time" + "unicode/utf8" + + "github.com/hhs/camtalk/internal/ai/llm" + "github.com/hhs/camtalk/internal/ai/stt" + "github.com/hhs/camtalk/internal/ai/tts" + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" + "github.com/hhs/camtalk/internal/session" +) + +// Pipeline 实现 Orchestrator 接口，管理 STT → LLM → TTS 流式管道。 +type Pipeline struct { + sttService stt.Service + llmService llm.Service + ttsService tts.Service + sessionMgr session.Manager +} + +// New 创建 Pipeline 实例。 +func New( + sttService stt.Service, + llmService llm.Service, + ttsService tts.Service, + sessionMgr session.Manager, +) *Pipeline { + return &Pipeline{ + sttService: sttService, + llmService: llmService, + ttsService: ttsService, + sessionMgr: sessionMgr, + } +} + +// ProcessQuery 实现 Orchestrator 接口。 +func (p *Pipeline) ProcessQuery( + ctx context.Context, + sessionID string, + req models.WsQuery, + history []models.Message, + sender Sender, +) error { + log := logger.Log + startTime := time.Now() + + // 解码音频数据 + audio, err := base64.StdEncoding.DecodeString(req.Audio) + if err != nil { + log.Errorw("音频解码失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "INVALID_MESSAGE", + Message: "音频数据解码失败", + }) + return err + } + + // 解码图片数据（可选） + var image []byte + if req.Image != "" { + image, err = base64.StdEncoding.DecodeString(req.Image) + if err != nil { + log.Errorw("图片解码失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "INVALID_MESSAGE", + Message: "图片数据解码失败", + }) + return err + } + } + + // 设置活跃请求 + if err := p.sessionMgr.SetActiveRequest(ctx, sessionID, req.RequestID); err != nil { + log.Errorw("设置活跃请求失败", "error", err) + } + defer p.sessionMgr.ClearActiveRequest(ctx, sessionID) + + // 获取会话配置 + sess, err := p.sessionMgr.Get(ctx, sessionID) + if err != nil { + log.Errorw("获取会话失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "SESSION_NOT_FOUND", + Message: "会话不存在", + }) + return err + } + + // Step 1: STT 语音识别 + log.Infow("开始语音识别", "request_id", req.RequestID) + sttResult, err := p.sttService.Recognize(ctx, audio, stt.Options{ + Encoding: "pcm_s16le", + SampleRate: 16000, + Language: sess.Config.Language, + }) + if err != nil { + log.Errorw("语音识别失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "STT_ERROR", + Message: "语音识别失败", + }) + return err + } + + // 发送 STT 结果 + if err := sender.SendSTTResult(models.WsSTTResult{ + Type: "stt_result", + RequestID: req.RequestID, + Text: sttResult, + IsFinal: true, + }); err != nil { + log.Errorw("发送 STT 结果失败", "error", err) + } + + // 追加用户消息到历史 + p.sessionMgr.AppendMessage(ctx, sessionID, models.Message{ + Role: "user", + Content: sttResult, + }) + + // Step 2+3: LLM 流式推理 + TTS 并行合成 + log.Infow("开始 LLM 推理", "request_id", req.RequestID) + llmReq := llm.Request{ + Image: image, + Text: sttResult, + History: history, + Language: sess.Config.Language, + } + + llmStream, err := p.llmService.ChatStream(ctx, llmReq) + if err != nil { + log.Errorw("LLM 流式推理启动失败", "error", err) + sender.SendError(models.WsError{ + Type: "error", + RequestID: req.RequestID, + Code: "LLM_ERROR", + Message: "LLM 推理失败", + }) + return err + } + + // 创建句子切分器 + sentenceCh := make(chan string, 4) + splitter := NewSplitter(sentenceCh) + + // 并行：LLM 消费 + TTS 合成 + var wg sync.WaitGroup + var fullText string + var ttsErr error + + // goroutine 1: 消费 LLM token + 句子切分 + wg.Add(1) + go func() { + defer wg.Done() + defer close(sentenceCh) + fullText = p.consumeLLMStream(ctx, llmStream, req.RequestID, sender, splitter) + }() + + // goroutine 2: TTS 合成（如果启用） + if sess.Config.TTSEnabled { + wg.Add(1) + go func() { + defer wg.Done() + log.Infow("开始 TTS 合成", "request_id", req.RequestID) + ttsErr = p.synthesizeTTS(ctx, sentenceCh, req.RequestID, sender) + }() + } else { + // 如果 TTS 未启用，需要消费 sentenceCh 防止阻塞 + go func() { + for range sentenceCh { + } + }() + } + + // 等待所有 goroutine 完成 + wg.Wait() + + // TTS 失败静默跳过 + if ttsErr != nil { + log.Warnw("TTS 合成失败（已跳过）", "error", ttsErr) + } + + // 追加助手消息到历史 + p.sessionMgr.AppendMessage(ctx, sessionID, models.Message{ + Role: "assistant", + Content: fullText, + }) + + // 发送 llm_done + latency := time.Since(startTime).Milliseconds() + if err := sender.SendLLMDone(models.WsLLMDone{ + Type: "llm_done", + RequestID: req.RequestID, + FullText: fullText, + Model: "gpt-4o", + LatencyMs: latency, + }); err != nil { + log.Errorw("发送 llm_done 失败", "error", err) + } + + log.Infow("查询处理完成", + "request_id", req.RequestID, + "latency_ms", latency, + "text_length", utf8.RuneCountInString(fullText), + ) + + return nil +} + +// consumeLLMStream 消费 LLM 流式输出，发送 llm_chunk 并进行句子切分。 +func (p *Pipeline) consumeLLMStream( + ctx context.Context, + stream <-chan llm.Chunk, + requestID string, + sender Sender, + splitter *Splitter, +) string { + log := logger.Log + var fullText strings.Builder + + for chunk := range stream { + // 检查上下文是否已取消 + select { + case <-ctx.Done(): + log.Infow("LLM 流被中断", "request_id", requestID) + return fullText.String() + default: + } + + if chunk.Done { + // 流结束 + if chunk.TokensUsed != nil { + log.Infow("LLM 用量统计", + "request_id", requestID, + "prompt_tokens", chunk.TokensUsed.Prompt, + "completion_tokens", chunk.TokensUsed.Completion, + "total_tokens", chunk.TokensUsed.Total, + ) + } + break + } + + // 累积全文 + fullText.WriteString(chunk.Delta) + + // 发送 llm_chunk + if err := sender.SendLLMChunk(models.WsLLMChunk{ + Type: "llm_chunk", + RequestID: requestID, + Delta: chunk.Delta, + Role: "assistant", + }); err != nil { + log.Errorw("发送 llm_chunk 失败", "error", err) + } + + // 句子切分 + splitter.Feed(chunk.Delta) + } + + // 刷新切分器中的剩余文本 + splitter.Flush() + + return fullText.String() +} + +// synthesizeTTS 从句子 channel 读取文本，进行 TTS 合成并发送音频。 +func (p *Pipeline) synthesizeTTS( + ctx context.Context, + sentenceCh <-chan string, + requestID string, + sender Sender, +) error { + log := logger.Log + + ttsStream, err := p.ttsService.SynthesizeStream(ctx, sentenceCh, tts.Options{ + Voice: "alloy", + Speed: 1.0, + OutputFmt: "mp3", + SampleRate: 24000, + }) + if err != nil { + log.Errorw("TTS 合成启动失败", "error", err) + return err + } + + // 消费 TTS 音频流 + for chunk := range ttsStream { + // 检查上下文是否已取消 + select { + case <-ctx.Done(): + log.Infow("TTS 流被中断", "request_id", requestID) + return ctx.Err() + default: + } + + // Base64 编码音频数据 + audioBase64 := base64.StdEncoding.EncodeToString(chunk.Audio) + + if err := sender.SendTTSAudio(models.WsTTSAudio{ + Type: "tts_audio", + RequestID: requestID, + Audio: audioBase64, + MimeType: "audio/mp3", + IsLast: chunk.IsLast, + }); err != nil { + log.Errorw("发送 tts_audio 失败", "error", err) + } + } + + return nil +} diff --git a/backend/internal/orchestrator/pipeline_test.go b/backend/internal/orchestrator/pipeline_test.go new file mode 100644 index 0000000..4e493d8 --- /dev/null +++ b/backend/internal/orchestrator/pipeline_test.go @@ -0,0 +1,661 @@ +package orchestrator + +import ( + "context" + "encoding/base64" + "errors" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + + "github.com/hhs/camtalk/internal/ai/llm" + "github.com/hhs/camtalk/internal/ai/stt" + "github.com/hhs/camtalk/internal/ai/tts" + "github.com/hhs/camtalk/internal/logger" + "github.com/hhs/camtalk/internal/models" +) + +func init() { + logger.Init("debug", "console") +} + +// MockSTTService mock STT 服务 +type MockSTTService struct { + mock.Mock +} + +func (m *MockSTTService) Recognize(ctx context.Context, audio []byte, opts stt.Options) (string, error) { + args := m.Called(ctx, audio, opts) + return args.String(0), args.Error(1) +} + +// MockLLMService mock LLM 服务 +type MockLLMService struct { + mock.Mock +} + +func (m *MockLLMService) ChatStream(ctx context.Context, req llm.Request) (<-chan llm.Chunk, error) { + args := m.Called(ctx, req) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).(<-chan llm.Chunk), args.Error(1) +} + +// MockTTSService mock TTS 服务 +type MockTTSService struct { + mock.Mock +} + +func (m *MockTTSService) SynthesizeStream(ctx context.Context, textStream <-chan string, opts tts.Options) (<-chan tts.Chunk, error) { + args := m.Called(ctx, textStream, opts) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).(<-chan tts.Chunk), args.Error(1) +} + +// MockSessionManager mock 会话管理器 +type MockSessionManager struct { + mock.Mock +} + +func (m *MockSessionManager) Create(ctx context.Context, config models.SessionConfig) (string, error) { + args := m.Called(ctx, config) + return args.String(0), args.Error(1) +} + +func (m *MockSessionManager) Get(ctx context.Context, sessionID string) (*models.Session, error) { + args := m.Called(ctx, sessionID) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).(*models.Session), args.Error(1) +} + +func (m *MockSessionManager) UpdateConfig(ctx context.Context, sessionID string, patch models.SessionConfigPatch) error { + args := m.Called(ctx, sessionID, patch) + return args.Error(0) +} + +func (m *MockSessionManager) GetHistory(ctx context.Context, sessionID string, limit int) ([]models.Message, error) { + args := m.Called(ctx, sessionID, limit) + return args.Get(0).([]models.Message), args.Error(1) +} + +func (m *MockSessionManager) AppendMessage(ctx context.Context, sessionID string, msg models.Message) error { + args := m.Called(ctx, sessionID, msg) + return args.Error(0) +} + +func (m *MockSessionManager) SetActiveRequest(ctx context.Context, sessionID string, requestID string) error { + args := m.Called(ctx, sessionID, requestID) + return args.Error(0) +} + +func (m *MockSessionManager) GetActiveRequestID(ctx context.Context, sessionID string) (string, error) { + args := m.Called(ctx, sessionID) + return args.String(0), args.Error(1) +} + +func (m *MockSessionManager) ClearActiveRequest(ctx context.Context, sessionID string) error { + args := m.Called(ctx, sessionID) + return args.Error(0) +} + +func (m *MockSessionManager) Touch(ctx context.Context, sessionID string) error { + args := m.Called(ctx, sessionID) + return args.Error(0) +} + +func (m *MockSessionManager) Destroy(ctx context.Context, sessionID string) error { + args := m.Called(ctx, sessionID) + return args.Error(0) +} + +func (m *MockSessionManager) ActiveCount() int { + args := m.Called() + return args.Int(0) +} + +// MockSender mock WebSocket 发送器 +type MockSender struct { + mock.Mock + STTResults []models.WsSTTResult + LLMChunks []models.WsLLMChunk + LLMDones []models.WsLLMDone + TTSAudios []models.WsTTSAudio + Errors []models.WsError +} + +func NewMockSender() *MockSender { + return &MockSender{ + STTResults: make([]models.WsSTTResult, 0), + LLMChunks: make([]models.WsLLMChunk, 0), + LLMDones: make([]models.WsLLMDone, 0), + TTSAudios: make([]models.WsTTSAudio, 0), + Errors: make([]models.WsError, 0), + } +} + +func (m *MockSender) SendSTTResult(result models.WsSTTResult) error { + m.STTResults = append(m.STTResults, result) + args := m.Called(result) + return args.Error(0) +} + +func (m *MockSender) SendLLMChunk(chunk models.WsLLMChunk) error { + m.LLMChunks = append(m.LLMChunks, chunk) + args := m.Called(chunk) + return args.Error(0) +} + +func (m *MockSender) SendLLMDone(done models.WsLLMDone) error { + m.LLMDones = append(m.LLMDones, done) + args := m.Called(done) + return args.Error(0) +} + +func (m *MockSender) SendTTSAudio(audio models.WsTTSAudio) error { + m.TTSAudios = append(m.TTSAudios, audio) + args := m.Called(audio) + return args.Error(0) +} + +func (m *MockSender) SendError(err models.WsError) error { + m.Errors = append(m.Errors, err) + args := m.Called(err) + return args.Error(0) +} + +// 辅助函数：创建 LLM 流式响应 +func createLLMStream(chunks []llm.Chunk) <-chan llm.Chunk { + ch := make(chan llm.Chunk, len(chunks)) + for _, chunk := range chunks { + ch <- chunk + } + close(ch) + return ch +} + +// 辅助函数：创建 TTS 流式响应 +func createTTSStream(chunks []tts.Chunk) <-chan tts.Chunk { + ch := make(chan tts.Chunk, len(chunks)) + for _, chunk := range chunks { + ch <- chunk + } + close(ch) + return ch +} + +// TestProcessQuery_Success 测试完整流程 +func TestProcessQuery_Success(t *testing.T) { + // 准备测试数据 + audioData := []byte("test audio") + imageData := []byte("test image") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + imageBase64 := base64.StdEncoding.EncodeToString(imageData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Image: imageBase64, + Audio: audioBase64, + } + + session := &models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + TTSEnabled: true, + DetailLevel: "low", + Language: "zh-CN", + }, + } + + // 创建 mock + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + // 设置 mock 期望 + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(session, nil) + mockSession.On("AppendMessage", mock.Anything, "session-123", mock.Anything).Return(nil) + + mockSTT.On("Recognize", mock.Anything, audioData, stt.Options{ + Encoding: "pcm_s16le", + SampleRate: 16000, + Language: "zh-CN", + }).Return("你好，世界", nil) + + mockSender.On("SendSTTResult", mock.Anything).Return(nil) + + llmChunks := []llm.Chunk{ + {Delta: "你好"}, + {Delta: "，世界！"}, + {Done: true, TokensUsed: &llm.TokenUsage{Prompt: 10, Completion: 5, Total: 15}}, + } + mockLLM.On("ChatStream", mock.Anything, mock.Anything).Return(createLLMStream(llmChunks), nil) + + mockSender.On("SendLLMChunk", mock.Anything).Return(nil) + mockSender.On("SendLLMDone", mock.Anything).Return(nil) + + ttsChunks := []tts.Chunk{ + {Audio: []byte("audio1"), IsLast: false}, + {Audio: []byte("audio2"), IsLast: true}, + } + mockTTS.On("SynthesizeStream", mock.Anything, mock.Anything, mock.Anything).Return(createTTSStream(ttsChunks), nil) + + mockSender.On("SendTTSAudio", mock.Anything).Return(nil) + + // 创建 Pipeline + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + // 执行 + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + // 验证 + assert.NoError(t, err) + assert.Len(t, mockSender.STTResults, 1) + assert.Equal(t, "你好，世界", mockSender.STTResults[0].Text) + assert.Len(t, mockSender.LLMChunks, 2) + assert.Len(t, mockSender.LLMDones, 1) + assert.Len(t, mockSender.TTSAudios, 2) + + mockSTT.AssertExpectations(t) + mockLLM.AssertExpectations(t) + mockTTS.AssertExpectations(t) + mockSession.AssertExpectations(t) +} + +// TestProcessQuery_STTError 测试 STT 失败降级 +func TestProcessQuery_STTError(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(&models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + Language: "zh-CN", + }, + }, nil) + + mockSTT.On("Recognize", mock.Anything, audioData, mock.Anything). + Return("", errors.New("STT service unavailable")) + + mockSender.On("SendError", mock.Anything).Return(nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + assert.Error(t, err) + assert.Len(t, mockSender.Errors, 1) + assert.Equal(t, "STT_ERROR", mockSender.Errors[0].Code) + + mockSTT.AssertExpectations(t) + mockLLM.AssertNotCalled(t, "ChatStream") + mockTTS.AssertNotCalled(t, "SynthesizeStream") +} + +// TestProcessQuery_LLMError 测试 LLM 失败降级 +func TestProcessQuery_LLMError(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + session := &models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + TTSEnabled: true, + Language: "zh-CN", + }, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(session, nil) + mockSession.On("AppendMessage", mock.Anything, "session-123", mock.Anything).Return(nil) + + mockSTT.On("Recognize", mock.Anything, audioData, mock.Anything).Return("你好", nil) + mockSender.On("SendSTTResult", mock.Anything).Return(nil) + + mockLLM.On("ChatStream", mock.Anything, mock.Anything). + Return(nil, errors.New("LLM service unavailable")) + + mockSender.On("SendError", mock.Anything).Return(nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + assert.Error(t, err) + assert.Len(t, mockSender.Errors, 1) + assert.Equal(t, "LLM_ERROR", mockSender.Errors[0].Code) + + mockSTT.AssertExpectations(t) + mockLLM.AssertExpectations(t) + mockTTS.AssertNotCalled(t, "SynthesizeStream") +} + +// TestProcessQuery_TTSError 测试 TTS 失败静默跳过 +func TestProcessQuery_TTSError(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + session := &models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + TTSEnabled: true, + Language: "zh-CN", + }, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(session, nil) + mockSession.On("AppendMessage", mock.Anything, "session-123", mock.Anything).Return(nil) + + mockSTT.On("Recognize", mock.Anything, audioData, mock.Anything).Return("你好", nil) + mockSender.On("SendSTTResult", mock.Anything).Return(nil) + + llmChunks := []llm.Chunk{ + {Delta: "你好"}, + {Done: true}, + } + mockLLM.On("ChatStream", mock.Anything, mock.Anything).Return(createLLMStream(llmChunks), nil) + mockSender.On("SendLLMChunk", mock.Anything).Return(nil) + mockSender.On("SendLLMDone", mock.Anything).Return(nil) + + mockTTS.On("SynthesizeStream", mock.Anything, mock.Anything, mock.Anything). + Return(nil, errors.New("TTS service unavailable")) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + // TTS 失败应该静默跳过，不返回错误 + assert.NoError(t, err) + assert.Len(t, mockSender.LLMDones, 1) + assert.Len(t, mockSender.TTSAudios, 0) + + mockSTT.AssertExpectations(t) + mockLLM.AssertExpectations(t) + mockTTS.AssertExpectations(t) +} + +// TestProcessQuery_ContextCancelled 测试上下文取消（Interrupt） +func TestProcessQuery_ContextCancelled(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + session := &models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + TTSEnabled: true, + Language: "zh-CN", + }, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(session, nil) + mockSession.On("AppendMessage", mock.Anything, "session-123", mock.Anything).Return(nil) + + mockSTT.On("Recognize", mock.Anything, audioData, mock.Anything).Return("你好", nil) + mockSender.On("SendSTTResult", mock.Anything).Return(nil) + + // 创建一个会延迟的 LLM 流，以便我们可以取消上下文 + llmCh := make(chan llm.Chunk) + go func() { + time.Sleep(100 * time.Millisecond) + llmCh <- llm.Chunk{Delta: "你"} + time.Sleep(100 * time.Millisecond) + llmCh <- llm.Chunk{Delta: "好"} + close(llmCh) + }() + + mockLLM.On("ChatStream", mock.Anything, mock.Anything).Return((<-chan llm.Chunk)(llmCh), nil) + mockSender.On("SendLLMChunk", mock.Anything).Return(nil) + mockSender.On("SendLLMDone", mock.Anything).Return(nil) + + // 创建一个会延迟的 TTS 流 + ttsCh := make(chan tts.Chunk) + go func() { + time.Sleep(200 * time.Millisecond) + close(ttsCh) + }() + mockTTS.On("SynthesizeStream", mock.Anything, mock.Anything, mock.Anything).Return((<-chan tts.Chunk)(ttsCh), nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + // 创建可取消的上下文 + ctx, cancel := context.WithCancel(context.Background()) + + // 在 50ms 后取消 + go func() { + time.Sleep(50 * time.Millisecond) + cancel() + }() + + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + // 上下文取消后，流程应该正常完成（中断流但不返回错误） + assert.NoError(t, err) + + mockSTT.AssertExpectations(t) +} + +// TestProcessQuery_DisabledTTS 测试 TTS 未启用的情况 +func TestProcessQuery_DisabledTTS(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + session := &models.Session{ + ID: "session-123", + Config: models.SessionConfig{ + TTSEnabled: false, // TTS 未启用 + Language: "zh-CN", + }, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(session, nil) + mockSession.On("AppendMessage", mock.Anything, "session-123", mock.Anything).Return(nil) + + mockSTT.On("Recognize", mock.Anything, audioData, mock.Anything).Return("你好", nil) + mockSender.On("SendSTTResult", mock.Anything).Return(nil) + + llmChunks := []llm.Chunk{ + {Delta: "你好"}, + {Done: true}, + } + mockLLM.On("ChatStream", mock.Anything, mock.Anything).Return(createLLMStream(llmChunks), nil) + mockSender.On("SendLLMChunk", mock.Anything).Return(nil) + mockSender.On("SendLLMDone", mock.Anything).Return(nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + assert.NoError(t, err) + assert.Len(t, mockSender.LLMDones, 1) + assert.Len(t, mockSender.TTSAudios, 0) + + // TTS 不应该被调用 + mockTTS.AssertNotCalled(t, "SynthesizeStream") +} + +// TestSplitter 测试句子切分器 +func TestSplitter(t *testing.T) { + ch := make(chan string, 10) + splitter := NewSplitter(ch) + + // 输入包含多个句子的文本 + splitter.Feed("你好。") + splitter.Feed("世界！") + splitter.Feed("这是") + splitter.Feed("一个测试。") + splitter.Flush() + + // 应该有 3 个句子 + assert.Equal(t, 3, len(ch)) + assert.Equal(t, "你好。", <-ch) + assert.Equal(t, "世界！", <-ch) + assert.Equal(t, "这是一个测试。", <-ch) +} + +// TestSplitter_NoDelimiter 测试没有分隔符的情况 +func TestSplitter_NoDelimiter(t *testing.T) { + ch := make(chan string, 10) + splitter := NewSplitter(ch) + + splitter.Feed("没有分隔符的文本") + splitter.Flush() + + // 应该有 1 个句子（Flush 会发送剩余内容） + assert.Equal(t, 1, len(ch)) + assert.Equal(t, "没有分隔符的文本", <-ch) +} + +// TestSplitter_Empty 测试空输入 +func TestSplitter_Empty(t *testing.T) { + ch := make(chan string, 10) + splitter := NewSplitter(ch) + + splitter.Flush() + + // 应该没有句子 + assert.Equal(t, 0, len(ch)) +} + +// TestProcessQuery_InvalidAudio 测试无效音频数据 +func TestProcessQuery_InvalidAudio(t *testing.T) { + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: "invalid-base64!!!", + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSender.On("SendError", mock.Anything).Return(nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + assert.Error(t, err) + assert.Len(t, mockSender.Errors, 1) + assert.Equal(t, "INVALID_MESSAGE", mockSender.Errors[0].Code) +} + +// TestProcessQuery_SessionNotFound 测试会话不存在 +func TestProcessQuery_SessionNotFound(t *testing.T) { + audioData := []byte("test audio") + audioBase64 := base64.StdEncoding.EncodeToString(audioData) + + req := models.WsQuery{ + Type: "query", + RequestID: "req-123", + Audio: audioBase64, + } + + mockSTT := new(MockSTTService) + mockLLM := new(MockLLMService) + mockTTS := new(MockTTSService) + mockSession := new(MockSessionManager) + mockSender := NewMockSender() + + mockSession.On("SetActiveRequest", mock.Anything, "session-123", "req-123").Return(nil) + mockSession.On("ClearActiveRequest", mock.Anything, "session-123").Return(nil) + mockSession.On("Get", mock.Anything, "session-123").Return(nil, errors.New("session not found")) + + mockSender.On("SendError", mock.Anything).Return(nil) + + pipeline := New(mockSTT, mockLLM, mockTTS, mockSession) + + ctx := context.Background() + err := pipeline.ProcessQuery(ctx, "session-123", req, nil, mockSender) + + assert.Error(t, err) + assert.Len(t, mockSender.Errors, 1) + assert.Equal(t, "SESSION_NOT_FOUND", mockSender.Errors[0].Code) +} diff --git a/backend/internal/orchestrator/sender.go b/backend/internal/orchestrator/sender.go new file mode 100644 index 0000000..4a9bbcd --- /dev/null +++ b/backend/internal/orchestrator/sender.go @@ -0,0 +1,22 @@ +package orchestrator + +import "github.com/hhs/camtalk/internal/models" + +// Sender 抽象 WebSocket 消息推送能力。 +// 便于测试时 mock，避免依赖真实 WebSocket 连接。 +type Sender interface { + // SendSTTResult 发送语音识别结果。 + SendSTTResult(result models.WsSTTResult) error + + // SendLLMChunk 发送 LLM 流式文本增量。 + SendLLMChunk(chunk models.WsLLMChunk) error + + // SendLLMDone 发送 LLM 流结束信号。 + SendLLMDone(done models.WsLLMDone) error + + // SendTTSAudio 发送 TTS 音频数据。 + SendTTSAudio(audio models.WsTTSAudio) error + + // SendError 发送错误消息。 + SendError(err models.WsError) error +} diff --git a/backend/internal/orchestrator/splitter.go b/backend/internal/orchestrator/splitter.go new file mode 100644 index 0000000..57b8b26 --- /dev/null +++ b/backend/internal/orchestrator/splitter.go @@ -0,0 +1,55 @@ +package orchestrator + +import "strings" + +// sentenceDelimiters 句子分隔符集合。 +var sentenceDelimiters = map[rune]bool{ + '。': true, + '！': true, + '？': true, + '\n': true, + '.': true, + '!': true, + '?': true, +} + +// Splitter 句子切分器。 +// 将流式文本按句子边界切分，发送到 channel 供 TTS 合成。 +type Splitter struct { + ch chan<- string + buffer strings.Builder +} + +// NewSplitter 创建句子切分器。 +// ch 用于接收切分后的句子文本。 +func NewSplitter(ch chan<- string) *Splitter { + return &Splitter{ + ch: ch, + } +} + +// Feed 输入增量文本，遇到句子分隔符时发送完整句子。 +func (s *Splitter) Feed(delta string) { + for _, r := range delta { + s.buffer.WriteRune(r) + if sentenceDelimiters[r] { + s.flushBuffer() + } + } +} + +// Flush 刷新缓冲区中的剩余文本（即使没有句子分隔符）。 +func (s *Splitter) Flush() { + if s.buffer.Len() > 0 { + s.flushBuffer() + } +} + +// flushBuffer 将缓冲区内容发送到 channel 并清空。 +func (s *Splitter) flushBuffer() { + text := strings.TrimSpace(s.buffer.String()) + if text != "" { + s.ch <- text + } + s.buffer.Reset() +}