cfy666
fd5c7712f8
- 引入 cloudwego/eino v0.9.9 和 eino-ext/components/model/openai v0.1.13 - 新增 internal/eino/ 包： - types.go: PipelineInput/Output、STTOutput、TokenUsage 类型定义 - state.go: PipelineState 跨节点状态收集（线程安全） - callback.go: ChatModel OnEndWithStreamOutput 回调，逐 token 推送 llm_chunk - nodes_stt.go: STT Lambda，支持文本/语音输入模式 - nodes_history.go: 历史组装 Lambda，含多模态图片支持 - nodes_splitter.go: 句子分割 Transform Lambda - nodes_tts.go: TTS Lambda，逐句合成推送音频 - nodes_done.go: Done Lambda，发送 llm_done 并追加历史 Co-Authored-By: Claude <noreply@anthropic.com>
93 lines
2.6 KiB
Go
93 lines
2.6 KiB
Go
package eino
|
||
|
||
import (
|
||
"context"
|
||
"encoding/base64"
|
||
|
||
"github.com/cloudwego/eino/compose"
|
||
|
||
"github.com/hhs/camtalk/internal/ai/tts"
|
||
"github.com/hhs/camtalk/internal/logger"
|
||
"github.com/hhs/camtalk/internal/models"
|
||
)
|
||
|
||
// NewTTSLambda 创建 TTS Lambda 节点。
|
||
// 输入: []string（句子数组，框架自动从 StreamReader concat）→ 输出: struct{}
|
||
//
|
||
// 将句子数组转为 channel，调用 ttsService.SynthesizeStream() 流式合成，
|
||
// 逐 chunk 推送 tts_audio 到客户端。TTS 失败静默跳过。
|
||
func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, ttsOutputFmt string, ttsSampleRate int) *compose.Lambda {
|
||
return compose.InvokableLambda(func(ctx context.Context, sentences []string) (struct{}, error) {
|
||
log := logger.Log
|
||
sender := senderFromCtx(ctx)
|
||
requestID := requestIDFromCtx(ctx)
|
||
state := stateFromCtx(ctx)
|
||
|
||
// 检查 TTS 是否启用（从 State 或 context 获取）
|
||
// TTSEnabled 信息在 PipelineInput 中，通过 State 传递
|
||
if state != nil {
|
||
state.mu.Lock()
|
||
ttsEnabled := true // 默认启用，由适配器通过 State 设置
|
||
state.mu.Unlock()
|
||
if !ttsEnabled {
|
||
return struct{}{}, nil
|
||
}
|
||
}
|
||
|
||
if len(sentences) == 0 {
|
||
return struct{}{}, nil
|
||
}
|
||
|
||
if sender == nil || requestID == "" {
|
||
return struct{}{}, nil
|
||
}
|
||
|
||
log.Infow("开始 TTS 合成", "request_id", requestID, "sentence_count", len(sentences))
|
||
|
||
// 将句子数组转为 channel（ttsService.SynthesizeStream 需要 <-chan string）
|
||
sentenceCh := make(chan string, len(sentences))
|
||
for _, s := range sentences {
|
||
sentenceCh <- s
|
||
}
|
||
close(sentenceCh)
|
||
|
||
// 调用 TTS 服务
|
||
ttsStream, err := ttsService.SynthesizeStream(ctx, sentenceCh, tts.Options{
|
||
Voice: ttsVoice,
|
||
Speed: ttsSpeed,
|
||
OutputFmt: ttsOutputFmt,
|
||
SampleRate: ttsSampleRate,
|
||
})
|
||
if err != nil {
|
||
log.Errorw("TTS 合成启动失败（已跳过）", "error", err, "request_id", requestID)
|
||
return struct{}{}, nil // TTS 失败不中断流程
|
||
}
|
||
|
||
// 消费 TTS 音频流，推送到客户端
|
||
for chunk := range ttsStream {
|
||
select {
|
||
case <-ctx.Done():
|
||
log.Infow("TTS 流被中断", "request_id", requestID)
|
||
return struct{}{}, ctx.Err()
|
||
default:
|
||
}
|
||
|
||
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,
|
||
Final: chunk.Final,
|
||
}); err != nil {
|
||
log.Errorw("发送 tts_audio 失败", "error", err)
|
||
}
|
||
}
|
||
|
||
log.Infow("TTS 合成完成", "request_id", requestID)
|
||
return struct{}{}, nil
|
||
})
|
||
}
|