feat: 实现 AI Orchestrator（核心编排）
- 实现 Orchestrator 接口（4.1） - 实现 Sender 接口（4.2） - 实现 STT → LLM → TTS 流式并行管道（4.3） - 实现句子切分器（4.4） - 实现错误降级处理（4.5） - 实现 Interrupt 支持（4.6） - 编写完整的单元测试（4.7）
This commit is contained in:
325
backend/internal/orchestrator/pipeline.go
Normal file
325
backend/internal/orchestrator/pipeline.go
Normal file
@@ -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
|
||||
}
|
||||
Reference in New Issue
Block a user