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

update:优化日志对象

Browse Source
This commit is contained in:
hrz
2025-05-07 18:06:13 +08:00
parent a26bee3696
commit ea5f54e421
16 changed files with 249 additions and 243 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main/xiaozhi-server/core/connection.py
+54 -70
View File
@@ -18,6 +18,8 @@ from core.utils.util import (
get_string_no_punctuation_or_emoji,
extract_json_from_string,
initialize_modules,
check_vad_update,
check_asr_update,
)
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from core.handle.sendAudioHandle import sendAudioMessage
@@ -52,7 +54,9 @@ class ConnectionHandler:
_intent,
server=None,
):
self.config = config
self.common_config = config
self.config = copy.deepcopy(config)
self.session_id = str(uuid.uuid4())
self.logger = setup_logging()
self.auth = AuthMiddleware(config)
self.server = server # 保存server实例的引用
@@ -66,7 +70,6 @@ class ConnectionHandler:
self.device_id = None
self.client_ip = None
self.client_ip_info = {}
self.session_id = None
self.prompt = None
self.welcome_msg = None
self.max_output_size = 0
@@ -87,8 +90,10 @@ class ConnectionHandler:
self.tts_report_thread = None
# 依赖的组件
self.vad = _vad
self.asr = _asr
self.vad = None
self.asr = None
self._asr = _asr
self._vad = _vad
self.llm = _llm
self.tts = _tts
self.memory = _memory
@@ -166,7 +171,6 @@ class ConnectionHandler:
# 认证通过,继续处理
self.websocket = ws
self.device_id = self.headers.get("device-id", None)
self.session_id = str(uuid.uuid4())
# 启动超时检查任务
self.timeout_task = asyncio.create_task(self._check_timeout())
@@ -176,9 +180,9 @@ class ConnectionHandler:
await self.websocket.send(json.dumps(self.welcome_msg))
# 获取差异化配置
private_config = self._initialize_private_config()
self._initialize_private_config()
# 异步初始化
self.executor.submit(self._initialize_components, private_config)
self.executor.submit(self._initialize_components)
# tts 消化线程
self.tts_priority_thread = threading.Thread(
target=self._tts_priority_thread, daemon=True
@@ -233,13 +237,17 @@ class ConnectionHandler:
elif isinstance(message, bytes):
await handleAudioMessage(self, message)
def _initialize_components(self, private_config):
def _initialize_components(self):
"""初始化组件"""
if private_config is not None:
self._initialize_models(private_config)
else:
self.prompt = self.config["prompt"]
self.change_system_prompt(self.prompt)
self.prompt = self.config["prompt"]
self.change_system_prompt(self.prompt)
self.logger.bind(tag=TAG).info(f"初始化组件: prompt成功 {self.prompt[:50]}...")
"""初始化本地组件"""
if self.vad is None:
self.vad = self._vad
if self.asr is None:
self.asr = self._asr
"""加载记忆"""
self._initialize_memory()
"""加载意图识别"""
@@ -286,54 +294,21 @@ class ConnectionHandler:
self.logger.bind(tag=TAG).error(f"获取差异化配置失败: {e}")
private_config = {}
init_tts = False
if private_config.get("TTS", None) is not None:
init_tts = True
self.config["TTS"] = private_config["TTS"]
self.config["selected_module"]["TTS"] = private_config["selected_module"][
"TTS"
]
try:
modules = initialize_modules(
self.logger,
private_config,
False,
False,
False,
init_tts,
False,
False,
)
except Exception as e:
self.logger.bind(tag=TAG).error(f"初始化组件失败: {e}")
modules = {}
if modules.get("tts", None) is not None:
self.tts = modules["tts"]
if modules.get("prompt", None) is not None:
self.change_system_prompt(modules["prompt"])
private_config["prompt"] = None
return private_config
def _initialize_models(self, private_config):
init_vad, init_asr, init_llm, init_memory, init_intent = (
False,
init_llm, init_tts, init_memory, init_intent = (
False,
False,
False,
False,
)
if private_config.get("VAD", None) is not None:
init_vad = True
self.config["VAD"] = private_config["VAD"]
self.config["selected_module"]["VAD"] = private_config["selected_module"][
"VAD"
]
if private_config.get("ASR", None) is not None:
init_asr = True
self.config["ASR"] = private_config["ASR"]
self.config["selected_module"]["ASR"] = private_config["selected_module"][
"ASR"
init_vad = check_vad_update(self.common_config, private_config)
init_asr = check_asr_update(self.common_config, private_config)
if private_config.get("TTS", None) is not None:
init_tts = True
self.config["TTS"] = private_config["TTS"]
self.config["selected_module"]["TTS"] = private_config["selected_module"][
"TTS"
]
if private_config.get("LLM", None) is not None:
init_llm = True
@@ -353,8 +328,11 @@ class ConnectionHandler:
self.config["selected_module"]["Intent"] = private_config[
"selected_module"
]["Intent"]
if private_config.get("prompt", None) is not None:
self.config["prompt"] = private_config["prompt"]
if private_config.get("device_max_output_size", None) is not None:
self.max_output_size = int(private_config["device_max_output_size"])
try:
modules = initialize_modules(
self.logger,
@@ -362,13 +340,15 @@ class ConnectionHandler:
init_vad,
init_asr,
init_llm,
False,
init_tts,
init_memory,
init_intent,
)
except Exception as e:
self.logger.bind(tag=TAG).error(f"初始化组件失败: {e}")
modules = {}
if modules.get("tts", None) is not None:
self.tts = modules["tts"]
if modules.get("vad", None) is not None:
self.vad = modules["vad"]
if modules.get("asr", None) is not None:
@@ -446,10 +426,12 @@ class ConnectionHandler:
processed_chars = 0 # 跟踪已处理的字符位置
try:
# 使用带记忆的对话
future = asyncio.run_coroutine_threadsafe(
self.memory.query_memory(query), self.loop
)
memory_str = future.result()
memory_str = None
if self.memory is not None:
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(
@@ -510,7 +492,7 @@ class ConnectionHandler:
future = self.executor.submit(
self.speak_and_play, segment_text, text_index
)
self.tts_queue.put(future)
self.tts_queue.put((future, text_index))
self.llm_finish_task = True
self.dialogue.put(Message(role="assistant", content="".join(response_message)))
@@ -537,10 +519,12 @@ class ConnectionHandler:
start_time = time.time()
# 使用带记忆的对话
future = asyncio.run_coroutine_threadsafe(
self.memory.query_memory(query), self.loop
)
memory_str = future.result()
memory_str = None
if self.memory is not None:
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)}")
@@ -685,7 +669,7 @@ class ConnectionHandler:
future = self.executor.submit(
self.speak_and_play, segment_text, text_index
)
self.tts_queue.put(future)
self.tts_queue.put((future, text_index))
# 存储对话内容
if len(response_message) > 0:
@@ -747,7 +731,7 @@ class ConnectionHandler:
text = result.response
self.recode_first_last_text(text, text_index)
future = self.executor.submit(self.speak_and_play, text, text_index)
self.tts_queue.put(future)
self.tts_queue.put((future, text_index))
self.dialogue.put(Message(role="assistant", content=text))
elif result.action == Action.REQLLM: # 调用函数后再请求llm生成回复
text = result.result
@@ -780,7 +764,7 @@ class ConnectionHandler:
text = result.result
self.recode_first_last_text(text, text_index)
future = self.executor.submit(self.speak_and_play, text, text_index)
self.tts_queue.put(future)
self.tts_queue.put((future, text_index))
self.dialogue.put(Message(role="assistant", content=text))
else:
pass
@@ -801,7 +785,7 @@ class ConnectionHandler:
if future is None:
continue
text = None
opus_datas, tts_file = [], None
opus_datas, tts_file = [], None
try:
self.logger.bind(tag=TAG).debug("正在处理TTS任务...")
tts_timeout = int(self.config.get("tts_timeout", 10))
main/xiaozhi-server/core/handle/abortHandle.py
+5 -4
View File
@@ -3,15 +3,16 @@ import queue
from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
async def handleAbortMessage(conn):
logger.bind(tag=TAG).info("Abort message received")
conn.logger.bind(tag=TAG).info("Abort message received")
# 设置成打断状态，会自动打断llm、tts任务
conn.client_abort = True
conn.clear_queues()
# 打断客户端说话状态
await conn.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": conn.session_id}))
await conn.websocket.send(
json.dumps({"type": "tts", "state": "stop", "session_id": conn.session_id})
)
conn.clearSpeakStatus()
logger.bind(tag=TAG).info("Abort message received-end")
conn.logger.bind(tag=TAG).info("Abort message received-end")
main/xiaozhi-server/core/handle/functionHandler.py
+7 -4
View File
@@ -4,7 +4,6 @@ from plugins_func.register import FunctionRegistry, ActionResponse, Action, Tool
from plugins_func.functions.hass_init import append_devices_to_prompt
TAG = __name__
logger = setup_logging()
class FunctionHandler:
@@ -40,7 +39,9 @@ class FunctionHandler:
for func in self.functions_desc:
func_names.append(func["function"]["name"])
# 打印当前支持的函数列表
logger.bind(tag=TAG).info(f"当前支持的函数列表: {func_names}")
self.conn.logger.bind(tag=TAG, session_id=self.conn.session_id).info(
f"当前支持的函数列表: {func_names}"
)
return func_names
def get_functions(self):
@@ -79,7 +80,9 @@ class FunctionHandler:
func = funcItem.func
arguments = function_call_data["arguments"]
arguments = json.loads(arguments) if arguments else {}
logger.bind(tag=TAG).debug(f"调用函数: {function_name}, 参数: {arguments}")
self.conn.logger.bind(tag=TAG).debug(
f"调用函数: {function_name}, 参数: {arguments}"
)
if (
funcItem.type == ToolType.SYSTEM_CTL
or funcItem.type == ToolType.IOT_CTL
@@ -94,6 +97,6 @@ class FunctionHandler:
action=Action.NOTFOUND, result="没有找到对应的函数", response=""
)
except Exception as e:
logger.bind(tag=TAG).error(f"处理function call错误: {e}")
self.conn.logger.bind(tag=TAG).error(f"处理function call错误: {e}")
return None
main/xiaozhi-server/core/handle/helloHandle.py
+1 -2
View File
@@ -9,7 +9,6 @@ import random
import time
TAG = __name__
logger = setup_logging()
WAKEUP_CONFIG = {
"dir": "config/assets/",
@@ -75,7 +74,7 @@ async def wakeupWordsResponse(conn):
await asyncio.sleep(1)
wait_max_time -= 1
if wait_max_time <= 0:
logger.bind(tag=TAG).error("连接对象没有llm")
conn.logger.bind(tag=TAG).error("连接对象没有llm")
return
"""唤醒词响应"""
main/xiaozhi-server/core/handle/intentHandler.py
+6 -8
View File
@@ -5,10 +5,8 @@ from core.handle.sendAudioHandle import send_stt_message
from core.handle.helloHandle import checkWakeupWords
from core.utils.util import remove_punctuation_and_length
from core.utils.dialogue import Message
from loguru import logger
TAG = __name__
logger = setup_logging()
async def handle_user_intent(conn, text):
@@ -36,7 +34,7 @@ async def check_direct_exit(conn, text):
cmd_exit = conn.cmd_exit
for cmd in cmd_exit:
if text == cmd:
logger.bind(tag=TAG).info(f"识别到明确的退出命令: {text}")
conn.logger.bind(tag=TAG).info(f"识别到明确的退出命令: {text}")
await send_stt_message(conn, text)
await conn.close()
return True
@@ -46,7 +44,7 @@ async def check_direct_exit(conn, text):
async def analyze_intent_with_llm(conn, text):
"""使用LLM分析用户意图"""
if not hasattr(conn, "intent") or not conn.intent:
logger.bind(tag=TAG).warning("意图识别服务未初始化")
conn.logger.bind(tag=TAG).warning("意图识别服务未初始化")
return None
# 对话历史记录
@@ -55,7 +53,7 @@ async def analyze_intent_with_llm(conn, text):
intent_result = await conn.intent.detect_intent(conn, dialogue.dialogue, text)
return intent_result
except Exception as e:
logger.bind(tag=TAG).error(f"意图识别失败: {str(e)}")
conn.logger.bind(tag=TAG).error(f"意图识别失败: {str(e)}")
return None
@@ -69,7 +67,7 @@ async def process_intent_result(conn, intent_result, original_text):
# 检查是否有function_call
if "function_call" in intent_data:
# 直接从意图识别获取了function_call
logger.bind(tag=TAG).debug(
conn.logger.bind(tag=TAG).debug(
f"检测到function_call格式的意图结果: {intent_data['function_call']['name']}"
)
function_name = intent_data["function_call"]["name"]
@@ -118,7 +116,7 @@ async def process_intent_result(conn, intent_result, original_text):
conn.speak_and_play, text, text_index
)
conn.llm_finish_task = True
conn.tts_queue.put(future)
conn.tts_queue.put((future, text_index))
conn.dialogue.put(Message(role="assistant", content=text))
# 将函数执行放在线程池中
@@ -126,7 +124,7 @@ async def process_intent_result(conn, intent_result, original_text):
return True
return False
except json.JSONDecodeError as e:
logger.bind(tag=TAG).error(f"处理意图结果时出错: {e}")
conn.logger.bind(tag=TAG).error(f"处理意图结果时出错: {e}")
return False
main/xiaozhi-server/core/handle/iotHandle.py
+22 -19
View File
@@ -10,7 +10,6 @@ from plugins_func.register import (
)
TAG = __name__
logger = setup_logging()
def wrap_async_function(async_func):
@@ -21,7 +20,7 @@ def wrap_async_function(async_func):
# 获取连接对象（第一个参数）
conn = args[0]
if not hasattr(conn, "loop"):
logger.bind(tag=TAG).error("Connection对象没有loop属性")
conn.logger.bind(tag=TAG).error("Connection对象没有loop属性")
return ActionResponse(
Action.ERROR,
"Connection对象没有loop属性",
@@ -35,7 +34,7 @@ def wrap_async_function(async_func):
# 等待结果返回
return future.result()
except Exception as e:
logger.bind(tag=TAG).error(f"运行异步函数时出错: {e}")
conn.logger.bind(tag=TAG).error(f"运行异步函数时出错: {e}")
return ActionResponse(Action.ERROR, str(e), f"执行操作时出错: {e}")
return wrapper
@@ -57,7 +56,7 @@ def create_iot_function(device_name, method_name, method_info):
response_failure = "操作失败"
# 打印响应参数
logger.bind(tag=TAG).debug(
conn.logger.bind(tag=TAG).debug(
f"控制函数接收到的响应参数: success='{response_success}', failure='{response_failure}'"
)
@@ -86,7 +85,9 @@ def create_iot_function(device_name, method_name, method_info):
return ActionResponse(Action.RESPONSE, result, response)
except Exception as e:
logger.bind(tag=TAG).error(f"执行{device_name}的{method_name}操作失败: {e}")
conn.logger.bind(tag=TAG).error(
f"执行{device_name}的{method_name}操作失败: {e}"
)
# 操作失败时使用大模型提供的失败响应
response = response_failure
@@ -104,7 +105,7 @@ def create_iot_query_function(device_name, prop_name, prop_info):
async def iot_query_function(conn, response_success=None, response_failure=None):
try:
# 打印响应参数
logger.bind(tag=TAG).info(
conn.logger.bind(tag=TAG).info(
f"查询函数接收到的响应参数: success='{response_success}', failure='{response_failure}'"
)
@@ -122,7 +123,9 @@ def create_iot_query_function(device_name, prop_name, prop_info):
return ActionResponse(Action.ERROR, f"属性{prop_name}不存在", response)
except Exception as e:
logger.bind(tag=TAG).error(f"查询{device_name}的{prop_name}时出错: {e}")
conn.logger.bind(tag=TAG).error(
f"查询{device_name}的{prop_name}时出错: {e}"
)
# 查询出错时使用大模型提供的失败响应
response = response_failure
@@ -280,7 +283,7 @@ async def handleIotDescriptors(conn, descriptors):
await asyncio.sleep(1)
wait_max_time -= 1
if wait_max_time <= 0:
logger.bind(tag=TAG).debug("连接对象没有func_handler")
conn.logger.bind(tag=TAG).debug("连接对象没有func_handler")
return
"""处理物联网描述"""
functions_changed = False
@@ -323,7 +326,7 @@ async def handleIotDescriptors(conn, descriptors):
if hasattr(conn, "func_handler"):
for func_name in device_functions:
conn.func_handler.function_registry.register_function(func_name)
logger.bind(tag=TAG).info(
conn.logger.bind(tag=TAG).info(
f"注册IOT函数到function handler: {func_name}"
)
functions_changed = True
@@ -332,8 +335,8 @@ async def handleIotDescriptors(conn, descriptors):
if functions_changed and hasattr(conn, "func_handler"):
conn.func_handler.upload_functions_desc()
func_names = conn.func_handler.current_support_functions()
logger.bind(tag=TAG).info(f"设备类型: {type_id}")
logger.bind(tag=TAG).info(
conn.logger.bind(tag=TAG).info(f"设备类型: {type_id}")
conn.logger.bind(tag=TAG).info(
f"更新function描述列表完成，当前支持的函数: {func_names}"
)
@@ -347,13 +350,13 @@ async def handleIotStatus(conn, states):
for k, v in state["state"].items():
if property_item["name"] == k:
if type(v) != type(property_item["value"]):
logger.bind(tag=TAG).error(
conn.logger.bind(tag=TAG).error(
f"属性{property_item['name']}的值类型不匹配"
)
break
else:
property_item["value"] = v
logger.bind(tag=TAG).info(
conn.logger.bind(tag=TAG).info(
f"物联网状态更新: {key} , {property_item['name']} = {v}"
)
break
@@ -367,7 +370,7 @@ async def get_iot_status(conn, name, property_name):
for property_item in value.properties:
if property_item["name"] == property_name:
return property_item["value"]
logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
conn.logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
return None
@@ -378,16 +381,16 @@ async def set_iot_status(conn, name, property_name, value):
for property_item in iot_descriptor.properties:
if property_item["name"] == property_name:
if type(value) != type(property_item["value"]):
logger.bind(tag=TAG).error(
conn.logger.bind(tag=TAG).error(
f"属性{property_item['name']}的值类型不匹配"
)
return
property_item["value"] = value
logger.bind(tag=TAG).info(
conn.logger.bind(tag=TAG).info(
f"物联网状态更新: {name} , {property_name} = {value}"
)
return
logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
conn.logger.bind(tag=TAG).warning(f"未找到设备 {name} 的属性 {property_name}")
async def send_iot_conn(conn, name, method_name, parameters):
@@ -409,6 +412,6 @@ async def send_iot_conn(conn, name, method_name, parameters):
command["parameters"] = parameters
send_message = json.dumps({"type": "iot", "commands": [command]})
await conn.websocket.send(send_message)
logger.bind(tag=TAG).info(f"发送物联网指令: {send_message}")
conn.logger.bind(tag=TAG).info(f"发送物联网指令: {send_message}")
return
logger.bind(tag=TAG).error(f"未找到方法{method_name}")
conn.logger.bind(tag=TAG).error(f"未找到方法{method_name}")
main/xiaozhi-server/core/handle/receiveAudioHandle.py
+6 -6
View File
@@ -1,4 +1,3 @@
from config.logger import setup_logging
import time
import copy
from core.utils.util import remove_punctuation_and_length
@@ -9,12 +8,13 @@ from core.handle.ttsReportHandle import enqueue_tts_report
from core.providers.tts.base import audio_to_opus_data
TAG = __name__
logger = setup_logging()
async def handleAudioMessage(conn, audio):
if conn.vad is None:
return
if not conn.asr_server_receive:
logger.bind(tag=TAG).debug(f"前期数据处理中，暂停接收")
conn.logger.bind(tag=TAG).debug(f"前期数据处理中，暂停接收")
return
if conn.client_listen_mode == "auto":
have_voice = conn.vad.is_vad(conn, audio)
@@ -40,7 +40,7 @@ async def handleAudioMessage(conn, audio):
conn.asr_server_receive = True
else:
text, _ = await conn.asr.speech_to_text(conn.asr_audio, conn.session_id)
logger.bind(tag=TAG).info(f"识别文本: {text}")
conn.logger.bind(tag=TAG).info(f"识别文本: {text}")
text_len, _ = remove_punctuation_and_length(text)
if text_len > 0:
# 使用自定义模块进行上报
@@ -120,7 +120,7 @@ async def check_bind_device(conn):
if conn.bind_code:
# 确保bind_code是6位数字
if len(conn.bind_code) != 6:
logger.bind(tag=TAG).error(f"无效的绑定码格式: {conn.bind_code}")
conn.logger.bind(tag=TAG).error(f"无效的绑定码格式: {conn.bind_code}")
text = "绑定码格式错误，请检查配置。"
await send_stt_message(conn, text)
return
@@ -144,7 +144,7 @@ async def check_bind_device(conn):
num_packets, _ = audio_to_opus_data(num_path)
conn.audio_play_queue.put((num_packets, None, i + 1))
except Exception as e:
logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
conn.logger.bind(tag=TAG).error(f"播放数字音频失败: {e}")
continue
else:
text = f"没有找到该设备的版本信息，请正确配置 OTA地址，然后重新编译固件。"
main/xiaozhi-server/core/handle/sendAudioHandle.py
+2 -4
View File
@@ -1,11 +1,9 @@
from config.logger import setup_logging
import json
import asyncio
import time
from core.utils.util import get_string_no_punctuation_or_emoji, analyze_emotion
TAG = __name__
logger = setup_logging()
emoji_map = {
"neutral": "😶",
@@ -49,10 +47,10 @@ async def sendAudioMessage(conn, audios, text, text_index=0):
)
if text_index == conn.tts_first_text_index:
logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
await send_tts_message(conn, "sentence_start", text)
is_first_audio = (text_index == conn.tts_first_text_index)
is_first_audio = text_index == conn.tts_first_text_index
await sendAudio(conn, audios, pre_buffer=is_first_audio)
await send_tts_message(conn, "sentence_end", text)
main/xiaozhi-server/core/handle/textHandle.py
+4 -4
View File
@@ -1,4 +1,3 @@
from config.logger import setup_logging
import json
from core.handle.abortHandle import handleAbortMessage
from core.handle.helloHandle import handleHelloMessage
@@ -10,12 +9,11 @@ from core.handle.ttsReportHandle import enqueue_tts_report
import asyncio
TAG = __name__
logger = setup_logging()
async def handleTextMessage(conn, message):
"""处理文本消息"""
logger.bind(tag=TAG).info(f"收到文本消息：{message}")
conn.logger.bind(tag=TAG).info(f"收到文本消息：{message}")
try:
msg_json = json.loads(message)
if isinstance(msg_json, int):
@@ -28,7 +26,9 @@ async def handleTextMessage(conn, message):
elif msg_json["type"] == "listen":
if "mode" in msg_json:
conn.client_listen_mode = msg_json["mode"]
logger.bind(tag=TAG).debug(f"客户端拾音模式：{conn.client_listen_mode}")
conn.logger.bind(tag=TAG).debug(
f"客户端拾音模式：{conn.client_listen_mode}"
)
if msg_json["state"] == "start":
conn.client_have_voice = True
conn.client_voice_stop = False
main/xiaozhi-server/core/handle/ttsReportHandle.py
+6 -11
View File
@@ -9,16 +9,11 @@ TTS上报功能已集成到ConnectionHandler类中。
具体实现请参考core/connection.py中的相关代码。
"""
import os
import uuid
import wave
import opuslib_next
from config.logger import setup_logging
from config.manage_api_client import report
TAG = __name__
logger = setup_logging()
def report_tts(conn, type, text, opus_data):
@@ -32,7 +27,7 @@ def report_tts(conn, type, text, opus_data):
"""
try:
if opus_data:
audio_data = opus_to_wav(opus_data)
audio_data = opus_to_wav(conn, opus_data)
else:
audio_data = None
# 执行上报
@@ -44,10 +39,10 @@ def report_tts(conn, type, text, opus_data):
audio=audio_data,
)
except Exception as e:
logger.bind(tag=TAG).error(f"TTS上报失败: {e}")
conn.logger.bind(tag=TAG).error(f"TTS上报失败: {e}")
def opus_to_wav(opus_data):
def opus_to_wav(conn, opus_data):
"""将Opus数据转换为WAV格式的字节流
Args:
@@ -65,7 +60,7 @@ def opus_to_wav(opus_data):
pcm_frame = decoder.decode(opus_packet, 960) # 960 samples = 60ms
pcm_data.append(pcm_frame)
except opuslib_next.OpusError as e:
logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
conn.logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
if not pcm_data:
raise ValueError("没有有效的PCM数据")
@@ -108,8 +103,8 @@ def enqueue_tts_report(conn, type, text, opus_data):
# 使用连接对象的队列，传入文本和二进制数据而非文件路径
conn.tts_report_queue.put((type, text, opus_data))
logger.bind(tag=TAG).debug(
conn.logger.bind(tag=TAG).debug(
f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} "
)
except Exception as e:
logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}")
conn.logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}")
main/xiaozhi-server/core/mcp/manager.py
+11 -12
View File
@@ -1,9 +1,9 @@
"""MCP服务管理器"""
import asyncio
import os, json
from typing import Dict, Any, List
from .MCPClient import MCPClient
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType
from config.config_loader import get_project_dir
@@ -18,11 +18,10 @@ class MCPManager:
初始化MCP管理器
"""
self.conn = conn
self.logger = setup_logging()
self.config_path = get_project_dir() + "data/.mcp_server_settings.json"
if os.path.exists(self.config_path) == False:
self.config_path = ""
self.logger.bind(tag=TAG).warning(
self.conn.logger.bind(tag=TAG).warning(
f"请检查mcp服务配置文件：data/.mcp_server_settings.json"
)
self.client: Dict[str, MCPClient] = {}
@@ -41,7 +40,7 @@ class MCPManager:
config = json.load(f)
return config.get("mcpServers", {})
except Exception as e:
self.logger.bind(tag=TAG).error(
self.conn.logger.bind(tag=TAG).error(
f"Error loading MCP config from {self.config_path}: {e}"
)
return {}
@@ -51,7 +50,7 @@ class MCPManager:
config = self.load_config()
for name, srv_config in config.items():
if not srv_config.get("command") and not srv_config.get("url"):
self.logger.bind(tag=TAG).warning(
self.conn.logger.bind(tag=TAG).warning(
f"Skipping server {name}: neither command nor url specified"
)
continue
@@ -60,7 +59,7 @@ class MCPManager:
client = MCPClient(srv_config)
await client.initialize()
self.client[name] = client
self.logger.bind(tag=TAG).info(f"Initialized MCP client: {name}")
self.conn.logger.bind(tag=TAG).info(f"Initialized MCP client: {name}")
client_tools = client.get_available_tools()
self.tools.extend(client_tools)
for tool in client_tools:
@@ -73,7 +72,7 @@ class MCPManager:
)
except Exception as e:
self.logger.bind(tag=TAG).error(
self.conn.logger.bind(tag=TAG).error(
f"Failed to initialize MCP server {name}: {e}"
)
self.conn.func_handler.upload_functions_desc()
@@ -94,8 +93,8 @@ class MCPManager:
"""
for tool in self.tools:
if (
tool.get("function") != None
and tool["function"].get("name") == tool_name
tool.get("function") != None
and tool["function"].get("name") == tool_name
):
return True
return False
@@ -110,7 +109,7 @@ class MCPManager:
Raises:
ValueError: 工具未找到时抛出
"""
self.logger.bind(tag=TAG).info(
self.conn.logger.bind(tag=TAG).info(
f"Executing tool {tool_name} with arguments: {arguments}"
)
for client in self.client.values():
@@ -124,9 +123,9 @@ class MCPManager:
for name, client in list(self.client.items()):
try:
await asyncio.wait_for(client.cleanup(), timeout=20)
self.logger.bind(tag=TAG).info(f"MCP client closed: {name}")
self.conn.logger.bind(tag=TAG).info(f"MCP client closed: {name}")
except (asyncio.TimeoutError, Exception) as e:
self.logger.bind(tag=TAG).error(
self.conn.logger.bind(tag=TAG).error(
f"Error closing MCP client {name}: {e}"
)
self.client.clear()
main/xiaozhi-server/core/utils/dialogue.py
+18 -9
View File
@@ -4,7 +4,14 @@ from datetime import datetime
class Message:
def __init__(self, role: str, content: str = None, uniq_id: str = None, tool_calls = None, tool_call_id=None):
def __init__(
self,
role: str,
content: str = None,
uniq_id: str = None,
tool_calls=None,
tool_call_id=None,
):
self.uniq_id = uniq_id if uniq_id is not None else str(uuid.uuid4())
self.role = role
self.content = content
@@ -16,7 +23,7 @@ class Dialogue:
def __init__(self):
self.dialogue: List[Message] = []
# 获取当前时间
self.current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
self.current_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def put(self, message: Message):
self.dialogue.append(message)
@@ -25,7 +32,9 @@ class Dialogue:
if m.tool_calls is not None:
dialogue.append({"role": m.role, "tool_calls": m.tool_calls})
elif m.role == "tool":
dialogue.append({"role": m.role, "tool_call_id": m.tool_call_id, "content": m.content})
dialogue.append(
{"role": m.role, "tool_call_id": m.tool_call_id, "content": m.content}
)
else:
dialogue.append({"role": m.role, "content": m.content})
@@ -44,23 +53,23 @@ class Dialogue:
else:
self.put(Message(role="system", content=new_content))
def get_llm_dialogue_with_memory(self, memory_str: str = None) -> List[Dict[str, str]]:
def get_llm_dialogue_with_memory(
self, memory_str: str = None
) -> List[Dict[str, str]]:
if memory_str is None or len(memory_str) == 0:
return self.get_llm_dialogue()
# 构建带记忆的对话
dialogue = []
# 添加系统提示和记忆
system_message = next(
(msg for msg in self.dialogue if msg.role == "system"), None
)
if system_message:
enhanced_system_prompt = (
f"{system_message.content}\n\n"
f"相关记忆：\n{memory_str}"
f"{system_message.content}\n\n" f"相关记忆：\n{memory_str}"
)
dialogue.append({"role": "system", "content": enhanced_system_prompt})
main/xiaozhi-server/core/utils/util.py
+42 -6
View File
@@ -350,12 +350,6 @@ def initialize_modules(
str(config.get("delete_audio", True)).lower() in ("true", "1", "yes"),
)
logger.bind(tag=TAG).info(f"初始化组件: asr成功 {select_asr_module}")
# 初始化自定义prompt
if config.get("prompt", None) is not None:
modules["prompt"] = config["prompt"]
logger.bind(tag=TAG).info(f"初始化组件: prompt成功 {modules['prompt'][:50]}...")
return modules
@@ -913,3 +907,45 @@ def audio_to_opus_data(audio_file_path):
opus_datas.append(opus_data)
return opus_datas, duration
def check_vad_update(before_config, new_config):
if new_config["selected_module"].get("VAD") is None:
return False
update_vad = False
current_vad_module = before_config["selected_module"]["VAD"]
new_vad_module = new_config["selected_module"]["VAD"]
current_vad_type = (
current_vad_module
if "type" not in before_config["VAD"][current_vad_module]
else before_config["VAD"][current_vad_module]["type"]
)
new_vad_type = (
new_vad_module
if "type" not in new_config["VAD"][new_vad_module]
else new_config["VAD"][new_vad_module]["type"]
)
print(f"前vad:{current_vad_type}，后vad:{new_vad_type}")
update_vad = current_vad_type != new_vad_type
return update_vad
def check_asr_update(before_config, new_config):
if new_config["selected_module"].get("ASR") is None:
return False
update_asr = False
current_asr_module = before_config["selected_module"]["ASR"]
new_asr_module = new_config["selected_module"]["ASR"]
current_asr_type = (
current_asr_module
if "type" not in before_config["ASR"][current_asr_module]
else before_config["ASR"][current_asr_module]["type"]
)
new_asr_type = (
new_asr_module
if "type" not in new_config["ASR"][new_asr_module]
else new_config["ASR"][new_asr_module]["type"]
)
print(f"前asr:{current_asr_type}，后asr:{new_asr_type}")
update_asr = current_asr_type != new_asr_type
return update_asr
main/xiaozhi-server/core/websocket_server.py
+15 -39
View File
@@ -2,7 +2,7 @@ import asyncio
import websockets
from config.logger import setup_logging
from core.connection import ConnectionHandler
from core.utils.util import initialize_modules
from core.utils.util import initialize_modules, check_vad_update, check_asr_update
from config.config_loader import get_config_from_api
TAG = __name__
@@ -84,38 +84,8 @@ class WebSocketServer:
return False
# 检查 VAD 和 ASR 类型是否需要更新
update_vad = False
update_asr = False
# 获取当前和新的 VAD 类型
current_vad_module = self.config["selected_module"]["VAD"]
new_vad_module = new_config["selected_module"]["VAD"]
current_vad_type = (
current_vad_module
if "type" not in self.config["VAD"][current_vad_module]
else self.config["VAD"][current_vad_module]["type"]
)
new_vad_type = (
new_vad_module
if "type" not in new_config["VAD"][new_vad_module]
else new_config["VAD"][new_vad_module]["type"]
)
update_vad = current_vad_type != new_vad_type
# 获取当前和新的 ASR 类型
current_asr_module = self.config["selected_module"]["ASR"]
new_asr_module = new_config["selected_module"]["ASR"]
current_asr_type = (
current_asr_module
if "type" not in self.config["ASR"][current_asr_module]
else self.config["ASR"][current_asr_module]["type"]
)
new_asr_type = (
new_asr_module
if "type" not in new_config["ASR"][new_asr_module]
else new_config["ASR"][new_asr_module]["type"]
)
update_asr = current_asr_type != new_asr_type
update_vad = check_vad_update(self.config, new_config)
update_asr = check_asr_update(self.config, new_config)
# 更新配置
self.config = new_config
@@ -132,12 +102,18 @@ class WebSocketServer:
)
# 更新组件实例
self._vad = modules["vad"] if "vad" in modules else None
self._asr = modules["asr"] if "asr" in modules else None
self._tts = modules["tts"] if "tts" in modules else None
self._llm = modules["llm"] if "llm" in modules else None
self._intent = modules["intent"] if "intent" in modules else None
self._memory = modules["memory"] if "memory" in modules else None
if "vad" in modules:
self._vad = modules["vad"]
if "asr" in modules:
self._asr = modules["asr"]
if "tts" in modules:
self._tts = modules["tts"]
if "llm" in modules:
self._llm = modules["llm"]
if "intent" in modules:
self._intent = modules["intent"]
if "memory" in modules:
self._memory = modules["memory"]
return True
except Exception as e:
main/xiaozhi-server/plugins_func/functions/play_music.py
+13 -13
View File
@@ -11,9 +11,7 @@ from core.utils import p3
from core.handle.sendAudioHandle import send_stt_message
from plugins_func.register import register_function, ToolType, ActionResponse, Action
TAG = __name__
logger = setup_logging()
MUSIC_CACHE = {}
@@ -45,7 +43,7 @@ def play_music(conn, song_name: str):
# 检查事件循环状态
if not conn.loop.is_running():
logger.bind(tag=TAG).error("事件循环未运行，无法提交任务")
conn.logger.bind(tag=TAG).error("事件循环未运行，无法提交任务")
return ActionResponse(
action=Action.RESPONSE, result="系统繁忙", response="请稍后再试"
)
@@ -59,9 +57,9 @@ def play_music(conn, song_name: str):
def handle_done(f):
try:
f.result() # 可在此处理成功逻辑
logger.bind(tag=TAG).info("播放完成")
conn.logger.bind(tag=TAG).info("播放完成")
except Exception as e:
logger.bind(tag=TAG).error(f"播放失败: {e}")
conn.logger.bind(tag=TAG).error(f"播放失败: {e}")
future.add_done_callback(handle_done)
@@ -69,7 +67,7 @@ def play_music(conn, song_name: str):
action=Action.NONE, result="指令已接收", response="正在为您播放音乐"
)
except Exception as e:
logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
conn.logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
return ActionResponse(
action=Action.RESPONSE, result=str(e), response="播放音乐时出错了"
)
@@ -150,7 +148,7 @@ async def handle_music_command(conn, text):
"""处理音乐播放指令"""
clean_text = re.sub(r"[^\w\s]", "", text).strip()
logger.bind(tag=TAG).debug(f"检查是否是音乐命令: {clean_text}")
conn.logger.bind(tag=TAG).debug(f"检查是否是音乐命令: {clean_text}")
# 尝试匹配具体歌名
if os.path.exists(MUSIC_CACHE["music_dir"]):
@@ -165,7 +163,7 @@ async def handle_music_command(conn, text):
if potential_song:
best_match = _find_best_match(potential_song, MUSIC_CACHE["music_files"])
if best_match:
logger.bind(tag=TAG).info(f"找到最匹配的歌曲: {best_match}")
conn.logger.bind(tag=TAG).info(f"找到最匹配的歌曲: {best_match}")
await play_local_music(conn, specific_file=best_match)
return True
# 检查是否是通用播放音乐命令
@@ -195,7 +193,9 @@ async def play_local_music(conn, specific_file=None):
"""播放本地音乐文件"""
try:
if not os.path.exists(MUSIC_CACHE["music_dir"]):
logger.bind(tag=TAG).error(f"音乐目录不存在: " + MUSIC_CACHE["music_dir"])
conn.logger.bind(tag=TAG).error(
f"音乐目录不存在: " + MUSIC_CACHE["music_dir"]
)
return
# 确保路径正确性
@@ -204,13 +204,13 @@ async def play_local_music(conn, specific_file=None):
music_path = os.path.join(MUSIC_CACHE["music_dir"], specific_file)
else:
if not MUSIC_CACHE["music_files"]:
logger.bind(tag=TAG).error("未找到MP3音乐文件")
conn.logger.bind(tag=TAG).error("未找到MP3音乐文件")
return
selected_music = random.choice(MUSIC_CACHE["music_files"])
music_path = os.path.join(MUSIC_CACHE["music_dir"], selected_music)
if not os.path.exists(music_path):
logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
conn.logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
return
text = _get_random_play_prompt(selected_music)
await send_stt_message(conn, text)
@@ -233,5 +233,5 @@ async def play_local_music(conn, specific_file=None):
conn.audio_play_queue.put((opus_packets, None, conn.tts_last_text_index))
except Exception as e:
logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")
logger.bind(tag=TAG).error(f"详细错误: {traceback.format_exc()}")
conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")
conn.logger.bind(tag=TAG).error(f"详细错误: {traceback.format_exc()}")
main/xiaozhi-server/plugins_func/functions/plugin_loader.py
+37 -32
View File
@@ -1,51 +1,56 @@
from plugins_func.register import register_function,ToolType, ActionResponse, Action
from config.logger import setup_logging
TAG = __name__
logger = setup_logging()
from plugins_func.register import register_function, ToolType, ActionResponse, Action
plugin_loader_function_desc = {
"type": "function",
"function": {
"name": "plugin_loader",
"description": "当用户想加载或卸载插件/function时，调用此函数：支持的插件列表为[plugins]",
"parameters": {
"type": "object",
"properties": {
"oper": {
"type": "string",
"description": "load or unload"
},
"name":{
"type": "string",
"description": "要加载或卸载的插件名字"
}
},
"required": ["oper","name"]
}
}
}
"type": "function",
"function": {
"name": "plugin_loader",
"description": "当用户想加载或卸载插件/function时，调用此函数：支持的插件列表为[plugins]",
"parameters": {
"type": "object",
"properties": {
"oper": {"type": "string", "description": "load or unload"},
"name": {"type": "string", "description": "要加载或卸载的插件名字"},
},
"required": ["oper", "name"],
},
},
}
@register_function('plugin_loader', plugin_loader_function_desc, ToolType.SYSTEM_CTL)
@register_function("plugin_loader", plugin_loader_function_desc, ToolType.SYSTEM_CTL)
def plugin_loader(conn, oper: str, name: str):
"""插件加载"""
if oper not in ["load", "unload"]:
return ActionResponse(action=Action.RESPONSE, result="插件操作失败", response="不支持的操作")
return ActionResponse(
action=Action.RESPONSE, result="插件操作失败", response="不支持的操作"
)
cur_support = conn.func_handler.current_support_functions()
if oper == "load":
if name in cur_support:
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response=f"{name}插件已加载,无需重复加载")
return ActionResponse(
action=Action.RESPONSE,
result="插件加载失败",
response=f"{name}插件已加载,无需重复加载",
)
func = conn.func_handler.function_registry.register_function(name)
if not func:
return ActionResponse(action=Action.RESPONSE, result="插件加载失败", response="插件未找到")
return ActionResponse(
action=Action.RESPONSE, result="插件加载失败", response="插件未找到"
)
res = f"{name}插件加载成功"
else:
if name not in cur_support:
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response=f"{name}插件未加载")
return ActionResponse(
action=Action.RESPONSE,
result="插件卸载失败",
response=f"{name}插件未加载",
)
bOK = conn.func_handler.function_registry.unregister_function(name)
if not bOK:
return ActionResponse(action=Action.RESPONSE, result="插件卸载失败", response="插件未找到")
return ActionResponse(
action=Action.RESPONSE, result="插件卸载失败", response="插件未找到"
)
res = f"{name}插件卸载成功"
conn.func_handler.upload_functions_desc()
return ActionResponse(action=Action.RESPONSE, result="插件操作成功", response=res)