Merge pull request '实现 AI Orchestrator（核心编排）' (#37) from feature/phase4 into develop

Reviewed-on: http://8.161.227.145:3000/XEngineers/CamTalk/pulls/37

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

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=

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"
+}

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
+}

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)
+}

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
+}

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()
+}