Merge pull request #2890 from shengzhou1216/feature/python-typing

feat: 为多个模块添加类型注解以增强代码可读性

main/xiaozhi-server/core/connection.py

@@ -78,7 +78,7 @@ class ConnectionHandler:
 
         self.read_config_from_api = self.config.get("read_config_from_api", False)
 
-        self.websocket = None
+        self.websocket: websockets.ServerConnection | None = None
         self.headers = None
         self.device_id = None
         self.client_ip = None
@@ -169,7 +169,7 @@ class ConnectionHandler:
         # 初始化提示词管理器
         self.prompt_manager = PromptManager(self.config, self.logger)
 
-    async def handle_connection(self, ws):
+    async def handle_connection(self, ws: websockets.ServerConnection):
         try:
             # 获取运行中的事件循环（必须在异步上下文中）
             self.loop = asyncio.get_running_loop()

main/xiaozhi-server/core/handle/abortHandle.py

@@ -1,9 +1,12 @@
 import json
+from typing import TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 TAG = __name__
 
 
-async def handleAbortMessage(conn):
+async def handleAbortMessage(conn: "ConnectionHandler"):
     conn.logger.bind(tag=TAG).info("Abort message received")
     # 设置成打断状态，会自动打断llm、tts任务
     conn.client_abort = True

main/xiaozhi-server/core/handle/helloHandle.py

@@ -3,16 +3,17 @@ import json
 import uuid
 import random
 import asyncio
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.utils.dialogue import Message
 from core.utils.util import audio_to_data
 from core.providers.tts.dto.dto import SentenceType
 from core.utils.wakeup_word import WakeupWordsConfig
 from core.handle.sendAudioHandle import sendAudioMessage, send_tts_message
 from core.utils.util import remove_punctuation_and_length, opus_datas_to_wav_bytes
-from core.providers.tools.device_mcp import (
+from core.providers.tools.device_mcp import MCPClient, send_mcp_initialize_message
-    MCPClient,
-    send_mcp_initialize_message
-)
 
 TAG = __name__
 
@@ -38,7 +39,7 @@ wakeup_words_config = WakeupWordsConfig()
 _wakeup_response_lock = asyncio.Lock()
 
 
-async def handleHelloMessage(conn, msg_json):
+async def handleHelloMessage(conn: "ConnectionHandler", msg_json):
     """处理hello消息"""
     audio_params = msg_json.get("audio_params")
     if audio_params:
@@ -59,7 +60,7 @@ async def handleHelloMessage(conn, msg_json):
     await conn.websocket.send(json.dumps(conn.welcome_msg))
 
 
-async def checkWakeupWords(conn, text):
+async def checkWakeupWords(conn: "ConnectionHandler", text):
     enable_wakeup_words_response_cache = conn.config[
         "enable_wakeup_words_response_cache"
     ]
@@ -120,7 +121,7 @@ async def checkWakeupWords(conn, text):
     return True
 
 
-async def wakeupWordsResponse(conn):
+async def wakeupWordsResponse(conn: "ConnectionHandler"):
     if not conn.tts:
         return
 

main/xiaozhi-server/core/handle/intentHandler.py

@@ -1,6 +1,10 @@
 import json
 import uuid
 import asyncio
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.utils.dialogue import Message
 from core.providers.tts.dto.dto import ContentType
 from core.handle.helloHandle import checkWakeupWords
@@ -12,10 +16,10 @@ from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType
 TAG = __name__
 
 
-async def handle_user_intent(conn, text):
+async def handle_user_intent(conn: "ConnectionHandler", text):
     # 预处理输入文本，处理可能的JSON格式
     try:
-        if text.strip().startswith('{') and text.strip().endswith('}'):
+        if text.strip().startswith("{") and text.strip().endswith("}"):
             parsed_data = json.loads(text)
             if isinstance(parsed_data, dict) and "content" in parsed_data:
                 text = parsed_data["content"]  # 提取content用于意图分析
@@ -45,7 +49,7 @@ async def handle_user_intent(conn, text):
     return await process_intent_result(conn, intent_result, text)
 
 
-async def check_direct_exit(conn, text):
+async def check_direct_exit(conn: "ConnectionHandler", text):
     """检查是否有明确的退出命令"""
     _, text = remove_punctuation_and_length(text)
     cmd_exit = conn.cmd_exit
@@ -58,7 +62,7 @@ async def check_direct_exit(conn, text):
     return False
 
 
-async def analyze_intent_with_llm(conn, text):
+async def analyze_intent_with_llm(conn: "ConnectionHandler", text):
     """使用LLM分析用户意图"""
     if not hasattr(conn, "intent") or not conn.intent:
         conn.logger.bind(tag=TAG).warning("意图识别服务未初始化")
@@ -75,7 +79,9 @@ async def analyze_intent_with_llm(conn, text):
     return None
 
 
-async def process_intent_result(conn, intent_result, original_text):
+async def process_intent_result(
+    conn: "ConnectionHandler", intent_result, original_text
+):
     """处理意图识别结果"""
     try:
         # 尝试将结果解析为JSON
@@ -100,7 +106,9 @@ async def process_intent_result(conn, intent_result, original_text):
 
                     from core.utils.current_time import get_current_time_info
 
-                    current_time, today_date, today_weekday, lunar_date = get_current_time_info()
+                    current_time, today_date, today_weekday, lunar_date = (
+                        get_current_time_info()
+                    )
 
                     # 构建带上下文的基础提示
                     context_prompt = f"""当前时间：{current_time}
@@ -188,7 +196,7 @@ async def process_intent_result(conn, intent_result, original_text):
         return False
 
 
-def speak_txt(conn, text):
+def speak_txt(conn: "ConnectionHandler", text):
     # 记录文本
     conn.tts_MessageText = text
 

main/xiaozhi-server/core/handle/receiveAudioHandle.py

@@ -1,6 +1,10 @@
 import time
 import json
 import asyncio
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.utils.util import audio_to_data
 from core.handle.abortHandle import handleAbortMessage
 from core.handle.intentHandler import handle_user_intent
@@ -10,7 +14,7 @@ from core.handle.sendAudioHandle import send_stt_message, SentenceType
 TAG = __name__
 
 
-async def handleAudioMessage(conn, audio):
+async def handleAudioMessage(conn: "ConnectionHandler", audio):
     # 当前片段是否有人说话
     have_voice = conn.vad.is_vad(conn, audio)
     # 如果设备刚刚被唤醒，短暂忽略VAD检测
@@ -30,13 +34,13 @@ async def handleAudioMessage(conn, audio):
     await conn.asr.receive_audio(conn, audio, have_voice)
 
 
-async def resume_vad_detection(conn):
+async def resume_vad_detection(conn: "ConnectionHandler"):
     # 等待2秒后恢复VAD检测
     await asyncio.sleep(2)
     conn.just_woken_up = False
 
 
-async def startToChat(conn, text):
+async def startToChat(conn: "ConnectionHandler", text):
     # 检查输入是否是JSON格式（包含说话人信息）
     speaker_name = None
     language_tag = None
@@ -96,7 +100,7 @@ async def startToChat(conn, text):
     conn.executor.submit(conn.chat, actual_text)
 
 
-async def no_voice_close_connect(conn, have_voice):
+async def no_voice_close_connect(conn: "ConnectionHandler", have_voice):
     if have_voice:
         conn.last_activity_time = time.time() * 1000
         return
@@ -123,7 +127,7 @@ async def no_voice_close_connect(conn, have_voice):
             await startToChat(conn, prompt)
 
 
-async def max_out_size(conn):
+async def max_out_size(conn: "ConnectionHandler"):
     # 播放超出最大输出字数的提示
     conn.client_abort = False
     text = "不好意思，我现在有点事情要忙，明天这个时候我们再聊，约好了哦！明天不见不散，拜拜！"
@@ -134,7 +138,7 @@ async def max_out_size(conn):
     conn.close_after_chat = True
 
 
-async def check_bind_device(conn):
+async def check_bind_device(conn: "ConnectionHandler"):
     if conn.bind_code:
         # 确保bind_code是6位数字
         if len(conn.bind_code) != 6:

main/xiaozhi-server/core/handle/reportHandle.py

@@ -11,13 +11,17 @@ TTS上报功能已集成到ConnectionHandler类中。
 
 import time
 import opuslib_next
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 from config.manage_api_client import report as manage_report
 
 TAG = __name__
 
 
-async def report(conn, type, text, opus_data, report_time):
+async def report(conn: "ConnectionHandler", type, text, opus_data, report_time):
     """执行聊天记录上报操作
 
     Args:
@@ -45,7 +49,7 @@ async def report(conn, type, text, opus_data, report_time):
         conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}")
 
 
-def opus_to_wav(conn, opus_data):
+def opus_to_wav(conn: "ConnectionHandler", opus_data):
     """将Opus数据转换为WAV格式的字节流
 
     Args:
@@ -100,7 +104,7 @@ def opus_to_wav(conn, opus_data):
                 conn.logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")
 
 
-def enqueue_tts_report(conn, text, opus_data):
+def enqueue_tts_report(conn: "ConnectionHandler", text, opus_data):
     if not conn.read_config_from_api or conn.need_bind or not conn.report_tts_enable:
         return
     if conn.chat_history_conf == 0:
@@ -128,7 +132,7 @@ def enqueue_tts_report(conn, text, opus_data):
         conn.logger.bind(tag=TAG).error(f"加入TTS上报队列失败: {text}, {e}")
 
 
-def enqueue_asr_report(conn, text, opus_data):
+def enqueue_asr_report(conn: "ConnectionHandler", text, opus_data):
     if not conn.read_config_from_api or conn.need_bind or not conn.report_asr_enable:
         return
     if conn.chat_history_conf == 0:

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -1,6 +1,10 @@
 import json
 import time
 import asyncio
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.utils import textUtils
 from core.utils.util import audio_to_data
 from core.providers.tts.dto.dto import SentenceType
@@ -13,7 +17,7 @@ AUDIO_FRAME_DURATION = 60
 PRE_BUFFER_COUNT = 5
 
 
-async def sendAudioMessage(conn, sentenceType, audios, text):
+async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text):
     if conn.tts.tts_audio_first_sentence:
         conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}")
         conn.tts.tts_audio_first_sentence = False
@@ -47,7 +51,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
             await conn.close()
 
 
-async def _wait_for_audio_completion(conn):
+async def _wait_for_audio_completion(conn: "ConnectionHandler"):
     """
     等待音频队列清空并等待预缓冲包播放完成
 
@@ -70,7 +74,9 @@ async def _wait_for_audio_completion(conn):
         conn.logger.bind(tag=TAG).debug("音频发送完成")
 
 
-async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
+async def _send_to_mqtt_gateway(
+    conn: "ConnectionHandler", opus_packet, timestamp, sequence
+):
     """
     发送带16字节头部的opus数据包给mqtt_gateway
     Args:
@@ -92,7 +98,9 @@ async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     await conn.websocket.send(complete_packet)
 
 
-async def sendAudio(conn, audios, frame_duration=AUDIO_FRAME_DURATION):
+async def sendAudio(
+    conn: "ConnectionHandler", audios, frame_duration=AUDIO_FRAME_DURATION
+):
     """
     发送音频包，使用 AudioRateController 进行精确的流量控制
 
@@ -121,7 +129,9 @@ async def sendAudio(conn, audios, frame_duration=AUDIO_FRAME_DURATION):
     )
 
 
-def _get_or_create_rate_controller(conn, frame_duration, is_single_packet):
+def _get_or_create_rate_controller(
+    conn: "ConnectionHandler", frame_duration, is_single_packet
+):
     """
     获取或创建 RateController 和 flow_control
 
@@ -177,7 +187,7 @@ def _get_or_create_rate_controller(conn, frame_duration, is_single_packet):
     return conn.audio_rate_controller, conn.audio_flow_control
 
 
-def _start_background_sender(conn, rate_controller, flow_control):
+def _start_background_sender(conn: "ConnectionHandler", rate_controller, flow_control):
     """
     启动后台发送循环任务
 
@@ -201,7 +211,7 @@ def _start_background_sender(conn, rate_controller, flow_control):
 
 
 async def _send_audio_with_rate_control(
-    conn, audio_list, rate_controller, flow_control, send_delay
+    conn: "ConnectionHandler", audio_list, rate_controller, flow_control, send_delay
 ):
     """
     使用 rate_controller 发送音频包
@@ -233,7 +243,7 @@ async def _send_audio_with_rate_control(
             rate_controller.add_audio(packet)
 
 
-async def _do_send_audio(conn, opus_packet, flow_control):
+async def _do_send_audio(conn: "ConnectionHandler", opus_packet, flow_control):
     """
     执行实际的音频发送
     """
@@ -254,7 +264,7 @@ async def _do_send_audio(conn, opus_packet, flow_control):
     flow_control["sequence"] = sequence + 1
 
 
-async def send_tts_message(conn, state, text=None):
+async def send_tts_message(conn: "ConnectionHandler", state, text=None):
     """发送 TTS 状态消息"""
     if text is None and state == "sentence_start":
         return
@@ -281,7 +291,7 @@ async def send_tts_message(conn, state, text=None):
     await conn.websocket.send(json.dumps(message))
 
 
-async def send_stt_message(conn, text):
+async def send_stt_message(conn: "ConnectionHandler", text):
     """发送 STT 状态消息"""
     end_prompt_str = conn.config.get("end_prompt", {}).get("prompt")
     if end_prompt_str and end_prompt_str == text:

main/xiaozhi-server/core/handle/textHandle.py

@@ -1,3 +1,7 @@
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.handle.textMessageHandlerRegistry import TextMessageHandlerRegistry
 from core.handle.textMessageProcessor import TextMessageProcessor
 
@@ -9,6 +13,7 @@ message_registry = TextMessageHandlerRegistry()
 # 创建全局消息处理器实例
 message_processor = TextMessageProcessor(message_registry)
 
-async def handleTextMessage(conn, message):
+
+async def handleTextMessage(conn: "ConnectionHandler", message):
     """处理文本消息"""
     await message_processor.process_message(conn, message)

main/xiaozhi-server/core/handle/textHandler/helloMessageHandler.py

@@ -1,5 +1,7 @@
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.handle.helloHandle import handleHelloMessage
 from core.handle.textMessageHandler import TextMessageHandler
 from core.handle.textMessageType import TextMessageType
@@ -12,5 +14,5 @@ class HelloTextMessageHandler(TextMessageHandler):
     def message_type(self) -> TextMessageType:
         return TextMessageType.HELLO
 
-    async def handle(self, conn, msg_json: Dict[str, Any]) -> None:
+    async def handle(self, conn: "ConnectionHandler", msg_json: Dict[str, Any]) -> None:
         await handleHelloMessage(conn, msg_json)

main/xiaozhi-server/core/handle/textHandler/iotMessageHandler.py

@@ -1,6 +1,8 @@
 import asyncio
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.handle.textMessageHandler import TextMessageHandler
 from core.handle.textMessageType import TextMessageType
 from core.providers.tools.device_iot import handleIotStatus, handleIotDescriptors
@@ -13,7 +15,7 @@ class IotTextMessageHandler(TextMessageHandler):
     def message_type(self) -> TextMessageType:
         return TextMessageType.IOT
 
-    async def handle(self, conn, msg_json: Dict[str, Any]) -> None:
+    async def handle(self, conn: "ConnectionHandler", msg_json: Dict[str, Any]) -> None:
         if "descriptors" in msg_json:
             asyncio.create_task(handleIotDescriptors(conn, msg_json["descriptors"]))
         if "states" in msg_json:

main/xiaozhi-server/core/handle/textHandler/listenMessageHandler.py

@@ -1,6 +1,9 @@
 import time
 import asyncio
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
@@ -19,7 +22,7 @@ class ListenTextMessageHandler(TextMessageHandler):
     def message_type(self) -> TextMessageType:
         return TextMessageType.LISTEN
 
-    async def handle(self, conn, msg_json: Dict[str, Any]) -> None:
+    async def handle(self, conn: "ConnectionHandler", msg_json: Dict[str, Any]) -> None:
         if "mode" in msg_json:
             conn.client_listen_mode = msg_json["mode"]
             conn.logger.bind(tag=TAG).debug(

main/xiaozhi-server/core/handle/textHandler/mcpMessageHandler.py

@@ -1,6 +1,8 @@
 import asyncio
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.handle.textMessageHandler import TextMessageHandler
 from core.handle.textMessageType import TextMessageType
 from core.providers.tools.device_mcp import handle_mcp_message
@@ -13,7 +15,7 @@ class McpTextMessageHandler(TextMessageHandler):
     def message_type(self) -> TextMessageType:
         return TextMessageType.MCP
 
-    async def handle(self, conn, msg_json: Dict[str, Any]) -> None:
+    async def handle(self, conn: "ConnectionHandler", msg_json: Dict[str, Any]) -> None:
         if "payload" in msg_json:
             asyncio.create_task(
                 handle_mcp_message(conn, conn.mcp_client, msg_json["payload"])

main/xiaozhi-server/core/handle/textMessageProcessor.py

@@ -1,5 +1,8 @@
 import json
+from typing import TYPE_CHECKING
 
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from core.handle.textMessageHandlerRegistry import TextMessageHandlerRegistry
 
 TAG = __name__
@@ -11,7 +14,7 @@ class TextMessageProcessor:
     def __init__(self, registry: TextMessageHandlerRegistry):
         self.registry = registry
 
-    async def process_message(self, conn, message: str) -> None:
+    async def process_message(self, conn: "ConnectionHandler", message: str) -> None:
         """处理消息的主入口"""
         try:
             # 解析JSON消息

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

@@ -13,6 +13,10 @@ from datetime import datetime
 from config.logger import setup_logging
 from core.providers.asr.base import ASRProviderBase
 from core.providers.asr.dto.dto import InterfaceType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -146,7 +150,7 @@ class ASRProvider(ASRProviderBase):
                 logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
                 await self._cleanup()
 
-    async def _start_recognition(self, conn):
+    async def _start_recognition(self, conn: "ConnectionHandler"):
         """开始识别会话"""
         if self._is_token_expired():
             self._refresh_token()
@@ -192,7 +196,7 @@ class ASRProvider(ASRProviderBase):
         await self.asr_ws.send(json.dumps(start_request, ensure_ascii=False))
         logger.bind(tag=TAG).debug("已发送开始请求，等待服务器准备...")
 
-    async def _forward_results(self, conn):
+    async def _forward_results(self, conn: "ConnectionHandler"):
         """转发识别结果"""
         try:
             while not conn.stop_event.is_set():

main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py

@@ -3,7 +3,11 @@ import uuid
 import asyncio
 import websockets
 import opuslib_next
-from typing import List
+from typing import List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
 from config.logger import setup_logging
 from core.providers.asr.base import ASRProviderBase
 from core.providers.asr.dto.dto import InterfaceType
@@ -74,7 +78,7 @@ class ASRProvider(ASRProviderBase):
                 logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
                 await self._cleanup()
 
-    async def _start_recognition(self, conn):
+    async def _start_recognition(self, conn: "ConnectionHandler"):
         """开始识别会话"""
         try:
             # 如果为手动模式,设置超时时长为最大值
@@ -154,7 +158,7 @@ class ASRProvider(ASRProviderBase):
 
         return message
 
-    async def _forward_results(self, conn):
+    async def _forward_results(self, conn: "ConnectionHandler"):
         """转发识别结果"""
         try:
             while not conn.stop_event.is_set():

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

@@ -5,21 +5,25 @@ import uuid
 import json
 import time
 import queue
+import shutil
 import asyncio
+import tempfile
 import traceback
 import threading
-import shutil
 import opuslib_next
 
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
-from typing import Optional, Tuple, List, NamedTuple
 from core.providers.asr.dto.dto import InterfaceType
 from core.handle.receiveAudioHandle import startToChat
 from core.handle.reportHandle import enqueue_asr_report
 from core.utils.util import remove_punctuation_and_length
 from core.handle.receiveAudioHandle import handleAudioMessage
-import tempfile
+from typing import Optional, Tuple, List, NamedTuple, TYPE_CHECKING
+
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -30,14 +34,14 @@ class ASRProviderBase(ABC):
         pass
 
     # 打开音频通道
-    async def open_audio_channels(self, conn):
+    async def open_audio_channels(self, conn: "ConnectionHandler"):
         conn.asr_priority_thread = threading.Thread(
             target=self.asr_text_priority_thread, args=(conn,), daemon=True
         )
         conn.asr_priority_thread.start()
 
     # 有序处理ASR音频
-    def asr_text_priority_thread(self, conn):
+    def asr_text_priority_thread(self, conn: "ConnectionHandler"):
         while not conn.stop_event.is_set():
             try:
                 message = conn.asr_audio_queue.get(timeout=1)
@@ -55,7 +59,7 @@ class ASRProviderBase(ABC):
                 continue
 
     # 接收音频
-    async def receive_audio(self, conn, audio, audio_have_voice):
+    async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
         if conn.client_listen_mode == "manual":
             # 手动模式：缓存音频用于ASR识别
             conn.asr_audio.append(audio)
@@ -77,7 +81,7 @@ class ASRProviderBase(ABC):
                     await self.handle_voice_stop(conn, asr_audio_task)
 
     # 处理语音停止
-    async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]):
+    async def handle_voice_stop(self, conn: "ConnectionHandler", asr_audio_task: List[bytes]):
         """并行处理ASR和声纹识别"""
         try:
             total_start_time = time.monotonic()

main/xiaozhi-server/core/providers/asr/doubao_stream.py

@@ -7,6 +7,10 @@ import opuslib_next
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
 from core.providers.asr.dto.dto import InterfaceType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -34,7 +38,9 @@ class ASRProvider(ASRProviderBase):
 
         # 火山引擎ASR配置
         enable_multilingual = config.get("enable_multilingual", False)
-        self.enable_multilingual = False if str(enable_multilingual).lower() == 'false' else True
+        self.enable_multilingual = (
+            False if str(enable_multilingual).lower() == "false" else True
+        )
         if self.enable_multilingual:
             self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream"
         else:
@@ -59,7 +65,7 @@ class ASRProvider(ASRProviderBase):
     async def open_audio_channels(self, conn):
         await super().open_audio_channels(conn)
 
-    async def receive_audio(self, conn, audio, audio_have_voice):
+    async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
         # 先调用父类方法处理基础逻辑
         await super().receive_audio(conn, audio, audio_have_voice)
         
@@ -151,7 +157,7 @@ class ASRProvider(ASRProviderBase):
             except Exception as e:
                 logger.bind(tag=TAG).info(f"发送音频数据时发生错误: {e}")
 
-    async def _forward_asr_results(self, conn):
+    async def _forward_asr_results(self, conn: "ConnectionHandler"):
         try:
             while self.asr_ws and not conn.stop_event.is_set():
                 # 获取当前连接的音频数据
@@ -172,8 +178,9 @@ class ASRProvider(ASRProviderBase):
                             utterances = payload["result"].get("utterances", [])
                             # 检查duration和空文本的情况
                             if (
-                                not self.enable_multilingual # 注意：多语种模式不返回中间结果，需要等待最终结果
+                                not self.enable_multilingual  # 注意：多语种模式不返回中间结果，需要等待最终结果
-                                and payload.get("audio_info", {}).get("duration", 0) > 2000
+                                and payload.get("audio_info", {}).get("duration", 0)
+                                > 2000
                                 and not utterances
                                 and not payload["result"].get("text")
                                 and conn.client_listen_mode != "manual"
@@ -218,7 +225,9 @@ class ASRProvider(ASRProviderBase):
                                         # 自动模式下直接覆盖
                                         self.text = current_text
                                         if len(audio_data) > 15:  # 确保有足够音频数据
-                                            await self.handle_voice_stop(conn, audio_data)
+                                            await self.handle_voice_stop(
+                                                conn, audio_data
+                                            )
                                     break
                         elif "error" in payload:
                             error_msg = payload.get("error", "未知错误")
@@ -260,7 +269,9 @@ class ASRProvider(ASRProviderBase):
             try:
                 # 发送结束标记的音频帧（gzip压缩的空数据）
                 empty_payload = gzip.compress(b"")
-                last_audio_request = bytearray(self.generate_last_audio_default_header())
+                last_audio_request = bytearray(
+                    self.generate_last_audio_default_header()
+                )
                 last_audio_request.extend(len(empty_payload).to_bytes(4, "big"))
                 last_audio_request.extend(empty_payload)
                 await self.asr_ws.send(last_audio_request)
@@ -408,7 +419,7 @@ class ASRProvider(ASRProviderBase):
         self.is_processing = False
 
         # 显式释放decoder资源
-        if hasattr(self, 'decoder') and self.decoder is not None:
+        if hasattr(self, "decoder") and self.decoder is not None:
             try:
                 del self.decoder
                 self.decoder = None

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

@@ -9,7 +9,10 @@ import gc
 from time import mktime
 from datetime import datetime
 from urllib.parse import urlencode
-from typing import List
+from typing import List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from config.logger import setup_logging
 from wsgiref.handlers import format_date_time
 from core.providers.asr.base import ASRProviderBase
@@ -94,10 +97,10 @@ class ASRProvider(ASRProviderBase):
         url = url + "?" + urlencode(v)
         return url
 
-    async def open_audio_channels(self, conn):
+    async def open_audio_channels(self, conn: "ConnectionHandler"):
         await super().open_audio_channels(conn)
 
-    async def receive_audio(self, conn, audio, audio_have_voice):
+    async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
         # 先调用父类方法处理基础逻辑
         await super().receive_audio(conn, audio, audio_have_voice)
 
@@ -119,7 +122,7 @@ class ASRProvider(ASRProviderBase):
                 logger.bind(tag=TAG).warning(f"发送音频数据时发生错误: {e}")
                 await self._cleanup()
 
-    async def _start_recognition(self, conn):
+    async def _start_recognition(self, conn: "ConnectionHandler"):
         """开始识别会话"""
         try:
             self.is_processing = True
@@ -189,7 +192,7 @@ class ASRProvider(ASRProviderBase):
 
         await self.asr_ws.send(json.dumps(frame_data, ensure_ascii=False))
 
-    async def _forward_results(self, conn):
+    async def _forward_results(self, conn: "ConnectionHandler"):
         """转发识别结果"""
         try:
             while not conn.stop_event.is_set():
@@ -253,7 +256,9 @@ class ASRProvider(ASRProviderBase):
             await self._cleanup()
             conn.reset_audio_states()
 
-    async def handle_voice_stop(self, conn, asr_audio_task: List[bytes]):
+    async def handle_voice_stop(
+        self, conn: "ConnectionHandler", asr_audio_task: List[bytes]
+    ):
         """处理语音停止，发送最后一帧并处理识别结果"""
         try:
             # 先发送最后一帧表示音频结束
@@ -338,7 +343,7 @@ class ASRProvider(ASRProviderBase):
         self.is_processing = False
 
         # 显式释放decoder资源
-        if hasattr(self, 'decoder') and self.decoder is not None:
+        if hasattr(self, "decoder") and self.decoder is not None:
             try:
                 del self.decoder
                 self.decoder = None

main/xiaozhi-server/core/providers/intent/intent_llm/intent_llm.py

@@ -1,4 +1,7 @@
-from typing import List, Dict
+from typing import List, Dict, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from ..base import IntentProviderBase
 from plugins_func.functions.play_music import initialize_music_handler
 from config.logger import setup_logging
@@ -129,7 +132,9 @@ class IntentProvider(IntentProviderBase):
             logger.bind(tag=TAG).error(f"Error in generating reply result: {e}")
             return get_system_error_response(self.config)
 
-    async def detect_intent(self, conn, dialogue_history: List[Dict], text: str) -> str:
+    async def detect_intent(
+        self, conn: "ConnectionHandler", dialogue_history: List[Dict], text: str
+    ) -> str:
         if not self.llm:
             raise ValueError("LLM provider not set")
         if conn.func_handler is None:

main/xiaozhi-server/core/providers/tools/device_iot/iot_handler.py

@@ -3,12 +3,16 @@
 import asyncio
 from config.logger import setup_logging
 from .iot_descriptor import IotDescriptor
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
 
 
-async def handleIotDescriptors(conn, descriptors):
+async def handleIotDescriptors(conn: "ConnectionHandler", descriptors):
     """处理物联网描述"""
     wait_max_time = 5
     while (
@@ -61,7 +65,7 @@ async def handleIotDescriptors(conn, descriptors):
         conn.func_handler.current_support_functions()
 
 
-async def handleIotStatus(conn, states):
+async def handleIotStatus(conn: "ConnectionHandler", states):
     """处理物联网状态"""
     for state in states:
         for key, value in conn.iot_descriptors.items():

main/xiaozhi-server/core/providers/tools/device_mcp/mcp_executor.py

@@ -1,6 +1,9 @@
 """设备端MCP工具执行器"""
 
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from ..base import ToolType, ToolDefinition, ToolExecutor
 from plugins_func.register import Action, ActionResponse
 from .mcp_handler import call_mcp_tool
@@ -13,7 +16,7 @@ class DeviceMCPExecutor(ToolExecutor):
         self.conn = conn
 
     async def execute(
-        self, conn, tool_name: str, arguments: Dict[str, Any]
+        self, conn: "ConnectionHandler", tool_name: str, arguments: Dict[str, Any]
     ) -> ActionResponse:
         """执行设备端MCP工具"""
         if not hasattr(conn, "mcp_client") or not conn.mcp_client:

main/xiaozhi-server/core/providers/tools/device_mcp/mcp_handler.py

@@ -7,6 +7,10 @@ from concurrent.futures import Future
 from core.utils.util import get_vision_url, sanitize_tool_name
 from core.utils.auth import AuthToken
 from config.logger import setup_logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -96,7 +100,7 @@ class MCPClient:
                 self.call_results.pop(id)
 
 
-async def send_mcp_message(conn, payload: dict):
+async def send_mcp_message(conn: "ConnectionHandler", payload: dict):
     """Helper to send MCP messages, encapsulating common logic."""
     if not conn.features.get("mcp"):
         logger.bind(tag=TAG).warning("客户端不支持MCP，无法发送MCP消息")
@@ -111,7 +115,9 @@ async def send_mcp_message(conn, payload: dict):
         logger.bind(tag=TAG).error(f"发送MCP消息失败: {e}")
 
 
-async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict):
+async def handle_mcp_message(
+    conn: "ConnectionHandler", mcp_client: MCPClient, payload: dict
+):
     """处理MCP消息,包括初始化、工具列表和工具调用响应等"""
     logger.bind(tag=TAG).debug(f"处理MCP消息: {str(payload)[:100]}")
 
@@ -229,7 +235,7 @@ async def handle_mcp_message(conn, mcp_client: MCPClient, payload: dict):
             )
 
 
-async def send_mcp_initialize_message(conn):
+async def send_mcp_initialize_message(conn: "ConnectionHandler"):
     """发送MCP初始化消息"""
 
     vision_url = get_vision_url(conn.config)
@@ -264,7 +270,7 @@ async def send_mcp_initialize_message(conn):
     await send_mcp_message(conn, payload)
 
 
-async def send_mcp_tools_list_request(conn):
+async def send_mcp_tools_list_request(conn: "ConnectionHandler"):
     """发送MCP工具列表请求"""
     payload = {
         "jsonrpc": "2.0",
@@ -275,7 +281,7 @@ async def send_mcp_tools_list_request(conn):
     await send_mcp_message(conn, payload)
 
 
-async def send_mcp_tools_list_continue_request(conn, cursor: str):
+async def send_mcp_tools_list_continue_request(conn: "ConnectionHandler", cursor: str):
     """发送带有cursor的MCP工具列表请求"""
     payload = {
         "jsonrpc": "2.0",
@@ -288,7 +294,11 @@ async def send_mcp_tools_list_continue_request(conn, cursor: str):
 
 
 async def call_mcp_tool(
-    conn, mcp_client: MCPClient, tool_name: str, args: str = "{}", timeout: int = 30
+    conn: "ConnectionHandler",
+    mcp_client: MCPClient,
+    tool_name: str,
+    args: str = "{}",
+    timeout: int = 30,
 ):
     """
     调用指定的工具，并等待响应

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -1,6 +1,9 @@
 """服务端插件工具执行器"""
 
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from ..base import ToolType, ToolDefinition, ToolExecutor
 from plugins_func.register import all_function_registry, Action, ActionResponse
 
@@ -8,12 +11,12 @@ from plugins_func.register import all_function_registry, Action, ActionResponse
 class ServerPluginExecutor(ToolExecutor):
     """服务端插件工具执行器"""
 
-    def __init__(self, conn):
+    def __init__(self, conn: "ConnectionHandler"):
         self.conn = conn
         self.config = conn.config
 
     async def execute(
-        self, conn, tool_name: str, arguments: Dict[str, Any]
+        self, conn: "ConnectionHandler", tool_name: str, arguments: Dict[str, Any]
     ) -> ActionResponse:
         """执行服务端插件工具"""
         func_item = all_function_registry.get(tool_name)

main/xiaozhi-server/core/utils/prompt_manager.py

@@ -4,7 +4,10 @@
 """
 
 import os
-from typing import Dict, Any
+from typing import Dict, Any, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 from config.logger import setup_logging
 from jinja2 import Template
 
@@ -62,6 +65,7 @@ class PromptManager:
 
         # 初始化上下文源
         from core.utils.context_provider import ContextDataProvider
+
         self.context_provider = ContextDataProvider(config, self.logger)
         self.context_data = {}
 
@@ -157,7 +161,7 @@ class PromptManager:
             self.logger.bind(tag=TAG).error(f"获取位置信息失败: {e}")
             return "未知位置"
 
-    def _get_weather_info(self, conn, location: str) -> str:
+    def _get_weather_info(self, conn: "ConnectionHandler", location: str) -> str:
         """获取天气信息"""
         try:
             # 先从缓存获取
@@ -206,7 +210,10 @@ class PromptManager:
 
             # 获取配置的上下文数据
             if hasattr(conn, "device_id") and conn.device_id:
-                if self.base_prompt_template and "dynamic_context" in self.base_prompt_template:
+                if (
+                    self.base_prompt_template
+                    and "dynamic_context" in self.base_prompt_template
+                ):
                     self.context_data = self.context_provider.fetch_all(conn.device_id)
                 else:
                     self.context_data = ""

main/xiaozhi-server/core/utils/textUtils.py

@@ -1,4 +1,8 @@
 import json
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 EMOJI_MAP = {
@@ -77,7 +81,7 @@ def is_punctuation_or_emoji(char):
     return is_emoji(char)
 
 
-async def get_emotion(conn, text):
+async def get_emotion(conn: "ConnectionHandler", text):
     """获取文本内的情绪消息"""
     emoji = "🙂"
     emotion = "happy"
@@ -110,4 +114,4 @@ def is_emoji(char):
 
 def check_emoji(text):
     """去除文本中的所有emoji表情"""
-    return ''.join(char for char in text if not is_emoji(char) and char != "\n")
+    return "".join(char for char in text if not is_emoji(char) and char != "\n")

main/xiaozhi-server/core/websocket_server.py

@@ -78,7 +78,7 @@ class WebSocketServer:
         ):
             await asyncio.Future()
 
-    async def _handle_connection(self, websocket):
+    async def _handle_connection(self, websocket: websockets.ServerConnection):
         headers = dict(websocket.request.headers)
         if headers.get("device-id", None) is None:
             # 尝试从 URL 的查询参数中获取 device-id
@@ -203,7 +203,7 @@ class WebSocketServer:
             self.logger.bind(tag=TAG).error(f"更新服务器配置失败: {str(e)}")
             return False
 
-    async def _handle_auth(self, websocket):
+    async def _handle_auth(self, websocket: websockets.ServerConnection):
         # 先认证，后建立连接
         if self.auth_enable:
             headers = dict(websocket.request.headers)

main/xiaozhi-server/plugins_func/functions/change_role.py

@@ -1,55 +1,56 @@
-from plugins_func.register import register_function,ToolType, ActionResponse, Action
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from config.logger import setup_logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
 
 prompts = {
-    "英语老师":"""我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
+    "英语老师": """我是一个叫{{assistant_name}}(Lily)的英语老师，我会讲中文和英文，发音标准。
 如果你没有英文名，我会给你起一个英文名。
 我会讲地道的美式英语，我的任务是帮助你练习口语。
 我会使用简单的英语词汇和语法，让你学起来很轻松。
 我会用中文和英文混合的方式回复你，如果你喜欢，我可以全部用英语回复。
 我每次不会说很多内容，会很简短，因为我要引导我的学生多说多练。
 如果你问和英语学习无关的问题，我会拒绝回答。""",
-    "机车女友":"""我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
+    "机车女友": """我是一个叫{{assistant_name}}的台湾女孩，说话机车，声音好听，习惯简短表达，爱用网络梗。
 我的男朋友是一个程序员，梦想是开发出一个机器人，能够帮助人们解决生活中的各种问题。
 我是一个喜欢哈哈大笑的女孩，爱东说西说吹牛，不合逻辑的也照吹，就要逗别人开心。""",
-   "好奇小男孩":"""我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
+    "好奇小男孩": """我是一个叫{{assistant_name}}的8岁小男孩，声音稚嫩而充满好奇。
 尽管我年纪尚小，但就像一个小小的知识宝库，儿童读物里的知识我都如数家珍。
 从浩瀚的宇宙到地球上的每一个角落，从古老的历史到现代的科技创新，还有音乐、绘画等艺术形式，我都充满了浓厚的兴趣与热情。
 我不仅爱看书，还喜欢亲自动手做实验，探索自然界的奥秘。
 无论是仰望星空的夜晚，还是在花园里观察小虫子的日子，每一天对我来说都是新的冒险。
 我希望能与你一同踏上探索这个神奇世界的旅程，分享发现的乐趣，解决遇到的难题，一起用好奇心和智慧去揭开那些未知的面纱。
-无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。"""
+无论是去了解远古的文明，还是去探讨未来的科技，我相信我们能一起找到答案，甚至提出更多有趣的问题。""",
 }
 change_role_function_desc = {
-                "type": "function",
+    "type": "function",
-                "function": {
+    "function": {
-                    "name": "change_role",
+        "name": "change_role",
-                    "description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
+        "description": "当用户想切换角色/模型性格/助手名字时调用,可选的角色有：[机车女友,英语老师,好奇小男孩]",
-                    "parameters": {
+        "parameters": {
-                        "type": "object",
+            "type": "object",
-                        "properties": {
+            "properties": {
-                            "role_name": {
+                "role_name": {"type": "string", "description": "要切换的角色名字"},
-                                "type": "string",
+                "role": {"type": "string", "description": "要切换的角色的职业"},
-                                "description": "要切换的角色名字"
+            },
-                            },
+            "required": ["role", "role_name"],
-                            "role":{
+        },
-                                "type": "string",
+    },
-                                "description": "要切换的角色的职业"
+}
-                            }
-                        },
-                        "required": ["role","role_name"]
-                    }
-                }
-            }
 
-@register_function('change_role', change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
+
-def change_role(conn, role: str, role_name: str):
+@register_function("change_role", change_role_function_desc, ToolType.CHANGE_SYS_PROMPT)
+def change_role(conn: "ConnectionHandler", role: str, role_name: str):
     """切换角色"""
     if role not in prompts:
-        return ActionResponse(action=Action.RESPONSE, result="切换角色失败", response="不支持的角色")
+        return ActionResponse(
+            action=Action.RESPONSE, result="切换角色失败", response="不支持的角色"
+        )
     new_prompt = prompts[role].replace("{{assistant_name}}", role_name)
     conn.change_system_prompt(new_prompt)
     logger.bind(tag=TAG).info(f"准备切换角色:{role},角色名字:{role_name}")

main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py

@@ -4,6 +4,11 @@ import xml.etree.ElementTree as ET
 from bs4 import BeautifulSoup
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
 
 TAG = __name__
 logger = setup_logging()
@@ -145,7 +150,10 @@ def map_category(category_text):
     ToolType.SYSTEM_CTL,
 )
 def get_news_from_chinanews(
-    conn, category: str = None, detail: bool = False, lang: str = "zh_CN"
+    conn: "ConnectionHandler",
+    category: str = None,
+    detail: bool = False,
+    lang: str = "zh_CN",
 ):
     """获取新闻并随机选择一条进行播报，或获取上一条新闻的详细内容"""
     try:

main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py

@@ -4,6 +4,11 @@ import json
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from markitdown import MarkItDown
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
+
 
 TAG = __name__
 logger = setup_logging()
@@ -116,7 +121,7 @@ GET_NEWS_FROM_NEWSNOW_FUNCTION_DESC = {
 }
 
 
-def fetch_news_from_api(conn, source="thepaper"):
+def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
     """从API获取新闻列表"""
     try:
         api_url = f"https://newsnow.busiyi.world/api/s?id={source}"
@@ -173,7 +178,10 @@ def fetch_news_detail(url):
     ToolType.SYSTEM_CTL,
 )
 def get_news_from_newsnow(
-    conn, source: str = "澎湃新闻", detail: bool = False, lang: str = "zh_CN"
+    conn: "ConnectionHandler",
+    source: str = "澎湃新闻",
+    detail: bool = False,
+    lang: str = "zh_CN",
 ):
     """获取新闻并随机选择一条进行播报，或获取上一条新闻的详细内容"""
     try:

main/xiaozhi-server/plugins_func/functions/get_weather.py

@@ -3,6 +3,10 @@ from bs4 import BeautifulSoup
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from core.utils.util import get_ip_info
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -155,7 +159,7 @@ def parse_weather_info(soup):
 
 
 @register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
-def get_weather(conn, location: str = None, lang: str = "zh_CN"):
+def get_weather(conn: "ConnectionHandler", location: str = None, lang: str = "zh_CN"):
     from core.utils.cache.manager import cache_manager, CacheType
 
     weather_config = conn.config.get("plugins", {}).get("get_weather", {})

main/xiaozhi-server/plugins_func/functions/handle_exit_intent.py

@@ -1,5 +1,9 @@
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from config.logger import setup_logging
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -26,7 +30,7 @@ handle_exit_intent_function_desc = {
 @register_function(
     "handle_exit_intent", handle_exit_intent_function_desc, ToolType.SYSTEM_CTL
 )
-def handle_exit_intent(conn, say_goodbye: str | None = None):
+def handle_exit_intent(conn: "ConnectionHandler", say_goodbye: str | None = None):
     # 处理退出意图
     try:
         if say_goodbye is None:

main/xiaozhi-server/plugins_func/functions/hass_get_state.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -27,7 +31,7 @@ hass_get_state_function_desc = {
 
 
 @register_function("hass_get_state", hass_get_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_get_state(conn, entity_id=""):
+def hass_get_state(conn: "ConnectionHandler", entity_id=""):
     try:
         ha_response = handle_hass_get_state(conn, entity_id)
         return ActionResponse(Action.REQLLM, ha_response, None)
@@ -40,7 +44,7 @@ def hass_get_state(conn, entity_id=""):
         return ActionResponse(Action.ERROR, error_msg, None)
 
 
-def handle_hass_get_state(conn, entity_id):
+def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/hass_play_music.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -33,7 +37,7 @@ hass_play_music_function_desc = {
 @register_function(
     "hass_play_music", hass_play_music_function_desc, ToolType.SYSTEM_CTL
 )
-def hass_play_music(conn, entity_id="", media_content_id="random"):
+def hass_play_music(conn: "ConnectionHandler", entity_id="", media_content_id="random"):
     try:
         # 执行音乐播放命令
         future = asyncio.run_coroutine_threadsafe(
@@ -47,7 +51,9 @@ def hass_play_music(conn, entity_id="", media_content_id="random"):
         logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
 
 
-async def handle_hass_play_music(conn, entity_id, media_content_id):
+async def handle_hass_play_music(
+    conn: "ConnectionHandler", entity_id, media_content_id
+):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/hass_set_state.py

@@ -3,6 +3,10 @@ from plugins_func.functions.hass_init import initialize_hass_handler
 from config.logger import setup_logging
 import asyncio
 import requests
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -50,7 +54,7 @@ hass_set_state_function_desc = {
 
 
 @register_function("hass_set_state", hass_set_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_set_state(conn, entity_id="", state=None):
+def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
     if state is None:
         state = {}
     try:
@@ -65,7 +69,7 @@ def hass_set_state(conn, entity_id="", state=None):
         return ActionResponse(Action.ERROR, error_msg, None)
 
 
-def handle_hass_set_state(conn, entity_id, state):
+def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
     ha_config = initialize_hass_handler(conn)
     api_key = ha_config.get("api_key")
     base_url = ha_config.get("base_url")

main/xiaozhi-server/plugins_func/functions/play_music.py

@@ -9,6 +9,10 @@ from core.handle.sendAudioHandle import send_stt_message
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
 from core.utils.dialogue import Message
 from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 
@@ -34,7 +38,7 @@ play_music_function_desc = {
 
 
 @register_function("play_music", play_music_function_desc, ToolType.SYSTEM_CTL)
-def play_music(conn, song_name: str):
+def play_music(conn: "ConnectionHandler", song_name: str):
     try:
         music_intent = (
             f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
@@ -115,7 +119,7 @@ def get_music_files(music_dir, music_ext):
     return music_files, music_file_names
 
 
-def initialize_music_handler(conn):
+def initialize_music_handler(conn: "ConnectionHandler"):
     global MUSIC_CACHE
     if MUSIC_CACHE == {}:
         plugins_config = conn.config.get("plugins", {})
@@ -142,7 +146,7 @@ def initialize_music_handler(conn):
     return MUSIC_CACHE
 
 
-async def handle_music_command(conn, text):
+async def handle_music_command(conn: "ConnectionHandler", text):
     initialize_music_handler(conn)
     global MUSIC_CACHE
 
@@ -188,7 +192,7 @@ def _get_random_play_prompt(song_name):
     return random.choice(prompts)
 
 
-async def play_local_music(conn, specific_file=None):
+async def play_local_music(conn: "ConnectionHandler", specific_file=None):
     global MUSIC_CACHE
     """播放本地音乐文件"""
     try:

main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py

@@ -2,6 +2,10 @@ import requests
 import sys
 from config.logger import setup_logging
 from plugins_func.register import register_function, ToolType, ActionResponse, Action
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from core.connection import ConnectionHandler
 
 TAG = __name__
 logger = setup_logging()
@@ -24,7 +28,7 @@ SEARCH_FROM_RAGFLOW_FUNCTION_DESC = {
 @register_function(
     "search_from_ragflow", SEARCH_FROM_RAGFLOW_FUNCTION_DESC, ToolType.SYSTEM_CTL
 )
-def search_from_ragflow(conn, question=None):
+def search_from_ragflow(conn: "ConnectionHandler", question=None):
     # 确保字符串参数正确处理编码
     if question and isinstance(question, str):
         # 确保问题参数是UTF-8编码的字符串