update code for readability

config.py

@@ -17,6 +17,10 @@ ASSERT_DIR = BASE_DIR / "assets"
 # 标点
 SENTENCE_END_MARKERS =  ['.', '!', '?', '。', '！', '？', ';', '；', ':', '：']
 PAUSE_END_MARKERS = [',', '，', '、']
+# 合并所有标点
+ALL_MARKERS = SENTENCE_END_MARKERS + PAUSE_END_MARKERS
+# 构造正则表达式字符类
+REGEX_MARKERS = re.compile(r'[' + re.escape(''.join(ALL_MARKERS)) + r']')
 
 sentence_end_chars = ''.join([re.escape(char) for char in SENTENCE_END_MARKERS])
 SENTENCE_END_PATTERN = re.compile(f'[{sentence_end_chars}]')

main.py

@@ -1,6 +1,6 @@
 from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 from urllib.parse import urlparse, parse_qsl
-from transcribe.whisper_llm_serve import PyWhiperCppServe
+from transcribe.whisper_llm_serve import WhisperTranscriptionService
 from uuid import uuid1
 from logging import getLogger
 import numpy as np
@@ -57,7 +57,7 @@ async def root():
 async def translate(websocket: WebSocket):
     query_parameters_dict = websocket.query_params
     from_lang, to_lang = query_parameters_dict.get('from'), query_parameters_dict.get('to')
-    client = PyWhiperCppServe(
+    client = WhisperTranscriptionService(
         websocket,
         pipe,
         language="en",

transcribe/strategy.py

@@ -1,153 +1,280 @@
-
-
-from logging import getLogger
-from difflib import SequenceMatcher
+import re
 import collections
-import config
-import numpy as np
+import logging
+from difflib import SequenceMatcher
 from itertools import chain
+from dataclasses import dataclass
+from typing import List, Tuple, Optional, Deque, Any, Iterator
+from config import SENTENCE_END_MARKERS, ALL_MARKERS,SENTENCE_END_PATTERN,REGEX_MARKERS
+import numpy as np
 
-logger = getLogger("Stragegy")
+logger = logging.getLogger("TranscriptionStrategy")
 
-class TripleTextBuffer:
-    def __init__(self, size=2):
-        self.history = collections.deque(maxlen=size)
 
-    def add_entry(self, text, index):
+@dataclass
+class TranscriptSegment:
+    """表示一个转录片段，包含文本和时间信息"""
+    text: str
+    t0: float  # 开始时间（百分之一秒）
+    t1: float  # 结束时间（百分之一秒）
+
+
+class TextStabilityBuffer:
+    """
+    通过比较连续文本样本的相似度来确定转录文本的稳定性。
+    当连续样本的相似度超过阈值时，认为文本已稳定。
+    """
+    def __init__(self, max_history: int = 2):
+        self.history: Deque[Tuple[str, int]] = collections.deque(maxlen=max_history)
+
+    def add_entry(self, text: str, index: int) -> None:
         """
-        text: 文本
-        index: 当前buffer的相对下标 数组索引
+        添加新的文本和索引到历史记录中
+        
+        Args:
+            text: 文本内容
+            index: 当前buffer的相对下标
         """
         self.history.append((text, index))
 
-
-    def get_final_index(self, similarity_threshold=0.7):
-        """根据文本变化，返回可靠的标点的buffer的位置下标"""
+    def get_stable_index(self, similarity_threshold: float = 0.7) -> Optional[int]:
+        """
+        根据文本相似度，判断文本是否稳定，返回稳定文本的索引
+        
+        Args:
+            similarity_threshold: 相似度阈值，超过此值认为文本稳定
+            
+        Returns:
+            稳定文本的索引，如果没有找到稳定文本则返回None
+        """
         if len(self.history) < 2:
             return None
 
-        # 获取三次的文本
         text1, _ = self.history[0]
         text2, idx2 = self.history[1]
-        # text3, idx3 = self.history[2]
 
-        # 计算变化程度
-        sim_12 = self.text_similarity(text1, text2)
-        # print("比较： ", text1, text2," => ", sim_12)
-        # sim_23 = self.text_similarity(text2, text3)
-        if sim_12 >= similarity_threshold:
+        similarity = self._calculate_similarity(text1, text2)
+        
+        if similarity >= similarity_threshold:
             self.history.clear()
             return idx2
         return None
 
     @staticmethod
-    def text_similarity(text1, text2):
+    def _calculate_similarity(text1: str, text2: str) -> float:
+        """计算两段文本的相似度"""
         return SequenceMatcher(None, text1, text2).ratio()
 
 
 
-class SegmentManager:
-    def __init__(self) -> None:
-        self._commited_segments = [] # 确定后的段落
-        self._commited_short_sentences = [] # 确定后的序列
-        self._temp_string = "" # 存储当前临时的文本字符串，直到以句号结尾
-
-    def handle(self, string):
-        self._temp_string = string
-        return self
+class TranscriptionManager:
+    """
+    管理转录文本的分级结构：临时字符串 -> 短句 -> 完整段落
 
+    |-- 已确认文本 --|-- 观察窗口 --|-- 新输入 --|
+    """
+    def __init__(self):
+        self._committed_segments: List[str] = []  # 确认的完整段落
+        self._committed_sentences: List[str] = []  # 确认的短句
+        self._temp_string: str = ""  # 临时字符串缓冲
+        
     @property
-    def short_sentence(self) -> str:
-        return "".join(self._commited_short_sentences)
-
+    def current_sentence(self) -> str:
+        """当前已确认的短句组合"""
+        return "".join(self._committed_sentences)
+    
     @property
-    def segment(self):
-        return self._commited_segments[-1] if len(self._commited_segments) > 0 else ""
-
-    def get_seg_id(self):
-        return len(self._commited_segments)
-
+    def latest_segment(self) -> str:
+        """最新确认的完整段落"""
+        return self._committed_segments[-1] if self._committed_segments else ""
+    
     @property
-    def string(self):
-        return self._temp_string
-
-
-    def commit_short_sentence(self):
-        """将临时字符串 提交到临时短句"""
-        self._commited_short_sentences.append(self._temp_string)
-        self._temp_string = ""
-
-    def commit_segment(self):
-        """将短句 合并 到长句中"""
-        self._commited_segments.append(self.short_sentence)
-        self._commited_short_sentences = []
-
-    def commit(self, is_end_sentence=False):
+    def segment_count(self) -> int:
+        """已确认的段落数量"""
+        return len(self._committed_segments)
+    
+    @property
+    def sentence_length(self) -> int:
+        """当前短句的总字符长度"""
+        return sum(len(s) for s in self._committed_sentences)
+    
+    def update_temp(self, text: str) -> 'TranscriptionManager':
+        """更新临时字符串"""
+        self._temp_string = text
+        return self
+    
+    def commit_sentence(self) -> None:
+        """将临时字符串提交到短句列表"""
+        if self._temp_string:
+            self._committed_sentences.append(self._temp_string)
+            self._temp_string = ""
+    
+    def commit_segment(self, is_end_of_sentence: bool = False) -> None:
         """
-        当需要切掉的音频部分的时候，将句子提交到短句队列中，并移除临时字符串
-        当完成一个整句的时候提交到段落中
+        提交当前内容到适当的层级
+        
+        Args:
+            is_end_of_sentence: 是否为完整句子的结束
         """
-        self.commit_short_sentence()
-        if is_end_sentence:
-            self.commit_segment()
-
-def segement_merge(segments):
-    """根据标点符号分整句"""
-    sequences = []
-    temp_seq = []
-
-    for seg in segments:
-        temp_seq.append(seg)
-        if any([mk in seg.text for mk in config.SENTENCE_END_MARKERS]):
-            sequences.append(temp_seq.copy())
-            temp_seq = []
-    if temp_seq:
-        sequences.append(temp_seq)
-    return sequences
-
-def segments_split(segments, audio_buffer: np.ndarray, sample_rate=16000):
-    """根据左边第一个标点符号来将序列拆分成 观察段 和 剩余部分"""
-    left_watch_sequences = []
-    left_watch_idx = 0
-    right_watch_sequences = []
-    is_end = False
-
-    if (len(audio_buffer) / sample_rate) < 12:
-        # 低于12s 使用短句符号比如逗号作为判断依据
-        markers = config.PAUSE_END_MARKERS + config.SENTENCE_END_MARKERS
-        is_end = False
-
+        self.commit_sentence()
+        if is_end_of_sentence and self._committed_sentences:
+            self._committed_segments.append(self.current_sentence)
+            self._committed_sentences = []
+            
+    def get_all_text(self) -> str:
+        """获取所有已提交的文本"""
+        all_segments = self._committed_segments.copy()
+        if self.current_sentence:
+            all_segments.append(self.current_sentence)
+        if self._temp_string:
+            all_segments.append(self._temp_string)
+        return "\n".join(all_segments)
+
+
+class TranscriptionSplitter:
+    """负责根据语音和文本特征拆分转录片段"""
+    
+    @staticmethod
+    def group_by_sentences(segments: List[TranscriptSegment]) -> List[List[TranscriptSegment]]:
+        """将片段按照完整句子分组"""
+        sequences = []
+        temp_seq = []
+
+        for seg in segments:
+            temp_seq.append(seg)
+            if any(marker in seg.text for marker in SENTENCE_END_MARKERS):
+                sequences.append(temp_seq.copy())
+                temp_seq = []
+                
+        if temp_seq:
+            sequences.append(temp_seq)
+        return sequences
+    
+    @staticmethod
+    def split_by_punctuation(
+        segments: List[TranscriptSegment], 
+        audio_buffer: np.ndarray, 
+        sample_rate: int = 16000
+    ) -> Tuple[int, List[TranscriptSegment], List[TranscriptSegment], bool]:
+        """
+        根据标点符号将片段分为左侧（已确认）和右侧（待确认）
+        
+        Returns:
+            (分割索引, 左侧片段, 右侧片段, 是否为句子结束)
+        """
+        left_segments = []
+        right_segments = []
+        split_index = 0
+        is_sentence_end = False
+        
+        # 短音频使用所有标点符号作为分割依据
+        buffer_duration = len(audio_buffer) / sample_rate
+        markers = ALL_MARKERS if buffer_duration < 12 else SENTENCE_END_MARKERS
+        
         for idx, seg in enumerate(segments):
-            # print('>>>>>>>>>>>>>>>> seg : ', seg)
-            left_watch_sequences.append(seg)
+            left_segments.append(seg)
             if seg.text and seg.text[-1] in markers:
-                seg_index = int(seg.t1 / 100 * sample_rate)
-                # rest_buffer_duration = (len(audio_buffer) - seg_index) / sample_rate
-                is_end = config.SENTENCE_END_PATTERN.search(seg.text)
+                split_index = int(seg.t1 / 100 * sample_rate)
+                is_sentence_end = bool(SENTENCE_END_PATTERN.search(seg.text))
+                right_segments = segments[min(idx+1, len(segments)):]
+                break
                 
-                right_watch_sequences = segments[min(idx+1, len(segments)):]
-                # if rest_buffer_duration >= 1.5:
-                left_watch_idx = seg_index
+        return split_index, left_segments, right_segments, is_sentence_end
+    
+    @staticmethod
+    def split_by_sequences(
+        segments: List[TranscriptSegment], 
+        audio_buffer: np.ndarray, 
+        sample_rate: int = 16000
+    ) -> Tuple[int, Iterator[TranscriptSegment], Iterator[TranscriptSegment], bool]:
+        """
+        对于长文本，按照句子组保留最新的两句
+        
+        Returns:
+            (分割索引, 左侧片段, 右侧片段, 是否为句子结束)
+        """
+        sequences = TranscriptionSplitter.group_by_sentences(segments)
+        
+        if len(sequences) > 2:
+            logger.info(f"Buffer clip via sequence, current length: {len(sequences)}")
+            left_segments = chain(*sequences[:-2])
+            right_segments = chain(*sequences[-2:])
+            
+            # 确定切分点
+            last_sequence = sequences[-3]
+            last_segment = last_sequence[-1]
+            split_index = int(last_segment.t1 / 100 * sample_rate)
+            
+            return split_index, left_segments, right_segments, True
+            
+        return 0, iter([]), iter(segments), False
+
+
+class TranscriptionStabilizer:
+    """
+    转录结果稳定器，负责确认和管理转录片段
+    """
+    def __init__(self, sample_rate: int = 16000):
+        self.manager = TranscriptionManager()
+        self.stability_buffer = TextStabilityBuffer(max_history=2)
+        self.sample_rate = sample_rate
+        
+    def process_segments(self, segments: List[TranscriptSegment]) -> Tuple[Optional[int], bool]:
+        """
+        处理转录片段，确认稳定的文本
+        
+        Args:
+            segments: 转录片段列表
+            
+        Returns:
+            (音频分割点索引, 是否达到足够长度需要换行)
+        """
+        # 查找第一个包含标点的片段作为分割点
+        split_index = None
+        stable_segments = []
+        
+        for idx, seg in enumerate(segments):
+            stable_segments.append(seg)
+            if REGEX_MARKERS.search(seg.text):
+                split_index = int(seg.t1 / 100 * self.sample_rate)
+                stable_idx = min(idx + 1, len(segments))
                 break
-    return left_watch_idx, left_watch_sequences, right_watch_sequences, is_end
-
-
-def sequences_split(segments, audio_buffer: np.ndarray, sample_rate=16000):
-    # 长句 保留最后两句即可
-    left_watch_sequences = []
-    right_watch_sequences = []
-    left_watch_idx = 0
-    is_end = False
-    sequences = segement_merge(segments)
-
-    if len(sequences) > 2:
-        logger.info(f"buffer clip via sequence, current length: {len(sequences)}")
-        is_end = True
-        left_watch_sequences = chain(*sequences[:-2])
-        right_watch_sequences = chain(*sequences[-2:])
-        last_sequence_segment = sequences[-3]
-        last_segment = last_sequence_segment[-1]
-        left_watch_idx = int(last_segment.t1 / 100 * sample_rate)
-    return left_watch_idx, left_watch_sequences, right_watch_sequences, is_end
-
-
+        
+        if split_index:  # 找到标点，确认标点前的内容
+            stable_text = self._join_segment_text(segments[:stable_idx])
+            self.manager.update_temp(stable_text).commit_sentence()
+            
+            # 更新剩余文本
+            remaining_text = self._join_segment_text(segments[stable_idx:])
+            self.manager.update_temp(remaining_text)
+        else:
+            # 没有找到标点，全部作为临时文本
+            self.manager.update_temp(self._join_segment_text(segments))
+            
+        # 检查是否达到换行标准
+        should_linebreak = self.manager.sentence_length >= 20
+        
+        return split_index, should_linebreak
+    
+    def check_stability(self, text: str, index: int) -> Optional[int]:
+        """
+        检查文本是否稳定
+        
+        Args:
+            text: 当前文本
+            index: 当前索引
+            
+        Returns:
+            如果文本稳定，返回稳定的索引；否则返回None
+        """
+        self.stability_buffer.add_entry(text, index)
+        return self.stability_buffer.get_stable_index()
+    
+    def commit_segment(self, is_end_of_sentence: bool) -> None:
+        """提交转录片段"""
+        self.manager.commit_segment(is_end_of_sentence)
+    
+    @staticmethod
+    def _join_segment_text(segments: List[TranscriptSegment], separator: str = "") -> str:
+        """连接多个片段的文本"""
+        return separator.join(seg.text for seg in segments)

transcribe/whisper_llm_serve.py

@@ -1,261 +1,309 @@
-
-
-import numpy as np
-from logging import getLogger
 import asyncio
-from .utils import save_to_wave
-import time
 import json
-import threading
-from .server import ServeClientBase
 import queue
-import collections
+import threading
+import time
+from logging import getLogger
+from typing import List, Optional, Iterator, Tuple, Any
+
+import numpy as np
+
 from api_model import TransResult, Message
-from .utils import log_block
+from .server import ServeClientBase
+from .utils import log_block, save_to_wave
 from .translatepipes import TranslatePipes
-from .strategy import TripleTextBuffer, SegmentManager, segments_split, sequences_split
-
-logger = getLogger("TranslatorApp")
+from .strategy import TextStabilityBuffer, TranscriptionManager, TranscriptionSplitter, TranscriptSegment
 
+logger = getLogger("TranscriptionService")
 
 
-class PyWhiperCppServe(ServeClientBase):
-   
-    def __init__(self, websocket, pipe:TranslatePipes,language=None, dst_lang=None, client_uid=None,):
+class WhisperTranscriptionService(ServeClientBase):
+    """
+    Whisper语音转录服务类，处理音频流转录和翻译
+    """
+    
+    def __init__(self, websocket, pipe: TranslatePipes, language=None, dst_lang=None, client_uid=None):
         super().__init__(client_uid, websocket)
-        self.language = language
-        self.dst_lang = dst_lang # 目标翻译语言
-        # 设置观察字符串 对比上下次的文字来判断字符串的输出是否固定
-        self._text_buffer = TripleTextBuffer()
-        # 存储转录数据
-        self._segment_manager = SegmentManager()
-        self._translate_pipes = pipe
-        self.lock = threading.Lock()
+        self.source_language = language  # 源语言
+        self.target_language = dst_lang  # 目标翻译语言
+        
+        # 转录结果稳定性管理
+        self._text_stability_buffer = TextStabilityBuffer()
+        self._transcription_manager = TranscriptionManager()
+        self._translate_pipe = pipe
+        
+        # 音频处理相关
+        self.sample_rate = 16000
         self.frames_np = None
+        self.lock = threading.Lock()
         self._frame_queue = queue.Queue()
-        self.sample_rate = 16000
- 
+        
+        # 文本分隔符，根据语言设置
+        self.text_separator = self._get_text_separator(language)
+        
+        # 发送就绪状态
         self.send_ready_state()
+        
+        # 启动处理线程
         self._translate_thread_stop = threading.Event()
-        self._frame_to_queue_thread_stop = threading.Event()
-        self.translate_thread = self.run_in_thread(self.speech_to_text)
-        self.frame_to_queue_thread = self.run_in_thread(self.get_frame_from_queue)
+        self._frame_processing_thread_stop = threading.Event()
+        self.translate_thread = self._start_thread(self._transcription_processing_loop)
+        self.frame_processing_thread = self._start_thread(self._frame_processing_loop)
 
-        self.text_sep = ""
+    def _start_thread(self, target_function) -> threading.Thread:
+        """启动守护线程执行指定函数"""
+        thread = threading.Thread(target=target_function)
+        thread.daemon = True
+        thread.start()
+        return thread
 
-    def run_in_thread(self, func):
-        t = threading.Thread(target=func)
-        t.daemon = True
-        t.start()
-        return t
+    def _get_text_separator(self, language: str) -> str:
+        """根据语言返回适当的文本分隔符"""
+        return "" if language == "zh" else " "
 
-    def send_ready_state(self):
+    def send_ready_state(self) -> None:
+        """发送服务就绪状态消息"""
         self.websocket.send(json.dumps({
             "uid": self.client_uid,
             "message": self.SERVER_READY,
-            "backend": "pywhispercpp"
+            "backend": "whisper_transcription"
         }))
 
-    def set_lang(self, src_lang, dst_lang):
-        self.language = src_lang
-        self.dst_lang = dst_lang
-        self.text_sep = "" if self.language == "zh" else " "
+    def set_language(self, source_lang: str, target_lang: str) -> None:
+        """设置源语言和目标语言"""
+        self.source_language = source_lang
+        self.target_language = target_lang
+        self.text_separator = self._get_text_separator(source_lang)
 
-    def add_frames(self, frame_np):
+    def add_audio_frames(self, frame_np: np.ndarray) -> None:
+        """添加音频帧到处理队列"""
         self._frame_queue.put(frame_np)
 
-    def vad_merge(self):
-        with self.lock:
-            frame = self.frames_np.copy()
-            item = self._translate_pipes.voice_detect(frame.tobytes())
-            frame_np = np.frombuffer(item.audio, dtype=np.float32)
-            self.frames_np = frame_np.copy()
-     
-    
-    def get_frame_from_queue(self,):
-        while not self._frame_to_queue_thread_stop.is_set():
+    def _frame_processing_loop(self) -> None:
+        """从队列获取音频帧并合并到缓冲区"""
+        while not self._frame_processing_thread_stop.is_set():
             try:
                 frame_np = self._frame_queue.get(timeout=0.1)
                 with self.lock:
                     if self.frames_np is None:
                         self.frames_np = frame_np.copy()
                     else:
-                        self.frames_np = np.append(self.frames_np,frame_np)
+                        self.frames_np = np.append(self.frames_np, frame_np)
             except queue.Empty:
                 pass
-          
 
-    def update_audio_buffer(self, last_offset):
+    def _apply_voice_activity_detection(self) -> None:
+        """应用语音活动检测来优化音频缓冲区"""
         with self.lock:
-            self.frames_np = self.frames_np[last_offset:]
+            if self.frames_np is not None:
+                frame = self.frames_np.copy()
+                processed_audio = self._translate_pipe.voice_detect(frame.tobytes())
+                self.frames_np = np.frombuffer(processed_audio.audio, dtype=np.float32).copy()
 
-    def transcribe_audio(self, audio_buffer):
-        """
-         Transcribe the audio chunk and send the results to the client.
+    def _update_audio_buffer(self, offset: int) -> None:
+        """从音频缓冲区中移除已处理的部分"""
+        with self.lock:
+            if self.frames_np is not None and offset > 0:
+                self.frames_np = self.frames_np[offset:]
 
-        Args:
-            audio_buffer (np.array): The audio chunk to transcribe.
-        """
+    def _get_audio_for_processing(self) -> Optional[np.ndarray]:
+        """准备用于处理的音频块"""
+        # 应用VAD处理
+        self._apply_voice_activity_detection()
+        
+        # 没有音频帧
+        if self.frames_np is None:
+            return None
+            
+        frames = self.frames_np.copy()
+        
+        # 音频过短时的处理
+        if len(frames) <= 100:
+            # 极短音频段，清空并返回None
+            self._update_audio_buffer(len(frames))
+            return None
+        elif len(frames) < self.sample_rate:
+            # 不足一秒的音频，补充静音
+            silence_audio = np.zeros((self.sample_rate + 1000,), dtype=np.float32)
+            silence_audio[-len(frames):] = frames
+            return silence_audio.copy()
+        
+        return frames.copy()
 
+    def _transcribe_audio(self, audio_buffer: np.ndarray) -> List[TranscriptSegment]:
+        """转录音频并返回转录片段"""
         log_block("Audio buffer length", f"{audio_buffer.shape[0]/self.sample_rate:.2f}", "s")
         start_time = time.perf_counter()
 
-        item = self._translate_pipes.transcrible(audio_buffer.tobytes(), self.language)
-        segments = item.segments
-        log_block("Whisper transcrible out", f"{''.join(seg.text for seg in segments)}", "")
-        log_block("Whisper transcrible time", f"{(time.perf_counter() - start_time):.3f}", "s")
-       
+        result = self._translate_pipe.transcrible(audio_buffer.tobytes(), self.source_language)
+        segments = result.segments
+        
+        log_block("Whisper transcription output", f"{''.join(seg.text for seg in segments)}", "")
+        log_block("Whisper transcription time", f"{(time.perf_counter() - start_time):.3f}", "s")
+        
         return segments
-    
-    def translate_text(self, text):
-        """
-        translate the text to dst lang"""
-        # return "sample english"
-        log_block("LLM translate input", f"{text}")
+
+    def _translate_text(self, text: str) -> str:
+        """将文本翻译为目标语言"""
+        if not text.strip():
+            return ""
+            
+        log_block("Translation input", f"{text}")
         start_time = time.perf_counter()
-        ret =  self._translate_pipes.translate(text, self.language, self.dst_lang)
-        translated_text = ret.translate_content
-        log_block("LLM translate time", f"{(time.perf_counter() - start_time):.3f}", "s")
-        log_block("LLM translate out", f"{translated_text}")
+        
+        result = self._translate_pipe.translate(text, self.source_language, self.target_language)
+        translated_text = result.translate_content
+        
+        log_block("Translation time", f"{(time.perf_counter() - start_time):.3f}", "s")
+        log_block("Translation output", f"{translated_text}")
+        
         return translated_text
-    
 
-    
-    def analysis_segments(self, segments,  audio_buffer: np.ndarray):
-          # 找到第一个标点符号作为锚点 左边为确认段，右边为观察段，
-        #   当左边确认后，右边段才会进入观察
-        #   当左边确认后，会从缓冲区中删除对应的buffer，减少下次输入的数据量
-        left_watch_idx, left_watch_sequences, right_watch_sequences, is_end_sentence = segments_split(segments, audio_buffer)
-        left_watch_string = self.text_sep.join(i.text for i in left_watch_sequences)
-        right_watch_string = self.text_sep.join(i.text for i in right_watch_sequences)
+    def _analyze_segments(self, segments: List[TranscriptSegment], audio_buffer: np.ndarray) -> Tuple[Optional[int], str, str, bool]:
+        """
+        分析转录片段，确定稳定部分和需要继续观察的部分
+        
+        Returns:
+            (分割索引, 左侧稳定文本, 右侧观察文本, 是否为句子结束)
+        """
+        # 尝试基于标点符号进行分割
+        left_idx, left_segments, right_segments, is_end = TranscriptionSplitter.split_by_punctuation(
+            segments, audio_buffer, self.sample_rate
+        )
+        
+        left_text = self.text_separator.join(seg.text for seg in left_segments)
+        right_text = self.text_separator.join(seg.text for seg in right_segments)
         
-        if left_watch_idx != 0:
-            # 将观察字符串临时存储
-            self._text_buffer.add_entry(left_watch_string, left_watch_idx)
-            audio_cut_index = self._text_buffer.get_final_index()
-            if audio_cut_index:
-                return audio_cut_index, left_watch_string, right_watch_string, is_end_sentence
+        # 如果找到分割点，检查左侧文本稳定性
+        if left_idx != 0:
+            self._text_stability_buffer.add_entry(left_text, left_idx)
+            stable_idx = self._text_stability_buffer.get_stable_index()
+            if stable_idx:
+                return stable_idx, left_text, right_text, is_end
         
-        # 整句消除 后两句之前的内容
-        left_watch_idx, left_watch_sequences, right_watch_sequences, is_end_sentence = sequences_split(segments, audio_buffer)
-        left_watch_string = self.text_sep.join(i.text for i in left_watch_sequences)
-        right_watch_string = self.text_sep.join(i.text for i in right_watch_sequences)
-        if left_watch_idx != 0:
-            return left_watch_idx, left_watch_string, right_watch_string, is_end_sentence
+        # 如果基于标点的方法未找到稳定点，尝试基于句子序列的方法
+        left_idx, left_segments, right_segments, is_end = TranscriptionSplitter.split_by_sequences(
+            segments, audio_buffer, self.sample_rate
+        )
         
-        return None, left_watch_string, right_watch_string, is_end_sentence
+        if left_idx != 0:
+            left_text = self.text_separator.join(seg.text for seg in left_segments)
+            right_text = self.text_separator.join(seg.text for seg in right_segments)
+            return left_idx, left_text, right_text, is_end
+        
+        # 如果都没有找到分割点
+        return None, left_text, right_text, is_end
 
-    def speech_to_text(self):
-        c = 0
+    def _transcription_processing_loop(self) -> None:
+        """主转录处理循环"""
         while not self._translate_thread_stop.is_set():
             if self.exit:
-                logger.info("Exiting speech to text thread")
+                logger.info("Exiting transcription thread")
                 break
-    
-            if self.frames_np is None :
-                time.sleep(0.02)  # wait for any audio to arrive
-                logger.info("waiting for client data...")
+                
+            # 等待音频数据
+            if self.frames_np is None:
+                time.sleep(0.02)
+                logger.info("Waiting for audio data...")
                 continue
-
-            audio_buffer = self.get_audio_chunk_for_processing()
+                
+            # 获取音频块进行处理
+            audio_buffer = self._get_audio_for_processing()
             if audio_buffer is None:
-                time.sleep(0.02)  # wait for any audio to arrive
+                time.sleep(0.02)
                 continue
-            
-            # c+= 1
-            # name = f"dev-{c}.wav"
-            # save_to_wave(name, audio_buffer)
-            # try:
-            logger.info(f"Audio buffer length: {len(audio_buffer) / self.sample_rate:.2f}s")
-            segments = self.transcribe_audio(audio_buffer)
-            for tran_result in self.handle_transcription_output(segments, audio_buffer):
-                self.send_to_client(tran_result)
-            # except KeyboardInterrupt:
-            #     break
-            # except Exception as e:
-            #     logger.error(f"{e}")
-            # if (time_delay := (1 - audio_duration)) > 0:
-            #     time.sleep(time_delay)
-    
-    def handle_transcription_output(self, segments, audio_buffer):
-        texts  = self.text_sep.join(i.text for i in segments)
-        if not len(texts):
+                
+            try:
+                logger.info(f"Processing audio buffer: {len(audio_buffer)/self.sample_rate:.2f}s")
+                segments = self._transcribe_audio(audio_buffer)
+                
+                # 处理转录结果并发送到客户端
+                for result in self._process_transcription_results(segments, audio_buffer):
+                    self._send_result_to_client(result)
+                    
+            except Exception as e:
+                logger.error(f"Error processing audio: {e}")
+
+    def _process_transcription_results(self, segments: List[TranscriptSegment], audio_buffer: np.ndarray) -> Iterator[TransResult]:
+        """
+        处理转录结果，生成翻译结果
+        
+        Returns:
+            TransResult对象的迭代器
+        """
+        # 合并所有片段的文本
+        full_text = self.text_separator.join(seg.text for seg in segments)
+        if not full_text:
             return
-        self._segment_manager.handle(texts)
-         # 分析句子
-        last_cut_index, left_string, right_string, is_end_sentence = self.analysis_segments(segments, audio_buffer)
-        # print(last_cut_index, left_string, right_string, is_end_sentence)
-        if last_cut_index:
-            self.update_audio_buffer(last_cut_index)
-            # 句子或者短句的提交
-            log_block("Whisper string lock ", f"{left_string}",)
-            self._segment_manager.handle(left_string).commit(is_end_sentence)
-            self._segment_manager.handle(right_string)
+            
+        # 更新转录管理器中的临时文本
+        self._transcription_manager.update_temp(full_text)
         
-        if is_end_sentence and last_cut_index:
-            message = self._segment_manager.segment
-            seg_id = self._segment_manager.get_seg_id() - 1
-            # logger.info(f"{seg_id}, {message}")
-            yield TransResult(
-                seg_id=seg_id,
-                context=message,
-                from_=self.language,
-                to=self.dst_lang,
-                tran_content=self.translate_text(message),
-                partial=False
-            )
-            if self._segment_manager.string.strip():
-                message = self._segment_manager.string.strip()
-                # logger.info(f"{seg_id + 1}, {message}")
-                yield TransResult(
-                    seg_id=seg_id+1,
-                    context=self._segment_manager.string,
-                    from_=self.language,
-                    to=self.dst_lang,
-                    tran_content=self.translate_text(message),
-            )
+        # 分析片段，确定稳定部分和需要继续观察的部分
+        cut_index, stable_text, remaining_text, is_sentence_end = self._analyze_segments(segments, audio_buffer)
         
+        # 如果找到稳定的分割点
+        if cut_index:
+            # 更新音频缓冲区，移除已处理部分
+            self._update_audio_buffer(cut_index)
+            
+            # 提交稳定的文本
+            log_block("Stable transcription", f"{stable_text}")
+            self._transcription_manager.update_temp(stable_text).commit_segment(is_sentence_end)
+            self._transcription_manager.update_temp(remaining_text)
+            
+            # 如果是句子结束，发送完整句子的翻译结果
+            if is_sentence_end:
+                segment_text = self._transcription_manager.latest_segment
+                segment_id = self._transcription_manager.segment_count - 1
+                
+                # 生成已确认句子的翻译结果
+                yield TransResult(
+                    seg_id=segment_id,
+                    context=segment_text,
+                    from_=self.source_language,
+                    to=self.target_language,
+                    tran_content=self._translate_text(segment_text),
+                    partial=False
+                )
+                
+                # 如果还有剩余部分，生成临时翻译结果
+                if self._transcription_manager.current_sentence.strip():
+                    yield TransResult(
+                        seg_id=segment_id + 1,
+                        context=self._transcription_manager.current_sentence,
+                        from_=self.source_language,
+                        to=self.target_language,
+                        tran_content=self._translate_text(self._transcription_manager.current_sentence.strip()),
+                        partial=True
+                    )
         else:
-            seg_id = self._segment_manager.get_seg_id()
-            message = self._segment_manager.short_sentence + self._segment_manager.string
-            # logger.info(f"{seg_id}, {message}")
+            # 没有找到稳定点，发送当前所有内容的临时翻译结果
+            segment_id = self._transcription_manager.segment_count
+            current_text = self._transcription_manager.current_sentence + self._transcription_manager.update_temp(remaining_text)._temp_string
+            
             yield TransResult(
-                seg_id=seg_id,
-                context=message,
-                from_=self.language,
-                to=self.dst_lang,
-                tran_content=self.translate_text(message),
+                seg_id=segment_id,
+                context=current_text,
+                from_=self.source_language,
+                to=self.target_language,
+                tran_content=self._translate_text(current_text),
+                partial=True
             )
-    
-    def send_to_client(self, data:TransResult):
+
+    def _send_result_to_client(self, result: TransResult) -> None:
+        """发送翻译结果到客户端"""
         try:
-            coro = self.websocket.send_text(
-                Message(result=data, request_id=self.client_uid).model_dump_json(by_alias=True)
-            )
+            message = Message(result=result, request_id=self.client_uid).model_dump_json(by_alias=True)
+            coro = self.websocket.send_text(message)
             asyncio.run(coro)
-        except RuntimeError as e:
+        except RuntimeError:
             self.stop()
-            return 
         except Exception as e:
-            logger.error(e)
-
-    
-
-    def get_audio_chunk_for_processing(self):
-        self.vad_merge()
-        silence_audio = np.zeros((self.sample_rate+1000,), dtype=np.float32)
-        frames = self.frames_np.copy()
-            # 添加对非常短音频的处理
-        if len(frames) <= 100:
-            # 对于极短的音频段(<=100帧)，直接返回空音频
-            self.update_audio_buffer(len(frames))
-            return None
-        elif len(frames) < self.sample_rate:
-            silence_audio[-len(frames):] = frames
-            return silence_audio.copy()
-        return frames.copy()
-
+            logger.error(f"Error sending result to client: {e}")
 
-    def stop(self):
+    def stop(self) -> None:
+        """停止所有处理线程并清理资源"""
         self._translate_thread_stop.set()
-        self._frame_to_queue_thread_stop.set()
+        self._frame_processing_thread_stop.set()
+        logger.info(f"Stopping transcription service for client: {self.client_uid}")