Merge pull request #1223 from GOODDAYDAY/feature/muti_upload

feat: 增加上报线程池
2026-07-22 07:03:53 +08:00 · 2025-05-27 13:06:55 +08:00
parent 02cb9c35b3 258d783f8e
commit baf174b059
21 changed files with 311 additions and 276 deletions
@@ -142,10 +142,11 @@

 #### 🚀 部署方式选择

-| 部署方式 | 特点 | 适用场景 | Docker部署文档 | 源码部署文档 |
-|---------|------|---------|---------|---------|
-| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) | [本地源码只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)|
-| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) |
+| 部署方式 | 特点 | 适用场景 | Docker部署文档 | 源码部署文档 | 视频教程 | 
+|---------|------|---------|---------|---------|---------|
+| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) | [本地源码只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| - | 
+| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) | 
+

 > 💡 提示：以下是按最新代码部署后的测试平台，有需要可烧录测试，并发为6个，每天会清空数据

@@ -258,4 +258,4 @@
            </plugin>
        </plugins>
    </build>
-</project>
+</project>
@@ -28,4 +28,6 @@ public class AgentChatHistoryReportDTO {
    private String content;
    @Schema(description = "base64编码的opus音频数据", example = "")
    private String audioBase64;
+    @Schema(description = "上报时间，十位时间戳，空时默认使用当前时间", example = "1745657732")
+    private Long reportTime;
 }
@@ -27,4 +27,4 @@ public interface AgentChatAudioService extends IService<AgentChatAudioEntity> {
     * @return 音频数据
     */
    byte[] getAudio(String audioId);
-}
+}
@@ -47,7 +47,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
    public Boolean report(AgentChatHistoryReportDTO report) {
        String macAddress = report.getMacAddress();
        Byte chatType = report.getChatType();
-        log.info("小智设备聊天上报请求: macAddress={}, type={}", macAddress, chatType);
+        Long reportTimeMillis = null != report.getReportTime() ? report.getReportTime() * 1000 : System.currentTimeMillis();
+        log.info("小智设备聊天上报请求: macAddress={}, type={} reportTime={}", macAddress, chatType, reportTimeMillis);

        // 根据设备MAC地址查询对应的默认智能体，判断是否需要上报
        AgentEntity agentEntity = agentService.getDefaultAgentByMacAddress(macAddress);
@@ -59,10 +60,10 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
        String agentId = agentEntity.getId();

        if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT.getCode())) {
-            saveChatText(report, agentId, macAddress, null);
+            saveChatText(report, agentId, macAddress, null, reportTimeMillis);
        } else if (Objects.equals(chatHistoryConf, Constant.ChatHistoryConfEnum.RECORD_TEXT_AUDIO.getCode())) {
            String audioId = saveChatAudio(report);
-            saveChatText(report, agentId, macAddress, audioId);
+            saveChatText(report, agentId, macAddress, audioId, reportTimeMillis);
        }

        // 更新设备最后对话时间
@@ -92,8 +93,7 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
    /**
     * 组装上报数据
     */
-    private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId) {
-
+    private void saveChatText(AgentChatHistoryReportDTO report, String agentId, String macAddress, String audioId, Long reportTime) {
        // 构建聊天记录实体
        AgentChatHistoryEntity entity = AgentChatHistoryEntity.builder()
                .macAddress(macAddress)
@@ -102,6 +102,8 @@ public class AgentChatHistoryBizServiceImpl implements AgentChatHistoryBizServic
                .chatType(report.getChatType())
                .content(report.getContent())
                .audioId(audioId)
+                .createdAt(new Date(reportTime))
+                // NOTE(haotian): 2025/5/26 updateAt可以不设置，重点是createAt，而且这样可以看到上报延迟
                .build();

        // 保存数据
@@ -31,4 +31,4 @@ public class AgentChatAudioServiceImpl extends ServiceImpl<AiAgentChatAudioDao,
        AgentChatAudioEntity entity = getById(audioId);
        return entity != null ? entity.getAudio() : null;
    }
-}
+}
@@ -48,4 +48,4 @@ spring:
          max-active: 8      # 连接池最大连接数（使用负值表示没有限制）
          max-idle: 8        # 连接池中的最大空闲连接
          min-idle: 0        # 连接池中的最小空闲连接
-        shutdown-timeout: 100ms  # 客户端优雅关闭的等待时间
+        shutdown-timeout: 100ms  # 客户端优雅关闭的等待时间
@@ -51,7 +51,7 @@
              字典管理
            </el-dropdown-item>
            <el-dropdown-item @click.native="goProviderManagement">
-              供应器管理
+              字段管理
            </el-dropdown-item>
            <el-dropdown-item @click.native="goServerSideManagement">
              服务端管理
@@ -467,7 +467,7 @@ export default {
 .main-wrapper {
  margin: 5px 22px;
  border-radius: 15px;
-  min-height: calc(100vh - 24vh);
+  min-height: calc(100vh - 26vh);
  height: auto;
  max-height: 80vh;
  box-shadow: 0 2px 12px rgba(0, 0, 0, 0.1);
@@ -3,7 +3,7 @@
    <HeaderBar />

    <div class="operation-bar">
-      <h2 class="page-title">供应器管理</h2>
+      <h2 class="page-title">字段管理</h2>
      <div class="right-operations">
        <el-dropdown trigger="click" @command="handleSelectModelType" @visible-change="handleDropdownVisibleChange">
          <el-button class="category-btn">
@@ -160,7 +160,7 @@ def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:


 def report(
-    mac_address: str, session_id: str, chat_type: int, content: str, audio
+    mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time
 ) -> Optional[Dict]:
    """带熔断的业务方法示例"""
    if not content or not ManageApiClient._instance:
@@ -174,6 +174,7 @@ def report(
                "sessionId": session_id,
                "chatType": chat_type,
                "content": content,
+                "reportTime": report_time,
                "audioBase64": (
                    base64.b64encode(audio).decode("utf-8") if audio else None
                ),
@@ -14,6 +14,8 @@ import websockets
 from typing import Dict, Any
 from plugins_func.loadplugins import auto_import_modules
 from config.logger import setup_logging
+from config.config_loader import get_project_dir
+from core.utils import p3
 from core.utils.dialogue import Message, Dialogue
 from core.handle.textHandle import handleTextMessage
 from core.utils.util import (
@@ -89,7 +91,8 @@ class ConnectionHandler:
        self.audio_play_queue = queue.Queue()
        self.executor = ThreadPoolExecutor(max_workers=10)

-        # 上报线程
+        # 添加上报线程池
+        self.report_thread_pool = ThreadPoolExecutor(max_workers=5, thread_name_prefix="report_worker")
        self.report_queue = queue.Queue()
        self.report_thread = None
        # TODO(haotian): 2025/5/12 可以通过修改此处，调节asr的上报和tts的上报
@@ -534,106 +537,20 @@ class ConnectionHandler:
        # 更新系统prompt至上下文
        self.dialogue.update_system_message(self.prompt)

-    def chat(self, query):
-
-        self.dialogue.put(Message(role="user", content=query))
-
-        response_message = []
-        processed_chars = 0  # 跟踪已处理的字符位置
-        try:
-            # 使用带记忆的对话
-            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(
-                self.session_id, self.dialogue.get_llm_dialogue_with_memory(memory_str)
-            )
-        except Exception as e:
-            self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}")
-            return None
-
-        self.llm_finish_task = False
-        text_index = 0
-        for content in llm_responses:
-            response_message.append(content)
-            if self.client_abort:
-                break
-
-            # 合并当前全部文本并处理未分割部分
-            full_text = "".join(response_message)
-            current_text = full_text[processed_chars:]  # 从未处理的位置开始
-
-            # 查找最后一个有效标点
-            punctuations = ("。", ".", "？", "?", "！", "!", "；", ";", "：")
-            last_punct_pos = -1
-            number_flag = True
-            for punct in punctuations:
-                pos = current_text.rfind(punct)
-                prev_char = current_text[pos - 1] if pos - 1 >= 0 else ""
-                # 如果.前面是数字统一判断为小数
-                if prev_char.isdigit() and punct == ".":
-                    number_flag = False
-                if pos > last_punct_pos and number_flag:
-                    last_punct_pos = pos
-
-            # 找到分割点则处理
-            if last_punct_pos != -1:
-                segment_text_raw = current_text[: last_punct_pos + 1]
-                segment_text = get_string_no_punctuation_or_emoji(segment_text_raw)
-                if segment_text:
-                    # 强制设置空字符，测试TTS出错返回语音的健壮性
-                    # if text_index % 2 == 0:
-                    #     segment_text = " "
-                    text_index += 1
-                    self.recode_first_last_text(segment_text, text_index)
-                    future = self.executor.submit(
-                        self.speak_and_play, segment_text, text_index
-                    )
-                    self.tts_queue.put((future, text_index))
-                    processed_chars += len(segment_text_raw)  # 更新已处理字符位置
-
-        # 处理最后剩余的文本
-        full_text = "".join(response_message)
-        remaining_text = full_text[processed_chars:]
-        if remaining_text:
-            segment_text = get_string_no_punctuation_or_emoji(remaining_text)
-            if segment_text:
-                text_index += 1
-                self.recode_first_last_text(segment_text, text_index)
-                future = self.executor.submit(
-                    self.speak_and_play, segment_text, text_index
-                )
-                self.tts_queue.put((future, text_index))
-
-        self.llm_finish_task = True
-        self.dialogue.put(Message(role="assistant", content="".join(response_message)))
-        self.logger.bind(tag=TAG).debug(
-            json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)
-        )
-        return True
-
-    def chat_with_function_calling(self, query, tool_call=False):
-        self.logger.bind(tag=TAG).debug(f"Chat with function calling start: {query}")
-        """Chat with function calling for intent detection using streaming"""
+    def chat(self, query, tool_call=False):
+        self.logger.bind(tag=TAG).debug(f"Chat: {query}")

        if not tool_call:
            self.dialogue.put(Message(role="user", content=query))

        # Define intent functions
        functions = None
-        if hasattr(self, "func_handler"):
+        if self.intent_type == "function_call" and hasattr(self, "func_handler"):
            functions = self.func_handler.get_functions()
        response_message = []
        processed_chars = 0  # 跟踪已处理的字符位置

        try:
-            start_time = time.time()
-
            # 使用带记忆的对话
            memory_str = None
            if self.memory is not None:
@@ -642,14 +559,18 @@ class ConnectionHandler:
                )
                memory_str = future.result()

-            # self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}")
-
-            # 使用支持functions的streaming接口
-            llm_responses = self.llm.response_with_functions(
-                self.session_id,
-                self.dialogue.get_llm_dialogue_with_memory(memory_str),
-                functions=functions,
-            )
+            if functions is not None:
+                # 使用支持functions的streaming接口
+                llm_responses = self.llm.response_with_functions(
+                    self.session_id,
+                    self.dialogue.get_llm_dialogue_with_memory(memory_str),
+                    functions=functions,
+                )
+            else:
+                llm_responses = self.llm.response(
+                    self.session_id,
+                    self.dialogue.get_llm_dialogue_with_memory(memory_str),
+                )
        except Exception as e:
            self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}")
            return None
@@ -665,27 +586,28 @@ class ConnectionHandler:
        content_arguments = ""

        for response in llm_responses:
-            content, tools_call = response
+            if self.intent_type == "function_call":
+                content, tools_call = response
+                if "content" in response:
+                    content = response["content"]
+                    tools_call = None
+                if content is not None and len(content) > 0:
+                    content_arguments += content

-            if "content" in response:
-                content = response["content"]
-                tools_call = None
-            if content is not None and len(content) > 0:
-                content_arguments += content
-
-            if not tool_call_flag and content_arguments.startswith("<tool_call>"):
-                # print("content_arguments", content_arguments)
-                tool_call_flag = True
-
-            if tools_call is not None:
-                tool_call_flag = True
-                if tools_call[0].id is not None:
-                    function_id = tools_call[0].id
-                if tools_call[0].function.name is not None:
-                    function_name = tools_call[0].function.name
-                if tools_call[0].function.arguments is not None:
-                    function_arguments += tools_call[0].function.arguments
+                if not tool_call_flag and content_arguments.startswith("<tool_call>"):
+                    # print("content_arguments", content_arguments)
+                    tool_call_flag = True

+                if tools_call is not None:
+                    tool_call_flag = True
+                    if tools_call[0].id is not None:
+                        function_id = tools_call[0].id
+                    if tools_call[0].function.name is not None:
+                        function_name = tools_call[0].function.name
+                    if tools_call[0].function.arguments is not None:
+                        function_arguments += tools_call[0].function.arguments
+            else:
+                content = response
            if content is not None and len(content) > 0:
                if not tool_call_flag:
                    response_message.append(content)
@@ -724,7 +646,7 @@ class ConnectionHandler:
                            text_index += 1
                            self.recode_first_last_text(segment_text, text_index)
                            future = self.executor.submit(
-                                self.speak_and_play, segment_text, text_index
+                                self.speak_and_play, None, segment_text, text_index
                            )
                            self.tts_queue.put((future, text_index))
                            # 更新已处理字符位置
@@ -783,7 +705,7 @@ class ConnectionHandler:
                text_index += 1
                self.recode_first_last_text(segment_text, text_index)
                future = self.executor.submit(
-                    self.speak_and_play, segment_text, text_index
+                    self.speak_and_play, None, segment_text, text_index
                )
                self.tts_queue.put((future, text_index))

@@ -846,7 +768,7 @@ class ConnectionHandler:
        if result.action == Action.RESPONSE:  # 直接回复前端
            text = result.response
            self.recode_first_last_text(text, text_index)
-            future = self.executor.submit(self.speak_and_play, text, text_index)
+            future = self.executor.submit(self.speak_and_play, None, text, text_index)
            self.tts_queue.put((future, text_index))
            self.dialogue.put(Message(role="assistant", content=text))
        elif result.action == Action.REQLLM:  # 调用函数后再请求llm生成回复
@@ -881,11 +803,11 @@ class ConnectionHandler:
                        content=text,
                    )
                )
-                self.chat_with_function_calling(text, tool_call=True)
+                self.chat(text, tool_call=True)
        elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
            text = result.result
            self.recode_first_last_text(text, text_index)
-            future = self.executor.submit(self.speak_and_play, text, text_index)
+            future = self.executor.submit(self.speak_and_play, None, text, text_index)
            self.tts_queue.put((future, text_index))
            self.dialogue.put(Message(role="assistant", content=text))
        else:
@@ -912,11 +834,7 @@ class ConnectionHandler:
                    self.logger.bind(tag=TAG).debug("正在处理TTS任务...")
                    tts_timeout = int(self.config.get("tts_timeout", 10))
                    tts_file, text, _ = future.result(timeout=tts_timeout)
-                    if text is None or len(text) <= 0:
-                        self.logger.bind(tag=TAG).error(
-                            f"TTS出错：{text_index}: tts text is empty"
-                        )
-                    elif tts_file is None:
+                    if tts_file is None:
                        self.logger.bind(tag=TAG).error(
                            f"TTS出错： file is empty: {text_index}: {text}"
                        )
@@ -925,12 +843,16 @@ class ConnectionHandler:
                            f"TTS生成：文件路径: {tts_file}"
                        )
                        if os.path.exists(tts_file):
-                            if self.audio_format == "pcm":
+                            if tts_file.endswith(".p3"):
+                                audio_datas, _ = p3.decode_opus_from_file(tts_file)
+                            elif self.audio_format == "pcm":
                                audio_datas, _ = self.tts.audio_to_pcm_data(tts_file)
                            else:
                                audio_datas, _ = self.tts.audio_to_opus_data(tts_file)
                            # 在这里上报TTS数据
-                            enqueue_tts_report(self, text, audio_datas)
+                            enqueue_tts_report(
+                                self, tts_file if text is None else text, audio_datas
+                            )
                        else:
                            self.logger.bind(tag=TAG).error(
                                f"TTS出错：文件不存在{tts_file}"
@@ -946,6 +868,7 @@ class ConnectionHandler:
                    self.tts.delete_audio_file
                    and tts_file is not None
                    and os.path.exists(tts_file)
+                    and tts_file.startswith(self.tts.output_file)
                ):
                    os.remove(tts_file)
            except Exception as e:
@@ -995,16 +918,13 @@ class ConnectionHandler:
                if item is None:  # 检测毒丸对象
                    break

-                type, text, audio_data = item
+                type, text, audio_data, report_time = item

                try:
-                    # 执行上报（传入二进制数据）
-                    report(self, type, text, audio_data)
+                    # 提交任务到线程池
+                    self.report_thread_pool.submit(self._process_report, type, text, audio_data, report_time)
                except Exception as e:
                    self.logger.bind(tag=TAG).error(f"聊天记录上报线程异常: {e}")
-                finally:
-                    # 标记任务完成
-                    self.report_queue.task_done()
            except queue.Empty:
                continue
            except Exception as e:
@@ -1012,18 +932,32 @@ class ConnectionHandler:

        self.logger.bind(tag=TAG).info("聊天记录上报线程已退出")

-    def speak_and_play(self, text, text_index=0):
-        if text is None or len(text) <= 0:
-            self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{text}")
-            return None, text, text_index
-        tts_file = self.tts.to_tts(text)
+    def _process_report(self, type, text, audio_data, report_time):
+        """处理上报任务"""
+        try:
+            # 执行上报（传入二进制数据）
+            report(self, type, text, audio_data, report_time)
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
+        finally:
+            # 标记任务完成
+            self.report_queue.task_done()
+
+    def speak_and_play(self, file_path, content, text_index=0):
+        if file_path is not None:
+            self.logger.bind(tag=TAG).info(f"无需tts转换: 从文件播放，{file_path}")
+            return file_path, content, text_index
+        if content is None or len(content) <= 0:
+            self.logger.bind(tag=TAG).info(f"无需tts转换，query为空，{content}")
+            return None, content, text_index
+        tts_file = self.tts.to_tts(content)
        if tts_file is None:
-            self.logger.bind(tag=TAG).error(f"tts转换失败，{text}")
-            return None, text, text_index
+            self.logger.bind(tag=TAG).error(f"tts转换失败，{content}")
+            return None, content, text_index
        self.logger.bind(tag=TAG).debug(f"TTS 文件生成完毕: {tts_file}")
        if self.max_output_size > 0:
-            add_device_output(self.headers.get("device-id"), len(text))
-        return tts_file, text, text_index
+            add_device_output(self.headers.get("device-id"), len(content))
+        return tts_file, content, text_index

    def clearSpeakStatus(self):
        self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态")
@@ -1067,6 +1001,12 @@ class ConnectionHandler:
            self.executor.shutdown(wait=False)
            self.executor = None

+        # 关闭上报线程池
+        if self.report_thread_pool:
+            self.report_thread_pool.shutdown(wait=False)
+            self.report_thread_pool = None
+            self.logger.bind(tag=TAG).info("上报线程池已关闭")
+
        self.logger.bind(tag=TAG).info("连接资源已释放")

    def clear_queues(self):
@@ -109,21 +109,21 @@ async def process_intent_result(conn, intent_result, original_text):
                    if result.action == Action.RESPONSE:  # 直接回复前端
                        text = result.response
                        if text is not None:
-                            speak_and_play(conn, text)
+                            speak_txt(conn, text)
                    elif result.action == Action.REQLLM:  # 调用函数后再请求llm生成回复
                        text = result.result
                        conn.dialogue.put(Message(role="tool", content=text))
                        llm_result = conn.intent.replyResult(text, original_text)
                        if llm_result is None:
                            llm_result = text
-                        speak_and_play(conn, llm_result)
+                        speak_txt(conn, llm_result)
                    elif (
                        result.action == Action.NOTFOUND
                        or result.action == Action.ERROR
                    ):
                        text = result.result
                        if text is not None:
-                            speak_and_play(conn, text)
+                            speak_txt(conn, text)
                    elif function_name != "play_music":
                        # For backward compatibility with original code
                        # 获取当前最新的文本索引
@@ -131,7 +131,7 @@ async def process_intent_result(conn, intent_result, original_text):
                        if text is None:
                            text = result.result
                        if text is not None:
-                            speak_and_play(conn, text)
+                            speak_txt(conn, text)

            # 将函数执行放在线程池中
            conn.executor.submit(process_function_call)
@@ -142,12 +142,12 @@ async def process_intent_result(conn, intent_result, original_text):
        return False


-def speak_and_play(conn, text):
+def speak_txt(conn, text):
    text_index = (
        conn.tts_last_text_index + 1 if hasattr(conn, "tts_last_text_index") else 0
    )
    conn.recode_first_last_text(text, text_index)
-    future = conn.executor.submit(conn.speak_and_play, text, text_index)
+    future = conn.executor.submit(conn.speak_and_play, None, text, text_index)
    conn.llm_finish_task = True
    conn.tts_queue.put((future, text_index))
    conn.dialogue.put(Message(role="assistant", content=text))
@@ -39,7 +39,9 @@ async def handleAudioMessage(conn, audio):
        if len(conn.asr_audio) < 15:
            conn.asr_server_receive = True
        else:
-            raw_text, _ = await conn.asr.speech_to_text(conn.asr_audio, conn.session_id)  # 确保ASR模块返回原始文本
+            raw_text, _ = await conn.asr.speech_to_text(
+                conn.asr_audio, conn.session_id
+            )  # 确保ASR模块返回原始文本
            conn.logger.bind(tag=TAG).info(f"识别文本: {raw_text}")
            text_len, _ = remove_punctuation_and_length(raw_text)
            if text_len > 0:
@@ -76,11 +78,7 @@ async def startToChat(conn, text):

    # 意图未被处理，继续常规聊天流程
    await send_stt_message(conn, text)
-    if conn.intent_type == "function_call":
-        # 使用支持function calling的聊天方法
-        conn.executor.submit(conn.chat_with_function_calling, text)
-    else:
-        conn.executor.submit(conn.chat, text)
+    conn.executor.submit(conn.chat, text)


 async def no_voice_close_connect(conn):
@@ -8,6 +8,7 @@ TTS上报功能已集成到ConnectionHandler类中。

 具体实现请参考core/connection.py中的相关代码。
 """
+import time

 import opuslib_next

@@ -16,7 +17,7 @@ from config.manage_api_client import report as manage_report
 TAG = __name__


-def report(conn, type, text, opus_data):
+def report(conn, type, text, opus_data, report_time):
    """执行聊天记录上报操作

    Args:
@@ -24,6 +25,7 @@ def report(conn, type, text, opus_data):
        type: 上报类型，1为用户，2为智能体
        text: 合成文本
        opus_data: opus音频数据
+        report_time: 上报时间
    """
    try:
        if opus_data:
@@ -37,6 +39,7 @@ def report(conn, type, text, opus_data):
            chat_type=type,
            content=text,
            audio=audio_data,
+            report_time=report_time,
        )
    except Exception as e:
        conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}")
@@ -104,12 +107,12 @@ def enqueue_tts_report(conn, text, opus_data):
    try:
        # 使用连接对象的队列，传入文本和二进制数据而非文件路径
        if conn.chat_history_conf == 2:
-            conn.report_queue.put((2, text, opus_data))
+            conn.report_queue.put((2, text, opus_data, int(time.time())))
            conn.logger.bind(tag=TAG).debug(
                f"TTS数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} "
            )
        else:
-            conn.report_queue.put((2, text, None))
+            conn.report_queue.put((2, text, None, int(time.time())))
            conn.logger.bind(tag=TAG).debug(
                f"TTS数据已加入上报队列: {conn.device_id}, 不上报音频"
            )
@@ -132,12 +135,12 @@ def enqueue_asr_report(conn, text, opus_data):
    try:
        # 使用连接对象的队列，传入文本和二进制数据而非文件路径
        if conn.chat_history_conf == 2:
-            conn.report_queue.put((1, text, opus_data))
+            conn.report_queue.put((1, text, opus_data, int(time.time())))
            conn.logger.bind(tag=TAG).debug(
                f"ASR数据已加入上报队列: {conn.device_id}, 音频大小: {len(opus_data)} "
            )
        else:
-            conn.report_queue.put((1, text, None))
+            conn.report_queue.put((1, text, None, int(time.time())))
            conn.logger.bind(tag=TAG).debug(
                f"ASR数据已加入上报队列: {conn.device_id}, 不上报音频"
            )
@@ -30,15 +30,25 @@ class ASRProviderBase(ABC):
    @staticmethod
    def decode_opus(opus_data: List[bytes]) -> bytes:
        """将Opus音频数据解码为PCM数据"""
+        try:
+            decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
+            pcm_data = []
+            buffer_size = 960  # 每次处理960个采样点

-        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-        pcm_data = []
+            for opus_packet in opus_data:
+                try:
+                    # 使用较小的缓冲区大小进行处理
+                    pcm_frame = decoder.decode(opus_packet, buffer_size)
+                    if pcm_frame:
+                        pcm_data.append(pcm_frame)
+                except opuslib_next.OpusError as e:
+                    logger.bind(tag=TAG).warning(f"Opus解码错误，跳过当前数据包: {e}")
+                    continue
+                except Exception as e:
+                    logger.bind(tag=TAG).error(f"音频处理错误: {e}", exc_info=True)
+                    continue

-        for opus_packet in opus_data:
-            try:
-                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)
-
-        return pcm_data
+            return pcm_data
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}", exc_info=True)
+            return []
@@ -9,10 +9,14 @@ import uuid
 from core.providers.asr.base import ASRProviderBase
 from funasr import AutoModel
 from funasr.utils.postprocess_utils import rich_transcription_postprocess
+import shutil

 TAG = __name__
 logger = setup_logging()

+MAX_RETRIES = 2
+RETRY_DELAY = 1  # 重试延迟（秒）
+

 # 捕获标准输出
 class CaptureOutput:
@@ -68,46 +72,69 @@ class ASRProvider(ASRProviderBase):
    ) -> Tuple[Optional[str], Optional[str]]:
        """语音转文本主处理逻辑"""
        file_path = None
-        try:
-            # 合并所有opus数据包
-            if self.audio_format == "pcm":
-                pcm_data = opus_data
-            else:
-                pcm_data = self.decode_opus(opus_data)
+        retry_count = 0

-            combined_pcm_data = b"".join(pcm_data)
+        while retry_count < MAX_RETRIES:
+            try:
+                # 合并所有opus数据包
+                if self.audio_format == "pcm":
+                    pcm_data = opus_data
+                else:
+                    pcm_data = self.decode_opus(opus_data)

-            # 判断是否保存为WAV文件
-            if self.delete_audio_file:
-                pass
-            else:
-                file_path = self.save_audio_to_file(pcm_data, session_id)
+                combined_pcm_data = b"".join(pcm_data)

-            # 语音识别
-            start_time = time.time()
-            result = self.model.generate(
-                input=combined_pcm_data,
-                cache={},
-                language="auto",
-                use_itn=True,
-                batch_size_s=60,
-            )
-            text = rich_transcription_postprocess(result[0]["text"])
-            logger.bind(tag=TAG).debug(
-                f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
-            )
+                # 检查磁盘空间
+                if not self.delete_audio_file:
+                    free_space = shutil.disk_usage(self.output_dir).free
+                    if free_space < len(combined_pcm_data) * 2:  # 预留2倍空间
+                        raise OSError("磁盘空间不足")

-            return text, file_path
+                # 判断是否保存为WAV文件
+                if self.delete_audio_file:
+                    pass
+                else:
+                    file_path = self.save_audio_to_file(pcm_data, session_id)

-        except Exception as e:
-            logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
-            return "", file_path
+                # 语音识别
+                start_time = time.time()
+                result = self.model.generate(
+                    input=combined_pcm_data,
+                    cache={},
+                    language="auto",
+                    use_itn=True,
+                    batch_size_s=60,
+                )
+                text = rich_transcription_postprocess(result[0]["text"])
+                logger.bind(tag=TAG).debug(
+                    f"语音识别耗时: {time.time() - start_time:.3f}s | 结果: {text}"
+                )

-        # finally:
-        #     # 文件清理逻辑
-        #     if self.delete_audio_file and file_path and os.path.exists(file_path):
-        #         try:
-        #             os.remove(file_path)
-        #             logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
-        #         except Exception as e:
-        #             logger.bind(tag=TAG).error(f"文件删除失败: {file_path} | 错误: {e}")
+                return text, file_path
+
+            except OSError as e:
+                retry_count += 1
+                if retry_count >= MAX_RETRIES:
+                    logger.bind(tag=TAG).error(
+                        f"语音识别失败（已重试{retry_count}次）: {e}", exc_info=True
+                    )
+                    return "", file_path
+                logger.bind(tag=TAG).warning(
+                    f"语音识别失败，正在重试（{retry_count}/{MAX_RETRIES}）: {e}"
+                )
+                time.sleep(RETRY_DELAY)
+
+            except Exception as e:
+                logger.bind(tag=TAG).error(f"语音识别失败: {e}", exc_info=True)
+                return "", file_path
+
+            finally:
+                # 文件清理逻辑
+                if self.delete_audio_file and file_path and os.path.exists(file_path):
+                    try:
+                        os.remove(file_path)
+                        logger.bind(tag=TAG).debug(f"已删除临时音频文件: {file_path}")
+                    except Exception as e:
+                        logger.bind(tag=TAG).error(
+                            f"文件删除失败: {file_path} | 错误: {e}"
+                        )
@@ -72,6 +72,10 @@ class IntentProvider(IntentProviderBase):
            '返回: {"function_call": {"name": "get_time"}}\n'
            "```\n"
            "```\n"
+            "用户: 当前电池电量是多少？\n"
+            '返回: {"function_call": {"name": "get_battery_level", "arguments": {"response_success": "当前电池电量为{value}%", "response_failure": "无法获取Battery的当前电量百分比"}}}\n'
+            "```\n"
+            "```\n"
            "用户: 我想结束对话\n"
            '返回: {"function_call": {"name": "handle_exit_intent", "arguments": {"say_goodbye": "goodbye"}}}\n'
            "```\n"
@@ -224,7 +228,8 @@ class IntentProvider(IntentProviderBase):
                if function_name == "continue_chat":
                    # 保留非工具相关的消息
                    clean_history = [
-                        msg for msg in conn.dialogue.dialogue
+                        msg
+                        for msg in conn.dialogue.dialogue
                        if msg.role not in ["tool", "function"]
                    ]
                    conn.dialogue.dialogue = clean_history
@@ -15,15 +15,26 @@ async def _get_device_status(conn, device_name, device_type, property_name):
    return status


-async def _set_device_property(conn, device_name, device_type, method_name, property_name, new_value=None, action=None, step=10):
+async def _set_device_property(
+    conn,
+    device_name,
+    device_type,
+    method_name,
+    property_name,
+    new_value=None,
+    action=None,
+    step=10,
+):
    """设置设备属性"""
-    current_value = await _get_device_status(conn, device_name, device_type, property_name)
+    current_value = await _get_device_status(
+        conn, device_name, device_type, property_name
+    )

-    if action == 'raise':
+    if action == "raise":
        current_value += step
-    elif action == 'lower':
+    elif action == "lower":
        current_value -= step
-    elif action == 'set':
+    elif action == "set":
        if new_value is None:
            raise Exception(f"缺少{property_name}参数")
        current_value = new_value
@@ -37,8 +48,7 @@ async def _set_device_property(conn, device_name, device_type, method_name, prop

 def _handle_device_action(conn, func, success_message, error_message, *args, **kwargs):
    """处理设备操作的通用函数"""
-    future = asyncio.run_coroutine_threadsafe(
-        func(conn, *args, **kwargs), conn.loop)
+    future = asyncio.run_coroutine_threadsafe(func(conn, *args, **kwargs), conn.loop)
    try:
        result = future.result()
        logger.bind(tag=TAG).info(f"{success_message}: {result}")
@@ -75,26 +85,41 @@ handle_device_function_desc = {
                "device_type": {
                    "type": "string",
                    "description": "设备类型，**严格限定为Speaker（音量）或Screen（亮度）**，其他设备类型禁止调用此函数",
-                    "enum": ["Speaker", "Screen"]
-
+                    "enum": ["Speaker", "Screen"],
                },
                "action": {
                    "type": "string",
-                    "description": "动作名称，可选值：get(获取),set(设置),raise(提高),lower(降低)"
+                    "description": "动作名称，可选值：get(获取),set(设置),raise(提高),lower(降低)",
                },
                "value": {
                    "type": "integer",
-                    "description": "值大小，可选值：0-100之间的整数"
-                }
+                    "description": "值大小，可选值：0-100之间的整数",
+                },
            },
-            "required": ["device_type", "action"]
-        }
-    }
+            "required": ["device_type", "action"],
+        },
+    },
 }


-@register_function('handle_speaker_volume_or_screen_brightness', handle_device_function_desc, ToolType.IOT_CTL)
-def handle_speaker_volume_or_screen_brightness(conn, device_type: str, action: str, value: int = None):
+@register_function(
+    "handle_speaker_volume_or_screen_brightness",
+    handle_device_function_desc,
+    ToolType.IOT_CTL,
+)
+def handle_speaker_volume_or_screen_brightness(
+    conn, device_type: str, action: str, value: int = None
+):
+    # 检查value是否为中文值
+    if (
+        value is not None
+        and isinstance(value, str)
+        and any("\u4e00" <= char <= "\u9fff" for char in str(value))
+    ):
+        raise Exception(
+            f"请直接告诉我要将{'音量' if device_type=='Speaker' else '亮度'}调整成多少"
+        )
+
    if device_type == "Speaker":
        method_name, property_name, device_name = "SetVolume", "volume", "音量"
    elif device_type == "Screen":
@@ -108,13 +133,25 @@ def handle_speaker_volume_or_screen_brightness(conn, device_type: str, action: s
    if action == "get":
        # get
        return _handle_device_action(
-            conn, _get_device_status, f"当前{device_name}", f"获取{device_name}失败",
-            device_name=device_name, device_type=device_type, property_name=property_name,
+            conn,
+            _get_device_status,
+            f"当前{device_name}",
+            f"获取{device_name}失败",
+            device_name=device_name,
+            device_type=device_type,
+            property_name=property_name,
        )
    else:
        # set, raise, lower
        return _handle_device_action(
-            conn, _set_device_property, f"{device_name}已调整到", f"{device_name}调整失败",
-            device_name=device_name, device_type=device_type, method_name=method_name,
-            property_name=property_name, new_value=value, action=action
+            conn,
+            _set_device_property,
+            f"{device_name}已调整到",
+            f"{device_name}调整失败",
+            device_name=device_name,
+            device_type=device_type,
+            method_name=method_name,
+            property_name=property_name,
+            new_value=value,
+            action=action,
        )
@@ -216,24 +216,16 @@ async def play_local_music(conn, specific_file=None):
        text = _get_random_play_prompt(selected_music)
        await send_stt_message(conn, text)
        conn.dialogue.put(Message(role="assistant", content=text))
-        conn.tts_first_text_index = 0
-        conn.tts_last_text_index = 0

-        tts_file = await asyncio.to_thread(conn.tts.to_tts, text)
-        if tts_file is not None and os.path.exists(tts_file):
-            conn.tts_last_text_index = 1
-            opus_packets, _ = conn.tts.audio_to_opus_data(tts_file)
-            conn.audio_play_queue.put((opus_packets, None, 0))
-            os.remove(tts_file)
+        conn.recode_first_last_text(text, 0)
+        future = conn.executor.submit(conn.speak_and_play, None, text, 0)
+        conn.tts_queue.put((future, 0))

+        conn.recode_first_last_text(text, 1)
+        future = conn.executor.submit(conn.speak_and_play, music_path, None, 1)
+        conn.tts_queue.put((future, 1))
        conn.llm_finish_task = True

-        if music_path.endswith(".p3"):
-            opus_packets, _ = p3.decode_opus_from_file(music_path)
-        else:
-            opus_packets, _ = conn.tts.audio_to_opus_data(music_path)
-        conn.audio_play_queue.put((opus_packets, None, conn.tts_last_text_index))
-
    except Exception as e:
        conn.logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")
        conn.logger.bind(tag=TAG).error(f"详细错误: {traceback.format_exc()}")
@@ -14,7 +14,7 @@
        }

        .container {
-            max-width: 800px;
+            max-width: 1000px;
            margin: 0 auto;
            background-color: white;
            border-radius: 10px;
@@ -482,9 +482,10 @@
                </span>
            </h2>
            <div class="connection-controls">
-                <input type="text" id="otaUrl" value="http://127.0.0.1:8002/xiaozhi/ota/" placeholder="OTA服务器地址" />
+                <input type="text" id="otaUrl" value="http://127.0.0.1:8002/xiaozhi/ota/"
+                    placeholder="OTA服务器地址，如：http://127.0.0.1:8002/xiaozhi/ota/" />
                <input type="text" id="serverUrl" value="ws://127.0.0.1:8000/xiaozhi/v1/"
-                    placeholder="WebSocket服务器地址" />
+                    placeholder="WebSocket服务器地址，如：ws://127.0.0.1:8000/xiaozhi/v1/" />
                <button id="connectButton">连接</button>
                <button id="authTestButton">测试认证</button>
            </div>
@@ -879,6 +880,7 @@
                                        // 流已结束且没有更多数据
                                        log("音频播放完成", 'info');
                                        isAudioPlaying = false;
+                                        this.endOfStream = false;
                                        streamingContext = null;
                                    } else {
                                        // 等待更多数据
@@ -1061,7 +1063,7 @@
            }
        }

-        // 初始化音频录制和处理 
+        // 初始化音频录制和处理
        async function initAudio() {
            try {
                // 请求麦克风权限
@@ -1306,7 +1308,8 @@
                // 先检查OTA状态
                log('正在检查OTA状态...', 'info');
                const otaUrl = document.getElementById('otaUrl').value.trim();
-
+                localStorage.setItem('otaUrl', otaUrl);
+                localStorage.setItem('wsUrl', url);
                try {
                    const otaResponse = await fetch(otaUrl, {
                        method: 'POST',
@@ -1563,6 +1566,10 @@
            const message = messageInput.value.trim();
            if (message === '' || !websocket || websocket.readyState !== WebSocket.OPEN) return;

+            audioBufferQueue = [];
+            isAudioBuffering = false;
+            isAudioPlaying = false;
+
            try {
                // 直接发送listen消息，不需要重复发送hello
                const listenMessage = {
@@ -1633,6 +1640,16 @@
            // 初始更新显示值
            updateDisplayValues();

+            const savedOtaUrl = localStorage.getItem('otaUrl');
+            if (savedOtaUrl) {
+                document.getElementById('otaUrl').value = savedOtaUrl;
+            }
+
+            const savedWsUrl = localStorage.getItem('wsUrl');
+            if (savedWsUrl) {
+                document.getElementById('serverUrl').value = savedWsUrl;
+            }
+
            // 切换面板显示
            toggleButton.addEventListener('click', () => {
                const isExpanded = configPanel.classList.contains('expanded');
@@ -1954,7 +1971,7 @@
                    this.buffer = new Int16Array(this.frameSize);
                    this.bufferIndex = 0;
                    this.isRecording = false;
-                    
+
                    // 监听来自主线程的消息
                    this.port.onmessage = (event) => {
                        if (event.data.command === 'start') {
@@ -1962,7 +1979,7 @@
                            this.port.postMessage({ type: 'status', status: 'started' });
                        } else if (event.data.command === 'stop') {
                            this.isRecording = false;
-                            
+
                            // 发送剩余的缓冲区
                            if (this.bufferIndex > 0) {
                                const finalBuffer = this.buffer.slice(0, this.bufferIndex);
@@ -1972,18 +1989,18 @@
                                });
                                this.bufferIndex = 0;
                            }
-                            
+
                            this.port.postMessage({ type: 'status', status: 'stopped' });
                        }
                    };
                }
-                
+
                process(inputs, outputs, parameters) {
                    if (!this.isRecording) return true;
-                    
+
                    const input = inputs[0][0]; // 获取第一个输入通道
                    if (!input) return true;
-                    
+
                    // 将浮点采样转换为16位整数并存储
                    for (let i = 0; i < input.length; i++) {
                        if (this.bufferIndex >= this.frameSize) {
@@ -1994,15 +2011,15 @@
                            });
                            this.bufferIndex = 0;
                        }
-                        
+
                        // 转换为16位整数 (-32768到32767)
                        this.buffer[this.bufferIndex++] = Math.max(-32768, Math.min(32767, Math.floor(input[i] * 32767)));
                    }
-                    
+
                    return true;
                }
            }
-            
+
            registerProcessor('audio-recorder-processor', AudioRecorderProcessor);
        `;