Merge branch 'refs/heads/main' into perf-tool-call-optimization

main/xiaozhi-server/core/connection.py

@@ -116,8 +116,6 @@ class ConnectionHandler:
         self.memory = _memory
         self.intent = _intent
 
-        self.is_exiting = False  # 标记是否正在执行退出流程
-
         # 为每个连接单独管理声纹识别
         self.voiceprint_provider = None
 
@@ -312,10 +310,6 @@ class ConnectionHandler:
 
     async def _route_message(self, message):
         """消息路由"""
-        # 退出状态丢弃所有消息
-        if self.is_exiting:
-           return
-
         # 检查是否已经获取到真实的绑定状态
         if not self.bind_completed_event.is_set():
             # 还没有获取到真实状态，等待直到获取到真实状态或超时

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

@@ -7,12 +7,9 @@ TAG = __name__
 
 
 async def handleAbortMessage(conn: "ConnectionHandler"):
-    if conn.close_after_chat or conn.is_exiting:
-        conn.logger.bind(tag=TAG).info("退出流程中被打断，直接关闭连接")
-        return
-        
     conn.logger.bind(tag=TAG).info("Abort message received")
     # 设置成打断状态，会自动打断llm、tts任务
+    conn.close_after_chat = False
     conn.client_abort = True
     conn.clear_queues()
     # 打断客户端说话状态

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

@@ -33,10 +33,6 @@ async def handle_user_intent(conn: "ConnectionHandler", text):
     if await check_direct_exit(conn, filtered_text):
         return True
 
-    # 明确再见不被打断
-    if conn.is_exiting:
-        return True
-
     # 检查是否是唤醒词
     if await checkWakeupWords(conn, filtered_text):
         return True
@@ -62,7 +58,6 @@ async def check_direct_exit(conn: "ConnectionHandler", text):
         if text == cmd:
             conn.logger.bind(tag=TAG).info(f"识别到明确的退出命令: {text}")
             await send_stt_message(conn, text)
-            conn.is_exiting = True
             await conn.close()
             return True
     return False

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

@@ -15,8 +15,6 @@ TAG = __name__
 
 
 async def handleAudioMessage(conn: "ConnectionHandler", audio):
-    if conn.is_exiting:
-        return
     # 当前片段是否有人说话
     have_voice = conn.vad.is_vad(conn, audio)
     # 如果设备刚刚被唤醒，短暂忽略VAD检测

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

@@ -84,12 +84,6 @@ class ASRProviderBase(ABC):
     async def handle_voice_stop(self, conn: "ConnectionHandler", asr_audio_task: List[bytes]):
         """并行处理ASR和声纹识别"""
         try:
-            # 如果处于退出流程中，直接关闭连接，不处理新消息
-            if conn.close_after_chat or conn.is_exiting:
-                logger.bind(tag=TAG).info("退出流程中收到新消息，直接关闭连接")
-                await conn.close()
-                return
-
             total_start_time = time.monotonic()
 
             # 准备音频数据

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

@@ -26,11 +26,14 @@ class ASRProvider(ASRProviderBase):
         self.asr_ws = None
         self.forward_task = None
         self.is_processing = False  # 添加处理状态标志
+        self._is_stopping = False  # 添加停止标志，防止竞态条件
 
         # 配置参数
         self.appid = str(config.get("appid"))
-        self.cluster = config.get("cluster")
         self.access_token = config.get("access_token")
+        # 资源ID，用于区分不同的ASR模型（默认1.0模型小时版，v2版本使用seed-asr）
+        self.resource_id = config.get("resource_id", "volc.bigasr.sauc.duration")
+
         self.boosting_table_name = config.get("boosting_table_name", "")
         self.correct_table_name = config.get("correct_table_name", "")
         self.output_dir = config.get("output_dir", "tmp/")
@@ -44,7 +47,7 @@ class ASRProvider(ASRProviderBase):
         if self.enable_multilingual:
             self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream"
         else:
-            self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel"
+            self.ws_url = "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_async"
         self.uid = config.get("uid", "streaming_asr_service")
         self.workflow = config.get(
             "workflow", "audio_in,resample,partition,vad,fe,decode,itn,nlu_punctuate"
@@ -146,7 +149,7 @@ class ASRProvider(ASRProviderBase):
                 return
 
         # 发送当前音频数据
-        if self.asr_ws and self.is_processing:
+        if self.asr_ws and self.is_processing and not self._is_stopping:
             try:
                 pcm_frame = self.decoder.decode(audio, 960)
                 payload = gzip.compress(pcm_frame)
@@ -254,6 +257,7 @@ class ASRProvider(ASRProviderBase):
                 await self.asr_ws.close()
                 self.asr_ws = None
             self.is_processing = False
+            self._is_stopping = False
             # 重置所有音频相关状态
             conn.reset_audio_states()
 
@@ -262,9 +266,11 @@ class ASRProvider(ASRProviderBase):
             asyncio.create_task(self.asr_ws.close())
             self.asr_ws = None
         self.is_processing = False
+        self._is_stopping = False
 
     async def _send_stop_request(self):
         """发送最后一个音频帧以通知服务器结束"""
+        self._is_stopping = True  # 先标记为停止状态，阻止后续音频发送
         if self.asr_ws:
             try:
                 # 发送结束标记的音频帧（gzip压缩的空数据）
@@ -283,7 +289,6 @@ class ASRProvider(ASRProviderBase):
         req = {
             "app": {
                 "appid": self.appid,
-                "cluster": self.cluster,
                 "token": self.access_token,
             },
             "user": {"uid": self.uid},
@@ -322,7 +327,7 @@ class ASRProvider(ASRProviderBase):
         return {
             "X-Api-App-Key": self.appid,
             "X-Api-Access-Key": self.access_token,
-            "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
+            "X-Api-Resource-Id": self.resource_id,
             "X-Api-Connect-Id": str(uuid.uuid4()),
         }
 
@@ -385,9 +390,17 @@ class ASRProvider(ASRProviderBase):
                     "payload_msg": error_msg,
                 }
 
-            # 获取JSON数据（跳过12字节头部）
+            # 获取JSON数据
             try:
-                json_data = res[12:].decode("utf-8")
+                # 检查字节8-11是否为有效的JSON长度字段
+                # 格式：4字节头 + 4字节序列号 + 4字节长度 + JSON数据
+                length = int.from_bytes(res[8:12], "big")
+                if length > 0 and length <= len(res) - 12:
+                    # 有长度字段，从字节12开始读取指定长度的JSON
+                    json_data = res[12:12 + length].decode("utf-8")
+                else:
+                    # 无长度字段或长度无效，尝试直接解析
+                    json_data = res[8:].decode("utf-8")
                 result = json.loads(json_data)
                 logger.bind(tag=TAG).debug(f"成功解析JSON响应: {result}")
                 return {"payload_msg": result}

main/xiaozhi-server/core/providers/llm/homeassistant/homeassistant.py

@@ -39,13 +39,12 @@ class LLMProvider(LLMProviderBase):
         }
 
         # 发起 POST 请求
-        response = requests.post(self.api_url, json=payload, headers=headers)
+        with requests.post(self.api_url, json=payload, headers=headers) as response:
+            # 检查请求是否成功
+            response.raise_for_status()
 
-        # 检查请求是否成功
-        response.raise_for_status()
-
-        # 解析返回数据
-        data = response.json()
+            # 解析返回数据
+            data = response.json()
         speech = (
             data.get("response", {})
             .get("speech", {})

main/xiaozhi-server/core/providers/llm/ollama/ollama.py

@@ -50,43 +50,46 @@ class LLMProvider(LLMProviderBase):
         # 用于处理跨chunk的标签
         buffer = ""
 
-        for chunk in responses:
-            try:
-                delta = (
-                    chunk.choices[0].delta
-                    if getattr(chunk, "choices", None)
-                    else None
-                )
-                content = delta.content if hasattr(delta, "content") else ""
+        try:
+            for chunk in responses:
+                try:
+                    delta = (
+                        chunk.choices[0].delta
+                        if getattr(chunk, "choices", None)
+                        else None
+                    )
+                    content = delta.content if hasattr(delta, "content") else ""
 
-                if content:
-                    # 将内容添加到缓冲区
-                    buffer += content
+                    if content:
+                        # 将内容添加到缓冲区
+                        buffer += content
 
-                    # 处理缓冲区中的标签
-                    while "<think>" in buffer and "</think>" in buffer:
-                        # 找到完整的<think></think>标签并移除
-                        pre = buffer.split("<think>", 1)[0]
-                        post = buffer.split("</think>", 1)[1]
-                        buffer = pre + post
+                        # 处理缓冲区中的标签
+                        while "<think>" in buffer and "</think>" in buffer:
+                            # 找到完整的<think></think>标签并移除
+                            pre = buffer.split("<think>", 1)[0]
+                            post = buffer.split("</think>", 1)[1]
+                            buffer = pre + post
 
-                    # 处理只有开始标签的情况
-                    if "<think>" in buffer:
-                        is_active = False
-                        buffer = buffer.split("<think>", 1)[0]
+                        # 处理只有开始标签的情况
+                        if "<think>" in buffer:
+                            is_active = False
+                            buffer = buffer.split("<think>", 1)[0]
 
-                    # 处理只有结束标签的情况
-                    if "</think>" in buffer:
-                        is_active = True
-                        buffer = buffer.split("</think>", 1)[1]
+                        # 处理只有结束标签的情况
+                        if "</think>" in buffer:
+                            is_active = True
+                            buffer = buffer.split("</think>", 1)[1]
 
-                    # 如果当前处于活动状态且缓冲区有内容，则输出
-                    if is_active and buffer:
-                        yield buffer
-                        buffer = ""  # 清空缓冲区
+                        # 如果当前处于活动状态且缓冲区有内容，则输出
+                        if is_active and buffer:
+                            yield buffer
+                            buffer = ""  # 清空缓冲区
 
-            except Exception as e:
-                logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"Error processing chunk: {e}")
+        finally:
+            responses.close()
 
     def response_with_functions(self, session_id, dialogue, functions=None):
         # 如果是qwen3模型，在用户最后一条消息中添加/no_think指令
@@ -117,49 +120,52 @@ class LLMProvider(LLMProviderBase):
         is_active = True
         buffer = ""
 
-        for chunk in stream:
-            try:
-                delta = (
-                    chunk.choices[0].delta
-                    if getattr(chunk, "choices", None)
-                    else None
-                )
-                content = delta.content if hasattr(delta, "content") else None
-                tool_calls = (
-                    delta.tool_calls if hasattr(delta, "tool_calls") else None
-                )
+        try:
+            for chunk in stream:
+                try:
+                    delta = (
+                        chunk.choices[0].delta
+                        if getattr(chunk, "choices", None)
+                        else None
+                    )
+                    content = delta.content if hasattr(delta, "content") else None
+                    tool_calls = (
+                        delta.tool_calls if hasattr(delta, "tool_calls") else None
+                    )
 
-                # 如果是工具调用，直接传递
-                if tool_calls:
-                    yield None, tool_calls
+                    # 如果是工具调用，直接传递
+                    if tool_calls:
+                        yield None, tool_calls
+                        continue
+
+                    # 处理文本内容
+                    if content:
+                        # 将内容添加到缓冲区
+                        buffer += content
+
+                        # 处理缓冲区中的标签
+                        while "<think>" in buffer and "</think>" in buffer:
+                            # 找到完整的<think></think>标签并移除
+                            pre = buffer.split("<think>", 1)[0]
+                            post = buffer.split("</think>", 1)[1]
+                            buffer = pre + post
+
+                        # 处理只有开始标签的情况
+                        if "<think>" in buffer:
+                            is_active = False
+                            buffer = buffer.split("<think>", 1)[0]
+
+                        # 处理只有结束标签的情况
+                        if "</think>" in buffer:
+                            is_active = True
+                            buffer = buffer.split("</think>", 1)[1]
+
+                        # 如果当前处于活动状态且缓冲区有内容，则输出
+                        if is_active and buffer:
+                            yield buffer, None
+                            buffer = ""  # 清空缓冲区
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
                     continue
-
-                # 处理文本内容
-                if content:
-                    # 将内容添加到缓冲区
-                    buffer += content
-
-                    # 处理缓冲区中的标签
-                    while "<think>" in buffer and "</think>" in buffer:
-                        # 找到完整的<think></think>标签并移除
-                        pre = buffer.split("<think>", 1)[0]
-                        post = buffer.split("</think>", 1)[1]
-                        buffer = pre + post
-
-                    # 处理只有开始标签的情况
-                    if "<think>" in buffer:
-                        is_active = False
-                        buffer = buffer.split("<think>", 1)[0]
-
-                    # 处理只有结束标签的情况
-                    if "</think>" in buffer:
-                        is_active = True
-                        buffer = buffer.split("</think>", 1)[1]
-
-                    # 如果当前处于活动状态且缓冲区有内容，则输出
-                    if is_active and buffer:
-                        yield buffer, None
-                        buffer = ""  # 清空缓冲区
-            except Exception as e:
-                logger.bind(tag=TAG).error(f"Error processing function chunk: {e}")
-                continue
+        finally:
+            stream.close()

main/xiaozhi-server/core/providers/llm/openai/openai.py

@@ -115,21 +115,24 @@ class LLMProvider(LLMProviderBase):
         responses = self.client.chat.completions.create(**request_params)
 
         is_active = True
-        for chunk in responses:
-            try:
-                delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None
-                content = getattr(delta, "content", "") if delta else ""
-            except IndexError:
-                content = ""
-            if content:
-                if "<think>" in content:
-                    is_active = False
-                    content = content.split("<think>")[0]
-                if "</think>" in content:
-                    is_active = True
-                    content = content.split("</think>")[-1]
-                if is_active:
-                    yield content
+        try:            
+            for chunk in responses:
+                try:
+                    delta = chunk.choices[0].delta if getattr(chunk, "choices", None) else None
+                    content = getattr(delta, "content", "") if delta else ""
+                except IndexError:
+                    content = ""
+                if content:
+                    if "<think>" in content:
+                        is_active = False
+                        content = content.split("<think>")[0]
+                    if "</think>" in content:
+                        is_active = True
+                        content = content.split("</think>")[-1]
+                    if is_active:
+                        yield content
+        finally:
+            responses.close()
 
     def response_with_functions(self, session_id, dialogue, functions=None, **kwargs):
         dialogue = self.normalize_dialogue(dialogue)
@@ -157,16 +160,19 @@ class LLMProvider(LLMProviderBase):
 
         stream = self.client.chat.completions.create(**request_params)
 
-        for chunk in stream:
-            if getattr(chunk, "choices", None):
-                delta = chunk.choices[0].delta
-                content = getattr(delta, "content", "")
-                tool_calls = getattr(delta, "tool_calls", None)
-                yield content, tool_calls
-            elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
-                usage_info = getattr(chunk, "usage", None)
-                logger.bind(tag=TAG).info(
-                    f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
-                    f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
-                    f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
-                )
+        try:
+            for chunk in stream:
+                if getattr(chunk, "choices", None):
+                    delta = chunk.choices[0].delta
+                    content = getattr(delta, "content", "")
+                    tool_calls = getattr(delta, "tool_calls", None)
+                    yield content, tool_calls
+                elif isinstance(getattr(chunk, "usage", None), CompletionUsage):
+                    usage_info = getattr(chunk, "usage", None)
+                    logger.bind(tag=TAG).info(
+                        f"Token 消耗：输入 {getattr(usage_info, 'prompt_tokens', '未知')}，"
+                        f"输出 {getattr(usage_info, 'completion_tokens', '未知')}，"
+                        f"共计 {getattr(usage_info, 'total_tokens', '未知')}"
+                    )
+        finally:
+            stream.close()

main/xiaozhi-server/core/providers/llm/xinference/xinference.py

@@ -74,12 +74,15 @@ class LLMProvider(LLMProviderBase):
             tools=functions,
         )
 
-        for chunk in stream:
-            delta = chunk.choices[0].delta
-            content = delta.content
-            tool_calls = delta.tool_calls
+        try:
+            for chunk in stream:
+                delta = chunk.choices[0].delta
+                content = delta.content
+                tool_calls = delta.tool_calls
 
-            if content:
-                yield content, tool_calls
-            elif tool_calls:
-                yield None, tool_calls
+                if content:
+                    yield content, tool_calls
+                elif tool_calls:
+                    yield None, tool_calls
+        finally:
+            stream.close()