329 lines
7.6 KiB
Go
329 lines
7.6 KiB
Go
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
|
||
model string // LLM 模型名，用于 llm_done 上报
|
||
}
|
||
|
||
// New 创建 Pipeline 实例。
|
||
func New(
|
||
sttService stt.Service,
|
||
llmService llm.Service,
|
||
ttsService tts.Service,
|
||
sessionMgr session.Manager,
|
||
model string,
|
||
) *Pipeline {
|
||
return &Pipeline{
|
||
sttService: sttService,
|
||
llmService: llmService,
|
||
ttsService: ttsService,
|
||
sessionMgr: sessionMgr,
|
||
model: model,
|
||
}
|
||
}
|
||
|
||
// 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: p.model,
|
||
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
|
||
}
|