Hass/xiaozhi-esp32-server
1
0
Fork 0
mirror of https://github.com/xinnan-tech/xiaozhi-esp32-server.git synced 2026-07-28 01:53:53 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix：播放音乐不会停

Browse Source
This commit is contained in:
lizhongxiang
2025-04-02 11:11:16 +08:00
parent cd3e9ed494 8a5a9cc0b3
commit 078605c188
7 changed files with 493 additions and 218 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+178 -57
View File
@@ -14,7 +14,11 @@ from config.logger import setup_logging
from core.providers.tts.dto.dto import TTSMessageDTO, MsgType
from core.utils.dialogue import Message, Dialogue
from core.handle.textHandle import handleTextMessage
from core.utils.util import get_string_no_punctuation_or_emoji, extract_json_from_string, get_ip_info
from core.utils.util import (
get_string_no_punctuation_or_emoji,
extract_json_from_string,
get_ip_info,
)
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from core.handle.sendAudioHandle import sendAudioMessage
from core.handle.receiveAudioHandle import handleAudioMessage
@@ -32,7 +36,9 @@ class TTSException(RuntimeError):
class ConnectionHandler:
def __init__(self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _memory, _intent):
def __init__(
self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _memory, _intent
):
self.config = config
self.logger = setup_logging()
self.auth = AuthMiddleware(config)
@@ -104,7 +110,7 @@ class ConnectionHandler:
self.is_device_verified = False # 添加设备验证状态标志
self.close_after_chat = False # 是否在聊天结束后关闭连接
self.use_function_call_mode = False
if self.config["selected_module"]["Intent"] == 'function_call':
if self.config["selected_module"]["Intent"] == "function_call":
self.use_function_call_mode = True
async def handle_connection(self, ws):
@@ -113,7 +119,9 @@ class ConnectionHandler:
self.headers = dict(ws.request.headers)
# 获取客户端ip地址
self.client_ip = ws.remote_address[0]
self.logger.bind(tag=TAG).info(f"{self.client_ip} conn - Headers: {self.headers}")
self.logger.bind(tag=TAG).info(
f"{self.client_ip} conn - Headers: {self.headers}"
)
# 进行认证
await self.auth.authenticate(self.headers)
@@ -124,10 +132,14 @@ class ConnectionHandler:
# Load private configuration if device_id is provided
bUsePrivateConfig = self.config.get("use_private_config", False)
self.logger.bind(tag=TAG).info(f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}")
self.logger.bind(tag=TAG).info(
f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}"
)
if bUsePrivateConfig and device_id:
try:
self.private_config = PrivateConfig(device_id, self.config, self.auth_code_gen)
self.private_config = PrivateConfig(
device_id, self.config, self.auth_code_gen
)
await self.private_config.load_or_create()
# 判断是否已经绑定
owner = self.private_config.get_owner()
@@ -140,12 +152,18 @@ class ConnectionHandler:
if all([llm, tts]):
self.llm = llm
self.tts = tts
self.logger.bind(tag=TAG).info(f"Loaded private config and instances for device {device_id}")
self.logger.bind(tag=TAG).info(
f"Loaded private config and instances for device {device_id}"
)
else:
self.logger.bind(tag=TAG).error(f"Failed to create instances for device {device_id}")
self.logger.bind(tag=TAG).error(
f"Failed to create instances for device {device_id}"
)
self.private_config = None
except Exception as e:
self.logger.bind(tag=TAG).error(f"Error initializing private config: {e}")
self.logger.bind(tag=TAG).error(
f"Error initializing private config: {e}"
)
self.private_config = None
raise
@@ -162,7 +180,9 @@ class ConnectionHandler:
# 音频播放 消化线程
self.stop_event.clear()
audio_play_priority = threading.Thread(target=self._audio_play_priority_thread, daemon=True)
audio_play_priority = threading.Thread(
target=self._audio_play_priority_thread, daemon=True
)
audio_play_priority.start()
# 打开音频通道
@@ -236,7 +256,9 @@ class ConnectionHandler:
def chat(self, query):
if self.isNeedAuth():
self.llm_finish_task = True
future = asyncio.run_coroutine_threadsafe(self._check_and_broadcast_auth_code(), self.loop)
future = asyncio.run_coroutine_threadsafe(
self._check_and_broadcast_auth_code(), self.loop
)
future.result()
return True
@@ -247,13 +269,14 @@ class ConnectionHandler:
try:
start_time = time.time()
# 使用带记忆的对话
future = asyncio.run_coroutine_threadsafe(self.memory.query_memory(query), self.loop)
future = asyncio.run_coroutine_threadsafe(
self.memory.query_memory(query), self.loop
)
memory_str = future.result()
self.logger.bind(tag=TAG).debug(f"记忆内容: {memory_str}")
llm_responses = self.llm.response(
self.session_id,
self.dialogue.get_llm_dialogue_with_memory(memory_str)
self.session_id, self.dialogue.get_llm_dialogue_with_memory(memory_str)
)
except Exception as e:
self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}")
@@ -270,20 +293,34 @@ class ConnectionHandler:
break
end_time = time.time()
self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}")
self.logger.bind(tag=TAG).debug(
f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}"
)
if text_index == 0:
self.tts.tts_text_queue.put(
TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content=''))
TTSMessageDTO(
u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content=""
)
)
self.start_tts_request_flag = True
self.tts.tts_text_queue.put(
TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=content))
TTSMessageDTO(
u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=content
)
)
text_index += 1
if self.start_tts_request_flag:
self.start_tts_request_flag = False
self.tts.tts_text_queue.put(TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=''))
self.tts.tts_text_queue.put(
TTSMessageDTO(
u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=""
)
)
self.llm_finish_task = True
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False))
self.logger.bind(tag=TAG).debug(
json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)
)
return True
def chat_with_function_calling(self, query, tool_call=False):
@@ -291,7 +328,9 @@ class ConnectionHandler:
"""Chat with function calling for intent detection using streaming"""
if self.isNeedAuth():
self.llm_finish_task = True
future = asyncio.run_coroutine_threadsafe(self._check_and_broadcast_auth_code(), self.loop)
future = asyncio.run_coroutine_threadsafe(
self._check_and_broadcast_auth_code(), self.loop
)
future.result()
return True
@@ -308,7 +347,9 @@ class ConnectionHandler:
start_time = time.time()
# 使用带记忆的对话
future = asyncio.run_coroutine_threadsafe(self.memory.query_memory(query), self.loop)
future = asyncio.run_coroutine_threadsafe(
self.memory.query_memory(query), self.loop
)
memory_str = future.result()
# self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}")
@@ -317,7 +358,7 @@ class ConnectionHandler:
llm_responses = self.llm.response_with_functions(
self.session_id,
self.dialogue.get_llm_dialogue_with_memory(memory_str),
functions=functions
functions=functions,
)
except Exception as e:
self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}")
@@ -338,7 +379,9 @@ class ConnectionHandler:
for response in llm_responses:
content, tools_call = response
if content is not None and len(content) > 0:
if len(response_message) <= 0 and (content == "```" or "<tool_call>" in content):
if len(response_message) <= 0 and (
content == "```" or "<tool_call>" in content
):
tool_call_flag = True
if tools_call is not None:
@@ -360,17 +403,33 @@ class ConnectionHandler:
break
end_time = time.time()
self.logger.bind(tag=TAG).debug(f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}")
self.logger.bind(tag=TAG).debug(
f"大模型返回时间: {end_time - start_time} 秒, 生成token={content}"
)
if text_index == 0:
self.tts.tts_text_queue.put(
TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.START_TTS_REQUEST, content=''))
TTSMessageDTO(
u_id=uuid_str,
msg_type=MsgType.START_TTS_REQUEST,
content="",
)
)
self.start_tts_request_flag = True
self.tts.tts_text_queue.put(
TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.TTS_TEXT_REQUEST, content=content))
TTSMessageDTO(
u_id=uuid_str,
msg_type=MsgType.TTS_TEXT_REQUEST,
content=content,
)
)
text_index += 1
if self.start_tts_request_flag:
self.start_tts_request_flag = False
self.tts.tts_text_queue.put(TTSMessageDTO(u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=''))
self.tts.tts_text_queue.put(
TTSMessageDTO(
u_id=uuid_str, msg_type=MsgType.STOP_TTS_REQUEST, content=""
)
)
# 处理function call
if tool_call_flag:
@@ -381,7 +440,9 @@ class ConnectionHandler:
try:
content_arguments_json = json.loads(a)
function_name = content_arguments_json["name"]
function_arguments = json.dumps(content_arguments_json["arguments"], ensure_ascii=False)
function_arguments = json.dumps(
content_arguments_json["arguments"], ensure_ascii=False
)
function_id = str(uuid.uuid4().hex)
except Exception as e:
bHasError = True
@@ -390,26 +451,35 @@ class ConnectionHandler:
bHasError = True
response_message.append(content_arguments)
if bHasError:
self.logger.bind(tag=TAG).error(f"function call error: {content_arguments}")
self.logger.bind(tag=TAG).error(
f"function call error: {content_arguments}"
)
else:
function_arguments = json.loads(function_arguments)
if not bHasError:
self.logger.bind(tag=TAG).info(
f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}")
f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}"
)
function_call_data = {
"name": function_name,
"id": function_id,
"arguments": function_arguments
"arguments": function_arguments,
}
result = self.func_handler.handle_llm_function_call(self, function_call_data)
result = self.func_handler.handle_llm_function_call(
self, function_call_data
)
self._handle_function_result(result, function_call_data, text_index + 1)
# 存储对话内容
if len(response_message) > 0:
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
self.dialogue.put(
Message(role="assistant", content="".join(response_message))
)
self.llm_finish_task = True
self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False))
self.logger.bind(tag=TAG).debug(
json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)
)
return True
@@ -417,7 +487,9 @@ class ConnectionHandler:
if result.action == Action.RESPONSE: # 直接回复前端
text = result.response
self.recode_first_last_text(text, text_index)
asyncio.run_coroutine_threadsafe(self.tts.tts_one_sentence(text), loop=self.loop)
asyncio.run_coroutine_threadsafe(
self.tts.tts_one_sentence(text), loop=self.loop
)
self.dialogue.put(Message(role="assistant", content=text))
elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.result
@@ -425,24 +497,40 @@ class ConnectionHandler:
function_id = function_call_data["id"]
function_name = function_call_data["name"]
function_arguments = function_call_data["arguments"]
self.dialogue.put(Message(role='assistant',
tool_calls=[{"id": function_id,
"function": {"arguments": function_arguments,
"name": function_name},
"type": 'function',
"index": 0}]))
self.dialogue.put(
Message(
role="assistant",
tool_calls=[
{
"id": function_id,
"function": {
"arguments": function_arguments,
"name": function_name,
},
"type": "function",
"index": 0,
}
],
)
)
self.dialogue.put(Message(role="tool", tool_call_id=function_id, content=text))
self.dialogue.put(
Message(role="tool", tool_call_id=function_id, content=text)
)
self.chat_with_function_calling(text, tool_call=True)
elif result.action == Action.NOTFOUND:
text = result.result
self.recode_first_last_text(text, text_index)
asyncio.run_coroutine_threadsafe(self.tts.tts_one_sentence(text), loop=self.loop)
asyncio.run_coroutine_threadsafe(
self.tts.tts_one_sentence(text), loop=self.loop
)
self.dialogue.put(Message(role="assistant", content=text))
else:
text = result.result
self.recode_first_last_text(text, text_index)
asyncio.run_coroutine_threadsafe(self.tts.tts_one_sentence(text), loop=self.loop)
asyncio.run_coroutine_threadsafe(
self.tts.tts_one_sentence(text), loop=self.loop
)
self.dialogue.put(Message(role="assistant", content=text))
def _tts_priority_thread(self):
@@ -465,15 +553,23 @@ class ConnectionHandler:
tts_timeout = self.config.get("tts_timeout", 10)
tts_file, text, text_index = future.result(timeout=tts_timeout)
if text is None or len(text) <= 0:
self.logger.bind(tag=TAG).error(f"TTS出错：{text_index}: tts text is empty")
self.logger.bind(tag=TAG).error(
f"TTS出错：{text_index}: tts text is empty"
)
elif tts_file is None:
self.logger.bind(tag=TAG).error(f"TTS出错： file is empty: {text_index}: {text}")
self.logger.bind(tag=TAG).error(
f"TTS出错： file is empty: {text_index}: {text}"
)
else:
self.logger.bind(tag=TAG).debug(f"TTS生成：文件路径: {tts_file}")
self.logger.bind(tag=TAG).debug(
f"TTS生成：文件路径: {tts_file}"
)
if os.path.exists(tts_file):
opus_datas, duration = self.tts.audio_to_opus_data(tts_file)
else:
self.logger.bind(tag=TAG).error(f"TTS出错：文件不存在{tts_file}")
self.logger.bind(tag=TAG).error(
f"TTS出错：文件不存在{tts_file}"
)
except TimeoutError:
self.logger.bind(tag=TAG).error("TTS超时")
except Exception as e:
@@ -481,16 +577,30 @@ class ConnectionHandler:
if not self.client_abort:
# 如果没有中途打断就发送语音
self.audio_play_queue.put((opus_datas, text, text_index))
if self.tts.delete_audio_file and tts_file is not None and os.path.exists(tts_file):
if (
self.tts.delete_audio_file
and tts_file is not None
and os.path.exists(tts_file)
):
os.remove(tts_file)
except Exception as e:
self.logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}")
self.clearSpeakStatus()
asyncio.run_coroutine_threadsafe(
self.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": self.session_id})),
self.loop
self.websocket.send(
json.dumps(
{
"type": "tts",
"state": "stop",
"session_id": self.session_id,
}
)
),
self.loop,
)
self.logger.bind(tag=TAG).error(
f"tts_priority priority_thread: {text} {e}"
)
self.logger.bind(tag=TAG).error(f"tts_priority priority_thread: {text} {e}")
def _tts_priority_thread_stream(self):
while not self.stop_event.is_set():
@@ -513,8 +623,16 @@ class ConnectionHandler:
except Exception as e:
self.clearSpeakStatus()
asyncio.run_coroutine_threadsafe(
self.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": self.session_id})),
self.loop
self.websocket.send(
json.dumps(
{
"type": "tts",
"state": "stop",
"session_id": self.session_id,
}
)
),
self.loop,
)
self.logger.error(f"tts_priority priority_thread: {text}{e}")
@@ -523,11 +641,14 @@ class ConnectionHandler:
text = None
try:
ttsMessageDTO = self.tts.tts_audio_queue.get()
future = asyncio.run_coroutine_threadsafe(sendAudioMessage(self, ttsMessageDTO),
self.loop)
future = asyncio.run_coroutine_threadsafe(
sendAudioMessage(self, ttsMessageDTO), self.loop
)
future.result()
except Exception as e:
self.logger.bind(tag=TAG).error(f"audio_play_priority priority_thread: {text} {e}")
self.logger.bind(tag=TAG).error(
f"audio_play_priority priority_thread: {text} {e}"
)
def speak_and_play(self, text, text_index=0):
if text is None or len(text) <= 0: