|
|
@@ -23,9 +23,9 @@ from http_pool import http_pool # 🔥 使用连接池管理器
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# 🔥 TTS API选择
|
|
|
-# 注意: SpeechSynthesizer (tts_v2) 使用callback模式,不适合我们的generator流式场景
|
|
|
-# MultiModalConversation 已经是最适合流式语音合成的API
|
|
|
-USE_NEW_TTS_API = False # 保持使用 MultiModalConversation
|
|
|
+# 使用 SpeechSynthesizer (tts_v2) 支持 cosyvoice-v3-flash 模型
|
|
|
+# 使用 streaming_call 实现真正的流式传输
|
|
|
+from dashscope.audio.tts_v2 import SpeechSynthesizer, ResultCallback
|
|
|
|
|
|
# 设置API Key和endpoint(根据国际版/国内版自动选择)
|
|
|
dashscope.api_key = settings.active_dashscope_api_key
|
|
|
@@ -89,16 +89,17 @@ class AIService:
|
|
|
try:
|
|
|
start_time = time.time()
|
|
|
|
|
|
- response = dashscope.MultiModalConversation.call(
|
|
|
+ # 使用 SpeechSynthesizer 预热(同步调用即可)
|
|
|
+ synthesizer = SpeechSynthesizer(
|
|
|
model=settings.tts_model,
|
|
|
- text="Hi", # 极短的测试文本
|
|
|
voice=settings.tts_voice,
|
|
|
- stream=True,
|
|
|
+ speech_rate=settings.tts_speech_rate,
|
|
|
+ pitch_rate=settings.tts_pitch_rate,
|
|
|
+ language_hints=settings.tts_language_hints,
|
|
|
)
|
|
|
|
|
|
- # 只接收首块即停止
|
|
|
- for _ in response:
|
|
|
- break
|
|
|
+ # 发送极短文本进行预热(使用同步调用)
|
|
|
+ audio = synthesizer.call("Hi")
|
|
|
|
|
|
AIService._tts_warmed_up = True # 标记为已预热
|
|
|
|
|
|
@@ -465,104 +466,223 @@ class AIService:
|
|
|
) -> AsyncGenerator[bytes, None]:
|
|
|
"""
|
|
|
TTS流式调用的内部实现(不含重试逻辑)
|
|
|
+ 使用 SpeechSynthesizer.streaming_call() 实现真正的流式传输
|
|
|
|
|
|
Args:
|
|
|
text: 要合成的文本
|
|
|
emotion: 情感类型
|
|
|
|
|
|
Yields:
|
|
|
- audio_data: PCM 音频数据块(Base64 解码后的原始数据)
|
|
|
+ audio_data: PCM 音频数据块(原始音频数据)
|
|
|
"""
|
|
|
try:
|
|
|
import time
|
|
|
tts_start = time.time()
|
|
|
|
|
|
- # 根据情感选择音色
|
|
|
+ # 根据情感选择音色和构建 instruction
|
|
|
voice = self._get_voice_for_emotion(emotion)
|
|
|
- # 🔍 调试:打印实际使用的语速配置
|
|
|
- logger.info(f"🔍 TTS调用 - 使用语速 speech_rate={settings.tts_speech_rate}, 音色={voice}")
|
|
|
- # TODO: 添加语速调整 、测试语速调整是否有效
|
|
|
- def sync_streaming_tts():
|
|
|
- return dashscope.MultiModalConversation.call(
|
|
|
- model=settings.tts_model,
|
|
|
- text=text,
|
|
|
- voice=voice,
|
|
|
- speech_rate=settings.tts_speech_rate, # 语速调整
|
|
|
- #speech_rate=0.9,
|
|
|
- stream=True, # 流式输出
|
|
|
- # 注意: incremental_output 参数在此API中可能不生效
|
|
|
+ instruction = self._build_emotion_instruction(emotion)
|
|
|
+
|
|
|
+ # 🔍 调试:打印实际使用的配置
|
|
|
+ logger.info(
|
|
|
+ f"🔍 TTS调用 - 模型={settings.tts_model}, "
|
|
|
+ f"音色={voice}, "
|
|
|
+ f"语速={settings.tts_speech_rate}, "
|
|
|
+ f"音调={settings.tts_pitch_rate}, "
|
|
|
+ f"语言={settings.tts_language_hints}, "
|
|
|
+ f"情感={emotion}"
|
|
|
)
|
|
|
|
|
|
- # 🔥 修复:使用队列在线程和异步之间传递数据
|
|
|
+ # 🔥 使用队列在线程和异步之间传递数据
|
|
|
import queue
|
|
|
import threading
|
|
|
+ from pydub import AudioSegment
|
|
|
+ import io
|
|
|
|
|
|
audio_queue = queue.Queue()
|
|
|
error_holder = [None] # 用列表包装以便在线程间共享
|
|
|
+ completion_event = threading.Event() # 用于等待流式完成
|
|
|
|
|
|
- def stream_producer():
|
|
|
- """在独立线程中执行同步流式API调用"""
|
|
|
- nonlocal chunk_count, first_chunk_time, total_audio_size
|
|
|
- try:
|
|
|
- # 🔥 关键修复:在锁内完成整个API调用和迭代
|
|
|
- # 原因:DashScope SDK内部可能有全局状态,并发迭代会冲突
|
|
|
- with AIService._tts_call_lock:
|
|
|
- response = sync_streaming_tts()
|
|
|
+ # MP3 音频缓冲区(用于累积 MP3 数据,然后转换为 PCM)
|
|
|
+ mp3_buffer = io.BytesIO()
|
|
|
+
|
|
|
+ class TTSStreamCallback(ResultCallback):
|
|
|
+ """TTS流式回调处理器"""
|
|
|
+
|
|
|
+ def __init__(self, queue: queue.Queue, mp3_buffer: io.BytesIO,
|
|
|
+ error_holder: list, completion_event: threading.Event):
|
|
|
+ self.queue = queue
|
|
|
+ self.mp3_buffer = mp3_buffer
|
|
|
+ self.error_holder = error_holder
|
|
|
+ self.completion_event = completion_event
|
|
|
+ self.first_chunk_time = None
|
|
|
+ self.chunk_count = 0
|
|
|
+ self.total_audio_size = 0
|
|
|
+ self.mp3_min_size = 4096 # MP3最小解码大小(4KB),确保能解码一个完整帧
|
|
|
+
|
|
|
+ def on_open(self):
|
|
|
+ """连接打开"""
|
|
|
+ logger.debug("TTS流式连接已建立")
|
|
|
+ self.first_chunk_time = time.time()
|
|
|
+
|
|
|
+ def _process_mp3_buffer(self):
|
|
|
+ """处理累积的MP3数据,转换为PCM并发送"""
|
|
|
+ try:
|
|
|
+ buffer_size = self.mp3_buffer.tell()
|
|
|
+ if buffer_size < self.mp3_min_size:
|
|
|
+ return # 数据不足,继续累积
|
|
|
+
|
|
|
+ # 尝试解码MP3
|
|
|
+ self.mp3_buffer.seek(0)
|
|
|
+ audio_segment = AudioSegment.from_mp3(self.mp3_buffer)
|
|
|
+
|
|
|
+ # 转换为PCM格式:24kHz, 16bit, Mono
|
|
|
+ audio_segment = audio_segment.set_frame_rate(24000).set_channels(1).set_sample_width(2)
|
|
|
+ pcm_data = audio_segment.raw_data
|
|
|
|
|
|
- chunk_index = 0
|
|
|
- for chunk in response:
|
|
|
- chunk_index += 1
|
|
|
+ # 分块发送PCM数据(3200字节/块)
|
|
|
+ chunk_size = 3200
|
|
|
+ for i in range(0, len(pcm_data), chunk_size):
|
|
|
+ chunk = pcm_data[i:i + chunk_size]
|
|
|
+ self.chunk_count += 1
|
|
|
+ self.queue.put(('data', chunk))
|
|
|
+
|
|
|
+ # 清空缓冲区(已处理的数据)
|
|
|
+ self.mp3_buffer.seek(0)
|
|
|
+ self.mp3_buffer.truncate(0)
|
|
|
+
|
|
|
+ except FileNotFoundError as e:
|
|
|
+ # ffmpeg/ffprobe 未安装
|
|
|
+ error_msg = str(e)
|
|
|
+ if 'ffprobe' in error_msg or 'ffmpeg' in error_msg or 'avprobe' in error_msg:
|
|
|
+ logger.error(
|
|
|
+ f"[TTS错误] 系统缺少 ffmpeg,无法解码 MP3 音频。"
|
|
|
+ f"请在生产服务器安装 ffmpeg: apt-get install ffmpeg (Ubuntu/Debian) 或 yum install ffmpeg (CentOS/RHEL)"
|
|
|
+ )
|
|
|
+ self.error_holder[0] = Exception(
|
|
|
+ "系统缺少 ffmpeg,无法处理音频。请安装: apt-get install ffmpeg 或 yum install ffmpeg"
|
|
|
+ )
|
|
|
+ self.queue.put(('error', self.error_holder[0]))
|
|
|
+ return
|
|
|
+ raise
|
|
|
+ except Exception as e:
|
|
|
+ # MP3数据不完整或格式错误,继续累积
|
|
|
+ # 这是正常的,因为流式传输中MP3数据是分片到达的
|
|
|
+ # 但如果是其他错误,记录日志
|
|
|
+ error_str = str(e).lower()
|
|
|
+ if 'ffprobe' not in error_str and 'ffmpeg' not in error_str and 'avprobe' not in error_str:
|
|
|
+ # 不是 ffmpeg 相关错误,可能是数据不完整,继续累积
|
|
|
+ pass
|
|
|
+ else:
|
|
|
+ # ffmpeg 相关错误,记录并抛出
|
|
|
+ logger.error(f"[TTS错误] MP3解码失败: {e}")
|
|
|
+ raise
|
|
|
+
|
|
|
+ def on_data(self, data: bytes):
|
|
|
+ """接收音频数据(MP3格式)"""
|
|
|
+ try:
|
|
|
+ if data:
|
|
|
+ # 累积 MP3 数据
|
|
|
+ self.mp3_buffer.write(data)
|
|
|
+ self.total_audio_size += len(data)
|
|
|
|
|
|
- # 🔥 检查API错误(特别是速率限制)
|
|
|
- if hasattr(chunk, 'status_code') and chunk.status_code != 200:
|
|
|
- error_code = getattr(chunk, 'code', 'Unknown')
|
|
|
- error_msg = getattr(chunk, 'message', 'No error message')
|
|
|
-
|
|
|
- # 详细记录TTS API错误
|
|
|
- logger.error(
|
|
|
- f"[TTS异常] 错误类型=TTS API响应错误, "
|
|
|
- f"HTTP状态码={chunk.status_code}, "
|
|
|
- f"错误码={error_code}, "
|
|
|
- f"错误消息={error_msg}, "
|
|
|
- f"模型={settings.tts_model}, "
|
|
|
- f"音色={voice}, "
|
|
|
- f"文本长度={len(text)}, "
|
|
|
- f"chunk索引={chunk_index}, "
|
|
|
- f"完整chunk对象={chunk}"
|
|
|
- )
|
|
|
+ # 🔥 当累积足够数据时,尝试解码并发送PCM
|
|
|
+ # 策略:累积到最小解码大小(4KB)后尝试解码
|
|
|
+ self._process_mp3_buffer()
|
|
|
|
|
|
- # 构造友好的错误信息
|
|
|
- if chunk.status_code == 429 or 'Throttling' in error_code:
|
|
|
- raise Exception(f"TTS API速率限制: {error_msg} (code: {error_code})")
|
|
|
- else:
|
|
|
- raise Exception(f"TTS API错误 (status={chunk.status_code}): {error_msg} (code: {error_code})")
|
|
|
-
|
|
|
- if hasattr(chunk, 'output') and hasattr(chunk.output, 'audio'):
|
|
|
- audio = chunk.output.audio
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"[TTS回调异常] on_data错误: {e}")
|
|
|
+ self.error_holder[0] = e
|
|
|
+ self.queue.put(('error', e))
|
|
|
+
|
|
|
+ def on_complete(self):
|
|
|
+ """流式合成完成"""
|
|
|
+ try:
|
|
|
+ # 🔥 处理剩余的 MP3 数据
|
|
|
+ if self.mp3_buffer.tell() > 0:
|
|
|
+ self.mp3_buffer.seek(0)
|
|
|
+ try:
|
|
|
+ audio_segment = AudioSegment.from_mp3(self.mp3_buffer)
|
|
|
+ audio_segment = audio_segment.set_frame_rate(24000).set_channels(1).set_sample_width(2)
|
|
|
+ pcm_data = audio_segment.raw_data
|
|
|
|
|
|
- # 接收到音频数据块
|
|
|
- if audio.data is not None:
|
|
|
- # 记录首字节时间
|
|
|
- if first_chunk_time is None:
|
|
|
- first_chunk_time = time.time()
|
|
|
-
|
|
|
- # Base64 解码 PCM 数据
|
|
|
- pcm_bytes = base64.b64decode(audio.data)
|
|
|
- chunk_count += 1
|
|
|
- total_audio_size += len(pcm_bytes)
|
|
|
-
|
|
|
- # 放入队列
|
|
|
- audio_queue.put(('data', pcm_bytes))
|
|
|
+ # 发送剩余的PCM数据
|
|
|
+ chunk_size = 3200
|
|
|
+ for i in range(0, len(pcm_data), chunk_size):
|
|
|
+ chunk = pcm_data[i:i + chunk_size]
|
|
|
+ self.chunk_count += 1
|
|
|
+ self.queue.put(('data', chunk))
|
|
|
+ except FileNotFoundError as e:
|
|
|
+ # ffmpeg/ffprobe 未安装
|
|
|
+ error_msg = str(e)
|
|
|
+ if 'ffprobe' in error_msg or 'ffmpeg' in error_msg or 'avprobe' in error_msg:
|
|
|
+ logger.error(
|
|
|
+ f"[TTS错误] 系统缺少 ffmpeg,无法解码 MP3 音频。"
|
|
|
+ f"请在生产服务器安装 ffmpeg: apt-get install ffmpeg (Ubuntu/Debian) 或 yum install ffmpeg (CentOS/RHEL)"
|
|
|
+ )
|
|
|
+ self.error_holder[0] = Exception(
|
|
|
+ "系统缺少 ffmpeg,无法处理音频。请安装: apt-get install ffmpeg 或 yum install ffmpeg"
|
|
|
+ )
|
|
|
+ self.queue.put(('error', self.error_holder[0]))
|
|
|
+ return
|
|
|
+ raise
|
|
|
+ except Exception as e:
|
|
|
+ logger.warning(f"[TTS回调] 处理剩余数据失败: {e}")
|
|
|
+
|
|
|
+ logger.debug(f"TTS流式合成完成,共{self.chunk_count}个chunk,总大小{self.total_audio_size}字节")
|
|
|
+ self.queue.put(('done', None))
|
|
|
+ self.completion_event.set()
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logger.error(f"[TTS回调异常] on_complete错误: {e}")
|
|
|
+ self.error_holder[0] = e
|
|
|
+ self.queue.put(('error', e))
|
|
|
+
|
|
|
+ def on_error(self, message: str):
|
|
|
+ """发生错误"""
|
|
|
+ error = Exception(f"TTS流式合成错误: {message}")
|
|
|
+ logger.error(f"[TTS回调错误] {message}")
|
|
|
+ self.error_holder[0] = error
|
|
|
+ self.queue.put(('error', error))
|
|
|
+ self.completion_event.set()
|
|
|
+
|
|
|
+ def on_close(self):
|
|
|
+ """连接关闭"""
|
|
|
+ logger.debug("TTS流式连接已关闭")
|
|
|
+
|
|
|
+ def stream_producer():
|
|
|
+ """在独立线程中执行流式TTS API调用"""
|
|
|
+ nonlocal chunk_count, first_chunk_time, total_audio_size
|
|
|
+ try:
|
|
|
+ # 🔥 关键修复:在锁内完成整个API调用
|
|
|
+ with AIService._tts_call_lock:
|
|
|
+ # 创建回调实例
|
|
|
+ callback = TTSStreamCallback(audio_queue, mp3_buffer, error_holder, completion_event)
|
|
|
|
|
|
- # 检查是否结束
|
|
|
- if hasattr(chunk.output, 'finish_reason') and chunk.output.finish_reason == "stop":
|
|
|
- tts_end = time.time()
|
|
|
- total_duration = (tts_end - tts_start) * 1000
|
|
|
-
|
|
|
- break
|
|
|
-
|
|
|
- # 标记完成
|
|
|
- audio_queue.put(('done', None))
|
|
|
+ # 创建 SpeechSynthesizer 实例(使用回调)
|
|
|
+ # 🔥 注意:instruction 参数可能在某些情况下导致错误,先尝试不传
|
|
|
+ synthesizer_params = {
|
|
|
+ "model": settings.tts_model,
|
|
|
+ "voice": voice,
|
|
|
+ "speech_rate": settings.tts_speech_rate,
|
|
|
+ "pitch_rate": settings.tts_pitch_rate,
|
|
|
+ "language_hints": settings.tts_language_hints,
|
|
|
+ "callback": callback,
|
|
|
+ }
|
|
|
+ # 只在 instruction 不为空时添加(避免参数错误)
|
|
|
+ if instruction and instruction.strip():
|
|
|
+ synthesizer_params["instruction"] = instruction
|
|
|
+
|
|
|
+ synthesizer = SpeechSynthesizer(**synthesizer_params)
|
|
|
+
|
|
|
+ # 🔥 使用流式调用(真正的流式传输)
|
|
|
+ synthesizer.streaming_call(text)
|
|
|
+
|
|
|
+ # 🔥 重要:必须调用 streaming_complete() 确保结尾部分被合成
|
|
|
+ synthesizer.streaming_complete()
|
|
|
+
|
|
|
+ # 等待回调完成
|
|
|
+ completion_event.wait(timeout=30)
|
|
|
|
|
|
except Exception as e:
|
|
|
logger.error(
|
|
|
@@ -572,11 +692,11 @@ class AIService:
|
|
|
f"模型={settings.tts_model}, "
|
|
|
f"音色={voice}, "
|
|
|
f"文本长度={len(text)}, "
|
|
|
- f"已接收chunk数={chunk_index}, "
|
|
|
f"堆栈跟踪:\n{traceback.format_exc()}"
|
|
|
)
|
|
|
error_holder[0] = e
|
|
|
audio_queue.put(('error', e))
|
|
|
+ completion_event.set()
|
|
|
|
|
|
# 启动生产者线程
|
|
|
chunk_count = 0
|
|
|
@@ -603,6 +723,9 @@ class AIService:
|
|
|
raise data
|
|
|
|
|
|
except queue.Empty:
|
|
|
+ # 检查是否已完成
|
|
|
+ if completion_event.is_set():
|
|
|
+ break
|
|
|
error_msg = "TTS API 超时(30秒无响应)"
|
|
|
logger.error(
|
|
|
f"[TTS异常] 错误类型=TTS API超时, "
|
|
|
@@ -626,9 +749,7 @@ class AIService:
|
|
|
f"模型={settings.tts_model}, "
|
|
|
f"音色={voice}, "
|
|
|
f"文本长度={len(text)}, "
|
|
|
- f"文本内容={text[:100]}... (前100字符), "
|
|
|
- f"chunk计数={chunk_count}, "
|
|
|
- f"总音频大小={total_audio_size} bytes"
|
|
|
+ f"文本内容={text[:100]}... (前100字符)"
|
|
|
)
|
|
|
raise Exception(error_msg)
|
|
|
|
|
|
@@ -735,15 +856,16 @@ class AIService:
|
|
|
self._healthcheck_tasks.add("tts")
|
|
|
# 执行简单的健康检查:发送一个极短的测试TTS请求
|
|
|
def health_check():
|
|
|
- response = dashscope.MultiModalConversation.call(
|
|
|
+ # 使用 SpeechSynthesizer 进行健康检查
|
|
|
+ synthesizer = SpeechSynthesizer(
|
|
|
model=settings.tts_model,
|
|
|
- text="Hi", # 极短的测试文本
|
|
|
voice=settings.tts_voice,
|
|
|
- stream=True,
|
|
|
+ speech_rate=settings.tts_speech_rate,
|
|
|
+ pitch_rate=settings.tts_pitch_rate,
|
|
|
+ language_hints=settings.tts_language_hints,
|
|
|
)
|
|
|
- # 只接收首块即停止
|
|
|
- for _ in response:
|
|
|
- break
|
|
|
+ # 发送极短文本进行测试
|
|
|
+ audio = synthesizer.call("Hi")
|
|
|
return True
|
|
|
|
|
|
await asyncio.to_thread(health_check)
|
|
|
@@ -800,6 +922,29 @@ class AIService:
|
|
|
# 所有情感统一使用配置文件中的音色
|
|
|
return settings.tts_voice
|
|
|
|
|
|
+ def _build_emotion_instruction(self, emotion: EmotionType) -> str:
|
|
|
+ """根据情感构建 instruction 参数(用于 cosyvoice-v3)"""
|
|
|
+ # 🔥 根据官方文档,instruction 格式应为英文,且情感名称需小写
|
|
|
+ # 暂时返回空字符串,避免参数错误(428),后续可根据需要启用
|
|
|
+ # emotion_map = {
|
|
|
+ # "neutral": "You are having a casual conversation, your emotion is neutral.",
|
|
|
+ # "happy": "You are having a casual conversation, your emotion is happy.",
|
|
|
+ # "sad": "You are having a casual conversation, your emotion is sad.",
|
|
|
+ # "excited": "You are having a casual conversation, your emotion is excited.",
|
|
|
+ # "angry": "You are having a casual conversation, your emotion is angry.",
|
|
|
+ # "surprised": "You are having a casual conversation, your emotion is surprised.",
|
|
|
+ # "thoughtful": "You are having a casual conversation, your emotion is thoughtful.",
|
|
|
+ # "encouraging": "You are having a casual conversation, your emotion is encouraging.",
|
|
|
+ # "confused": "You are having a casual conversation, your emotion is confused.",
|
|
|
+ # "worried": "You are having a casual conversation, your emotion is worried.",
|
|
|
+ # "satisfied": "You are having a casual conversation, your emotion is satisfied.",
|
|
|
+ # "friendly": "You are having a casual conversation, your emotion is friendly.",
|
|
|
+ # "sorry": "You are having a casual conversation, your emotion is sorry.",
|
|
|
+ # "fearful": "You are having a casual conversation, your emotion is fearful.",
|
|
|
+ # }
|
|
|
+ # return emotion_map.get(emotion, emotion_map["neutral"])
|
|
|
+ return "" # 暂时禁用 instruction,避免 428 错误
|
|
|
+
|
|
|
def _build_system_prompt(self) -> str:
|
|
|
"""构建System Prompt"""
|
|
|
learner_proficiency = settings.LEARNER_PROFICIENCY
|