init main framework

README.md

@@ -1,5 +1,5 @@
 ---
-title: Whisper API Server
+title: 伪流式音频转写 + LLM优化系统
 emoji: 🎙️
 colorFrom: indigo
 colorTo: pink
@@ -8,17 +8,145 @@ app_file: app.py
 pinned: false
 ---
 
-# Whisper API Server with faster-whisper
+# 伪流式音频转写 + LLM优化系统
 
-This Space provides a REST API to transcribe audio using faster-whisper + FastAPI.
+这个项目实现了一个伪流式音频转写系统，包括VAD分段、Whisper转录、语义聚合、LLM优化和翻译等功能。系统采用模块化设计，各个组件可以独立工作，也可以组合使用。
 
-## API Endpoints
+## 系统架构
 
-- `GET /` → Health check  
-- `POST /transcribe` → Upload a `.wav/.mp3` file and receive transcript text
+```mermaid
+graph TD
+    A[音频流输入] --> B[VAD]
+    B --> C[Transcribe]
+    C --> D[语义聚合控制器]
 
-## Example usage (cURL)
+    D --> E[即时输出模块]
+    D --> F[LLM 优化调度器]
+
+    F --> G[优化后回填模块]
+    G --> E
+    E --> H[翻译模块]  
+```
+
+## 主要模块
+
+- **VAD分段器**: 根据能量、静音、说话边界等信号判断语音段落
+- **Whisper转录模块**: 对每段VAD输出进行whisper转写，输出文本+时间戳
+- **语义聚合控制器**: 维护segment缓冲池，判断是否组成完整语义单元，推送到下游
+- **即时输出模块**: 将聚合后的转写结果立即显示给用户
+- **LLM优化调度器**: 接收待优化句子，加入优化任务队列
+- **优化后回填模块**: 对照原句编号，将LLM优化结果回填替换
+- **翻译模块**: 接收所有来自即时输出模块的句子，将其翻译为目标语言
+
+## 语义聚合控制器
+
+语义聚合控制器是系统的核心模块，负责将多个音频片段的转录结果聚合成完整的语义单元（句子），并推送到下游模块（显示和翻译）。
+
+### 主要功能
+
+1. **维护转录片段缓冲池**：收集来自转录模块的片段，直到形成完整语义单元
+2. **判断语义完整性**：使用ChatGPT进行few-shot学习，判断多个片段是否组成完整句子
+3. **重新转录**：将多个片段的音频合并，进行整体重新转录，提高准确性
+4. **推送到下游**：将聚合结果发送到显示模块和翻译模块
+
+详细信息请参考 [aggregator/README.md](aggregator/README.md)。
+
+## 使用示例
+
+### 完整流程示例
+
+```python
+from vad.vad import VoiceActivityDetector
+from transcribe.transcribe import AudioTranscriber
+from display.display import OutputRenderer
+from translator.translator import NLLBTranslator
+from aggregator.semantic_aggregator import SemanticAggregator
+
+# 初始化各个模块
+vad = VoiceActivityDetector()
+transcriber = AudioTranscriber(model="small", device="cuda")
+renderer = OutputRenderer()
+translator = NLLBTranslator()
+
+# 回调函数
+def display_callback(sentence_id, text, state):
+    renderer.display(sentence_id, text, state)
+    
+def translate_callback(sentence_id, text):
+    translation = translator.translate(text)
+    print(f"[翻译] 句子 {sentence_id}: {translation}")
+
+# 初始化聚合器
+aggregator = SemanticAggregator(
+    on_display=display_callback,
+    on_translate=translate_callback,
+    transcriber=transcriber
+)
+
+# 处理音频
+audio_data, sample_rate = sf.read("audio.wav")
+segments = vad.detect_voice_segments(audio_data, sample_rate)
+
+for i, (start, end) in enumerate(segments):
+    segment_audio = audio_data[int(start * sample_rate):int(end * sample_rate)]
+    results = transcriber.transcribe_segment(segment_audio, start_time=start)
+    
+    for result in results:
+        result.segment_index = i + 1
+        aggregator.add_segment(result)
+
+# 最后强制刷新缓冲区
+aggregator.flush(force=True)
+```
+
+更详细的示例请参考 [aggregator/integration_example.py](aggregator/integration_example.py)。
+
+## API服务
+
+系统也提供了REST API服务，可以通过HTTP请求进行音频转写。
+
+### API端点
+
+- `GET /` → 健康检查  
+- `POST /transcribe` → 上传`.wav/.mp3`文件并接收转写文本
+
+### 使用示例 (cURL)
 
 ```bash
 curl -X POST https://your-space-name.hf.space/transcribe \
   -F "[email protected]"
+```
+
+## 安装与运行
+
+### 环境要求
+
+- Python 3.8+
+- PyTorch 1.12+
+- CUDA 11.6+ (如果使用GPU)
+
+### 安装依赖
+
+```bash
+pip install -r requirements.txt
+```
+
+### 运行API服务
+
+```bash
+python app.py
+```
+
+### 运行集成示例
+
+```bash
+# 设置OpenAI API密钥（用于句子完整性判断）
+export OPENAI_API_KEY=your_api_key
+
+# 运行集成示例
+python -m aggregator.integration_example
+```
+
+## 许可证
+
+[MIT License](LICENSE)

aggregator/README.md

@@ -0,0 +1,112 @@
+# 语义聚合控制器 (Semantic Aggregator)
+
+语义聚合控制器是伪流式音频转写系统的核心模块，负责将多个音频片段的转录结果聚合成完整的语义单元（句子），并推送到下游模块（显示和翻译）。
+
+## 主要功能
+
+1. **维护转录片段缓冲池**：收集来自转录模块的片段，直到形成完整语义单元
+2. **判断语义完整性**：使用ChatGPT进行few-shot学习，判断多个片段是否组成完整句子
+3. **重新转录**：将多个片段的音频合并，进行整体重新转录，提高准确性
+4. **推送到下游**：将聚合结果发送到显示模块和翻译模块
+
+## 核心组件
+
+### SentenceCompletionDetector
+
+使用ChatGPT进行few-shot学习，判断文本是否是一个完整的句子。
+
+```python
+detector = SentenceCompletionDetector()
+is_complete = detector.is_sentence_complete("你会学习到如何使用音频数据集")  # False
+is_complete = detector.is_sentence_complete("你会学习到如何使用音频数据集。")  # True
+```
+
+### SemanticAggregator
+
+主要聚合控制器，负责缓冲、判断、重新转录和推送。
+
+```python
+aggregator = SemanticAggregator(
+    on_display=display_callback,  # 显示回调
+    on_translate=translate_callback,  # 翻译回调
+    transcriber=transcriber,  # 转录器实例
+    segments_dir="dataset/audio/segments",  # 音频片段目录
+    max_window=5.0,  # 最大聚合时长（秒）
+    max_segments=5,  # 最大聚合片段数
+    min_gap=0.8,  # 触发聚合的最小间隔（秒）
+    force_flush_timeout=3.0  # 强制flush超时时间（秒）
+)
+```
+
+## 聚合判断逻辑
+
+聚合器使用以下逻辑判断是否应该聚合并输出：
+
+1. **语义完整性**：使用ChatGPT判断当前缓冲区中的文本是否形成完整句子
+2. **时间间隔**：如果相邻片段之间的间隔超过阈值，认为是不同的语义单元
+3. **最大窗口**：如果聚合的总时长超过阈值，强制聚合
+4. **最大片段数**：如果聚合的片段数超过阈值，强制聚合
+5. **超时机制**：如果长时间没有新片段，强制输出当前缓冲区内容
+
+## 重新转录流程
+
+1. 获取所有片段的音频数据
+2. 合并音频数据
+3. 使用转录器重新转录合并后的音频
+4. 比较重新转录结果与原始聚合结果
+5. 如果有差异，更新显示并发送到翻译模块
+
+## 使用示例
+
+```python
+from display.display import OutputRenderer
+from translator.translator import NLLBTranslator
+from transcribe.transcribe import AudioTranscriber, TranscriptionResult
+from aggregator.semantic_aggregator import SemanticAggregator
+
+# 初始化各个模块
+renderer = OutputRenderer()
+translator = NLLBTranslator()
+transcriber = AudioTranscriber(model="small", device="cuda")
+
+# 回调函数
+def display_callback(sentence_id, text, state):
+    renderer.display(sentence_id, text, state)
+    
+def translate_callback(sentence_id, text):
+    translation = translator.translate(text)
+    print(f"[翻译] 句子 {sentence_id}: {translation}")
+
+# 初始化聚合器
+aggregator = SemanticAggregator(
+    on_display=display_callback,
+    on_translate=translate_callback,
+    transcriber=transcriber
+)
+
+# 添加转录结果
+for result in transcription_results:
+    aggregator.add_segment(result)
+
+# 最后强制刷新缓冲区
+aggregator.flush(force=True)
+```
+
+## 测试
+
+可以使用 `test_aggregator.py` 脚本测试聚合器功能：
+
+```bash
+# 设置OpenAI API密钥
+export OPENAI_API_KEY=your_api_key
+
+# 运行测试脚本
+python -m aggregator.test_aggregator
+```
+
+## 注意事项
+
+1. 需要设置 `OPENAI_API_KEY` 环境变量才能使用ChatGPT进行句子完整性判断
+2. 音频片段目录需要包含所有需要重新转录的音频文件
+3. 转录器需要正确初始化，包括模型、设备和计算类型
+4. 回调函数需要正确处理聚合结果，包括显示和翻译

aggregator/__init__.py

@@ -0,0 +1,10 @@
+"""
+语义聚合控制器模块
+
+这个模块负责将多个音频片段的转录结果聚合成完整的语义单元（句子），
+并推送到下游模块（显示和翻译）。
+"""
+
+from .semantic_aggregator import SemanticAggregator, SentenceCompletionDetector
+
+__all__ = ['SemanticAggregator', 'SentenceCompletionDetector']

aggregator/semantic_aggregator.py

@@ -1,47 +1,140 @@
-from typing import List, Callable, Optional
-from vad.audio_transcriber import TranscriptionResult
+from typing import List, Callable, Optional, Dict, Tuple
 import uuid
 import time
+import os
+import numpy as np
+import soundfile as sf
+import logging
+from openai import OpenAI
+from transcribe.transcribe import TranscriptionResult, AudioTranscriber
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建日志记录器
+logger = setup_logger("aggregator")
+
+class SentenceCompletionDetector:
+    """
+    使用ChatGPT判断句子是否完整
+    """
+    def __init__(self, model="gpt-3.5-turbo"):
+        self.model = model
+        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+    
+    def build_prompt(self, text: str) -> str:
+        return (
+            "判断以下语句是否为一句话的结尾，如果是，返回 True，否则返回 False：\n"
+            "\"你会学习到如何使用音频数据集，包括音频数据加载\"\n"
+            "False\n\n"
+            "\"你会学习到如何使用音频数据集，包括音频数据加载，音频数据预处理，以及高效加载大规模音频数据集的流式加载方法\"\n"
+            "True\n\n"
+            "\"在开始学习之前，我们需要\"\n"
+            "False\n\n"
+            "\"在开始学习之前，我们需要了解一些基本概念\"\n"
+            "True\n\n"
+            "\"第一章，介绍基础知识\"\n"
+            "True\n\n"
+            f"\"{text}\"\n"
+        )
+    
+    def is_sentence_complete(self, text: str) -> bool:
+        """
+        判断文本是否是一个完整的句子
+        """
+        # # 简单规则：如果以标点符号结尾，认为是完整的句子
+        # if text.strip() and text.strip()[-1] in "。！？!?.;；":
+        #     return True
+            
+        # 使用ChatGPT进行更复杂的判断
+        prompt = self.build_prompt(text)
+        try:
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=[
+                    {"role": "system", "content": "你是一个语言专家，擅长判断句子是否完整。"},
+                    {"role": "user", "content": prompt}
+                ],
+                temperature=0.1,
+                max_tokens=10,
+            )
+            result = response.choices[0].message.content.strip()
+            logger.debug(f"ChatGPT判断结果: {result}")
+            return result.lower() == "true"
+        except Exception as e:
+            logger.error(f"调用ChatGPT出错: {str(e)}")
+            # 出错时使用简单规则判断
+            return len(text) > 20  # 如果文本较长，可能是完整句子
 
 class SemanticAggregator:
     """
     语义聚合控制器
     - 维护segment缓冲池
     - 判断是否组成完整语义单元
-    - 推送到下游（display/optimizer）
+    - 推送到下游（display/translator）
     """
 
     def __init__(
         self,
-        on_aggregate: Callable[[str, List[TranscriptionResult], str], None],
+        on_display: Callable[[str, str, str], None],
+        on_translate: Callable[[str, str], None],
+        transcriber: AudioTranscriber,
+        segments_dir: str = "dataset/audio/segments",
         max_window: float = 5.0,
         max_segments: int = 5,
         min_gap: float = 0.8,
         force_flush_timeout: float = 3.0
     ):
         """
-        :param on_aggregate: 聚合回调 (text, segments, sentence_id)
+        :param on_display: 显示回调 (sentence_id, text, state)
+        :param on_translate: 翻译回调 (sentence_id, text)
+        :param transcriber: 转录器实例
+        :param segments_dir: 音频片段目录
         :param max_window: 最大聚合时长（秒）
         :param max_segments: 最大聚合片段数
         :param min_gap: 触发聚合的最小间隔（秒）
         :param force_flush_timeout: 强制flush超时时间（秒）
         """
         self.buffer: List[TranscriptionResult] = []
-        self.on_aggregate = on_aggregate
+        self.on_display = on_display
+        self.on_translate = on_translate
+        self.transcriber = transcriber
+        self.segments_dir = segments_dir
         self.max_window = max_window
         self.max_segments = max_segments
         self.min_gap = min_gap
         self.force_flush_timeout = force_flush_timeout
         self.last_flush_time = time.time()
+        self.sentence_detector = SentenceCompletionDetector()
+        self.audio_cache: Dict[int, np.ndarray] = {}  # 缓存音频数据，避免重复读取
+        self.sample_rate = 16000  # 假设采样率为16kHz
+        logger.debug(f"语义聚合器初始化完成，参数: max_window={max_window}, max_segments={max_segments}")
 
     def add_segment(self, result: TranscriptionResult):
         """
         新增转写片段到缓冲池，自动判断是否聚合
         """
         self.buffer.append(result)
+        logger.debug(f"添加片段: {result.text}")
         if self._should_aggregate():
             self._aggregate_and_flush()
         elif time.time() - self.last_flush_time > self.force_flush_timeout:
+            logger.debug(f"超时强制刷新: {self.force_flush_timeout}秒")
             self.flush(force=True)
 
     def flush(self, force: bool = False):
@@ -49,6 +142,7 @@ class SemanticAggregator:
         强制输出当前聚合内容
         """
         if self.buffer:
+            logger.debug(f"强制刷新缓冲区，当前片段数: {len(self.buffer)}")
             self._aggregate_and_flush()
         self.last_flush_time = time.time()
 
@@ -58,49 +152,259 @@ class SemanticAggregator:
         """
         if not self.buffer:
             return False
-        # 1. 标点符号结尾
-        if self.buffer[-1].text.strip() and self.buffer[-1].text.strip()[-1] in "。！？!?":
+            
+        # 1. 使用ChatGPT判断是否是完整句子
+        # 使用逗号连接segments，与_aggregate_and_flush保持一致
+        segments = [seg.text for seg in self.buffer]
+        combined_text = "，".join(segments)
+        if self.sentence_detector.is_sentence_complete(combined_text):
+            logger.info(f"检测到完整句子: {combined_text}")
             return True
+            
         # 2. segment间隔
         if len(self.buffer) >= 2:
             gap = self.buffer[-1].start_time - self.buffer[-2].end_time
             if gap > self.min_gap:
+                logger.info(f"检测到较大间隔: {gap:.2f}秒")
                 return True
+                
         # 3. 最大窗口/片段数
         total_duration = self.buffer[-1].end_time - self.buffer[0].start_time
-        if total_duration > self.max_window or len(self.buffer) >= self.max_segments:
+        if total_duration > self.max_window:
+            logger.info(f"达到最大时间窗口: {total_duration:.2f}秒")
+            return True
+        if len(self.buffer) >= self.max_segments:
+            logger.info(f"达到最大片段数: {len(self.buffer)}")
             return True
+            
         return False
 
+    def _get_segment_audio(self, segment_index: int) -> np.ndarray:
+        """
+        获取指定索引的音频片段数据
+        """
+        if segment_index in self.audio_cache:
+            return self.audio_cache[segment_index]
+            
+        # 读取音频文件
+        audio_path = os.path.join(self.segments_dir, f"test1_segment_{segment_index}.wav")
+        try:
+            audio_data, sample_rate = sf.read(audio_path)
+            self.audio_cache[segment_index] = audio_data
+            logger.debug(f"读取音频片段: {audio_path}, 长度: {len(audio_data)/sample_rate:.2f}秒")
+            return audio_data
+        except Exception as e:
+            logger.error(f"读取音频文件失败: {audio_path}, 错误: {str(e)}")
+            return np.array([])
+
+    def _combine_audio_segments(self, segment_indices: List[int]) -> Tuple[np.ndarray, float]:
+        """
+        合并多个音频片段
+        返回: (合并后的音频数据, 起始时间)
+        """
+        if not segment_indices:
+            return np.array([]), 0.0
+            
+        # 获取所有片段的音频数据
+        audio_segments = []
+        for idx in segment_indices:
+            audio_data = self._get_segment_audio(idx)
+            if len(audio_data) > 0:
+                audio_segments.append(audio_data)
+                
+        if not audio_segments:
+            return np.array([]), 0.0
+            
+        # 合并音频数据
+        combined_audio = np.concatenate(audio_segments)
+        
+        # 获取第一个片段的起始时间
+        first_segment = self.buffer[0]
+        start_time = first_segment.start_time
+        
+        logger.debug(f"合并音频片段: {segment_indices}, 总长度: {len(combined_audio)/self.sample_rate:.2f}秒")
+        return combined_audio, start_time
+
+    def _retranscribe_segments(self, segment_indices: List[int]) -> List[TranscriptionResult]:
+        """
+        重新转录合并后的音频片段
+        """
+        combined_audio, start_time = self._combine_audio_segments(segment_indices)
+        if len(combined_audio) == 0:
+            logger.warning("没有有效的音频数据可以重新转录")
+            return []
+            
+        logger.debug(f"重新转录合并的音频片段, 长度: {len(combined_audio)/self.sample_rate:.2f}秒")
+        try:
+            results = self.transcriber.transcribe_segment(combined_audio, start_time=start_time)
+            logger.debug(f"重新转录结果: {len(results)}条")
+            return results
+        except Exception as e:
+            logger.error(f"重新转录失败: {str(e)}")
+            return []
+
     def _aggregate_and_flush(self):
         """
         聚合并推送到下游
         """
-        text = "".join([seg.text for seg in self.buffer])
+        if not self.buffer:
+            return
+            
+        # 获取所有片段的索引
+        segment_indices = []
+        for seg in self.buffer:
+            if hasattr(seg, 'segment_index') and seg.segment_index is not None:
+                if isinstance(seg.segment_index, list):
+                    segment_indices.extend(seg.segment_index)
+                else:
+                    segment_indices.append(seg.segment_index)
+            
+        # 去重并排序
+        segment_indices = sorted(list(set(segment_indices)))
+        
+        # 生成句子ID
         sentence_id = str(uuid.uuid4())
-        self.on_aggregate(text, self.buffer, sentence_id)
+        
+        # 1. 先使用原始文本进行输出，在segment之间添加逗号
+        original_segments = [seg.text for seg in self.buffer]
+        # 使用逗号连接segments，但不在最后添加句号
+        original_text = "，".join(original_segments)
+        logger.info(f"原始聚合文本: {original_text}")
+        self.on_display(sentence_id, original_text, "raw")
+        
+        # 2. 重新转录
+        if segment_indices:
+            retranscribed_results = self._retranscribe_segments(segment_indices)
+            if retranscribed_results:
+                # 合并重新转录的结果，在segment之间添加逗号
+                retranscribed_segments = [res.text for res in retranscribed_results]
+                retranscribed_text = "，".join(retranscribed_segments)
+                logger.info(f"重新转录文本: {retranscribed_text}")
+                
+                # 如果重新转录的结果与原始文本不同，则更新显示
+                if retranscribed_text != original_text:
+                    self.on_display(sentence_id, retranscribed_text, "optimized")
+                    
+                # 发送到翻译模块
+                self.on_translate(sentence_id, retranscribed_text)
+            else:
+                # 如果重新转录失败，使用原始文本进行翻译
+                logger.warning("重新转录失败，使用原始文本进行翻译")
+                self.on_translate(sentence_id, original_text)
+        else:
+            # 如果没有有效的片段索引，使用原始文本进行翻译
+            logger.warning("没有有效的片段索引，使用原始文本进行翻译")
+            self.on_translate(sentence_id, original_text)
+        
+        # 清空缓冲区
+        buffer_size = len(self.buffer)
         self.buffer.clear()
         self.last_flush_time = time.time()
+        logger.debug(f"清空缓冲区，释放 {buffer_size} 个片段")
 
 
+def load_transcription_results(json_path):
+    """从JSON文件加载转录结果"""
+    import json
+    with open(json_path, 'r', encoding='utf-8') as f:
+        data = json.load(f)
+    
+    results = []
+    for segment in data['segments']:
+        result = TranscriptionResult(
+            text=segment['text'],
+            start_time=segment['start_time'],
+            end_time=segment['end_time'],
+            confidence=segment['confidence'],
+            verified=segment['verified'],
+            verified_text=segment['verified_text'],
+            verification_notes=segment['verification_notes'],
+            segment_index=segment['segment_index'] if 'segment_index' in segment else None
+        )
+        results.append(result)
+    
+    return results
+
 if __name__ == "__main__":
-    # 示例：如何集成display和optimizer（无函数，主流程直写）
+    """测试聚合器功能"""
+    import os
+    import sys
+    import json
+    from pathlib import Path
+    
+    # 配置日志级别
+    logger.setLevel(logging.DEBUG)
+    
+    # 检查OpenAI API密钥
+    if not os.getenv("OPENAI_API_KEY"):
+        logger.warning("未设置OPENAI_API_KEY环境变量，句子完整性判断将使用备用方法")
+    
+    # 初始化各个模块
     from display.display import OutputRenderer
-    from optimizer.dispatcher import OptimizationDispatcher
-
+    from translator.translator import NLLBTranslator
+    from transcribe.transcribe import AudioTranscriber
+    
+    # 初始化显示器
     renderer = OutputRenderer()
-    dispatcher = OptimizationDispatcher(max_workers=2)
-
-    def aggregate_callback(text, segments, sentence_id):
-        # 直接在主流程内联调用
-        print(f"[聚合完成] sentence_id={sentence_id}")
-        print(f"聚合文本: {text}")
-        renderer.display(sentence_id, text, state="raw")
-        dispatcher.submit(sentence_id, text, callback=None)  # callback可自定义
-
-    aggregator = SemanticAggregator(on_aggregate=aggregate_callback)
-
-    # 假设有若干TranscriptionResult对象results
-    # for result in results:
-    #     aggregator.add_segment(result)
-    # aggregator.flush(force=True)
+    
+    # 初始化转录器
+    try:
+        transcriber = AudioTranscriber(model="small", device="cuda", compute_type="int8")
+        logger.info("使用GPU进行转录")
+    except Exception as e:
+        logger.warning(f"GPU初始化失败，使用CPU: {str(e)}")
+        transcriber = AudioTranscriber(model="small", device="cpu", compute_type="float32")
+    
+    # 初始化翻译器（可选）
+    try:
+        translator = NLLBTranslator()
+        translation_enabled = True
+    except Exception as e:
+        logger.warning(f"翻译器初始化失败: {str(e)}")
+        translation_enabled = False
+    
+    # 回调函数
+    def display_callback(sentence_id, text, state):
+        renderer.display(sentence_id, text, state)
+        
+    def translate_callback(sentence_id, text):
+        if translation_enabled:
+            try:
+                translation = translator.translate(text)
+                logger.info(f"[翻译] 句子 {sentence_id}: {translation}")
+            except Exception as e:
+                logger.error(f"翻译失败: {str(e)}")
+        else:
+            logger.info(f"[翻译已禁用] 句子 {sentence_id}: {text}")
+    
+    # 初始化聚合器
+    aggregator = SemanticAggregator(
+        on_display=display_callback,
+        on_translate=translate_callback,
+        transcriber=transcriber,
+        segments_dir="dataset/audio/segments",
+        max_window=10.0,  # 增大窗口以便测试
+        max_segments=10,  # 增大片段数以便测试
+        force_flush_timeout=5.0  # 增大超时以便测试
+    )
+    
+    # 加载测试数据
+    test_file = "dataset/transcripts/test1_segment_1_20250423_201934.json"
+    try:
+        results = load_transcription_results(test_file)
+        logger.info(f"加载了 {len(results)} 条转录结果")
+    except Exception as e:
+        logger.error(f"加载转录结果失败: {str(e)}")
+        sys.exit(1)
+    
+    # 模拟添加转录结果
+    for i, result in enumerate(results):
+        logger.info(f"添加第 {i+1}/{len(results)} 条转录结果: {result.text}")
+        aggregator.add_segment(result)
+        # 模拟处理延迟
+        # time.sleep(0.5)
+    
+    # 强制刷新缓冲区
+    aggregator.flush(force=True)
+    
+    logger.info("测试完成")

dataset/transcripts/test1_segment_1_20250423_201934.json

@@ -12,7 +12,7 @@
       "verification_notes": null
     },
     {
-      "text": "音频数据出来",
+      "text": "音频数据处理",
       "start_time": 4.34,
       "end_time": 5.56,
       "confidence": 0.4482421875,
@@ -84,7 +84,7 @@
       "verification_notes": null
     },
     {
-      "text": "包括波形,彩虹率和冰普渡",
+      "text": "包括波形,采样率和频谱图",
       "start_time": 26.28,
       "end_time": 28.28,
       "confidence": 0.732666015625,
@@ -111,7 +111,7 @@
       "verification_notes": null
     },
     {
-      "text": "高效加载大规模音频数据集的流逝加载方法。",
+      "text": "高效加载大规模音频数据集的流式加载方法。",
       "start_time": 33.54,
       "end_time": 36.5,
       "confidence": 0.88739013671875,
@@ -138,7 +138,7 @@
       "verification_notes": null
     },
     {
-      "text": "基础的音频相关数",
+      "text": "基础的音频相关术语",
       "start_time": 40.86,
       "end_time": 42.4,
       "confidence": 0.609619140625,

display/display.py

@@ -1,32 +1,87 @@
+"""
+显示模块 - 负责将转写结果显示给用户
+"""
+
 from rich.console import Console
 from rich.text import Text
 from typing import Literal
+import logging
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
 
+# 创建日志记录器
+logger = setup_logger("display")
+
+# 创建控制台对象
 console = Console()
 
 class OutputRenderer:
+    """
+    输出渲染器，负责将转写结果显示给用户
+    支持原始文本和优化后文本的不同样式显示
+    """
+    
     def __init__(self):
+        """初始化输出渲染器"""
         self.history = {}  # 用于更新同一条句子的优化内容
+        logger.debug("输出渲染器初始化完成")
 
     def display(self, sentence_id: str, text: str, state: Literal["raw", "optimized"]):
+        """
+        显示转写结果
+        
+        :param sentence_id: 句子ID
+        :param text: 文本内容
+        :param state: 状态，raw表示原始文本，optimized表示优化后文本
+        """
         if state == "raw":
             styled_text = Text(text, style="dim")  # 灰色表示原始输出
+            logger.debug(f"显示原始文本: {sentence_id}")
         elif state == "optimized":
             styled_text = Text(text, style="bold black")  # 深黑色加粗
+            logger.debug(f"显示优化文本: {sentence_id}")
         else:
-            raise ValueError("Unknown output state")
+            logger.error(f"未知的输出状态: {state}")
+            raise ValueError(f"未知的输出状态: {state}")
 
         # 打印新内容（或替换历史）
         if sentence_id in self.history:
             console.print(f"[更新] 句子 {sentence_id}：", styled_text)
+            logger.info(f"更新句子: {sentence_id}")
         else:
             console.print(f"[输出] 句子 {sentence_id}：", styled_text)
+            logger.info(f"输出句子: {sentence_id}")
 
+        # 记录历史
         self.history[sentence_id] = text
+        logger.debug(f"句子内容: {text}")
+
 
 if __name__ == "__main__":
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
+    
+    # 测试代码
     renderer = OutputRenderer()
 
+    # 显示原始文本
     renderer.display("s1", "I think we should start the meeting now.", "raw")
+    
     # 模拟优化回填
     renderer.display("s1", "I believe it's time to begin the meeting.", "optimized")

main.py

@@ -0,0 +1,300 @@
+"""
+伪流式音频转写 + LLM优化系统 主程序
+
+这个程序实现了完整的音频处理流水线，包括：
+1. VAD分段
+2. Whisper转录
+3. 语义聚合
+4. 即时输出
+5. LLM优化
+6. 翻译
+
+使用方法：
+python main.py [--audio_path AUDIO_PATH] [--use_gpu] [--enable_translation] [--enable_optimization]
+"""
+
+import os
+import sys
+import time
+import logging
+import argparse
+import numpy as np
+import soundfile as sf
+from pathlib import Path
+from typing import List, Dict, Optional, Tuple, Union
+import uuid
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建主日志记录器
+logger = setup_logger("main")
+
+# 导入各个模块
+from vad import VoiceActivityDetector
+from transcribe.transcribe import AudioTranscriber, TranscriptionResult
+from aggregator.semantic_aggregator import SemanticAggregator
+from display.display import OutputRenderer
+from optimizer.dispatcher import OptimizationDispatcher
+from translator.translator import NLLBTranslator
+
+class AudioProcessingPipeline:
+    """完整的音频处理流水线"""
+    
+    def __init__(
+        self, 
+        audio_path: str, 
+        use_gpu: bool = True,
+        enable_translation: bool = True,
+        enable_optimization: bool = True,
+        whisper_model: str = "large",
+        log_level: Union[int, str] = logging.INFO
+    ):
+        """
+        初始化处理流水线
+        
+        :param audio_path: 音频文件路径
+        :param use_gpu: 是否使用GPU
+        :param enable_translation: 是否启用翻译
+        :param enable_optimization: 是否启用LLM优化
+        :param whisper_model: Whisper模型大小 (tiny, base, small, medium, large)
+        :param log_level: 日志级别
+        """
+        # 设置日志级别
+        if isinstance(log_level, str):
+            log_level = getattr(logging, log_level.upper())
+        logger.setLevel(log_level)
+        
+        self.audio_path = audio_path
+        self.use_gpu = use_gpu
+        self.enable_translation = enable_translation
+        self.enable_optimization = enable_optimization
+        self.whisper_model = whisper_model
+        
+        # 检查设备
+        self.device = "cuda" if use_gpu and self._is_gpu_available() else "cpu"
+        logger.info(f"使用设备: {self.device}")
+        logger.debug(f"配置: whisper_model={whisper_model}, translation={enable_translation}, optimization={enable_optimization}")
+        
+        # 初始化各个模块
+        self._init_modules()
+        
+        # 加载音频数据
+        self.audio_data, self.sample_rate = sf.read(audio_path)
+        logger.info(f"加载音频: {os.path.basename(audio_path)}, 长度: {len(self.audio_data)/self.sample_rate:.2f}秒")
+        logger.debug(f"音频详情: 采样率={self.sample_rate}Hz, 形状={self.audio_data.shape}")
+        
+        # 存储句子ID到优化任务的映射
+        self.optimization_tasks: Dict[str, str] = {}
+    
+    def _is_gpu_available(self) -> bool:
+        """检查GPU是否可用"""
+        try:
+            import torch
+            if torch.cuda.is_available():
+                logger.debug(f"检测到GPU: {torch.cuda.get_device_name(0)}")
+                return True
+            return False
+        except ImportError:
+            logger.debug("未检测到PyTorch，将使用CPU")
+            return False
+    
+    def _init_modules(self):
+        """初始化各个处理模块"""
+        # 1. 初始化VAD
+        logger.debug("初始化VAD模块...")
+        self.vad = VoiceActivityDetector(save_audio=True, save_json=True)
+        
+        # 2. 初始化转录器
+        logger.debug(f"初始化Whisper转录模块 (model={self.whisper_model}, device={self.device})...")
+        self.transcriber = AudioTranscriber(
+            model=self.whisper_model,
+            device=self.device,
+            compute_type="int8" if self.device == "cuda" else "float32"
+        )
+        
+        # 3. 初始化显示器
+        logger.debug("初始化显示模块...")
+        self.renderer = OutputRenderer()
+        
+        # 4. 初始化优化调度器（如果启用）
+        if self.enable_optimization:
+            logger.debug("初始化LLM优化调度器...")
+            self.optimizer = OptimizationDispatcher(
+                max_workers=2,
+                callback=self._optimization_callback
+            )
+        else:
+            logger.debug("LLM优化已禁用")
+            self.optimizer = None
+        
+        # 5. 初始化翻译器（如果启用）
+        if self.enable_translation:
+            logger.debug("初始化翻译模块...")
+            try:
+                self.translator = NLLBTranslator()
+                self.translation_enabled = True
+            except Exception as e:
+                logger.warning(f"翻译器初始化失败: {str(e)}")
+                self.translation_enabled = False
+        else:
+            logger.debug("翻译已禁用")
+            self.translation_enabled = False
+            self.translator = None
+        
+        # 6. 初始化聚合器
+        logger.debug("初始化语义聚合控制器...")
+        self.aggregator = SemanticAggregator(
+            on_display=self._display_callback,
+            on_translate=self._translate_callback,
+            transcriber=self.transcriber,
+            segments_dir="dataset/audio/segments",
+            max_window=5.0,
+            max_segments=5,
+            min_gap=0.8,
+            force_flush_timeout=3.0
+        )
+        
+        logger.info("所有模块初始化完成")
+    
+    def _display_callback(self, sentence_id: str, text: str, state: str):
+        """显示回调函数"""
+        self.renderer.display(sentence_id, text, state)
+        
+        # 如果启用了优化，且是原始文本，则提交优化任务
+        if self.enable_optimization and state == "raw" and self.optimizer:
+            logger.debug(f"提交优化任务: {sentence_id}")
+            self.optimizer.submit(sentence_id, text)
+    
+    def _translate_callback(self, sentence_id: str, text: str):
+        """翻译回调函数"""
+        if self.translation_enabled and self.translator:
+            try:
+                # 翻译模块内部已经记录了原文和结果，这里只需调用翻译方法
+                self.translator.translate(text)
+                logger.debug(f"已翻译句子: {sentence_id}")
+            except Exception as e:
+                logger.error(f"翻译失败: {str(e)}")
+    
+    def _optimization_callback(self, sentence_id: str, original_text: str, optimized_text: str):
+        """优化回调函数"""
+        logger.debug(f"收到优化结果: {sentence_id}")
+        # 更新显示
+        self.renderer.display(sentence_id, optimized_text, "optimized")
+        # 如果启用了翻译，则翻译优化后的文本
+        if self.translation_enabled:
+            logger.debug(f"翻译优化后的文本: {sentence_id}")
+            self._translate_callback(sentence_id, optimized_text)
+    
+    def process(self):
+        """处理音频文件"""
+        logger.info("开始处理音频...")
+        
+        # 1. VAD分段
+        logger.debug("执行VAD分段...")
+        segments = self.vad.detect_voice_segments(self.audio_data, self.sample_rate)
+        logger.info(f"VAD分段完成: {len(segments)}个片段")
+        
+        # 2. 转录每个片段
+        for i, (start, end) in enumerate(segments):
+            logger.debug(f"转录片段 {i+1}/{len(segments)}: {start:.2f}s -> {end:.2f}s")
+            
+            # 提取片段音频数据
+            segment_audio = self.audio_data[int(start * self.sample_rate):int(end * self.sample_rate)]
+            
+            # 转录片段
+            results = self.transcriber.transcribe_segment(segment_audio, start_time=start)
+            
+            # 添加片段索引
+            for result in results:
+                result.segment_index = i + 1  # 片段索引从1开始
+            
+            # 添加到聚合器
+            for result in results:
+                logger.debug(f"添加转录结果: {result.text}")
+                self.aggregator.add_segment(result)
+            
+            # 模拟处理延迟
+            time.sleep(0.1)
+        
+        # 3. 最后强制刷新缓冲区
+        logger.debug("强制刷新缓冲区...")
+        self.aggregator.flush(force=True)
+        
+        # 4. 等待所有优化任务完成
+        if self.enable_optimization and self.optimizer:
+            logger.debug("等待所有优化任务完成...")
+            self.optimizer.wait_until_done()
+        
+        logger.info("音频处理完成")
+
+def parse_args():
+    """解析命令行参数"""
+    parser = argparse.ArgumentParser(description="伪流式音频转写 + LLM优化系统")
+    parser.add_argument("--audio_path", type=str, default="dataset/audio/test1.wav", 
+                        help="音频文件路径")
+    parser.add_argument("--use_gpu", action="store_true", default=True,
+                        help="是否使用GPU")
+    parser.add_argument("--enable_translation", action="store_true", default=True,
+                        help="是否启用翻译")
+    parser.add_argument("--enable_optimization", action="store_true", default=True,
+                        help="是否启用LLM优化")
+    parser.add_argument("--whisper_model", type=str, default="small",
+                        choices=["tiny", "base", "small", "medium", "large"],
+                        help="Whisper模型大小")
+    parser.add_argument("--log_level", type=str, default="INFO",
+                        choices=["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+                        help="日志级别")
+    return parser.parse_args()
+
+def main():
+    """主函数"""
+    # 解析命令行参数
+    args = parse_args()
+    
+    # 设置日志级别
+    log_level = getattr(logging, args.log_level)
+    
+    # 设置所有模块的日志级别
+    for module in ["main", "vad", "transcribe", "aggregator", "display", "optimizer", "translator"]:
+        setup_logger(module, log_level)
+    
+    # 检查OpenAI API密钥（用于句子完整性判断和优化）
+    if not os.getenv("OPENAI_API_KEY") and args.enable_optimization:
+        logger.warning("未设置OPENAI_API_KEY环境变量，句子完整性判断将使用备用方法")
+    
+    # 检查音频文件是否存在
+    if not os.path.exists(args.audio_path):
+        logger.error(f"音频文件不存在: {args.audio_path}")
+        return
+    
+    # 创建并运行处理流水线
+    pipeline = AudioProcessingPipeline(
+        audio_path=args.audio_path,
+        use_gpu=args.use_gpu,
+        enable_translation=args.enable_translation,
+        enable_optimization=args.enable_optimization,
+        whisper_model=args.whisper_model,
+        log_level=log_level
+    )
+    
+    # 处理音频
+    pipeline.process()
+
+if __name__ == "__main__":
+    main()

optimizer/dispatcher.py

@@ -1,46 +1,127 @@
-# optimizer/dispatcher.py
+"""
+优化调度器 - 负责管理LLM优化任务队列
+"""
+
 import asyncio
+import logging
 from concurrent.futures import ThreadPoolExecutor
-from optimizer.llm_runner import TinyLLaMARunner
+from typing import Callable, Optional
+from optimizer.llm_api_runner import ChatGPTRunner
 from optimizer.optimize_task import OptimizeTask
 
-class OptimizationDispatcher:
-    def __init__(self, max_workers: int = 1):
-        self.queue = asyncio.Queue()
-        self.executor = ThreadPoolExecutor(max_workers=max_workers)
-        self.model_runner = TinyLLaMARunner()
-
-    def submit(self, sentence_id: str, text: str, callback):
-        task = OptimizeTask(sentence_id, text, callback)
-        self.queue.put_nowait(task)
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
 
-    async def start(self):
-        print("[Dispatcher] 启动优化调度器...")
-        while True:
-            task: OptimizeTask = await self.queue.get()
-            asyncio.create_task(self._handle(task))
+# 创建日志记录器
+logger = setup_logger("optimizer")
 
-    async def _handle(self, task: OptimizeTask):
-        await asyncio.get_event_loop().run_in_executor(
-            self.executor,
-            task.run,
-            self.model_runner
-        )
+class OptimizationDispatcher:
+    """
+    优化调度器，负责管理LLM优化任务队列
+    支持异步处理多个优化任务
+    """
+    
+    def __init__(self, max_workers: int = 2, callback: Optional[Callable] = None):
+        """
+        初始化优化调度器
+        
+        :param max_workers: 最大工作线程数
+        :param callback: 优化完成后的回调函数
+        """
+        self.tasks = {}  # 存储任务ID到任务的映射
+        self.executor = ThreadPoolExecutor(max_workers=max_workers)
+        self.model_runner = ChatGPTRunner()
+        self.callback = callback
+        logger.debug(f"优化调度器初始化完成，最大工作线程数: {max_workers}")
+    
+    def submit(self, sentence_id: str, text: str, callback: Optional[Callable] = None):
+        """
+        提交优化任务
+        
+        :param sentence_id: 句子ID
+        :param text: 需要优化的文本
+        :param callback: 优化完成后的回调函数，如果为None则使用默认回调
+        """
+        task_callback = callback or self.callback
+        task = OptimizeTask(sentence_id, text, task_callback)
+        self.tasks[sentence_id] = task
+        logger.debug(f"提交优化任务: {sentence_id}")
+        
+        # 在线程池中执行任务
+        self.executor.submit(self._process_task, task)
+        logger.debug(f"任务已提交到线程池: {sentence_id}")
+    
+    def _process_task(self, task: OptimizeTask):
+        """
+        处理优化任务
+        
+        :param task: 优化任务
+        """
+        try:
+            logger.debug(f"开始处理任务: {task.sentence_id}")
+            # 使用模型运行器优化文本
+            optimized_text = self.model_runner.optimize(task.text)
+            logger.debug(f"任务处理完成: {task.sentence_id}")
+            
+            # 调用回调函数
+            if task.callback:
+                task.callback(task.sentence_id, task.text, optimized_text)
+                logger.debug(f"已调用回调函数: {task.sentence_id}")
+            
+            # 从任务列表中移除
+            if task.sentence_id in self.tasks:
+                del self.tasks[task.sentence_id]
+                
+            logger.info(f"优化任务完成: {task.sentence_id}")
+        except Exception as e:
+            logger.error(f"处理任务出错: {task.sentence_id}, 错误: {str(e)}")
+    
+    def wait_until_done(self, timeout: Optional[float] = None):
+        """
+        等待所有任务完成
+        
+        :param timeout: 超时时间（秒），如果为None则一直等待
+        :return: 是否所有任务都已完成
+        """
+        logger.debug(f"等待所有任务完成，当前任务数: {len(self.tasks)}")
+        self.executor.shutdown(wait=True, timeout=timeout)
+        # 创建新的线程池
+        self.executor = ThreadPoolExecutor(max_workers=self.executor._max_workers)
+        logger.debug("所有任务已完成")
+        return True
 
 
 if __name__ == "__main__":
-    import time
-
-    def test_callback(sid, result):
-        print(f"[回填] {sid}: {result}")
-
-    async def main():
-        dispatcher = OptimizationDispatcher()
-        asyncio.create_task(dispatcher.start())
-
-        dispatcher.submit("s001", "we maybe start tomorrow okay", test_callback)
-        dispatcher.submit("s002", "they need eat fast meeting now", test_callback)
-
-        await asyncio.sleep(5)
-
-    asyncio.run(main())
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
+    
+    # 测试回调函数
+    def test_callback(sentence_id, original_text, optimized_text):
+        logger.info(f"[回填] {sentence_id}: {optimized_text}")
+    
+    # 创建调度器
+    dispatcher = OptimizationDispatcher(callback=test_callback)
+    
+    # 提交测试任务
+    dispatcher.submit("s001", "we maybe start tomorrow okay")
+    dispatcher.submit("s002", "they need eat fast meeting now")
+    
+    # 等待任务完成
+    dispatcher.wait_until_done()
+    
+    logger.info("测试完成")

optimizer/llm_api_runner.py

@@ -1,15 +1,61 @@
-# optimizer/llm_api_runner.py
+"""
+ChatGPT优化器 - 使用OpenAI API优化转写结果
+"""
+
 from openai import OpenAI
 import os
+import logging
+import time
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建日志记录器
+logger = setup_logger("optimizer.api")
 
-MODEL_NAME = "gpt-3.5-turbo"  # 可改为 "gpt-3.5-turbo"
+# 默认模型
+MODEL_NAME = "gpt-3.5-turbo"
 
 class ChatGPTRunner:
-    def __init__(self, model="gpt-3.5-turbo"):
+    """
+    ChatGPT优化器，使用OpenAI API优化转写结果
+    """
+    
+    def __init__(self, model: str = MODEL_NAME):
+        """
+        初始化ChatGPT优化器
+        
+        :param model: 使用的模型名称
+        """
         self.model = model
-        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
+        api_key = os.getenv("OPENAI_API_KEY")
+        if not api_key:
+            logger.warning("未设置OPENAI_API_KEY环境变量")
+        self.client = OpenAI(api_key=api_key)
+        logger.debug(f"ChatGPT优化器初始化完成，使用模型: {model}")
 
     def build_prompt(self, text: str) -> str:
+        """
+        构建优化提示
+        
+        :param text: 需要优化的文本
+        :return: 构建好的提示
+        """
         return (
             "示例：\n"
             "原句：你门现在就得开会了，别迟到了。\n"
@@ -20,26 +66,60 @@ class ChatGPTRunner:
             "修改：本章节将为你介绍音频数据的基本概念，包括波形、采样、频谱、图像。\n"
             "原句：系统将进入留言模式，请耐行等待。\n"
             "修改：系统将进入留言模式，请耐心等待。\n"
+            "原句：你会学习到如何使用音频数据集，包括音频数据加载，音频数据预处理，以及高效加载大规模音频数据集的流逝加载方法。\n"
+            "修改：你会学习到如何使用音频数据集，包括音频数据加载，音频数据预处理，以及高效加载大规模音频数据集的流式加载方法。\n"
             f"原句：{text}\n"
             f"修改："
         )
 
     def optimize(self, text: str, max_tokens: int = 256) -> str:
+        """
+        优化文本
+        
+        :param text: 需要优化的文本
+        :param max_tokens: 最大生成token数
+        :return: 优化后的文本
+        """
+        logger.debug(f"开始优化文本: {text}")
+        start_time = time.time()
+        
+        # 构建提示
         prompt = self.build_prompt(text)
-        response = self.client.chat.completions.create(
-            model=self.model,
-            messages=[
-                {"role": "system", "content": "你是用于优化语音识别的转写结果的校对助手。请保留原始句子的结构，仅修正错别字、语义不通或专业术语使用错误的部分。不要增加、删减或合并句子，务必保留原文的信息表达，仅对用词错误做最小修改。"},
-                {"role": "user", "content": prompt}
-            ],
-            temperature=0.4,
-            max_tokens=max_tokens,
-        )
-        return response.choices[0].message.content.strip()
+        
+        try:
+            # 调用API
+            response = self.client.chat.completions.create(
+                model=self.model,
+                messages=[
+                    {"role": "system", "content": "你是用于优化语音识别的转写结果的校对助手。请保留原始句子的结构，仅修正错别字、语义不通或专业术语使用错误的部分。不要增加、删减或合并句子，务必保留原文的信息表达，仅对用词错误做最小修改。"},
+                    {"role": "user", "content": prompt}
+                ],
+                temperature=0.4,
+                max_tokens=max_tokens,
+            )
+            
+            # 提取结果
+            result = response.choices[0].message.content.strip()
+            
+            # 记录耗时
+            elapsed_time = time.time() - start_time
+            logger.debug(f"优化完成，耗时: {elapsed_time:.2f}秒")
+            logger.info(f"优化结果: {result}")
+            
+            return result
+        except Exception as e:
+            logger.error(f"优化失败: {str(e)}")
+            # 出错时返回原文
+            return text
     
 if __name__ == "__main__":
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
+    
+    # ��试优化
     runner = ChatGPTRunner(MODEL_NAME)
     test_input = "你会学习到如何使用音频数据集，包括音频数据加载，音频数据预处理，以及高效加载大规模音频数据集的流逝加载方法。"
+    
+    logger.info(f"优化前: {test_input}")
     result = runner.optimize(test_input)
-    print("优化前：", test_input)
-    print("优化后：", result)
+    logger.info(f"优化后: {result}")

optimizer/optimize_task.py

@@ -1,23 +1,74 @@
-# optimizer/optimize_task.py
-from typing import Callable
-from optimizer.llm_runner import TinyLLaMARunner
+"""
+优化任务 - 表示一个LLM优化任务
+"""
+
+import logging
+from typing import Callable, Optional
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建日志记录器
+logger = setup_logger("optimizer.task")
 
 class OptimizeTask:
-    def __init__(self, sentence_id: str, text: str, callback: Callable[[str, str], None]):
+    """
+    优化任务，表示一个需要LLM优化的文本任务
+    """
+    
+    def __init__(self, sentence_id: str, text: str, callback: Optional[Callable[[str, str, str], None]] = None):
+        """
+        初始化优化任务
+        
+        :param sentence_id: 句子ID
+        :param text: 需要优化的文本
+        :param callback: 优化完成后的回调函数，接收参数(sentence_id, original_text, optimized_text)
+        """
         self.sentence_id = sentence_id
         self.text = text
         self.callback = callback
-
-    def run(self, model_runner: TinyLLaMARunner):
-        optimized_text = model_runner.optimize(self.text)
-        self.callback(self.sentence_id, optimized_text)
+        logger.debug(f"创建优化任务: {sentence_id}")
+    
+    def __str__(self):
+        """字符串表示"""
+        return f"OptimizeTask(id={self.sentence_id}, text={self.text[:20]}...)"
 
 
 if __name__ == "__main__":
-    def fake_callback(sid, text):
-        print(f"[回调] 优化结果：({sid}) -> {text}")
-
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
+    
+    # 测试回调函数
+    def fake_callback(sid, original_text, optimized_text):
+        logger.info(f"[回调] 优化结果：({sid}) -> {optimized_text}")
+    
+    # 创建任务
     task = OptimizeTask("s001", "they go home maybe tomorrow", fake_callback)
-    from optimizer.llm_runner import TinyLLaMARunner
-    runner = TinyLLaMARunner()
-    task.run(runner)
+    
+    # 创建模型运行器
+    from optimizer.llm_api_runner import ChatGPTRunner
+    runner = ChatGPTRunner()
+    
+    # 优化文本
+    optimized_text = runner.optimize(task.text)
+    
+    # 调用回调
+    if task.callback:
+        task.callback(task.sentence_id, task.text, optimized_text)
+    
+    logger.info("测试完成")

transcribe/transcribe.py

@@ -21,11 +21,25 @@ class TranscriptionResult:
     segment_index: Optional[int] = None  # 添加片段索引字段
 
 # 配置日志
-logger = logging.getLogger("transcribe")
-handler = logging.StreamHandler()
-formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
-handler.setFormatter(formatter)
-logger.addHandler(handler)
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建日志记录器
+logger = setup_logger("transcribe")
 
 class AudioTranscriber:
     def __init__(self, model: str = "medium", device: str = "cuda", compute_type: str = "int8", 
@@ -44,12 +58,12 @@ class AudioTranscriber:
             log_level = getattr(logging, log_level.upper())
         logger.setLevel(log_level)
         
-        logger.debug("📥 Loading Whisper model...")
+        logger.debug(f"初始化转录器: model={model}, device={device}, compute_type={compute_type}")
             
         from faster_whisper import WhisperModel
         self.model = WhisperModel(model, device=device, compute_type=compute_type)
         
-        logger.debug("📥 Loading Whisper model successfully!!")
+        logger.debug("Whisper模型加载完成")
 
     def transcribe_segment(self, audio_data: np.ndarray, start_time: float = 0.0) -> List[TranscriptionResult]:
         """
@@ -64,7 +78,6 @@ class AudioTranscriber:
         """
         start_process_time = time.time()
         
-        logger.debug("Model transcribe...")
         logger.debug(f"开始转录音频片段，长度: {len(audio_data)} 采样点 ({len(audio_data)/16000:.2f}秒)")
             
         try:
@@ -74,9 +87,9 @@ class AudioTranscriber:
                 
             segments = list(segments_generator)
             
-            logger.debug(f"Model transcribe successfully! Segments count: {len(segments)}")
+            logger.debug(f"转录成功，片段数: {len(segments)}")
             if len(segments) > 0:
-                logger.debug(f"First segment: {segments[0]}")
+                logger.debug(f"第一个片段: {segments[0]}")
                     
             results = []
             for seg in segments:
@@ -141,6 +154,7 @@ class AudioTranscriber:
         with open(output_path, 'w', encoding='utf-8') as f:
             json.dump(data, f, ensure_ascii=False, indent=2)
             
+        logger.info(f"转录结果已保存到: {output_path}")
         return output_path
 
 
@@ -149,12 +163,11 @@ if __name__ == "__main__":
     audio_path = "dataset/audio/test1.wav"  # 替换为实际的音频文件路径
     import soundfile as sf
     
-    # 设置日志级别: DEBUG, INFO, WARNING, ERROR, CRITICAL
-    # 可以通过字符串或常量设置
-    processor = AudioTranscriber(log_level="DEBUG")  # 或 log_level=logging.INFO
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
     
-    # 也可以直接设置logger级别
-    # logger.setLevel(logging.DEBUG)  # 查看所有详细日志
+    # 初始化转录器
+    processor = AudioTranscriber(log_level="DEBUG")
     
     try:
         audio_data, sample_rate = sf.read(audio_path)

translator/translator.py

@@ -1,20 +1,56 @@
-# translator_nllb.py
+"""
+翻译模块 - 使用NLLB模型进行多语言翻译
+"""
 
 from transformers import AutoTokenizer, AutoModelForSeq2SeqLM
 from langdetect import detect
 import torch
 import time
-
+import logging
+
+# 配置日志
+def setup_logger(name, level=logging.INFO):
+    """设置日志记录器"""
+    logger = logging.getLogger(name)
+    # 清除所有已有的handler，避免重复
+    if logger.handlers:
+        logger.handlers.clear()
+    
+    # 添加新的handler
+    handler = logging.StreamHandler()
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    handler.setFormatter(formatter)
+    logger.addHandler(handler)
+    logger.setLevel(level)
+    # 禁止传播到父logger，避免重复日志
+    logger.propagate = False
+    return logger
+
+# 创建日志记录器
+logger = setup_logger("translator")
 
 class NLLBTranslator:
+    """
+    NLLB翻译器，使用Facebook的NLLB模型进行多语言翻译
+    """
+    
     def __init__(self, model_name="facebook/nllb-200-distilled-600M", default_target="eng_Latn"):
+        """
+        初始化NLLB翻译器
+        
+        :param model_name: 模型名称
+        :param default_target: 默认目标语言代码
+        """
         self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
-        print(f"[⚙️ 模型加载中] 使用设备: {self.device}")
+        logger.debug(f"使用设备: {self.device}")
+        
         if self.device.type == "cuda":
-            print(f"[GPU] 当前设备: {torch.cuda.get_device_name(0)}")
+            logger.debug(f"GPU设备: {torch.cuda.get_device_name(0)}")
             total_mem = torch.cuda.get_device_properties(0).total_memory / 1024**3
-            print(f"[GPU] 显存总量: {total_mem:.1f} GB")
+            logger.debug(f"GPU显存: {total_mem:.1f} GB")
 
+        # 加载模型和分词器
+        logger.debug(f"加载模型: {model_name}")
         self.tokenizer = AutoTokenizer.from_pretrained(model_name, use_fast=False)
         self.model = AutoModelForSeq2SeqLM.from_pretrained(
             model_name,
@@ -22,51 +58,92 @@ class NLLBTranslator:
         ).to(self.device)
 
         self.default_target = default_target
+        logger.debug(f"翻译器初始化完成，默认目标语言: {default_target}")
 
     def detect_lang_code(self, text: str) -> str:
+        """
+        检测文本语言并返回NLLB语言代码
+        
+        :param text: 要检测的文本
+        :return: NLLB语言代码
+        """
         try:
             lang = detect(text)
+            logger.debug(f"检测到语言: {lang}")
         except Exception:
-            print("⚠️ 检测失败，默认 zh")
+            logger.debug("语言检测失败，默认使用中文(zh)")
             lang = "zh-cn"
 
+        # 语言代码映射
         lang_map = {
             "zh-cn": "zho_Hans", "zh": "zho_Hans", "en": "eng_Latn", "fr": "fra_Latn",
             "de": "deu_Latn", "ja": "jpn_Jpan", "ko": "kor_Hang", "ar": "arb_Arab"
         }
+        
         lang_code = lang_map.get(lang.lower(), "eng_Latn")
-        print(f"[🔍 语言识别] Detected `{lang}`, mapped to `{lang_code}`")
+        logger.debug(f"映射语言代码: {lang} -> {lang_code}")
         return lang_code
 
     def translate(self, text: str, target_lang_code: str = None) -> str:
-        print("\n🌐 [翻译任务启动]")
-        print(f"原文：{text}")
-
+        """
+        翻译文本到目标语言
+        
+        :param text: 要翻译的文本
+        :param target_lang_code: 目标语言代码，如果为None则使用默认目标语言
+        :return: 翻译后的文本
+        """
+        logger.debug("开始翻译")
+        
+        # 记录原文（INFO级别）
+        logger.info(f"[翻译原文] {text}")
+
+        # 检测源语言
         src_lang = self.detect_lang_code(text)
         tgt_lang = target_lang_code or self.default_target
 
+        # 准备输入
         self.tokenizer.src_lang = src_lang
         inputs = self.tokenizer(text, return_tensors="pt", padding=True, truncation=True).to(self.device)
         inputs["forced_bos_token_id"] = self.tokenizer.convert_tokens_to_ids(tgt_lang)
 
+        # 执行翻译
         start = time.time()
         with torch.no_grad():
             output = self.model.generate(**inputs, max_new_tokens=80)
+        
+        # 解码结果
         result = self.tokenizer.decode(output[0], skip_special_tokens=True)
-        print(f"[✅ 翻译完成] {src_lang} → {tgt_lang}，耗时：{time.time() - start:.2f}s")
+        
+        # 记录耗时和结果
+        duration = time.time() - start
+        logger.debug(f"翻译完成: {src_lang} -> {tgt_lang}, 耗时: {duration:.2f}秒")
+        
+        # 记录翻译结果（INFO级别）
+        logger.info(f"[翻译结果] {result}")
+        
         return result
 
 
 if __name__ == "__main__":
+    # 设置日志级别为DEBUG以查看详细信息
+    logger.setLevel(logging.DEBUG)
+    
+    # 创建翻译器
     translator = NLLBTranslator()
 
+    # 测试中文到英文
     zh_text = "你会学习到如何使用音频数据集"
-    print("\n==== 中文 → 英文 ====")
-    print("翻译结果：", translator.translate(zh_text, target_lang_code="eng_Latn"))
+    logger.info("\n==== 中文 → 英文 ====")
+    result = translator.translate(zh_text, target_lang_code="eng_Latn")
+    logger.info(f"测试完成: {result}")
 
+    # 测试英文到法语
     en_text = "This audio processing pipeline is fast and accurate."
-    print("\n==== 英文 → 法语 ====")
-    print("翻译结果：", translator.translate(en_text, target_lang_code="fra_Latn"))
-
-    print("\n==== 英文 → 阿拉伯语 ====")
-    print("翻译结果：", translator.translate(en_text, target_lang_code="arb_Arab"))
+    logger.info("\n==== 英文 → 法语 ====")
+    result = translator.translate(en_text, target_lang_code="fra_Latn")
+    logger.info(f"测试完成: {result}")
+
+    # 测试英文到阿拉伯语
+    logger.info("\n==== 英文 → 阿拉伯语 ====")
+    result = translator.translate(en_text, target_lang_code="arb_Arab")
+    logger.info(f"测试完成: {result}")

vad/__init__.py

@@ -1,3 +1,36 @@
 from .vad import AudioVad, AudioSegment
+import numpy as np
+from typing import List, Tuple
 
-__all__ = ['AudioVad', 'AudioSegment']
+class VoiceActivityDetector:
+    """
+    VAD检测器，用于检测音频中的语音片段
+    这是一个包装类，内部使用AudioVad实现功能
+    """
+    
+    def __init__(self, save_audio=True, save_json=True):
+        """
+        初始化VAD检测器
+        
+        :param save_audio: 是否保存分段音频
+        :param save_json: 是否保存JSON元数据
+        """
+        self.vad = AudioVad(
+            save_audio=save_audio,
+            save_json=save_json,
+            output_dir="dataset/audio/segments",
+            json_dir="dataset/audio/metadata"
+        )
+    
+    def detect_voice_segments(self, audio_data: np.ndarray, sample_rate: int) -> List[Tuple[float, float]]:
+        """
+        检测音频中的语音片段
+        
+        :param audio_data: 音频数据
+        :param sample_rate: 采样率
+        :return: 语音片段列表，每个片段为(开始时间, 结束时间)的元组
+        """
+        segments = self.vad.process_audio_data(audio_data, sample_rate)
+        return [(segment.start_time, segment.end_time) for segment in segments]
+
+__all__ = ['AudioVad', 'AudioSegment', 'VoiceActivityDetector']