update realtime translate demo

ax_speech_translate_demo_qwen_api_realtime.py

@@ -0,0 +1,991 @@
+# import subprocess
+import tempfile
+import os
+# import json
+# import shutil
+import time
+import librosa
+import torch
+import argparse
+import soundfile as sf
+# from pathlib import Path
+import cn2an
+import requests
+import re
+import numpy as np
+import onnxruntime as ort
+import axengine as axe
+import threading
+import queue
+from collections import deque
+
+# 导入SenseVoice相关模块
+from model import SinusoidalPositionEncoder
+from utils.ax_model_bin import AX_SenseVoiceSmall
+from utils.ax_vad_bin import AX_Fsmn_vad 
+from utils.vad_utils import merge_vad
+from funasr.tokenizer.sentencepiece_tokenizer import SentencepiecesTokenizer
+
+# 导入MeloTTS相关模块
+from libmelotts.python.split_utils import split_sentence
+from libmelotts.python.text import cleaned_text_to_sequence
+from libmelotts.python.text.cleaner import clean_text
+from libmelotts.python.symbols import LANG_TO_SYMBOL_MAP
+
+# 配置参数
+# tts 参数
+TTS_MODEL_DIR = "libmelotts/models"
+TTS_MODEL_FILES = {
+    "g": "g-zh_mix_en.bin",
+    "encoder": "encoder-zh.onnx",
+    "decoder": "decoder-zh.axmodel"
+}
+
+# Qwen大模型翻译API参数
+QWEN_API_URL = "http://10.126.29.158:8000"  # API服务地址
+
+# TTS辅助函数
+def intersperse(lst, item):
+    result = [item] * (len(lst) * 2 + 1)
+    result[1::2] = lst
+    return result
+
+# def get_text_for_tts_infer(text, language_str, symbol_to_id=None):
+#     norm_text, phone, tone, word2ph = clean_text(text, language_str)
+#     phone, tone, language = cleaned_text_to_sequence(phone, tone, language_str, symbol_to_id)
+
+#     phone = intersperse(phone, 0)
+#     tone = intersperse(tone, 0)
+#     language = intersperse(language, 0)
+
+#     phone = np.array(phone, dtype=np.int32)
+#     tone = np.array(tone, dtype=np.int32)
+#     language = np.array(language, dtype=np.int32)
+#     word2ph = np.array(word2ph, dtype=np.int32) * 2
+#     word2ph[0] += 1
+
+#     return phone, tone, language, norm_text, word2ph
+
+# 处理字符无法不识别
+def get_text_for_tts_infer(text, language_str, symbol_to_id=None):
+    """修复版音素处理：确保所有数组长度一致"""
+    try:
+        norm_text, phone, tone, word2ph = clean_text(text, language_str)
+        
+        # 特殊音素直接映射为空字符串
+        phone_mapping = {
+            'ɛ': '', 'æ': '', 'ʌ': '', 'ʊ': '', 'ɔ': '', 'ɪ': '', 'ɝ': '', 'ɚ': '', 'ɑ': '',
+            'ʒ': '', 'θ': '', 'ð': '', 'ŋ': '', 'ʃ': '', 'ʧ': '', 'ʤ': '', 'ː': '', 'ˈ': '',
+            'ˌ': '', 'ʰ': '', 'ʲ': '', 'ʷ': '', 'ʔ': '', 'ɾ': '', 'ɹ': '', 'ɫ': '', 'ɡ': '',
+        }
+        
+        # 同步处理 phone 和 tone，确保它们长度一致
+        processed_phone = []
+        processed_tone = []
+        removed_symbols = set()
+        
+        for p, t in zip(phone, tone):
+            if p in phone_mapping:
+                # 特殊音素直接删除，同时删除对应的 tone
+                removed_symbols.add(p)
+            elif p in symbol_to_id:
+                # 正常音素保留，同时保留对应的 tone
+                processed_phone.append(p)
+                processed_tone.append(t)
+            else:
+                # 其他未知音素也删除
+                removed_symbols.add(p)
+        
+        # 记录被删除的音素
+        if removed_symbols:
+            print(f"[音素过滤] 删除了 {len(removed_symbols)} 个特殊音素: {sorted(removed_symbols)}")
+            print(f"[音素过滤] 处理后音素序列长度: {len(processed_phone)}")
+            print(f"[音素过滤] 处理后音调序列长度: {len(processed_tone)}")
+        
+        # 如果没有有效音素，使用默认音素，
+        if not processed_phone:
+            print("[警告] 没有有效音素，使用默认中文音素")
+            processed_phone = ['ni', 'hao']
+            processed_tone = ['1', '3']
+            word2ph = [1, 1]
+        
+        # 确保 word2ph 的长度与处理后的音素序列匹配
+        if len(processed_phone) != len(phone):
+            print(f"[警告] 音素序列长度变化: {len(phone)} -> {len(processed_phone)}")
+            # 简单处理：重新计算 word2ph
+            word2ph = [1] * len(processed_phone)
+        
+        phone, tone, language = cleaned_text_to_sequence(processed_phone, processed_tone, language_str, symbol_to_id)
+
+        phone = intersperse(phone, 0)
+        tone = intersperse(tone, 0)
+        language = intersperse(language, 0)
+
+        phone = np.array(phone, dtype=np.int32)
+        tone = np.array(tone, dtype=np.int32)
+        language = np.array(language, dtype=np.int32)
+        word2ph = np.array(word2ph, dtype=np.int32) * 2
+        word2ph[0] += 1
+
+        return phone, tone, language, norm_text, word2ph
+        
+    except Exception as e:
+        print(f"[错误] 文本处理失败: {e}")
+        import traceback
+        traceback.print_exc()
+        raise e
+
+def audio_numpy_concat(segment_data_list, sr, speed=1.):
+    audio_segments = []
+    for segment_data in segment_data_list:
+        audio_segments += segment_data.reshape(-1).tolist()
+        audio_segments += [0] * int((sr * 0.05) / speed)
+    audio_segments = np.array(audio_segments).astype(np.float32)
+    return audio_segments
+
+def merge_sub_audio(sub_audio_list, pad_size, audio_len):
+    if pad_size > 0:
+        for i in range(len(sub_audio_list) - 1):
+            sub_audio_list[i][-pad_size:] += sub_audio_list[i+1][:pad_size]
+            sub_audio_list[i][-pad_size:] /= 2
+            if i > 0:
+                sub_audio_list[i] = sub_audio_list[i][pad_size:]
+
+    sub_audio = np.concatenate(sub_audio_list, axis=-1)
+    return sub_audio[:audio_len]
+
+def calc_word2pronoun(word2ph, pronoun_lens):
+    indice = [0]
+    for ph in word2ph[:-1]:
+        indice.append(indice[-1] + ph)
+    word2pronoun = []
+    for i, ph in zip(indice, word2ph):
+        word2pronoun.append(np.sum(pronoun_lens[i : i + ph]))
+    return word2pronoun
+
+def generate_slices(word2pronoun, dec_len):
+    pn_start, pn_end = 0, 0
+    zp_start, zp_end = 0, 0
+    zp_len = 0
+    pn_slices = []
+    zp_slices = []
+    while pn_end < len(word2pronoun):
+        if pn_end - pn_start > 2 and np.sum(word2pronoun[pn_end - 2 : pn_end + 1]) <= dec_len:
+            zp_len = np.sum(word2pronoun[pn_end - 2 : pn_end])
+            zp_start = zp_end - zp_len
+            pn_start = pn_end - 2
+        else:
+            zp_len = 0
+            zp_start = zp_end
+            pn_start = pn_end
+            
+        while pn_end < len(word2pronoun) and zp_len + word2pronoun[pn_end] <= dec_len:
+            zp_len += word2pronoun[pn_end]
+            pn_end += 1
+        zp_end = zp_start + zp_len
+        pn_slices.append(slice(pn_start, pn_end))
+        zp_slices.append(slice(zp_start, zp_end))
+    return pn_slices, zp_slices
+
+# 确认中英文
+def lang_detect_with_regex(text):
+    text_without_digits = re.sub(r'\d+', '', text)
+    
+    if not text_without_digits:
+        return 'unknown'
+    
+    if re.search(r'[\u4e00-\u9fff]', text_without_digits):
+        return 'chinese'
+    else:
+        if re.search(r'[a-zA-Z]', text_without_digits):
+            return 'english'
+        else:
+            return 'unknown'
+
+class QwenTranslationAPI:
+    def __init__(self, api_url=QWEN_API_URL):
+        self.api_url = api_url
+        self.session_id = f"speech_translate_{int(time.time())}"
+        self.last_reset_time = time.time()
+        self.request_count = 0
+        self.max_requests_before_reset = 10
+        
+    def reset_context(self):
+        """重置API上下文"""
+        try:
+            reset_url = f"{self.api_url}/api/reset"
+            response = requests.post(reset_url, json={}, timeout=5)
+            if response.status_code == 200:
+                print("[翻译API] ✓ 上下文重置成功")
+                self.last_reset_time = time.time()
+                self.request_count = 0
+                return True
+            else:
+                print(f"[翻译API] ✗ 重置失败，状态码: {response.status_code}, 响应: {response.text}")
+        except Exception as e:
+            print(f"[翻译API] ✗ 重置上下文失败: {e}")
+        return False
+
+    def check_and_reset_if_needed(self):
+        """检查是否需要重置上下文"""
+        current_time = time.time()
+        if (self.request_count >= 10 or 
+            current_time - self.last_reset_time > 120):  # 2分钟
+            print(f"[翻译API] 重置 (请求数: {self.request_count}, 时间: {current_time - self.last_reset_time:.1f}秒)")
+            return self.reset_context()
+        return True
+        
+    def translate(self, text_content, max_retries=3, timeout=120):
+        if not text_content or text_content.strip() == "":
+            return "输入文本为空"
+        
+        # 过滤太短的文本
+        if len(text_content.strip()) < 3:
+            return text_content
+            
+        if lang_detect_with_regex(text_content)=='chinese':
+            prompt_f = "翻译成英文"
+        else:
+            prompt_f= "翻译成中文"
+
+        prompt = f"{prompt_f}：{text_content}"
+        print(f"[翻译API] 发送请求: {prompt}")
+        
+        # 检查是否需要重置
+        self.check_and_reset_if_needed()
+        
+        for attempt in range(max_retries):
+            try:
+                generate_url = f"{self.api_url}/api/generate"
+                payload = {
+                    "prompt": prompt,
+                    "temperature": 0.1,
+                    "repetition_penalty": 1.0,
+                    "top-p": 0.9,
+                    "top-k": 40,
+                    "max_new_tokens": 512
+                }
+                
+                print(f"[翻译API] 开始生成请求 (尝试 {attempt + 1}/{max_retries})")
+                response = requests.post(generate_url, json=payload, timeout=30)
+                response.raise_for_status()
+                print("[翻译API] 生成请求成功")
+                
+                result_url = f"{self.api_url}/api/generate_provider"
+                start_time = time.time()
+                full_translation = ""
+                error_detected = False
+                
+                while time.time() - start_time < timeout:
+                    try:
+                        result_response = requests.get(result_url, timeout=10)
+                        result_data = result_response.json()
+                        
+                        current_chunk = result_data.get("response", "")
+                        
+                        # 检查是否有错误
+                        if "error:" in current_chunk.lower() or "setkvcache failed" in current_chunk.lower():
+                            print(f"[翻译API] ✗ 检测到错误: {current_chunk}")
+                            error_detected = True
+                            print("[翻译API] 立即重置上下文...")
+                            self.reset_context()
+                            break
+                        
+                        full_translation += current_chunk
+                        
+                        if result_data.get("done", False):
+                            if full_translation and len(full_translation.strip()) > 0:
+                                self.request_count += 1
+                                print(f"[翻译API] ✓ 翻译完成: {full_translation}")
+                                return full_translation
+                            else:
+                                print(f"[翻译API] ✗ 翻译结果为空")
+                                break
+                        
+                        time.sleep(0.05)
+                        
+                    except requests.exceptions.RequestException as e:
+                        print(f"[翻译API] 轮询请求失败: {e}")
+                        if time.time() - start_time > timeout:
+                            break
+                        time.sleep(0.5)
+                        continue
+                
+                if error_detected:
+                    if attempt < max_retries - 1:
+                        wait_time = 1
+                        print(f"[翻译API] 等待 {wait_time} 秒后重试...")
+                        time.sleep(wait_time)
+                        continue
+                    else:
+                        print("[翻译API] 达到最大重试次数，返回原文")
+                        return text_content
+                
+                print(f"[翻译API] 轮询超时，尝试第 {attempt + 1} 次重试")
+                
+            except requests.exceptions.RequestException as e:
+                print(f"[翻译API] 请求失败 (尝试 {attempt + 1}/{max_retries}): {e}")
+                if attempt < max_retries - 1:
+                    wait_time = 2 ** attempt
+                    print(f"[翻译API] 等待 {wait_time} 秒后重试...")
+                    time.sleep(wait_time)
+                else:
+                    return text_content
+            except Exception as e:
+                print(f"[翻译API] 翻译过程出错: {e}")
+                if attempt < max_retries - 1:
+                    time.sleep(1)
+                    continue
+                return text_content
+        
+        print("[翻译API] 翻译超时，返回原文")
+        return text_content
+
+class AudioResampler:
+    """音频重采样器"""
+    def __init__(self, target_sr=16000):
+        self.target_sr = target_sr
+        
+    def resample_audio(self, audio_data, original_sr):
+        """重采样音频到目标采样率，asr统一输入16000Hz"""
+        if original_sr == self.target_sr:
+            return audio_data
+            
+        print(f"[重采样] {original_sr}Hz -> {self.target_sr}Hz")
+        return librosa.resample(y=audio_data, orig_sr=original_sr, target_sr=self.target_sr)
+    
+    def resample_chunk(self, audio_chunk, original_sr):
+        """重采样音频块：长音频进行过冲采样后，音频块可以不做重采样"""
+        if original_sr == self.target_sr:
+            return audio_chunk
+            
+        if len(audio_chunk) < 1000:
+            return self._linear_resample(audio_chunk, original_sr, self.target_sr)
+        else:
+            return librosa.resample(y=audio_chunk, orig_sr=original_sr, target_sr=self.target_sr)
+    
+    def _linear_resample(self, audio_chunk, original_sr, target_sr):
+        """线性插值重采样"""
+        ratio = target_sr / original_sr
+        old_length = len(audio_chunk)
+        new_length = int(old_length * ratio)
+        
+        old_indices = np.arange(old_length)
+        new_indices = np.linspace(0, old_length - 1, new_length)
+        
+        resampled = np.interp(new_indices, old_indices, audio_chunk)
+        return resampled
+
+class StreamProcessor:
+    """流式处理"""
+    def __init__(self, pipeline, chunk_duration=7.0, overlap_duration=0.01, target_sr=16000):
+        self.pipeline = pipeline
+        self.chunk_duration = chunk_duration  # 增加4->7秒
+        self.overlap_duration = overlap_duration  # 减少到0.1->0.01秒
+        self.target_sr = target_sr
+        self.chunk_samples = int(chunk_duration * target_sr)
+        self.overlap_samples = int(overlap_duration * target_sr)
+        self.audio_buffer = deque()
+        self.result_queue = queue.Queue()
+        self.is_running = False
+        self.processing_thread = None
+        self.resampler = AudioResampler(target_sr=target_sr)
+        self.segment_counter = 0  # 音频段计数器
+        self.processed_texts = set()  # 记录已处理的文本，避免重复
+        
+    def start_processing(self):
+        """开始流式处理"""
+        self.is_running = True
+        self.processing_thread = threading.Thread(target=self._process_loop)
+        self.processing_thread.daemon = True
+        self.processing_thread.start()
+        
+    def stop_processing(self):
+        """停止流式处理"""
+        self.is_running = False
+        if self.processing_thread:
+            self.processing_thread.join(timeout=5)
+            
+    def add_audio_chunk(self, audio_chunk, original_sr=None):
+        """添加音频块到缓冲区"""
+        if original_sr and original_sr != self.target_sr:
+            audio_chunk = self.resampler.resample_chunk(audio_chunk, original_sr)
+        
+        self.audio_buffer.append(audio_chunk)
+        
+    def get_next_result(self, timeout=1.0):
+        """获取下一个处理结果"""
+        try:
+            return self.result_queue.get(timeout=timeout)
+        except queue.Empty:
+            return None
+            
+    def _process_loop(self):
+        """处理循环"""
+        accumulated_audio = np.array([], dtype=np.float32)
+        last_asr_result = ""  # 记录上一次的ASR结果，防止重复处理
+        
+        while self.is_running:
+            if len(self.audio_buffer) > 0:
+                audio_chunk = self.audio_buffer.popleft()
+                accumulated_audio = np.concatenate([accumulated_audio, audio_chunk])
+                
+                # 当积累的音频足够处理时
+                if len(accumulated_audio) >= self.chunk_samples:
+                    # 提取处理块（减少重叠）
+                    process_chunk = accumulated_audio[:self.chunk_samples]
+                    accumulated_audio = accumulated_audio[self.chunk_samples - self.overlap_samples:]
+                    
+                    try:
+                        # 实时ASR识别
+                        asr_result = self._stream_asr(process_chunk)
+                        
+                        # 过滤条件：
+                        # # 1. 文本有效且足够长
+                        # 2. 与上次结果不同（避免重复）
+                        # 3. 不是已处理过的文本
+                        if (asr_result and asr_result.strip() and 
+                            # len(asr_result.strip()) >= 5 and
+                            asr_result != last_asr_result and
+                            asr_result not in self.processed_texts):
+                            
+                            print(f"[实时ASR] {asr_result}")
+                            last_asr_result = asr_result
+                            self.processed_texts.add(asr_result)
+                            
+                            # 实时翻译
+                            try:
+                                translation_result = self.pipeline.run_translation(asr_result)
+                                
+                                # 检查翻译结果是否有效
+                                if (translation_result and 
+                                    translation_result != asr_result and
+                                    "翻译失败" not in translation_result and 
+                                    "error:" not in translation_result.lower() and
+                                    "输入文本为空" not in translation_result):
+                                    
+                                    print(f"[实时翻译] {translation_result}")
+                                    
+                                    # TTS合成
+                                    try:
+                                        self.segment_counter += 1
+                                        tts_filename = f"stream_segment_{self.segment_counter:04d}.wav"
+                                        tts_start_time = time.time()
+                                        
+                                        tts_path = self.pipeline.run_tts(
+                                            translation_result, 
+                                            self.pipeline.output_dir, 
+                                            tts_filename
+                                        )
+                                        
+                                        tts_time = time.time() - tts_start_time
+                                        print(f"[实时TTS] 音频已保存: {tts_path} (耗时: {tts_time:.2f}秒)")
+                                        
+                                        # 将完整结果放入队列
+                                        self.result_queue.put({
+                                            'type': 'complete',
+                                            'original': asr_result,
+                                            'translated': translation_result,
+                                            'audio_path': tts_path,
+                                            'timestamp': time.time(),
+                                            'segment_id': self.segment_counter
+                                        })
+                                        
+                                    except Exception as tts_error:
+                                        print(f"[实时TTS错误] {tts_error}")
+                                        import traceback
+                                        traceback.print_exc()
+                                else:
+                                    print(f"[实时翻译] 翻译结果无效，已跳过")
+                                    
+                            except Exception as translation_error:
+                                print(f"[实时翻译错误] {translation_error}")
+                        else:
+                            if asr_result == last_asr_result:
+                                print(f"[实时ASR] 重复内容已跳过: {asr_result}")
+                                
+                    except Exception as e:
+                        print(f"[流式处理错误] {e}")
+                        import traceback
+                        traceback.print_exc()
+                        
+            time.sleep(0.01)
+            
+    def _stream_asr(self, audio_chunk):
+        """流式ASR识别（带VAD）"""
+        try:
+            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+            # 步骤1: VAD检测 - 过滤静音段
+            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+            res_vad = self.pipeline.model_vad(audio_chunk)[0]
+            vad_segments = merge_vad(res_vad, 15 * 1000)
+            
+            # 如果没有检测到语音段，直接返回空
+            if not vad_segments or len(vad_segments) == 0:
+                print(f"[VAD] 未检测到语音活动，跳过此音频块")
+                return ""
+            
+            print(f"[VAD] 检测到 {len(vad_segments)} 个语音段")
+            
+            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+            # 步骤2: 对检测到的语音段进行ASR识别
+            # ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
+            all_results = ""
+            
+            for i, segment in enumerate(vad_segments):
+                segment_start, segment_end = segment
+                start_sample = int(segment_start / 1000 * self.target_sr)
+                end_sample = min(int(segment_end / 1000 * self.target_sr), len(audio_chunk))
+                segment_audio = audio_chunk[start_sample:end_sample]
+                
+                # 跳过太短的片段，减少误识别（小于0.3秒）
+                if len(segment_audio) < int(0.3 * self.target_sr):
+                    continue
+                
+                # 写入临时文件
+                with tempfile.NamedTemporaryFile(suffix='.wav', delete=False) as temp_file:
+                    sf.write(temp_file.name, segment_audio, self.target_sr)
+                    temp_filename = temp_file.name
+                
+                try:
+                    # ASR识别
+                    segment_result = self.pipeline.model_bin(
+                        temp_filename, 
+                        "auto",
+                        True,
+                        self.pipeline.position_encoding, 
+                        tokenizer=self.pipeline.tokenizer,
+                    )
+                    
+                    if segment_result and segment_result.strip():
+                        all_results += segment_result + " "
+                    
+                    # 清理临时文件
+                    os.unlink(temp_filename)
+                    
+                except Exception as e:
+                    print(f"[ASR错误] 处理VAD段 {i} 时出错: {e}")
+                    if os.path.exists(temp_filename):
+                        os.unlink(temp_filename)
+                    continue
+            
+            return all_results.strip()
+                
+        except Exception as e:
+            print(f"[ASR错误] {e}")
+            return ""
+
+class SpeechTranslationPipeline:
+    def __init__(self, 
+                 tts_model_dir, tts_model_files,
+                 asr_model_dir="ax_model", seq_len=132,
+                 tts_dec_len=128, sample_rate=44100, tts_speed=0.8,
+                 qwen_api_url=QWEN_API_URL, target_sr=16000,
+                 output_dir="./output"):
+        self.tts_model_dir = tts_model_dir
+        self.tts_model_files = tts_model_files
+        self.asr_model_dir = asr_model_dir
+        self.seq_len = seq_len
+        self.tts_dec_len = tts_dec_len
+        self.sample_rate = sample_rate
+        self.tts_speed = tts_speed
+        self.qwen_api_url = qwen_api_url
+        self.target_sr = target_sr
+        self.output_dir = output_dir
+        
+        # 输出目录
+        os.makedirs(self.output_dir, exist_ok=True)
+        
+        # 初始化音频重采样器
+        self.resampler = AudioResampler(target_sr=target_sr)
+        
+        # 初始化ASR模型
+        self._init_asr_models()
+        
+        # 初始化TTS模型
+        self._init_tts_models()
+        
+        # 初始化翻译API
+        self.translator = QwenTranslationAPI(api_url=qwen_api_url)
+        
+        # 初始化流式处理器
+        self.stream_processor = StreamProcessor(self, target_sr=target_sr)
+        
+        # 验证所有必需文件存在
+        self._validate_files()
+        
+        # 初始化时重置API上下文
+        print("[初始化] 重置API上下文...")
+        self.translator.reset_context()
+    
+    def _init_asr_models(self):
+        """初始化语音识别相关模型"""
+        print("Initializing SenseVoice models...")
+        
+        self.model_vad = AX_Fsmn_vad(self.asr_model_dir)
+        
+        self.embed = SinusoidalPositionEncoder()
+        self.position_encoding = self.embed.get_position_encoding(
+            torch.randn(1, self.seq_len, 560)).numpy()
+        
+        self.model_bin = AX_SenseVoiceSmall(self.asr_model_dir, seq_len=self.seq_len)
+        
+        tokenizer_path = os.path.join(self.asr_model_dir, "chn_jpn_yue_eng_ko_spectok.bpe.model")
+        self.tokenizer = SentencepiecesTokenizer(bpemodel=tokenizer_path)
+        
+        print("SenseVoice models initialized successfully.")
+    
+    def _init_tts_models(self):
+        """初始化TTS相关模型"""
+        print("Initializing MeloTTS models...")
+        init_start = time.time()
+        
+        enc_model = os.path.join(self.tts_model_dir, self.tts_model_files["encoder"])
+        dec_model = os.path.join(self.tts_model_dir, self.tts_model_files["decoder"])
+        
+        model_load_start = time.time()
+        self.sess_enc = ort.InferenceSession(enc_model, providers=["CPUExecutionProvider"], sess_options=ort.SessionOptions())
+        self.sess_dec = axe.InferenceSession(dec_model)
+        print(f"  Load encoder/decoder models: {(time.time() - model_load_start)*1000:.2f}ms")
+        
+        g_file = os.path.join(self.tts_model_dir, self.tts_model_files["g"])
+        self.tts_g = np.fromfile(g_file, dtype=np.float32).reshape(1, 256, 1)
+        
+        self.tts_language = "ZH_MIX_EN"
+        self.symbol_to_id = {s: i for i, s in enumerate(LANG_TO_SYMBOL_MAP[self.tts_language])}
+        
+        print("  Warming up TTS modules...")
+        warmup_start = time.time()
+        
+        try:
+            warmup_text_mix = "这是一个test测试。"
+            _, _, _, _, _ = get_text_for_tts_infer(warmup_text_mix, self.tts_language, symbol_to_id=self.symbol_to_id)
+            print(f"  Mixed ZH-EN warm-up: {(time.time() - warmup_start)*1000:.2f}ms")
+        except Exception as e:
+            print(f"  Warning: Mixed warm-up failed: {e}")
+        
+        total_init_time = (time.time() - init_start) * 1000
+        print(f"MeloTTS models initialized successfully. Total init time: {total_init_time:.2f}ms")
+    
+    def _validate_files(self):
+        """验证所有必需的文件都存在"""
+        for key, filename in self.tts_model_files.items():
+            filepath = os.path.join(self.tts_model_dir, filename)
+            if not os.path.exists(filepath):
+                raise FileNotFoundError(f"TTS模型文件不存在: {filepath}")
+        
+        try:
+            response = requests.get(f"{self.qwen_api_url}/api/generate_provider", timeout=5)
+            print("[API检查] 千问API服务连接正常")
+        except:
+            print("[API警告] 无法连接到千问API服务，请确保已启动API服务")
+
+    def start_stream_processing(self):
+        """开始流式处理"""
+        self.stream_processor.start_processing()
+        print("[流式处理] 已启动")
+        
+    def stop_stream_processing(self):
+        """停止流式处理"""
+        self.stream_processor.stop_processing()
+        print("[流式处理] 已停止")
+        
+    def process_audio_stream(self, audio_chunk, original_sr=None):
+        """处理音频流数据"""
+        self.stream_processor.add_audio_chunk(audio_chunk, original_sr)
+        
+    def get_stream_results(self):
+        """获取流式处理结果"""
+        return self.stream_processor.get_next_result()
+
+    def load_and_resample_audio(self, audio_file):
+        """加载音频并重采样到目标��样率"""
+        print(f"加载音频文件: {audio_file}")
+        speech, original_sr = librosa.load(audio_file, sr=None)
+        
+        audio_duration = len(speech) / original_sr
+        print(f"原始音频: {original_sr}Hz, 时长: {audio_duration:.2f}秒")
+        
+        if original_sr != self.target_sr:
+            speech = self.resampler.resample_audio(speech, original_sr)
+            print(f"重采样后: {self.target_sr}Hz, 时长: {len(speech)/self.target_sr:.2f}秒")
+        
+        return speech, self.target_sr
+    
+    def run_translation(self, text_content):
+        """调用Qwen大模型API中英互译"""
+        print("Starting translation via API...")
+        translation_start_time = time.time()
+        
+        translate_content = self.translator.translate(text_content)
+        
+        translation_time_cost = time.time() - translation_start_time
+        print(f"Translation processing time: {translation_time_cost:.2f} seconds")
+        print(f"Translation Result: {translate_content}")
+        
+        return translate_content
+    
+    def run_tts(self, translate_content, output_dir, output_wav=None):
+        """使用TTS模型合成语音"""
+        output_path = os.path.join(output_dir, output_wav)
+        
+        try:
+            if lang_detect_with_regex(translate_content) == "chinese":
+                translate_content = cn2an.transform(translate_content, "an2cn")
+            
+            print(f"TTS synthesis for text: {translate_content}")
+            
+            sens = split_sentence(translate_content, language_str=self.tts_language)
+            print(f"Text split into {len(sens)} sentences")
+            
+            audio_list = []
+            
+            for n, se in enumerate(sens):
+                if self.tts_language in ['EN', 'ZH_MIX_EN']:
+                    se = re.sub(r'([a-z])([A-Z])', r'\1 \2', se)
+                
+                print(f"Processing sentence[{n}]: {se}")
+                
+                phones, tones, lang_ids, norm_text, word2ph = get_text_for_tts_infer(
+                    se, self.tts_language, symbol_to_id=self.symbol_to_id)
+                
+                encoder_start = time.time()
+                z_p, pronoun_lens, audio_len = self.sess_enc.run(None, input_feed={
+                    'phone': phones, 'g': self.tts_g,
+                    'tone': tones, 'language': lang_ids, 
+                    'noise_scale': np.array([0], dtype=np.float32),
+                    'length_scale': np.array([1.0 / self.tts_speed], dtype=np.float32),
+                    'noise_scale_w': np.array([0], dtype=np.float32),
+                    'sdp_ratio': np.array([0], dtype=np.float32)})
+                print(f"Encoder run time: {1000 * (time.time() - encoder_start):.2f}ms")
+                
+                word2pronoun = calc_word2pronoun(word2ph, pronoun_lens)
+                pn_slices, zp_slices = generate_slices(word2pronoun, self.tts_dec_len)
+                
+                audio_len = audio_len[0]
+                sub_audio_list = []
+                
+                for i, (ps, zs) in enumerate(zip(pn_slices, zp_slices)):
+                    zp_slice = z_p[..., zs]
+                    
+                    sub_dec_len = zp_slice.shape[-1]
+                    sub_audio_len = 512 * sub_dec_len
+                    
+                    if zp_slice.shape[-1] < self.tts_dec_len:
+                        zp_slice = np.concatenate((zp_slice, np.zeros((*zp_slice.shape[:-1], self.tts_dec_len - zp_slice.shape[-1]), dtype=np.float32)), axis=-1)
+                    
+                    decoder_start = time.time()
+                    audio = self.sess_dec.run(None, input_feed={"z_p": zp_slice, "g": self.tts_g})[0].flatten()
+                    
+                    audio_start = 0
+                    if len(sub_audio_list) > 0:
+                        if pn_slices[i - 1].stop > ps.start:
+                            audio_start = 512 * word2pronoun[ps.start]
+                    
+                    audio_end = sub_audio_len
+                    if i < len(pn_slices) - 1:
+                        if ps.stop > pn_slices[i + 1].start:
+                            audio_end = sub_audio_len - 512 * word2pronoun[ps.stop - 1]
+                    
+                    audio = audio[audio_start:audio_end]
+                    print(f"Decode slice[{i}]: decoder run time {1000 * (time.time() - decoder_start):.2f}ms")
+                    sub_audio_list.append(audio)
+                
+                sub_audio = merge_sub_audio(sub_audio_list, 0, audio_len)
+                audio_list.append(sub_audio)
+            
+            audio = audio_numpy_concat(audio_list, sr=self.sample_rate, speed=self.tts_speed)
+            
+            sf.write(output_path, audio, self.sample_rate)
+            print(f"TTS audio saved to {output_path}")
+            
+            return output_path
+            
+        except Exception as e:
+            print(f"TTS synthesis failed: {e}")
+            import traceback
+            traceback.print_exc()
+            raise e
+
+    def process_long_audio_stream(self, audio_file, chunk_size=64000):
+        """
+        处理长音频文件的流式模拟
+        chunk_size增加到64000（4秒 * 16000Hz），与StreamProcessor的chunk_duration匹配
+        4秒有点短，改到7秒感觉更好点
+        """
+        print(f"[流式处理] 开始处理长音频: {audio_file}")
+        
+        # 加载并重采样音频
+        speech, fs = self.load_and_resample_audio(audio_file)
+        
+        # 启动流式处理
+        self.start_stream_processing()
+        
+        total_chunks = (len(speech) + chunk_size - 1) // chunk_size
+        print(f"[流式处理] 音频总长度: {len(speech)/fs:.2f}秒, 分块数: {total_chunks}")
+        
+        # 收集所有结果
+        all_results = []
+        
+        # 模拟流式输入
+        chunk_count = 0
+        for i in range(0, len(speech), chunk_size):
+            chunk = speech[i:i+chunk_size]
+            chunk_count += 1
+            
+            # 处理最后一块：如果不足chunk_size，填零补齐
+            if len(chunk) < chunk_size:
+                padding_size = chunk_size - len(chunk)
+                chunk = np.concatenate([chunk, np.zeros(padding_size, dtype=np.float32)])
+                print(f"\n[流式处理] 处理音频块 {chunk_count}/{total_chunks} (最后一块，已填零 {padding_size} 样本)")
+            else:
+                print(f"\n[流式处理] 处理音频块 {chunk_count}/{total_chunks}")
+            
+            self.process_audio_stream(chunk, fs)
+            
+            # 获取并显示实时结果
+            result = self.get_stream_results()
+            while result:
+                print(f"\n{'='*70}")
+                print(f"[实时结果 #{len(all_results) + 1}]")
+                print(f"段落ID: {result['segment_id']}")
+                print(f"原文: {result['original']}")
+                print(f"翻译: {result['translated']}")
+                print(f"音频: {result['audio_path']}")
+                print(f"{'='*70}")
+                all_results.append(result)
+                result = self.get_stream_results()
+                
+            time.sleep(0.1)
+        
+        # 输出结果
+        # print(f"\n[流式处理] 等待处理剩余音频块...")
+        max_wait_time = 20  # 增加等待时间到20秒
+        wait_start = time.time()
+        
+        while time.time() - wait_start < max_wait_time:
+            result = self.get_stream_results()
+            if result:
+                print(f"\n{'='*70}")
+                print(f"[实时结果 #{len(all_results) + 1}]")
+                print(f"段落ID: {result['segment_id']}")
+                print(f"原文: {result['original']}")
+                print(f"翻译: {result['translated']}")
+                print(f"音频: {result['audio_path']}")
+                print(f"{'='*70}")
+                all_results.append(result)
+                wait_start = time.time()  # 重置等待时间
+            else:
+                time.sleep(0.2)
+                
+        # 停止流式处理
+        self.stop_stream_processing()
+        
+        print(f"\n[流式处理] 完成！共处理 {len(all_results)} 个有效结果")
+        return all_results
+
+def main():
+    parser = argparse.ArgumentParser(description="实时语音翻译pipeline")
+    parser.add_argument("--audio_file", type=str, required=True, help="输入音频文件路径")
+    parser.add_argument("--output_dir", type=str, default="./output", help="输出目录")
+    parser.add_argument("--api_url", type=str, default=QWEN_API_URL, help="Qwen API服务器URL")
+    parser.add_argument("--target_sr", type=int, default=16000, help="ASR目标采样率 (默认: 16000)")
+    parser.add_argument("--chunk_duration", type=float, default=7.0, help="音频块时长(秒) (默认: 7.0)")
+    parser.add_argument("--overlap_duration", type=float, default=0.01, help="重叠时长(秒) (默认: 0.1)")
+    
+    args = parser.parse_args()
+    print("-------------------实时语音翻译pipeline-------------------\n")
+    os.makedirs(args.output_dir, exist_ok=True)
+    
+    print(f"处理音频文件: {args.audio_file}")
+    print(f"输出目录: {args.output_dir}")
+    print(f"音频块时长: {args.chunk_duration}秒")
+    print(f"重叠时长: {args.overlap_duration}秒\n")
+    
+    # 初始化Pipeline
+    pipeline = SpeechTranslationPipeline(
+        tts_model_dir=TTS_MODEL_DIR,
+        tts_model_files=TTS_MODEL_FILES,
+        asr_model_dir="ax_model",
+        seq_len=132,
+        tts_dec_len=128,
+        sample_rate=44100,
+        tts_speed=0.8,
+        qwen_api_url=args.api_url,
+        target_sr=args.target_sr,
+        output_dir=args.output_dir
+    )
+    
+    # # 可选：调整流式处理参数
+    # if args.chunk_duration != 7.0 or args.overlap_duration != 0.01:
+    #     pipeline.stream_processor.chunk_duration = args.chunk_duration
+    #     pipeline.stream_processor.overlap_duration = args.overlap_duration
+    #     pipeline.stream_processor.chunk_samples = int(args.chunk_duration * args.target_sr)
+    #     pipeline.stream_processor.overlap_samples = int(args.overlap_duration * args.target_sr)
+    #     print(f"[配置] 已更新流式处理参数: chunk_duration={args.chunk_duration}s, overlap_duration={args.overlap_duration}s\n")
+    
+    start_time = time.time()
+    try:
+        # 流式处理模式
+        print("使用流式处理模式（包含TTS）...")
+        print("="*70 + "\n")
+        
+        # 计算chunk_size以匹配chunk_duration
+        chunk_size = int(args.chunk_duration * args.target_sr)
+        results = pipeline.process_long_audio_stream(args.audio_file, chunk_size=chunk_size)
+        
+        print("\n" + "="*70)
+        print("                   处理完成")
+        print("="*70)
+        print(f"\n  成功处理 {len(results)} 个有效翻译段落\n")
+        
+        # 显示所有结果
+        if results:
+            print("所有翻译结果：")
+            print("-" * 70)
+            for idx, result in enumerate(results, 1):
+                print(f"\n【段落 {idx}】(ID: {result['segment_id']})")
+                print(f"  原文: {result['original']}")
+                print(f"  译文: {result['translated']}")
+                print(f"  音频: {result['audio_path']}")
+                print(f"  时间: {time.strftime('%H:%M:%S', time.localtime(result['timestamp']))}")
+            print("-" * 70)
+            
+            # 保存结果到文件
+            result_file = os.path.join(args.output_dir, "stream_results.txt")
+            with open(result_file, 'w', encoding='utf-8') as f:
+                f.write(f"流式翻译+TTS结果 - {args.audio_file}\n")
+                f.write(f"处理时间: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
+                f.write(f"音频块时长: {args.chunk_duration}秒, 重叠时长: {args.overlap_duration}秒\n")
+                f.write("="*70 + "\n\n")
+                for idx, result in enumerate(results, 1):
+                    f.write(f"【段落 {idx}】(ID: {result['segment_id']})\n")
+                    f.write(f"原文: {result['original']}\n")
+                    f.write(f"译文: {result['translated']}\n")
+                    f.write(f"音频: {result['audio_path']}\n")
+                    f.write(f"时间: {time.strftime('%H:%M:%S', time.localtime(result['timestamp']))}\n")
+                    f.write("\n" + "-"*70 + "\n\n")
+            print(f"\n✓ 结果已保存到: {result_file}")
+            
+            # 统计音频文件
+            audio_files = [r['audio_path'] for r in results]
+            print(f"\n  生成 {len(audio_files)} 个TTS音频文件:")
+            for audio_file in audio_files:
+                print(f"  - {audio_file}")
+        else:
+            print("\n  未获取到有效的翻译结果")
+        
+        print("="*70)
+        
+        # 总耗时
+        total_time = time.time() - start_time
+        print(f"\n总处理时间: {total_time:.2f} 秒")
+        
+    except Exception as e:
+        print(f"Pipeline执行失败: {e}")
+        import traceback
+        traceback.print_exc()
+
+if __name__ == "__main__":
+    main()
+