fix: 修复流式情况下替换词失败

main/xiaozhi-server/core/providers/tts/alibl_stream.py

@@ -124,6 +124,8 @@ class TTSProvider(TTSProviderBase):
                 )
 
                 if message.sentence_type == SentenceType.FIRST:
+                    # 重置流式处理状态
+                    self.reset_stream_state()
                     # 初始化会话
                     try:
                         if not getattr(self.conn, "sentence_id", None): 
@@ -194,22 +196,24 @@ class TTSProvider(TTSProviderBase):
 
             # 过滤Markdown
             filtered_text = MarkdownCleaner.clean_markdown(text)
-            if self._correct_words_pattern:
-                filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text)
 
             if filtered_text:
-                # 发送continue-task消息
+                # 使用滑动窗口匹配处理跨分片的替换词
-                continue_task_message = {
+                confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text)
-                    "header": {
-                        "action": "continue-task",
-                        "task_id": self.conn.sentence_id,
-                        "streaming": "duplex",
-                    },
-                    "payload": {"input": {"text": filtered_text}},
-                }
 
-                await self.ws.send(json.dumps(continue_task_message))
+                # 发送每个确定的文本片段
-                self.last_active_time = time.time()
+                for txt in confirmed_texts:
+                    if txt and self.ws:
+                        continue_task_message = {
+                            "header": {
+                                "action": "continue-task",
+                                "task_id": self.conn.sentence_id,
+                                "streaming": "duplex",
+                            },
+                            "payload": {"input": {"text": txt}},
+                        }
+                        await self.ws.send(json.dumps(continue_task_message))
+                        self.last_active_time = time.time()
             return
         except Exception as e:
             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")

main/xiaozhi-server/core/providers/tts/aliyun_stream.py

@@ -233,6 +233,8 @@ class TTSProvider(TTSProviderBase):
                 )
 
                 if message.sentence_type == SentenceType.FIRST:
+                    # 重置流式处理状态
+                    self.reset_stream_state()
                     # 初始化参数
                     try:
                         logger.bind(tag=TAG).debug("开始启动TTS会话...")
@@ -295,21 +297,26 @@ class TTSProvider(TTSProviderBase):
                 logger.bind(tag=TAG).warning(f"WebSocket连接不存在，终止发送文本")
                 return
             filtered_text = MarkdownCleaner.clean_markdown(text)
-            if self._correct_words_pattern:
+
-                filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text)
             if filtered_text:
-                run_request = {
+                # 使用滑动窗口匹配处理跨分片的替换词
-                    "header": {
+                confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text)
-                        "message_id": uuid.uuid4().hex,
+
-                        "task_id": self.task_id,
+                # 发送每个确定的文本片段
-                        "namespace": "FlowingSpeechSynthesizer",
+                for txt in confirmed_texts:
-                        "name": "RunSynthesis",
+                    if txt and self.ws:
-                        "appkey": self.appkey,
+                        run_request = {
-                    },
+                            "header": {
-                    "payload": {"text": filtered_text},
+                                "message_id": uuid.uuid4().hex,
-                }
+                                "task_id": self.task_id,
-                await self.ws.send(json.dumps(run_request))
+                                "namespace": "FlowingSpeechSynthesizer",
-                self.last_active_time = time.time()
+                                "name": "RunSynthesis",
+                                "appkey": self.appkey,
+                            },
+                            "payload": {"text": txt},
+                        }
+                        await self.ws.send(json.dumps(run_request))
+                        self.last_active_time = time.time()
             return
 
         except Exception as e:

main/xiaozhi-server/core/providers/tts/base.py

@@ -64,11 +64,20 @@ class TTSProviderBase(ABC):
             reverse_pattern_str = "|".join(re.escape(k) for k in sorted_reverse_keys)
             self._reverse_words_pattern = re.compile(reverse_pattern_str)
             self._reverse_words_map = reverse_map
+            # 流式滑动窗口：按首字分组的替换词字典，用于快速查找
+            self._words_by_first_char = {}
+            for key in sorted_keys:  # 使用已按长度降序排列的keys，确保长词优先匹配
+                first_char = key[0] if key else ""
+                if first_char not in self._words_by_first_char:
+                    self._words_by_first_char[first_char] = []
+                self._words_by_first_char[first_char].append(key)
         else:
             self._correct_words_pattern = None
             self._reverse_words_pattern = None
             self._reverse_words_map = None
 
+        # 流式滑动窗口：待匹配的缓存文本
+        self._pending_prefix = ""
         self.tts_text_buff = []
         self.punctuations = (
             "。",
@@ -564,3 +573,64 @@ class TTSProviderBase(ABC):
             if config_key in config:
                 val = convert_percentage_to_range(config[config_key], min_val, max_val, base_val)
                 setattr(self, attr_name, transform(val) if transform else val)
+
+    def _match_stream_text(self, text):
+        """流式文本滑动窗口匹配，用于处理跨分片的替换词
+
+        Args:
+            text: 输入的文本片段
+
+        Returns:
+            tuple: (确定的文本列表, 剩余待匹配的前缀)
+        """
+        if not self.correct_words or not text:
+            return [text] if text else [], ""
+
+        result = []
+        pending = self._pending_prefix
+        i = 0
+
+        while i < len(text):
+            char = text[i]
+
+            # 尝试：pending + 当前字符 是否能匹配替换词
+            test_text = pending + char
+
+            matched = False
+            # 遍历可能匹配的替换词
+            candidates = self._words_by_first_char.get(pending[0], []) if pending else self._words_by_first_char.get(char, [])
+            for key in candidates:
+                if test_text == key:
+                    # 完整匹配，替换后发送
+                    result.append(self.correct_words[key])
+                    pending = ""
+                    matched = True
+                    break
+                elif key.startswith(test_text):
+                    # 是替换词的前缀，继续等待
+                    pending = test_text
+                    matched = True
+                    break
+
+            if matched:
+                i += 1
+                continue
+
+            # 没有匹配到更长的词，pending 的内容确定可以发送
+            if pending:
+                result.append(pending)
+                pending = ""
+
+            # 检查当前字符是否是某个替换词的开头
+            if char in self._words_by_first_char:
+                pending = char
+            else:
+                result.append(char)
+
+            i += 1
+
+        return result, pending
+
+    def reset_stream_state(self):
+        """重置流式处理状态，用于会话开始时清理残留状态"""
+        self._pending_prefix = ""

main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py

@@ -300,6 +300,8 @@ class TTSProvider(TTSProviderBase):
                 )
 
                 if message.sentence_type == SentenceType.FIRST:
+                    # 重置流式处理状态
+                    self.reset_stream_state()
                     # 初始化参数
                     try:
                         if not getattr(self.conn, "sentence_id", None): 
@@ -370,12 +372,16 @@ class TTSProvider(TTSProviderBase):
 
             #  过滤Markdown
             filtered_text = MarkdownCleaner.clean_markdown(text)
-            if self._correct_words_pattern:
-                filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text)
 
             if filtered_text:
-                # 发送文本
+                # 使用滑动窗口匹配处理跨分片的替换词
-                await self.send_text(self.voice, filtered_text, self.conn.sentence_id)
+                confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text)
+
+                # 发送每个确定的文本片段
+                for txt in confirmed_texts:
+                    if txt and self.ws:
+                        await self.send_text(self.voice, txt, self.conn.sentence_id)
+
             return
         except Exception as e:
             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")

main/xiaozhi-server/core/providers/tts/xunfei_stream.py

@@ -168,6 +168,8 @@ class TTSProvider(TTSProviderBase):
                 )
 
                 if message.sentence_type == SentenceType.FIRST:
+                    # 重置流式处理状态
+                    self.reset_stream_state()
                     # 重置序列号
                     self.text_seq = 0
                 # 增加序列号
@@ -245,12 +247,17 @@ class TTSProvider(TTSProviderBase):
                 return
 
             filtered_text = MarkdownCleaner.clean_markdown(text)
-            if self._correct_words_pattern:
+
-                filtered_text = self._correct_words_pattern.sub(lambda m: self.correct_words[m.group(0)], filtered_text)
             if filtered_text:
-                # 发送文本合成请求
+                # 使用滑动窗口匹配处理跨分片的替换词
-                run_request = self._build_base_request(status=1,text=filtered_text)
+                confirmed_texts, self._pending_prefix = self._match_stream_text(filtered_text)
-                await self.ws.send(json.dumps(run_request))
+
+                # 发送每个确定的文本片段
+                for txt in confirmed_texts:
+                    if txt and self.ws:
+                        # 发送文本合成请求
+                        run_request = self._build_base_request(status=1, text=txt)
+                        await self.ws.send(json.dumps(run_request))
             return
 
         except Exception as e: