diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 1573d5b1..fa862e8b 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -2,6 +2,7 @@ import os import sys import copy import json +import re import uuid import time import queue @@ -1016,21 +1017,28 @@ class ConnectionHandler: self._merge_tool_calls(tool_calls_list, tools_call) # 流式提取 direct_answer 的 response 参数,实时送 TTS + # 使用安全缓冲区,防止 JSON 闭合符号泄漏到 TTS + _DA_STREAM_BUFFER = 5 for tc in tool_calls_list: if tc["name"] == "direct_answer" and tc.get("arguments"): da_text = self._extract_direct_answer_response(tc["arguments"]) sent_len = tc.get("_da_sent", 0) if da_text and len(da_text) > sent_len: - new_part = da_text[sent_len:] - tc["_da_sent"] = len(da_text) - self.tts.tts_text_queue.put( - TTSMessageDTO( - sentence_id=current_sentence_id, - sentence_type=SentenceType.MIDDLE, - content_type=ContentType.TEXT, - content_detail=new_part, - ) - ) + safe_end = max(sent_len, len(da_text) - _DA_STREAM_BUFFER) + if safe_end > sent_len: + new_part = da_text[sent_len:safe_end] + # 清理 delta 中可能泄漏的 JSON 闭合垃圾 + new_part = self._clean_response_garbage(new_part) + if new_part: + tc["_da_sent"] = safe_end + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=current_sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=new_part, + ) + ) else: content = response @@ -1112,9 +1120,24 @@ class ConnectionHandler: f"模型选择 direct_answer,流式已播报,写入对话历史" ) for tc in direct_answer_calls: - # 文本已在流式阶段实时播报,这里只写入对话历史 da_response = self._extract_direct_answer_response(tc.get("arguments", "{}")) if da_response: + # 刷新流式缓冲区中未发送的部分 + sent_len = tc.get("_da_sent", 0) + remaining = da_response[sent_len:] + if remaining: + remaining = self._clean_response_garbage(remaining) + if remaining: + self.tts.tts_text_queue.put( + TTSMessageDTO( + sentence_id=current_sentence_id, + sentence_type=SentenceType.MIDDLE, + content_type=ContentType.TEXT, + content_detail=remaining, + ) + ) + # 写入对话历史 + da_response = self._clean_response_garbage(da_response) self.tts.store_tts_text(current_sentence_id, da_response) self.dialogue.put(Message(role="assistant", content=da_response)) @@ -1550,15 +1573,21 @@ class ConnectionHandler: @staticmethod def _extract_direct_answer_response(arguments_str): """从 direct_answer 的参数中提取 response 值。 - 支持完整的 JSON 和流式中不完整的 JSON 片段。 + 优先使用 json.loads 标准解析,流式阶段 fallback 到字符串提取。 """ if not arguments_str: return "" - # 找到 "response": " 的位置 + # 优先尝试标准 JSON 解析(适用于完整且格式正确的 JSON) + try: + data = json.loads(arguments_str) + if isinstance(data, dict) and "response" in data: + return data["response"] + except (json.JSONDecodeError, TypeError): + pass + # Fallback:流式阶段 JSON 可能不完整,使用字符串提取 marker = '"response": "' idx = arguments_str.find(marker) if idx < 0: - # 尝试不带空格的格式 marker = '"response":"' idx = arguments_str.find(marker) if idx < 0: @@ -1574,6 +1603,28 @@ class ConnectionHandler: raw = raw.replace('\\"', '"').replace('\\n', '\n').replace('\\\\', '\\') return raw + @staticmethod + def _clean_response_garbage(text): + """清理 response 中可能泄漏的 JSON 闭合符号。 + 模型有时会在 response 内容中生成 JSON 闭合字符(如 )"}} 或 '}), + 这些不是故事内容的一部分,需要去除。 + """ + if not text: + return text + # 清理独立一行的 JSON 闭合垃圾(如 )"}} '}} "}} }} } ) + _garbage_chars = frozenset('")\'})') + lines = text.split('\n') + cleaned = [] + for line in lines: + stripped = line.strip() + if stripped and len(stripped) <= 8 and all(c in _garbage_chars for c in stripped): + continue + cleaned.append(line) + result = '\n'.join(cleaned) + # 清理末尾残留的 JSON 闭合符号 + result = re.sub(r'["\'}\]]+$', '', result.rstrip()).rstrip() + return result + def _merge_tool_calls(self, tool_calls_list, tools_call): """合并工具调用列表