cfy666
eb1b90445f
- 添加 msg2str 转换节点解决 ChatModel 输出 *schema.Message 与 Splitter 期望 string 的类型不匹配 - 将多模态图片内容从 system 消息移到 user 消息（DashScope API 仅支持 user/tool 角色的多模态内容） - 修复 Content 和 UserInputMultiContent 不能同时设置的问题 - Splitter 输出改为 StreamReader[string]（单句），TTS 改为 TransformableLambda 流式消费 - 修复 .env 中 PostgreSQL DSN 和 Redis ADDR 的 http:// 前缀问题
117 lines
2.9 KiB
Go
117 lines
2.9 KiB
Go
package eino
|
|
|
|
import (
|
|
"context"
|
|
"encoding/base64"
|
|
"io"
|
|
|
|
"github.com/cloudwego/eino/compose"
|
|
"github.com/cloudwego/eino/schema"
|
|
|
|
"github.com/hhs/camtalk/internal/ai/tts"
|
|
"github.com/hhs/camtalk/internal/logger"
|
|
"github.com/hhs/camtalk/internal/models"
|
|
)
|
|
|
|
// NewTTSLambda 创建 TTS Transform Lambda 节点。
|
|
// 输入: StreamReader[string]（句子流）→ 输出: StreamReader[struct{}]（结果流）
|
|
//
|
|
// 流式消费每个句子，调用 ttsService.SynthesizeStream() 合成，
|
|
// 逐 chunk 推送 tts_audio 到客户端。TTS 失败静默跳过。
|
|
func NewTTSLambda(ttsService tts.Service, ttsVoice string, ttsSpeed float64, ttsOutputFmt string, ttsSampleRate int) *compose.Lambda {
|
|
return compose.TransformableLambda(func(ctx context.Context, input *schema.StreamReader[string]) (*schema.StreamReader[struct{}], error) {
|
|
sr, sw := schema.Pipe[struct{}](8)
|
|
|
|
go func() {
|
|
defer sw.Close()
|
|
defer input.Close()
|
|
|
|
log := logger.Log
|
|
sender := senderFromCtx(ctx)
|
|
requestID := requestIDFromCtx(ctx)
|
|
|
|
if sender == nil || requestID == "" {
|
|
// 消费并丢弃流
|
|
for {
|
|
_, err := input.Recv()
|
|
if err != nil {
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// 收集句子，按批次合成 TTS
|
|
var sentences []string
|
|
for {
|
|
sentence, err := input.Recv()
|
|
if err != nil {
|
|
if err == io.EOF {
|
|
break
|
|
}
|
|
log.Errorw("TTS: stream recv error", "error", err, "request_id", requestID)
|
|
break
|
|
}
|
|
if sentence != "" {
|
|
sentences = append(sentences, sentence)
|
|
}
|
|
}
|
|
|
|
if len(sentences) == 0 {
|
|
sw.Send(struct{}{}, nil)
|
|
return
|
|
}
|
|
|
|
log.Infow("开始 TTS 合成", "request_id", requestID, "sentence_count", len(sentences))
|
|
|
|
// 将句子数组转为 channel
|
|
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)
|
|
sw.Send(struct{}{}, nil)
|
|
return
|
|
}
|
|
|
|
// 消费 TTS 音频流，推送到客户端
|
|
for chunk := range ttsStream {
|
|
select {
|
|
case <-ctx.Done():
|
|
log.Infow("TTS 流被中断", "request_id", requestID)
|
|
sw.Send(struct{}{}, ctx.Err())
|
|
return
|
|
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)
|
|
sw.Send(struct{}{}, nil)
|
|
}()
|
|
|
|
return sr, nil
|
|
})
|
|
}
|