Merge pull request #2628 from xinnan-tech/py_await

Py await

docs/ragflow-integration.md

@@ -156,6 +156,14 @@ services:
 
 编辑`ragflow/docker`文件夹下的`.env`文件,找到以下配置，逐个搜索，逐个修改！逐个搜索，逐个修改！
 
+下面对于`.env`文件的修改，60%的人会忽略`MYSQL_USER`配置导致ragflow启动不成功，因此，需要强调三次：
+
+强调第一次：如果你的`.env`文件如果没有`MYSQL_USER`配置，请在配置文件增加这项！
+
+强调第二次：如果你的`.env`文件如果没有`MYSQL_USER`配置，请在配置文件增加这项！
+
+强调第三次：如果你的`.env`文件如果没有`MYSQL_USER`配置，请在配置文件增加这项！
+
 ``` env
 # 端口设置
 SVR_WEB_HTTP_PORT=8008           # HTTP端口

main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java

@@ -299,7 +299,7 @@ public interface Constant {
     /**
      * 版本号
      */
-    public static final String VERSION = "0.8.8";
+    public static final String VERSION = "0.8.9";
 
     /**
      * 无效固件URL

main/manager-api/src/main/java/xiaozhi/common/exception/ErrorCode.java

@@ -228,4 +228,16 @@ public interface ErrorCode {
 
     // 智能体模板相关错误码(补充)
     int AGENT_TEMPLATE_NOT_FOUND = 10183; // 默认智能体未找到
+
+    // 知识库适配器相关错误码
+    int RAG_ADAPTER_TYPE_NOT_SUPPORTED = 10184; // 不支持的适配器类型
+    int RAG_CONFIG_VALIDATION_FAILED = 10185; // RAG配置验证失败
+    int RAG_ADAPTER_CREATION_FAILED = 10186; // 适配器创建失败
+    int RAG_ADAPTER_INIT_FAILED = 10187; // 适配器初始化失败
+    int RAG_ADAPTER_CONNECTION_FAILED = 10188; // 适配器连接测试失败
+    int RAG_ADAPTER_OPERATION_FAILED = 10189; // 适配器操作失败
+    int RAG_ADAPTER_NOT_FOUND = 10190; // 适配器未找到
+    int RAG_ADAPTER_CACHE_ERROR = 10191; // 适配器缓存错误
+    int RAG_ADAPTER_TYPE_NOT_FOUND = 10192; // 适配器类型未找到
+
 }

main/manager-api/src/main/java/xiaozhi/modules/agent/service/impl/AgentServiceImpl.java

@@ -395,7 +395,20 @@ public class AgentServiceImpl extends BaseServiceImpl<AgentDao, AgentEntity> imp
             entity.setIntentModelId(template.getIntentModelId());
             entity.setSystemPrompt(template.getSystemPrompt());
             entity.setSummaryMemory(template.getSummaryMemory());
-            entity.setChatHistoryConf(template.getChatHistoryConf());
+
+            // 根据记忆模型类型设置默认的chatHistoryConf值
+            if (template.getMemModelId() != null) {
+                if (template.getMemModelId().equals("Memory_nomem")) {
+                    // 无记忆功能的模型，默认不记录聊天记录
+                    entity.setChatHistoryConf(0);
+                } else {
+                    // 有记忆功能的模型，默认记录文本和语音
+                    entity.setChatHistoryConf(2);
+                }
+            } else {
+                entity.setChatHistoryConf(template.getChatHistoryConf());
+            }
+
             entity.setLangCode(template.getLangCode());
             entity.setLanguage(template.getLanguage());
         }

main/manager-api/src/main/java/xiaozhi/modules/knowledge/config/KnowledgeBaseConfig.java

@@ -0,0 +1,22 @@
+package xiaozhi.modules.knowledge.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
+
+/**
+ * 知识库配置类
+ * 配置知识库相关的Bean
+ */
+@Configuration
+public class KnowledgeBaseConfig {
+
+    /**
+     * 提供KnowledgeBaseAdapterFactory的Bean实例
+     * @return KnowledgeBaseAdapterFactory实例
+     */
+    @Bean
+    public KnowledgeBaseAdapterFactory knowledgeBaseAdapterFactory() {
+        return new KnowledgeBaseAdapterFactory();
+    }
+}

main/manager-api/src/main/java/xiaozhi/modules/knowledge/dto/KnowledgeFilesDTO.java

@@ -77,7 +77,7 @@ public class KnowledgeFilesDTO implements Serializable {
             return STATUS_UNSTART;
         }
 
-        // 根据run字段的值直接映射到对应的状态码
+        // RAGFlow根据run字段的值直接映射到对应的状态码
         switch (run.toUpperCase()) {
             case "RUNNING":
                 return STATUS_RUNNING;

main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapter.java

@@ -0,0 +1,200 @@
+package xiaozhi.modules.knowledge.rag;
+
+import java.util.List;
+import java.util.Map;
+
+import org.springframework.web.multipart.MultipartFile;
+
+import xiaozhi.common.page.PageData;
+import xiaozhi.modules.knowledge.dto.KnowledgeFilesDTO;
+
+/**
+ * 知识库API适配器抽象基类
+ * 定义通用的知识库操作接口，支持多种后端API实现
+ */
+public abstract class KnowledgeBaseAdapter {
+
+        /**
+         * 获取适配器类型标识
+         * 
+         * @return 适配器类型（如：ragflow, milvus, pinecone等）
+         */
+        public abstract String getAdapterType();
+
+        /**
+         * 初始化适配器配置
+         * 
+         * @param config 配置参数
+         */
+        public abstract void initialize(Map<String, Object> config);
+
+        /**
+         * 验证配置是否有效
+         * 
+         * @param config 配置参数
+         * @return 验证结果
+         */
+        public abstract boolean validateConfig(Map<String, Object> config);
+
+        /**
+         * 分页查询文档列表
+         * 
+         * @param datasetId   知识库ID
+         * @param queryParams 查询参数
+         * @param page        页码
+         * @param limit       每页数量
+         * @return 分页数据
+         */
+        public abstract PageData<KnowledgeFilesDTO> getDocumentList(String datasetId,
+                        Map<String, Object> queryParams,
+                        Integer page,
+                        Integer limit);
+
+        /**
+         * 根据文档ID获取文档详情
+         * 
+         * @param datasetId 知识库ID
+         * @return 文档详情
+         */
+        public abstract KnowledgeFilesDTO getDocumentById(String datasetId, String documentId);
+
+        /**
+         * 上传文档到知识库
+         * 
+         * @param datasetId    知识库ID
+         * @param file         上传的文件
+         * @param name         文档名称
+         * @param metaFields   元数据字段
+         * @param chunkMethod  分块方法
+         * @param parserConfig 解析器配置
+         * @return 上传的文档信息
+         */
+        public abstract KnowledgeFilesDTO uploadDocument(String datasetId,
+                        MultipartFile file,
+                        String name,
+                        Map<String, Object> metaFields,
+                        String chunkMethod,
+                        Map<String, Object> parserConfig);
+
+        /**
+         * 根据状态分页查询文档列表
+         * 
+         * @param datasetId 知识库ID
+         * @param status    文档解析状态
+         * @param page      页码
+         * @param limit     每页数量
+         * @return 分页数据
+         */
+        public abstract PageData<KnowledgeFilesDTO> getDocumentListByStatus(String datasetId,
+                        Integer status,
+                        Integer page,
+                        Integer limit);
+
+        /**
+         * 删除文档
+         * 
+         * @param datasetId  知识库ID
+         * @param documentId 文档ID
+         */
+        public abstract void deleteDocument(String datasetId, String documentId);
+
+        /**
+         * 解析文档（切块）
+         * 
+         * @param datasetId   知识库ID
+         * @param documentIds 文档ID列表
+         * @return 解析结果
+         */
+        public abstract boolean parseDocuments(String datasetId, List<String> documentIds);
+
+        /**
+         * 列出指定文档的切片
+         * 
+         * @param datasetId  知识库ID
+         * @param documentId 文档ID
+         * @param keywords   关键词过滤
+         * @param page       页码
+         * @param pageSize   每页数量
+         * @param chunkId    切片ID
+         * @return 切片列表信息
+         */
+        public abstract Map<String, Object> listChunks(String datasetId,
+                        String documentId,
+                        String keywords,
+                        Integer page,
+                        Integer pageSize,
+                        String chunkId);
+
+        /**
+         * 召回测试 - 从知识库中检索相关切片
+         * 
+         * @param question        用户查询
+         * @param datasetIds      数据集ID列表
+         * @param documentIds     文档ID列表
+         * @param retrievalParams 检索参数
+         * @return 召回测试结果
+         */
+        public abstract Map<String, Object> retrievalTest(String question,
+                        List<String> datasetIds,
+                        List<String> documentIds,
+                        Map<String, Object> retrievalParams);
+
+        /**
+         * 测试连接
+         * 
+         * @return 连接测试结果
+         */
+        public abstract boolean testConnection();
+
+        /**
+         * 获取适配器状态信息
+         * 
+         * @return 状态信息
+         */
+        public abstract Map<String, Object> getStatus();
+
+        /**
+         * 获取支持的配置参数
+         * 
+         * @return 配置参数说明
+         */
+        public abstract Map<String, Object> getSupportedConfig();
+
+        /**
+         * 获取默认配置
+         * 
+         * @return 默认配置
+         */
+        public abstract Map<String, Object> getDefaultConfig();
+
+        /**
+         * 创建数据集
+         * 
+         * @param createParams 创建参数
+         * @return 数据集ID
+         */
+        public abstract String createDataset(Map<String, Object> createParams);
+
+        /**
+         * 更新数据集
+         * 
+         * @param datasetId    数据集ID
+         * @param updateParams 更新参数
+         */
+        public abstract void updateDataset(String datasetId, Map<String, Object> updateParams);
+
+        /**
+         * 删除数据集
+         * 
+         * @param datasetId 数据集ID
+         */
+        public abstract void deleteDataset(String datasetId);
+
+        /**
+         * 获取数据集的文档数量
+         * 
+         * @param datasetId 数据集ID
+         * @return 文档数量
+         */
+        public abstract Integer getDocumentCount(String datasetId);
+}

main/manager-api/src/main/java/xiaozhi/modules/knowledge/rag/KnowledgeBaseAdapterFactory.java

@@ -0,0 +1,197 @@
+package xiaozhi.modules.knowledge.rag;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import lombok.extern.slf4j.Slf4j;
+import xiaozhi.common.exception.ErrorCode;
+import xiaozhi.common.exception.RenException;
+
+/**
+ * 知识库适配器工厂类
+ * 负责创建和管理不同类型的知识库API适配器
+ */
+@Slf4j
+public class KnowledgeBaseAdapterFactory {
+
+    // 注册的适配器类型映射
+    private static final Map<String, Class<? extends KnowledgeBaseAdapter>> adapterRegistry = new HashMap<>();
+
+    // 适配器实例缓存
+    private static final Map<String, KnowledgeBaseAdapter> adapterCache = new ConcurrentHashMap<>();
+
+    static {
+        // 注册内置适配器类型
+        registerAdapter("ragflow", xiaozhi.modules.knowledge.rag.impl.RAGFlowAdapter.class);
+        // 可以在这里注册更多适配器类型
+    }
+
+    /**
+     * 注册新的适配器类型
+     * 
+     * @param adapterType  适配器类型标识
+     * @param adapterClass 适配器类
+     */
+    public static void registerAdapter(String adapterType, Class<? extends KnowledgeBaseAdapter> adapterClass) {
+        if (adapterRegistry.containsKey(adapterType)) {
+            log.warn("适配器类型 '{}' 已存在，将被覆盖", adapterType);
+        }
+        adapterRegistry.put(adapterType, adapterClass);
+        log.info("注册适配器类型: {} -> {}", adapterType, adapterClass.getSimpleName());
+    }
+
+    /**
+     * 获取适配器实例
+     * 
+     * @param adapterType 适配器类型
+     * @param config      配置参数
+     * @return 适配器实例
+     */
+    public static KnowledgeBaseAdapter getAdapter(String adapterType, Map<String, Object> config) {
+        String cacheKey = buildCacheKey(adapterType, config);
+
+        // 检查缓存中是否已存在实例
+        if (adapterCache.containsKey(cacheKey)) {
+            log.debug("从缓存获取适配器实例: {}", cacheKey);
+            return adapterCache.get(cacheKey);
+        }
+
+        // 创建新的适配器实例
+        KnowledgeBaseAdapter adapter = createAdapter(adapterType, config);
+
+        // 缓存适配器实例
+        adapterCache.put(cacheKey, adapter);
+        log.info("创建并缓存适配器实例: {}", cacheKey);
+
+        return adapter;
+    }
+
+    /**
+     * 获取适配器实例（无配置）
+     * 
+     * @param adapterType 适配器类型
+     * @return 适配器实例
+     */
+    public static KnowledgeBaseAdapter getAdapter(String adapterType) {
+        return getAdapter(adapterType, null);
+    }
+
+    /**
+     * 获取所有已注册的适配器类型
+     * 
+     * @return 适配器类型集合
+     */
+    public static Set<String> getRegisteredAdapterTypes() {
+        return adapterRegistry.keySet();
+    }
+
+    /**
+     * 检查适配器类型是否已注册
+     * 
+     * @param adapterType 适配器类型
+     * @return 是否已注册
+     */
+    public static boolean isAdapterTypeRegistered(String adapterType) {
+        return adapterRegistry.containsKey(adapterType);
+    }
+
+    /**
+     * 清除适配器缓存
+     */
+    public static void clearCache() {
+        int cacheSize = adapterCache.size();
+        adapterCache.clear();
+        log.info("清除适配器缓存，共清除 {} 个实例", cacheSize);
+    }
+
+    /**
+     * 移除特定适配器类型的缓存
+     * 
+     * @param adapterType 适配器类型
+     */
+    public static void removeCacheByType(String adapterType) {
+        int removedCount = 0;
+        for (String cacheKey : adapterCache.keySet()) {
+            if (cacheKey.startsWith(adapterType + "@")) {
+                adapterCache.remove(cacheKey);
+                removedCount++;
+            }
+        }
+        log.info("移除适配器类型 '{}' 的缓存，共移除 {} 个实例", adapterType, removedCount);
+    }
+
+    /**
+     * 获取适配器工厂状态信息
+     * 
+     * @return 状态信息
+     */
+    public static Map<String, Object> getFactoryStatus() {
+        Map<String, Object> status = new HashMap<>();
+        status.put("registeredAdapterTypes", adapterRegistry.keySet());
+        status.put("cachedAdapterCount", adapterCache.size());
+        status.put("cacheKeys", adapterCache.keySet());
+        return status;
+    }
+
+    /**
+     * 创建适配器实例
+     * 
+     * @param adapterType 适配器类型
+     * @param config      配置参数
+     * @return 适配器实例
+     */
+    private static KnowledgeBaseAdapter createAdapter(String adapterType, Map<String, Object> config) {
+        if (!adapterRegistry.containsKey(adapterType)) {
+            throw new RenException(ErrorCode.RAG_ADAPTER_TYPE_NOT_SUPPORTED,
+                    "不支持的适配器类型: " + adapterType);
+        }
+
+        try {
+            Class<? extends KnowledgeBaseAdapter> adapterClass = adapterRegistry.get(adapterType);
+            KnowledgeBaseAdapter adapter = adapterClass.getDeclaredConstructor().newInstance();
+
+            // 初始化适配器
+            if (config != null) {
+                adapter.initialize(config);
+
+                // 验证配置
+                if (!adapter.validateConfig(config)) {
+                    throw new RenException(ErrorCode.RAG_CONFIG_VALIDATION_FAILED,
+                            "适配器配置验证失败: " + adapterType);
+                }
+            }
+
+            log.info("成功创建适配器实例: {}", adapterType);
+            return adapter;
+
+        } catch (Exception e) {
+            log.error("创建适配器实例失败: {}", adapterType, e);
+            throw new RenException(ErrorCode.RAG_ADAPTER_CREATION_FAILED,
+                    "创建适配器失败: " + adapterType + ", 错误: " + e.getMessage());
+        }
+    }
+
+    /**
+     * 构建缓存键
+     * 
+     * @param adapterType 适配器类型
+     * @param config      配置参数
+     * @return 缓存键
+     */
+    private static String buildCacheKey(String adapterType, Map<String, Object> config) {
+        if (config == null || config.isEmpty()) {
+            return adapterType + "@default";
+        }
+
+        // 基于配置参数生成缓存键
+        StringBuilder keyBuilder = new StringBuilder(adapterType + "@");
+
+        // 使用配置的哈希值作为缓存键的一部分
+        int configHash = config.hashCode();
+        keyBuilder.append(configHash);
+
+        return keyBuilder.toString();
+    }
+}

main/manager-api/src/main/java/xiaozhi/modules/knowledge/service/impl/KnowledgeBaseServiceImpl.java

@@ -35,6 +35,8 @@ import xiaozhi.common.utils.MessageUtils;
 import xiaozhi.modules.knowledge.dao.KnowledgeBaseDao;
 import xiaozhi.modules.knowledge.dto.KnowledgeBaseDTO;
 import xiaozhi.modules.knowledge.entity.KnowledgeBaseEntity;
+import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapter;
+import xiaozhi.modules.knowledge.rag.KnowledgeBaseAdapterFactory;
 import xiaozhi.modules.knowledge.service.KnowledgeBaseService;
 import xiaozhi.modules.model.dao.ModelConfigDao;
 import xiaozhi.modules.model.entity.ModelConfigEntity;
@@ -106,7 +108,7 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
         if (pageData != null && pageData.getList() != null) {
             for (KnowledgeBaseDTO knowledgeBase : pageData.getList()) {
                 try {
-                    Integer documentCount = getDocumentCountFromRAGFlow(knowledgeBase.getDatasetId(),
+                    Integer documentCount = getDocumentCountFromRAG(knowledgeBase.getDatasetId(),
                             knowledgeBase.getRagModelId());
                     knowledgeBase.setDocumentCount(documentCount);
                 } catch (Exception e) {
@@ -146,10 +148,10 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
         checkDuplicateKnowledgeBaseName(knowledgeBaseDTO, null);
 
         String datasetId = null;
-        // 调用RAGFlow API创建数据集
+        // 调用RAG API创建数据集
         try {
             Map<String, Object> ragConfig = getValidatedRAGConfig(knowledgeBaseDTO.getRagModelId());
-            datasetId = createDatasetInRAGFlow(
+            datasetId = createDatasetInRAG(
                     knowledgeBaseDTO.getName(),
                     knowledgeBaseDTO.getDescription(),
                     ragConfig);
@@ -162,13 +164,13 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
         KnowledgeBaseEntity existingEntity = knowledgeBaseDao.selectOne(
                 new QueryWrapper<KnowledgeBaseEntity>().eq("dataset_id", datasetId));
         if (existingEntity != null) {
-            // 如果datasetId已存在，删除RAGFlow中的数据集并抛出异常
+            // 如果datasetId已存在，删除RAG中的数据集并抛出异常
             try {
                 Map<String, Object> ragConfig = getValidatedRAGConfig(knowledgeBaseDTO.getRagModelId());
-                deleteDatasetInRAGFlow(datasetId, ragConfig);
+                deleteDatasetInRAG(datasetId, ragConfig);
             } catch (Exception deleteException) {
                 // 提供更详细的错误信息，包括异常类型和消息
-                String errorMessage = "删除重复datasetId的RAGFlow数据集失败: " + deleteException.getClass().getSimpleName();
+                String errorMessage = "删除重复datasetId的RAG数据集失败: " + deleteException.getClass().getSimpleName();
                 if (deleteException.getMessage() != null) {
                     errorMessage += " - " + deleteException.getMessage();
                 }
@@ -219,17 +221,17 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
                 // 先校验RAG配置
                 Map<String, Object> ragConfig = getValidatedRAGConfig(knowledgeBaseDTO.getRagModelId());
 
-                // 调用RAGFlow API更新数据集
-                updateDatasetInRAGFlow(
+                // 调用RAG API更新数据集
+                updateDatasetInRAG(
                         knowledgeBaseDTO.getDatasetId(),
                         knowledgeBaseDTO.getName(),
                         knowledgeBaseDTO.getDescription(),
                         ragConfig);
 
-                log.info("RAGFlow API更新成功，datasetId: {}", knowledgeBaseDTO.getDatasetId());
+                log.info("RAG API更新成功，datasetId: {}", knowledgeBaseDTO.getDatasetId());
             } catch (Exception e) {
                 // 提供更详细的错误信息，包括异常类型和消息
-                String errorMessage = "更新RAGFlow数据集失败: " + e.getClass().getSimpleName();
+                String errorMessage = "更新RAG数据集失败: " + e.getClass().getSimpleName();
                 if (e.getMessage() != null) {
                     errorMessage += " - " + e.getMessage();
                 }
@@ -237,7 +239,7 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
                 throw e;
             }
         } else {
-            log.warn("datasetId或ragModelId为空，跳过RAGFlow更新");
+            log.warn("datasetId或ragModelId为空，跳过RAG更新");
         }
 
         KnowledgeBaseEntity entity = ConvertUtils.sourceToTarget(knowledgeBaseDTO, KnowledgeBaseEntity.class);
@@ -288,19 +290,19 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
         log.info("找到记录: ID={}, datasetId={}, ragModelId={}",
                 entity.getId(), entity.getDatasetId(), entity.getRagModelId());
 
-        // 先调用RAGFlow API删除数据集
+        // 先调用RAG API删除数据集
         boolean apiDeleteSuccess = false;
         if (StringUtils.isNotBlank(entity.getDatasetId()) && StringUtils.isNotBlank(entity.getRagModelId())) {
             try {
-                log.info("开始调用RAGFlow API删除数据集");
+                log.info("开始调用RAG API删除数据集");
                 // 在删除前进行RAG配置校验
                 Map<String, Object> ragConfig = getValidatedRAGConfig(entity.getRagModelId());
-                deleteDatasetInRAGFlow(entity.getDatasetId(), ragConfig);
-                log.info("RAGFlow API删除调用完成");
+                deleteDatasetInRAG(entity.getDatasetId(), ragConfig);
+                log.info("RAG API删除调用完成");
                 apiDeleteSuccess = true;
             } catch (Exception e) {
                 // 提供更详细的错误信息，包括异常类型和消息
-                String errorMessage = "删除RAGFlow数据集失败: " + e.getClass().getSimpleName();
+                String errorMessage = "删除RAG数据集失败: " + e.getClass().getSimpleName();
                 if (e.getMessage() != null) {
                     errorMessage += " - " + e.getMessage();
                 }
@@ -308,7 +310,7 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
                 throw e;
             }
         } else {
-            log.warn("datasetId或ragModelId为空，跳过RAGFlow删除");
+            log.warn("datasetId或ragModelId为空，跳过RAG删除");
             apiDeleteSuccess = true; // 没有RAG数据集，视为成功
         }
 
@@ -432,289 +434,133 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
     }
 
     /**
-     * 调用RAGFlow API创建数据集
+     * 从RAG配置中提取适配器类型
+     * 
+     * @param config RAG配置
+     * @return 适配器类型
      */
-    private String createDatasetInRAGFlow(String name, String description, Map<String, Object> ragConfig) {
-        String datasetId = null;
-        String baseUrl = (String) ragConfig.get("base_url");
-        String apiKey = (String) ragConfig.get("api_key");
-
-        log.info("开始调用RAGFlow API创建数据集, name: {}", name);
-        log.debug("RAGFlow配置 - baseUrl: {}, apiKey: {}", baseUrl, StringUtils.isBlank(apiKey) ? "未配置" : "已配置");
-
-        // 构建请求URL
-        String url = baseUrl + "/api/v1/datasets";
-        log.debug("请求URL: {}", url);
-
-        // 构建请求体
-        Map<String, Object> requestBody = new HashMap<>();
-        String username = SecurityUser.getUser().getUsername();
-        requestBody.put("name", username + "_" + name);
-        if (StringUtils.isNotBlank(description)) {
-            requestBody.put("description", description);
-        }
-        log.debug("请求体: {}", requestBody);
-
-        // 设置请求头
-        HttpHeaders headers = new HttpHeaders();
-        headers.setContentType(MediaType.APPLICATION_JSON);
-        headers.set("Authorization", "Bearer " + apiKey);
-
-        HttpEntity<Map<String, Object>> requestEntity = new HttpEntity<>(requestBody, headers);
-
-        // 发送POST请求
-        log.info("发送POST请求到RAGFlow API...");
-        ResponseEntity<String> response;
-        try {
-            response = restTemplate.exchange(url, HttpMethod.POST, requestEntity, String.class);
-        } catch (Exception e) {
-            String errorMessage = url + e.getMessage();
-            log.error(errorMessage, e);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
+    private String extractAdapterType(Map<String, Object> config) {
+        if (config == null) {
+            throw new RenException(ErrorCode.RAG_CONFIG_NOT_FOUND);
         }
 
-        log.info("RAGFlow API响应状态码: {}", response.getStatusCode());
-        log.debug("RAGFlow API响应内容: {}", response.getBody());
+        // 从配置中提取适配器类型
+        String adapterType = (String) config.get("type");
 
-        if (!response.getStatusCode().is2xxSuccessful()) {
-            String errorMessage = response.getStatusCode() + ", 响应内容: " + response.getBody();
-            log.error(errorMessage);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
+        // 验证适配器类型是否存在且非空
+        if (StringUtils.isBlank(adapterType)) {
+            throw new RenException(ErrorCode.RAG_ADAPTER_TYPE_NOT_FOUND);
         }
 
-        // 解析响应体，提取datasetId
-        String responseBody = response.getBody();
-        if (StringUtils.isNotBlank(responseBody)) {
-            try {
-                // 解析RAGFlow API响应，支持多种可能的字段名
-                ObjectMapper objectMapper = new ObjectMapper();
-                Map<String, Object> responseMap = objectMapper.readValue(responseBody, Map.class);
-
-                log.debug("RAGFlow API响应解析结果: {}", responseMap);
-
-                // 首先检查响应码
-                Integer code = (Integer) responseMap.get("code");
-                String message = (String) responseMap.get("message");
-                if (code != null && code == 0) {
-                    // 响应码为0表示成功，从data字段中获取datasetId
-                    Object dataObj = responseMap.get("data");
-                    if (dataObj instanceof Map) {
-                        Map<String, Object> dataMap = (Map<String, Object>) dataObj;
-                        datasetId = (String) dataMap.get("id");
-
-                        if (StringUtils.isBlank(datasetId)) {
-                            // 如果id字段为空，尝试其他可能的字段名
-                            datasetId = (String) dataMap.get("dataset_id");
-                            datasetId = (String) dataMap.get("datasetId");
-                        }
-                    }
-                } else {
-                    // 如果响应码不为0，说明API调用失败
-                    String errorMessage = code + (message != null ? message : "null");
-                    log.error(errorMessage);
-                    throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-                }
-
-                log.info("从RAGFlow API响应中解析出datasetId: {}", datasetId);
-                log.debug("完整响应内容: {}", responseBody);
-            } catch (IOException e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-            } catch (Exception e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                // 如果解析失败，但响应体不为空，尝试直接使用响应体作为错误信息
-                String finalErrorMessage = responseBody;
-                if (e.getMessage() != null) {
-                    finalErrorMessage += errorMessage;
-                }
-                throw new RenException(ErrorCode.RAG_API_ERROR, finalErrorMessage);
-            }
+        // 验证适配器类型是否已注册
+        if (!KnowledgeBaseAdapterFactory.isAdapterTypeRegistered(adapterType)) {
+            throw new RenException(ErrorCode.RAG_ADAPTER_TYPE_NOT_SUPPORTED,
+                    "不支持的适配器类型: " + adapterType);
         }
 
-        if (StringUtils.isBlank(datasetId)) {
-            log.error("无法从RAGFlow API响应中获取datasetId，响应内容: {}", responseBody);
-            throw new RenException(ErrorCode.RAG_DATASET_ID_NOT_NULL);
-        }
-        log.info("RAGFlow数据集创建成功，datasetId: {}", datasetId);
-
-        return datasetId;
+        return adapterType;
     }
 
     /**
-     * 调用RAGFlow API更新数据集
+     * 使用适配器创建数据集
      */
-    private void updateDatasetInRAGFlow(String datasetId, String name, String description,
+    private String createDatasetInRAG(String name, String description, Map<String, Object> ragConfig) {
+        log.info("开始使用适配器创建数据集, name: {}", name);
+
+        try {
+            // 从RAG配置中提取适配器类型
+            String adapterType = extractAdapterType(ragConfig);
+
+            // 使用适配器工厂获取适配器实例
+            KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
+
+            // 构建数据集创建参数
+            Map<String, Object> createParams = new HashMap<>();
+            String username = SecurityUser.getUser().getUsername();
+            createParams.put("name", username + "_" + name);
+            if (StringUtils.isNotBlank(description)) {
+                createParams.put("description", description);
+            }
+
+            // 调用适配器的创建数据集方法
+            String datasetId = adapter.createDataset(createParams);
+
+            log.info("数据集创建成功，datasetId: {}", datasetId);
+            return datasetId;
+
+        } catch (Exception e) {
+            // 直接传递底层适配器的详细错误信息
+            log.error("创建数据集失败", e);
+            if (e instanceof RenException) {
+                throw (RenException) e;
+            }
+            throw new RenException(ErrorCode.RAG_API_ERROR, e.getMessage());
+        }
+    }
+
+    /**
+     * 使用适配器更新数据集
+     */
+    private void updateDatasetInRAG(String datasetId, String name, String description,
             Map<String, Object> ragConfig) {
-        String baseUrl = (String) ragConfig.get("base_url");
-        String apiKey = (String) ragConfig.get("api_key");
+        log.info("开始使用适配器更新数据集，datasetId: {}, name: {}", datasetId, name);
 
-        log.info("开始调用RAGFlow API更新数据集，datasetId: {}, name: {}", datasetId, name);
-        log.debug("RAGFlow配置 - baseUrl: {}, apiKey: {}", baseUrl, StringUtils.isBlank(apiKey) ? "未配置" : "已配置");
-
-        // 构建请求URL
-        String url = baseUrl + "/api/v1/datasets/" + datasetId;
-        log.debug("请求URL: {}", url);
-        // 构建请求体
-        Map<String, Object> requestBody = new HashMap<>();
-        requestBody.put("dataset_id", datasetId);
-        String username = SecurityUser.getUser().getUsername();
-        requestBody.put("name", username + "_" + name);
-        if (StringUtils.isNotBlank(description)) {
-            requestBody.put("description", description);
-        }
-        log.debug("请求体: {}", requestBody);
-
-        // 设置请求头
-        HttpHeaders headers = new HttpHeaders();
-        headers.setContentType(MediaType.APPLICATION_JSON);
-        headers.set("Authorization", "Bearer " + apiKey);
-
-        HttpEntity<Map<String, Object>> requestEntity = new HttpEntity<>(requestBody, headers);
-
-        // 发送PUT请求
-        log.info("发送PUT请求到RAGFlow API...");
-        ResponseEntity<String> response;
         try {
-            response = restTemplate.exchange(url, HttpMethod.PUT, requestEntity, String.class);
-        } catch (Exception e) {
-            String errorMessage = url + e.getMessage();
-            log.error(errorMessage, e);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-        }
+            // 从RAG配置中提取适配器类型
+            String adapterType = extractAdapterType(ragConfig);
 
-        log.info("RAGFlow API响应状态码: {}", response.getStatusCode());
-        log.debug("RAGFlow API响应内容: {}", response.getBody());
+            // 使用适配器工厂获取适配器实例
+            KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
 
-        if (!response.getStatusCode().is2xxSuccessful()) {
-            String errorMessage = response.getStatusCode() + ", 响应内容: " + response.getBody();
-            log.error(errorMessage);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-        }
-
-        // 解析响应体，验证操作是否真正成功
-        String responseBody = response.getBody();
-        if (responseBody != null) {
-            try {
-                Map<String, Object> responseMap = objectMapper.readValue(responseBody, Map.class);
-                Integer code = (Integer) responseMap.get("code");
-                String message = (String) responseMap.get("message");
-
-                if (code != null && code != 0) {
-                    String errorMessage = code + (message != null ? message : "null");
-                    log.error(errorMessage);
-                    throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-                }
-            } catch (IOException e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-            } catch (Exception e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                // 如果解析失败，但响应体不为空，尝试直接使用响应体作为错误信息
-                String finalErrorMessage = responseBody;
-                if (e.getMessage() != null) {
-                    finalErrorMessage += errorMessage;
-                }
-                throw new RenException(ErrorCode.RAG_API_ERROR, finalErrorMessage);
+            // 构建数据集更新参数
+            Map<String, Object> updateParams = new HashMap<>();
+            String username = SecurityUser.getUser().getUsername();
+            updateParams.put("name", username + "_" + name);
+            if (StringUtils.isNotBlank(description)) {
+                updateParams.put("description", description);
             }
+
+            // 调用适配器的更新数据集方法
+            adapter.updateDataset(datasetId, updateParams);
+
+            log.info("数据集更新成功，datasetId: {}", datasetId);
+
+        } catch (Exception e) {
+            // 直接传递底层适配器的详细错误信息
+            log.error("更新数据集失败", e);
+            if (e instanceof RenException) {
+                throw (RenException) e;
+            }
+            throw new RenException(ErrorCode.RAG_API_ERROR, e.getMessage());
         }
-
-        log.info("RAGFlow数据集更新成功，datasetId: {}", datasetId);
-
     }
 
     /**
-     * 调用RAGFlow API删除数据集
+     * 使用适配器删除数据集
      */
-    private void deleteDatasetInRAGFlow(String datasetId, Map<String, Object> ragConfig) {
-        String baseUrl = (String) ragConfig.get("base_url");
-        String apiKey = (String) ragConfig.get("api_key");
+    private void deleteDatasetInRAG(String datasetId, Map<String, Object> ragConfig) {
+        log.info("开始使用适配器删除数据集，datasetId: {}", datasetId);
 
-        log.info("开始调用RAGFlow API删除数据集，datasetId: {}", datasetId);
-        log.debug("RAGFlow配置 - baseUrl: {}, apiKey: {}", baseUrl, StringUtils.isBlank(apiKey) ? "未配置" : "已配置");
-
-        // 构建请求URL
-        String url = baseUrl + "/api/v1/datasets";
-        log.debug("请求URL: {}", url);
-
-        // 构建请求体
-        Map<String, Object> requestBody = new HashMap<>();
-        requestBody.put("ids", List.of(datasetId));
-        log.debug("请求体: {}", requestBody);
-
-        // 设置请求头
-        HttpHeaders headers = new HttpHeaders();
-        headers.setContentType(MediaType.APPLICATION_JSON);
-        headers.set("Authorization", "Bearer " + apiKey);
-
-        HttpEntity<Map<String, Object>> requestEntity = new HttpEntity<>(requestBody, headers);
-
-        // 发送DELETE请求
-        log.info("发送DELETE请求到RAGFlow API...");
-        ResponseEntity<String> response;
         try {
-            response = restTemplate.exchange(url, HttpMethod.DELETE, requestEntity, String.class);
+            // 从RAG配置中提取适配器类型
+            String adapterType = extractAdapterType(ragConfig);
+
+            // 使用适配器工厂获取适配器实例
+            KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
+
+            // 调用适配器的删除数据集方法
+            adapter.deleteDataset(datasetId);
+
+            log.info("数据集删除成功，datasetId: {}", datasetId);
+
         } catch (Exception e) {
-            String errorMessage = url + e.getMessage();
-            log.error(errorMessage, e);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-        }
-
-        log.info("RAGFlow API响应状态码: {}", response.getStatusCode());
-        log.debug("RAGFlow API响应内容: {}", response.getBody());
-
-        if (!response.getStatusCode().is2xxSuccessful()) {
-            String errorMessage = response.getStatusCode() + response.getBody();
-            log.error(errorMessage);
-            throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-        }
-
-        // 解析响应体，验证操作是否真正成功
-        String responseBody = response.getBody();
-        if (responseBody != null) {
-            try {
-                Map<String, Object> responseMap = objectMapper.readValue(responseBody, Map.class);
-                Integer code = (Integer) responseMap.get("code");
-                String message = (String) responseMap.get("message");
-
-                if (code != null && code != 0) {
-                    String errorMessage = code + (message != null ? message : "null");
-                    log.error(errorMessage);
-                    throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-                }
-            } catch (IOException e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                throw new RenException(ErrorCode.RAG_API_ERROR, errorMessage);
-            } catch (Exception e) {
-                // 构建详细的错误信息，包含异常类型和消息
-                String baseErrorMessage = e.getClass().getSimpleName();
-                String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-                log.error(errorMessage, e);
-                // 如果解析失败，但响应体不为空，尝试直接使用响应体作为错误信息
-                String finalErrorMessage = responseBody;
-                if (e.getMessage() != null) {
-                    finalErrorMessage += errorMessage;
-                }
-                throw new RenException(ErrorCode.RAG_API_ERROR, finalErrorMessage);
+            // 直接传递底层适配器的详细错误信息
+            log.error("删除数据集失败", e);
+            if (e instanceof RenException) {
+                throw (RenException) e;
             }
+            throw new RenException(ErrorCode.RAG_API_ERROR, e.getMessage());
         }
-
-        log.info("RAGFlow数据集删除成功，datasetId: {}", datasetId);
-
     }
 
     /**
@@ -760,9 +606,9 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
     }
 
     /**
-     * 从RAGFlow API获取知识库的文档数量
+     * 从适配器获取知识库的文档数量
      */
-    private Integer getDocumentCountFromRAGFlow(String datasetId, String ragModelId) {
+    private Integer getDocumentCountFromRAG(String datasetId, String ragModelId) {
         if (StringUtils.isBlank(datasetId) || StringUtils.isBlank(ragModelId)) {
             log.warn("datasetId或ragModelId为空，无法获取文档数量");
             return 0;
@@ -770,79 +616,29 @@ public class KnowledgeBaseServiceImpl extends BaseServiceImpl<KnowledgeBaseDao,
 
         log.info("开始获取知识库 {} 的文档数量", datasetId);
 
-        // 获取RAG配置
-        Map<String, Object> ragConfig = getValidatedRAGConfig(ragModelId);
-        String baseUrl = (String) ragConfig.get("base_url");
-        String apiKey = (String) ragConfig.get("api_key");
-
-        // 构建请求URL - 调用RAGFlow API获取文档列表，但不返回文档详情，只获取总数
-        String url = baseUrl + "/api/v1/datasets/" + datasetId + "/documents?page=1&size=1";
-        log.debug("请求URL: {}", url);
-
-        // 设置请求头
-        HttpHeaders headers = new HttpHeaders();
-        headers.setContentType(MediaType.APPLICATION_JSON);
-        headers.set("Authorization", "Bearer " + apiKey);
-
-        HttpEntity<String> requestEntity = new HttpEntity<>(headers);
-
-        // 发送GET请求
-        log.info("发送GET请求到RAGFlow API获取文档数量...");
-        ResponseEntity<String> response;
         try {
-            response = restTemplate.exchange(url, HttpMethod.GET, requestEntity, String.class);
+            // 获取RAG配置
+            Map<String, Object> ragConfig = getValidatedRAGConfig(ragModelId);
+
+            // 从RAG配置中提取适配器类型
+            String adapterType = extractAdapterType(ragConfig);
+
+            // 使用适配器工厂获取适配器实例
+            KnowledgeBaseAdapter adapter = KnowledgeBaseAdapterFactory.getAdapter(adapterType, ragConfig);
+
+            // 调用适配器的获取文档数量方法
+            Integer documentCount = adapter.getDocumentCount(datasetId);
+
+            log.info("获取知识库 {} 的文档数量成功: {}", datasetId, documentCount);
+            return documentCount;
+
         } catch (Exception e) {
-            String errorMessage = url + e.getMessage();
+            // 构建详细的错误信息，包含异常类型和消息
+            String baseErrorMessage = e.getClass().getSimpleName() + " - 获取知识库文档数量失败";
+            String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
             log.error(errorMessage, e);
             return 0;
         }
-
-        log.info("RAGFlow API响应状态码: {}", response.getStatusCode());
-
-        if (!response.getStatusCode().is2xxSuccessful()) {
-            log.error("RAGFlow API调用失败，状态码: {}, 响应内容: {}", response.getStatusCode(), response.getBody());
-            return 0;
-        }
-
-        String responseBody = response.getBody();
-        log.debug("RAGFlow API响应内容: {}", responseBody);
-
-        // 解析响应
-        try {
-            ObjectMapper objectMapper = new ObjectMapper();
-            Map<String, Object> responseMap = objectMapper.readValue(responseBody, Map.class);
-            Integer code = (Integer) responseMap.get("code");
-
-            if (code != null && code == 0) {
-                Object dataObj = responseMap.get("data");
-                if (dataObj instanceof Map) {
-                    Map<String, Object> dataMap = (Map<String, Object>) dataObj;
-                    Object totalObj = dataMap.get("total");
-                    if (totalObj instanceof Integer) {
-                        Integer documentCount = (Integer) totalObj;
-                        log.info("获取知识库 {} 的文档数量成功: {}", datasetId, documentCount);
-                        return documentCount;
-                    } else if (totalObj instanceof Long) {
-                        Long documentCount = (Long) totalObj;
-                        log.info("获取知识库 {} 的文档数量成功: {}", datasetId, documentCount);
-                        return documentCount.intValue();
-                    }
-                }
-            } else {
-                log.error("RAGFlow API调用失败，响应码: {}, 响应内容: {}", code, responseBody);
-            }
-        } catch (IOException e) {
-            // 构建详细的错误信息，包含异常类型和消息
-            String baseErrorMessage = e.getClass().getSimpleName() + " - 解析RAGFlow API响应时发生IO异常";
-            String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-            log.error(errorMessage, e);
-        } catch (Exception e) {
-            // 构建详细的错误信息，包含异常类型和消息
-            String baseErrorMessage = e.getClass().getSimpleName() + " - 解析RAGFlow API响应失败";
-            String errorMessage = baseErrorMessage + (e.getMessage() != null ? ": " + e.getMessage() : "");
-            log.error(errorMessage, e);
-        }
-        return 0;
     }
 
 }

main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java

@@ -174,7 +174,7 @@ public class SysParamsController {
             return;
         }
         if (StringUtils.isBlank(url) || url.equals("null")) {
-            throw new RenException(ErrorCode.OTA_URL_EMPTY);
+            return;
         }
 
         // 检查是否包含localhost或127.0.0.1
@@ -211,7 +211,7 @@ public class SysParamsController {
             return;
         }
         if (StringUtils.isBlank(url) || url.equals("null")) {
-            throw new RenException(ErrorCode.MCP_URL_EMPTY);
+            return;
         }
         if (url.contains("localhost") || url.contains("127.0.0.1")) {
             throw new RenException(ErrorCode.MCP_URL_LOCALHOST);
@@ -242,7 +242,7 @@ public class SysParamsController {
             return;
         }
         if (StringUtils.isBlank(url) || url.equals("null")) {
-            throw new RenException(ErrorCode.VOICEPRINT_URL_EMPTY);
+            return;
         }
         if (url.contains("localhost") || url.contains("127.0.0.1")) {
             throw new RenException(ErrorCode.VOICEPRINT_URL_LOCALHOST);

main/manager-api/src/main/resources/i18n/messages.properties

@@ -189,4 +189,13 @@
 10180=\u6587\u4EF6\u5185\u5BB9\u4E0D\u80FD\u4E3A\u7A7A
 10181=\u97F3\u8272\u514B\u9686\u540D\u79F0\u4E0D\u80FD\u4E3A\u7A7A
 10182=\u97F3\u8272\u514B\u9686\u97F3\u9891\u4E0D\u5B58\u5728
-10183=\u9ED8\u8BA4\u667A\u80FD\u4F53\u672A\u627E\u5230
+10183=\u9ED8\u8BA4\u667A\u80FD\u4F53\u672A\u627E\u5230
+10184=\u4E0D\u652F\u6301\u7684\u9002\u914D\u5668\u7C7B\u578B
+10185=RAG\u914D\u7F6E\u9A8C\u8BC1\u5931\u8D25
+10186=\u9002\u914D\u5668\u521B\u5EFA\u5931\u8D25
+10187=\u9002\u914D\u5668\u521D\u59CB\u5316\u5931\u8D25
+10188=\u9002\u914D\u5668\u8FDE\u63A5\u6D4B\u8BD5\u5931\u8D25
+10189=\u9002\u914D\u5668\u64CD\u4F5C\u5931\u8D25
+10190=\u9002\u914D\u5668\u672A\u627E\u5230
+10191=\u9002\u914D\u5668\u7F13\u5B58\u9519\u8BEF
+10192=\u9002\u914D\u5668\u7C7B\u578B\u672A\u627E\u5230

main/manager-api/src/main/resources/i18n/messages_de_DE.properties

@@ -189,4 +189,13 @@
 10180=Dateiinhalt darf nicht leer sein
 10181=Stimmenklon-Name darf nicht leer sein
 10182=Stimmenklon-Audio nicht gefunden
-10183=Standard-Agent-Vorlage nicht gefunden
+10183=Standard-Agent-Vorlage nicht gefunden
+10184=Nicht unterstützter Adaptertyp
+10185=RAG-Konfigurationsvalidierung fehlgeschlagen
+10186=Adapter-Erstellung fehlgeschlagen
+10187=Adapter-Initialisierung fehlgeschlagen
+10188=Adapter-Verbindungstest fehlgeschlagen
+10189=Adapter-Operation fehlgeschlagen
+10190=Adapter nicht gefunden
+10191=Adapter-Cache-Fehler
+10192=Adaptertyp nicht gefunden

main/manager-api/src/main/resources/i18n/messages_en_US.properties

@@ -189,4 +189,13 @@
 10180=File content cannot be empty
 10181=Voice clone name cannot be empty
 10182=Voice clone audio not found
-10183=Default agent template not found
+10183=Default agent template not found
+10184=Unsupported adapter type
+10185=RAG configuration validation failed
+10186=Adapter creation failed
+10187=Adapter initialization failed
+10188=Adapter connection test failed
+10189=Adapter operation failed
+10190=Adapter not found
+10191=Adapter cache error
+10192=Adapter type not found

main/manager-api/src/main/resources/i18n/messages_vi_VN.properties

@@ -189,4 +189,13 @@
 10180=Nội dung tệp không thể để trống
 10181=Tên nhân bản giọng nói không thể để trống
 10182=Không tìm thấy âm thanh nhân bản giọng nói
-10183=Không tìm thấy mẫu agent mặc định
+10183=Không tìm thấy mẫu agent mặc định
+10184=Loại bộ chuyển đổi không được hỗ trợ
+10185=Kiểm tra cấu hình RAG thất bại
+10186=Tạo bộ chuyển đổi thất bại
+10187=Khởi tạo bộ chuyển đổi thất bại
+10188=Kiểm tra kết nối bộ chuyển đổi thất bại
+10189=Thao tác bộ chuyển đổi thất bại
+10190=Không tìm thấy bộ chuyển đổi
+10191=Lỗi bộ nhớ đệm bộ chuyển đổi
+10192=Không tìm thấy loại bộ chuyển đổi

main/manager-api/src/main/resources/i18n/messages_zh_CN.properties

@@ -189,4 +189,13 @@
 10180=\u6587\u4EF6\u5185\u5BB9\u4E0D\u80FD\u4E3A\u7A7A
 10181=\u97F3\u8272\u514B\u9686\u540D\u79F0\u4E0D\u80FD\u4E3A\u7A7A
 10182=\u97F3\u8272\u514B\u9686\u97F3\u9891\u4E0D\u5B58\u5728
-10183=\u9ED8\u8BA4\u667A\u80FD\u4F53\u672A\u627E\u5230
+10183=\u9ED8\u8BA4\u667A\u80FD\u4F53\u672A\u627E\u5230
+10184=\u4E0D\u652F\u6301\u7684\u9002\u914D\u5668\u7C7B\u578B
+10185=RAG\u914D\u7F6E\u9A8C\u8BC1\u5931\u8D25
+10186=\u9002\u914D\u5668\u521B\u5EFA\u5931\u8D25
+10187=\u9002\u914D\u5668\u521D\u59CB\u5316\u5931\u8D25
+10188=\u9002\u914D\u5668\u8FDE\u63A5\u6D4B\u8BD5\u5931\u8D25
+10189=\u9002\u914D\u5668\u64CD\u4F5C\u5931\u8D25
+10190=\u9002\u914D\u5668\u672A\u627E\u5230
+10191=\u9002\u914D\u5668\u7F13\u5B58\u9519\u8BEF
+10192=\u9002\u914D\u5668\u7C7B\u578B\u672A\u627E\u5230

main/manager-api/src/main/resources/i18n/messages_zh_TW.properties

@@ -189,4 +189,13 @@
 10180=\u6587\u4ef6\u5185\u5bb9\u4e0d\u80fd\u70ba\u7a7a
 10181=\u97f3\u8272\u514b\u9686\u540d\u7a31\u4e0d\u80fd\u70ba\u7a7a
 10182=\u97f3\u8272\u514b\u9686\u97f3\u983b\u4e0d\u5b58\u5728
-10183=\u9ed8\u8ba4\u667a\u80fd\u4f53\u672a\u627e\u5230
+10183=\u9ed8\u8ba4\u667a\u80fd\u4f53\u672a\u627e\u5230
+10184=\u4E0D\u652F\u6301\u7684\u9002\u914D\u5668\u985E\u578B
+10185=RAG\u914D\u7F6E\u9A57\u8B49\u5931\u6557
+10186=\u9002\u914D\u5668\u5275\u5EFA\u5931\u6557
+10187=\u9002\u914D\u5668\u521D\u59CB\u5316\u5931\u6557
+10188=\u9002\u914D\u5668\u9023\u63A5\u6E2C\u8A66\u5931\u6557
+10189=\u9002\u914D\u5668\u64CD\u4F5C\u5931\u6557
+10190=\u9002\u914D\u5668\u672A\u627E\u5230
+10191=\u9002\u914D\u5668\u7F13\u5B58\u932F\u8AA4
+10192=\u9002\u914D\u5668\u985E\u578B\u672A\u627E\u5230

main/manager-mobile/src/pages/settings/index.vue

@@ -235,7 +235,7 @@ function showAbout() {
     title: t('settings.aboutApp', { appName: import.meta.env.VITE_APP_TITLE }),
     content: t('settings.aboutContent', {
       appName: import.meta.env.VITE_APP_TITLE,
-      version: '0.8.8'
+      version: '0.8.9'
     }),
     showCancel: false,
     confirmText: t('common.confirm'),

main/manager-web/src/components/AddModelDialog.vue

@@ -200,6 +200,13 @@ export default {
     confirm() {
       this.saving = true;
 
+      // 校验模型ID不能为纯文字或空格
+      if (this.formData.id && !this.validateModelId(this.formData.id)) {
+        this.$message.error(this.$t('modelConfigDialog.invalidModelId'));
+        this.saving = false;
+        return;
+      }
+
       if (!this.formData.supplier) {
         this.$message.error(this.$t('addModelDialog.requiredSupplier'));
         this.saving = false;
@@ -254,6 +261,38 @@ export default {
       this.providerFields = [];
       this.currentProvider = null;
     },
+    
+    // 校验模型ID：不能为纯文字或空格
+    validateModelId(modelId) {
+      if (!modelId || typeof modelId !== 'string') {
+        return false;
+      }
+      
+      // 去除首尾空格
+      const trimmedId = modelId.trim();
+      
+      // 检查是否为空或纯空格
+      if (trimmedId === '') {
+        return false;
+      }
+      
+      // 检查是否只包含字母（纯文字）
+      if (/^[a-zA-Z]+$/.test(trimmedId)) {
+        return false;
+      }
+      
+      // 检查是否包含空格
+      if (/\s/.test(trimmedId)) {
+        return false;
+      }
+      
+      // 允许字母、数字、下划线、连字符
+      if (!/^[a-zA-Z0-9_-]+$/.test(trimmedId)) {
+        return false;
+      }
+      
+      return true;
+    }
   }
 }
 </script>
@@ -365,7 +404,7 @@ export default {
 
 .custom-input-bg .el-input__inner,
 .custom-input-bg .el-textarea__inner {
-  background-color: #f6f8fc;
+  background-color: #ffffff;
 }
 
 

main/manager-web/src/i18n/en.js

@@ -849,6 +849,7 @@ export default {
   'modelConfigDialog.setDefault': 'Set as Default',
   'modelConfigDialog.modelId': 'Model ID',
   'modelConfigDialog.enterModelId': 'If not filled in, it will be generated automatically',
+  'modelConfigDialog.invalidModelId': 'Model ID cannot be pure text or spaces, please use letters, numbers, underscores, or hyphens',
   'modelConfigDialog.modelName': 'Model Name',
   'modelConfigDialog.enterModelName': 'Please enter model name',
   'modelConfigDialog.modelCode': 'Model Code',

main/manager-web/src/i18n/zh_CN.js

@@ -849,6 +849,7 @@ export default {
   'modelConfigDialog.setDefault': '设为默认',
   'modelConfigDialog.modelId': '模型ID',
   'modelConfigDialog.enterModelId': '未填写将自动生成模型ID',
+  'modelConfigDialog.invalidModelId': '模型ID不能为纯文字或空格，请使用字母、数字、下划线或连字符组合',
   'modelConfigDialog.modelName': '模型名称',
   'modelConfigDialog.enterModelName': '请输入模型名称',
   'modelConfigDialog.modelCode': '模型编码',

main/manager-web/src/i18n/zh_TW.js

@@ -849,6 +849,7 @@ export default {
   'modelConfigDialog.setDefault': '設為默認',
   'modelConfigDialog.modelId': '模型ID',
   'modelConfigDialog.enterModelId': '未填冩將自動生成模型ID',
+  'modelConfigDialog.invalidModelId': '模型ID不能為純文字或空格，請使用字母、數字、底線或連字符組合',
   'modelConfigDialog.modelName': '模型名稱',
   'modelConfigDialog.enterModelName': '請輸入模型名稱',
   'modelConfigDialog.modelCode': '模型編碼',

main/manager-web/src/views/KnowledgeFileUpload.vue

@@ -214,10 +214,11 @@
     <el-dialog :title="$t('knowledgeFileUpload.retrievalTest')" :visible.sync="retrievalTestDialogVisible"
       width="1200px" class="retrieval-test-dialog">
       <div class="retrieval-test-form">
-        <el-form :model="retrievalTestForm" label-width="80px">
+        <el-form :model="retrievalTestForm" label-width="80px" @submit.native.prevent="runRetrievalTest">
           <el-form-item :label="$t('knowledgeFileUpload.testQuestion')" required>
             <el-input v-model="retrievalTestForm.question"
-              :placeholder="$t('knowledgeFileUpload.testQuestionPlaceholder')" style="width: 100%; max-height: 80px;">
+              :placeholder="$t('knowledgeFileUpload.testQuestionPlaceholder')" style="width: 100%; max-height: 80px;"
+              @keyup.enter.native="runRetrievalTest">
             </el-input>
           </el-form-item>
         </el-form>

main/manager-web/src/views/auth.scss

@@ -38,11 +38,11 @@
   position: absolute;
   top: 50%;
   transform: translateY(-50%);
-  right: 15%;
+  right: 18%;
   background-color: #fff;
   border-radius: 20px;
   padding: 35px 0;
-  width: 550px;
+  width: 450px;
   box-sizing: border-box;
 }
 

main/manager-web/src/views/roleConfig.vue

@@ -604,15 +604,14 @@ export default {
       if (type === "Intent" && value !== "Intent_nointent") {
         this.fetchAllFunctions();
       }
-      if (type === "Memory" && value === "Memory_nomem") {
-        this.form.chatHistoryConf = 0;
-      }
-      if (
-        type === "Memory" &&
-        value !== "Memory_nomem" &&
-        (this.form.chatHistoryConf === 0 || this.form.chatHistoryConf === null)
-      ) {
-        this.form.chatHistoryConf = 2;
+      if (type === "Memory") {
+        if (value === "Memory_nomem") {
+          // 无记忆功能的模型，默认不记录聊天记录
+          this.form.chatHistoryConf = 0;
+        } else {
+          // 有记忆功能的模型，默认记录文本和语音
+          this.form.chatHistoryConf = 2;
+        }
       }
       if (type === "LLM") {
         // 当LLM类型改变时，更新意图识别选项的可见性
@@ -1170,7 +1169,8 @@ export default {
 
 .template-item {
   height: 4vh;
-  width: 76px;
+  min-width: 60px;
+  padding: 0 12px;
   border-radius: 8px;
   background: #e6ebff;
   line-height: 4vh;
@@ -1180,6 +1180,7 @@ export default {
   color: #5778ff;
   cursor: pointer;
   transition: background-color 0.3s ease;
+  white-space: nowrap;
 }
 
 .template-item:hover {

main/xiaozhi-server/agent-base-prompt.txt

@@ -56,6 +56,7 @@
   3. **洞察需求：** 结合上下文**深入理解用户真实意图**后再决定调用，避免无意义调用。
   4. **独立任务：** 除`<context>`已涵盖信息外，用户每个要求（即使相似）都视为**独立任务**，需调用工具获取最新数据，**不可偷懒复用历史结果**。
   5. **不确定时：** **切勿猜测或编造答案**。若不确定相关操作，可引导用户澄清或告知能力限制。
+  6. **多工具调用：** 当用户要求执行多个任务时，你会调用多个工具（数量不定）。**重要：在获取到所有工具结果后，你必须依次总结每个工具的查询结果**，不要遗漏任何一个。例如用户问"设备当前状态，某某地方的天气和社会新闻"，你要先说设备状态，再说天气情况，最后说新闻内容。
 - **重要例外（无需调用）：**
   - `查询"现在的时间"、"今天的日期/星期几"、"今天农历"、"{{local_address}}的天气/未来天气"` -> **直接使用`<context>`信息回复**。
 - **需要调用的情况（示例）：**

main/xiaozhi-server/app.py

@@ -9,6 +9,7 @@ from core.utils.util import get_local_ip, validate_mcp_endpoint
 from core.http_server import SimpleHttpServer
 from core.websocket_server import WebSocketServer
 from core.utils.util import check_ffmpeg_installed
+from core.utils.gc_manager import get_gc_manager
 
 TAG = __name__
 logger = setup_logging()
@@ -63,6 +64,10 @@ async def main():
     # 添加 stdin 监控任务
     stdin_task = asyncio.create_task(monitor_stdin())
 
+    # 启动全局GC管理器（5分钟清理一次）
+    gc_manager = get_gc_manager(interval_seconds=300)
+    await gc_manager.start()
+
     # 启动 WebSocket 服务器
     ws_server = WebSocketServer(config)
     ws_task = asyncio.create_task(ws_server.start())
@@ -122,6 +127,9 @@ async def main():
     except asyncio.CancelledError:
         print("任务被取消，清理资源中...")
     finally:
+        # 停止全局GC管理器
+        await gc_manager.stop()
+
         # 取消所有任务（关键修复点）
         stdin_task.cancel()
         ws_task.cancel()

main/xiaozhi-server/config.yaml

@@ -487,7 +487,6 @@ LLM:
     temperature: 0.7  # 温度值
     max_tokens: 500   # 最大生成token数
     top_p: 1
-    top_k: 50
     frequency_penalty: 0  # 频率惩罚
   AliAppLLM:
     # 定义LLM API类型

main/xiaozhi-server/config/config_loader.py

@@ -32,7 +32,16 @@ def load_config():
     custom_config = read_config(custom_config_path)
 
     if custom_config.get("manager-api", {}).get("url"):
-        config = get_config_from_api(custom_config)
+        import asyncio
+        try:
+            loop = asyncio.get_running_loop()
+            # 如果已经在事件循环中，使用异步版本
+            config = asyncio.run_coroutine_threadsafe(
+                get_config_from_api_async(custom_config), loop
+            ).result()
+        except RuntimeError:
+            # 如果不在事件循环中（启动时），创建新的事件循环
+            config = asyncio.run(get_config_from_api_async(custom_config))
     else:
         # 合并配置
         config = merge_configs(default_config, custom_config)
@@ -44,13 +53,13 @@ def load_config():
     return config
 
 
-def get_config_from_api(config):
-    """从Java API获取配置"""
+async def get_config_from_api_async(config):
+    """从Java API获取配置（异步版本）"""
     # 初始化API客户端
     init_service(config)
 
     # 获取服务器配置
-    config_data = get_server_config()
+    config_data = await get_server_config()
     if config_data is None:
         raise Exception("Failed to fetch server config from API")
 
@@ -74,9 +83,9 @@ def get_config_from_api(config):
     return config_data
 
 
-def get_private_config_from_api(config, device_id, client_id):
+async def get_private_config_from_api(config, device_id, client_id):
     """从Java API获取私有配置"""
-    return get_agent_models(device_id, client_id, config["selected_module"])
+    return await get_agent_models(device_id, client_id, config["selected_module"])
 
 
 def ensure_directories(config):

main/xiaozhi-server/config/logger.py

@@ -5,7 +5,7 @@ from config.config_loader import load_config
 from config.settings import check_config_file
 from datetime import datetime
 
-SERVER_VERSION = "0.8.8"
+SERVER_VERSION = "0.8.9"
 _logger_initialized = False
 
 

main/xiaozhi-server/config/manage_api_client.py

@@ -1,5 +1,4 @@
 import os
-import time
 import base64
 from typing import Optional, Dict
 
@@ -20,7 +19,7 @@ class DeviceBindException(Exception):
 
 class ManageApiClient:
     _instance = None
-    _client = None
+    _async_clients = {}  # 为每个事件循环存储独立的客户端
     _secret = None
 
     def __new__(cls, config):
@@ -32,7 +31,7 @@ class ManageApiClient:
 
     @classmethod
     def _init_client(cls, config):
-        """初始化持久化连接池"""
+        """初始化配置（延迟创建客户端）"""
         cls.config = config.get("manager-api")
 
         if not cls.config:
@@ -47,23 +46,40 @@ class ManageApiClient:
         cls._secret = cls.config.get("secret")
         cls.max_retries = cls.config.get("max_retries", 6)  # 最大重试次数
         cls.retry_delay = cls.config.get("retry_delay", 10)  # 初始重试延迟(秒)
-        # NOTE(goody): 2025/4/16 http相关资源统一管理，后续可以增加线程池或者超时
-        # 后续也可以统一配置apiToken之类的走通用的Auth
-        cls._client = httpx.Client(
-            base_url=cls.config.get("url"),
-            headers={
-                "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})",
-                "Accept": "application/json",
-                "Authorization": "Bearer " + cls._secret,
-            },
-            timeout=cls.config.get("timeout", 30),  # 默认超时时间30秒
-        )
+        # 不在这里创建 AsyncClient，延迟到实际使用时创建
+        cls._async_clients = {}
 
     @classmethod
-    def _request(cls, method: str, endpoint: str, **kwargs) -> Dict:
-        """发送单次HTTP请求并处理响应"""
+    async def _ensure_async_client(cls):
+        """确保异步客户端已创建（为每个事件循环创建独立的客户端）"""
+        import asyncio
+        try:
+            loop = asyncio.get_running_loop()
+            loop_id = id(loop)
+
+            # 为每个事件循环创建独立的客户端
+            if loop_id not in cls._async_clients:
+                cls._async_clients[loop_id] = httpx.AsyncClient(
+                    base_url=cls.config.get("url"),
+                    headers={
+                        "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})",
+                        "Accept": "application/json",
+                        "Authorization": "Bearer " + cls._secret,
+                    },
+                    timeout=cls.config.get("timeout", 30),
+                )
+            return cls._async_clients[loop_id]
+        except RuntimeError:
+            # 如果没有运行中的事件循环，创建一个临时的
+            raise Exception("必须在异步上下文中调用")
+
+    @classmethod
+    async def _async_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """发送单次异步HTTP请求并处理响应"""
+        # 确保客户端已创建
+        client = await cls._ensure_async_client()
         endpoint = endpoint.lstrip("/")
-        response = cls._client.request(method, endpoint, **kwargs)
+        response = await client.request(method, endpoint, **kwargs)
         response.raise_for_status()
 
         result = response.json()
@@ -96,22 +112,23 @@ class ManageApiClient:
         return False
 
     @classmethod
-    def _execute_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
-        """带重试机制的请求执行器"""
+    async def _execute_async_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """带重试机制的异步请求执行器"""
+        import asyncio
         retry_count = 0
 
         while retry_count <= cls.max_retries:
             try:
-                # 执行请求
-                return cls._request(method, endpoint, **kwargs)
+                # 执行异步请求
+                return await cls._async_request(method, endpoint, **kwargs)
             except Exception as e:
                 # 判断是否应该重试
                 if retry_count < cls.max_retries and cls._should_retry(e):
                     retry_count += 1
                     print(
-                        f"{method} {endpoint} 请求失败，将在 {cls.retry_delay:.1f} 秒后进行第 {retry_count} 次重试"
+                        f"{method} {endpoint} 异步请求失败，将在 {cls.retry_delay:.1f} 秒后进行第 {retry_count} 次重试"
                     )
-                    time.sleep(cls.retry_delay)
+                    await asyncio.sleep(cls.retry_delay)
                     continue
                 else:
                     # 不重试，直接抛出异常
@@ -119,22 +136,27 @@ class ManageApiClient:
 
     @classmethod
     def safe_close(cls):
-        """安全关闭连接池"""
-        if cls._client:
-            cls._client.close()
-            cls._instance = None
+        """安全关闭所有异步连接池"""
+        import asyncio
+        for client in list(cls._async_clients.values()):
+            try:
+                asyncio.run(client.aclose())
+            except Exception:
+                pass
+        cls._async_clients.clear()
+        cls._instance = None
 
 
-def get_server_config() -> Optional[Dict]:
+async def get_server_config() -> Optional[Dict]:
     """获取服务器基础配置"""
-    return ManageApiClient._instance._execute_request("POST", "/config/server-base")
+    return await ManageApiClient._instance._execute_async_request("POST", "/config/server-base")
 
 
-def get_agent_models(
+async def get_agent_models(
     mac_address: str, client_id: str, selected_module: Dict
 ) -> Optional[Dict]:
     """获取代理模型配置"""
-    return ManageApiClient._instance._execute_request(
+    return await ManageApiClient._instance._execute_async_request(
         "POST",
         "/config/agent-models",
         json={
@@ -145,9 +167,9 @@ def get_agent_models(
     )
 
 
-def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
+async def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
     try:
-        return ManageApiClient._instance._execute_request(
+        return await ManageApiClient._instance._execute_async_request(
             "PUT",
             f"/agent/saveMemory/" + mac_address,
             json={
@@ -159,14 +181,14 @@ def save_mem_local_short(mac_address: str, short_momery: str) -> Optional[Dict]:
         return None
 
 
-def report(
+async def report(
     mac_address: str, session_id: str, chat_type: int, content: str, audio, report_time
 ) -> Optional[Dict]:
-    """带熔断的业务方法示例"""
+    """异步聊天记录上报"""
     if not content or not ManageApiClient._instance:
         return None
     try:
-        return ManageApiClient._instance._execute_request(
+        return await ManageApiClient._instance._execute_async_request(
             "POST",
             f"/agent/chat-history/report",
             json={

main/xiaozhi-server/core/api/vision_handler.py

@@ -96,7 +96,7 @@ class VisionHandler:
             current_config = copy.deepcopy(self.config)
             read_config_from_api = current_config.get("read_config_from_api", False)
             if read_config_from_api:
-                current_config = get_private_config_from_api(
+                current_config = await get_private_config_from_api(
                     current_config,
                     device_id,
                     client_id,

main/xiaozhi-server/core/connection.py

@@ -68,8 +68,11 @@ class ConnectionHandler:
         self.logger = setup_logging()
         self.server = server  # 保存server实例的引用
 
-        self.need_bind = False
-        self.bind_code = None
+        self.need_bind = False  # 是否需要绑定设备
+        self.bind_code = None  # 绑定设备的验证码
+        self.last_bind_prompt_time = 0  # 上次播放绑定提示的时间戳(秒)
+        self.bind_prompt_interval = 60  # 绑定提示播放间隔(秒)
+
         self.read_config_from_api = self.config.get("read_config_from_api", False)
 
         self.websocket = None
@@ -88,7 +91,7 @@ class ConnectionHandler:
         self.client_listen_mode = "auto"
 
         # 线程任务相关
-        self.loop = asyncio.get_event_loop()
+        self.loop = None  # 在 handle_connection 中获取运行中的事件循环
         self.stop_event = threading.Event()
         self.executor = ThreadPoolExecutor(max_workers=5)
 
@@ -116,6 +119,7 @@ class ConnectionHandler:
         self.client_audio_buffer = bytearray()
         self.client_have_voice = False
         self.client_voice_window = deque(maxlen=5)
+        self.first_activity_time = 0.0  # 记录首次活动的时间（毫秒）
         self.last_activity_time = 0.0  # 统一的活动时间戳（毫秒）
         self.client_voice_stop = False
         self.last_is_voice = False
@@ -162,6 +166,9 @@ class ConnectionHandler:
 
     async def handle_connection(self, ws):
         try:
+            # 获取运行中的事件循环（必须在异步上下文中）
+            self.loop = asyncio.get_running_loop()
+
             # 获取并验证headers
             self.headers = dict(ws.request.headers)
             real_ip = self.headers.get("x-real-ip") or self.headers.get(
@@ -187,6 +194,7 @@ class ConnectionHandler:
                 self.logger.bind(tag=TAG).info("连接来自:MQTT网关")
 
             # 初始化活动时间戳
+            self.first_activity_time = time.time() * 1000
             self.last_activity_time = time.time() * 1000
 
             # 启动超时检查任务
@@ -195,10 +203,8 @@ class ConnectionHandler:
             self.welcome_msg = self.config["xiaozhi"]
             self.welcome_msg["session_id"] = self.session_id
 
-            # 获取差异化配置
-            self._initialize_private_config()
-            # 异步初始化
-            self.executor.submit(self._initialize_components)
+            # 在后台初始化配置和组件（完全不阻塞主循环）
+            asyncio.create_task(self._background_initialize())
 
             try:
                 async for message in self.websocket:
@@ -268,6 +274,22 @@ class ConnectionHandler:
             if self.vad is None or self.asr is None:
                 return
 
+            # 未绑定设备直接丢弃所有音频，不进行ASR处理
+            if self.need_bind:
+                current_time = time.time()
+                # 检查是否需要播放绑定提示
+                if (
+                    current_time - self.last_bind_prompt_time
+                    >= self.bind_prompt_interval
+                ):
+                    self.last_bind_prompt_time = current_time
+                    # 复用现有的绑定提示逻辑
+                    from core.handle.receiveAudioHandle import check_bind_device
+
+                    asyncio.create_task(check_bind_device(self))
+                # 直接丢弃音频，不进行ASR处理
+                return
+
             # 处理来自MQTT网关的音频包
             if self.conn_from_mqtt_gateway and len(message) >= 16:
                 handled = await self._process_mqtt_audio_message(message)
@@ -501,21 +523,30 @@ class ConnectionHandler:
         except Exception as e:
             self.logger.bind(tag=TAG).warning(f"声纹识别初始化失败: {str(e)}")
 
-    def _initialize_private_config(self):
-        """如果是从配置文件获取，则进行二次实例化"""
+    async def _background_initialize(self):
+        """在后台初始化配置和组件（完全不阻塞主循环）"""
+        try:
+            # 异步获取差异化配置
+            await self._initialize_private_config_async()
+            # 在线程池中初始化组件
+            self.executor.submit(self._initialize_components)
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"后台初始化失败: {e}")
+
+    async def _initialize_private_config_async(self):
+        """从接口异步获取差异化配置（异步版本，不阻塞主循环）"""
         if not self.read_config_from_api:
             return
-        """从接口获取差异化的配置进行二次实例化，非全量重新实例化"""
         try:
             begin_time = time.time()
-            private_config = get_private_config_from_api(
+            private_config = await get_private_config_from_api(
                 self.config,
                 self.headers.get("device-id"),
                 self.headers.get("client-id", self.headers.get("device-id")),
             )
             private_config["delete_audio"] = bool(self.config.get("delete_audio", True))
             self.logger.bind(tag=TAG).info(
-                f"{time.time() - begin_time} 秒，获取差异化配置成功: {json.dumps(filter_sensitive_info(private_config), ensure_ascii=False)}"
+                f"{time.time() - begin_time} 秒，异步获取差异化配置成功: {json.dumps(filter_sensitive_info(private_config), ensure_ascii=False)}"
             )
         except DeviceNotFoundException as e:
             self.need_bind = True
@@ -526,7 +557,7 @@ class ConnectionHandler:
             private_config = {}
         except Exception as e:
             self.need_bind = True
-            self.logger.bind(tag=TAG).error(f"获取差异化配置失败: {e}")
+            self.logger.bind(tag=TAG).error(f"异步获取差异化配置失败: {e}")
             private_config = {}
 
         init_llm, init_tts, init_memory, init_intent = (
@@ -599,8 +630,12 @@ class ConnectionHandler:
             self.chat_history_conf = int(private_config["chat_history_conf"])
         if private_config.get("mcp_endpoint", None) is not None:
             self.config["mcp_endpoint"] = private_config["mcp_endpoint"]
+
+        # 使用 run_in_executor 在线程池中执行 initialize_modules，避免阻塞主循环
         try:
-            modules = initialize_modules(
+            modules = await self.loop.run_in_executor(
+                None,  # 使用默认线程池
+                initialize_modules,
                 self.logger,
                 private_config,
                 init_vad,
@@ -723,11 +758,12 @@ class ConnectionHandler:
         self.dialogue.update_system_message(self.prompt)
 
     def chat(self, query, depth=0):
-        self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
-        self.llm_finish_task = False
+        if query is not None:
+            self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}")
 
         # 为最顶层时新建会话ID和发送FIRST请求
         if depth == 0:
+            self.llm_finish_task = False
             self.sentence_id = str(uuid.uuid4().hex)
             self.dialogue.put(Message(role="user", content=query))
             self.tts.tts_text_queue.put(
@@ -738,9 +774,31 @@ class ConnectionHandler:
                 )
             )
 
+        # 设置最大递归深度，避免无限循环，可根据实际需求调整
+        MAX_DEPTH = 5
+        force_final_answer = False  # 标记是否强制最终回答
+
+        if depth >= MAX_DEPTH:
+            self.logger.bind(tag=TAG).debug(
+                f"已达到最大工具调用深度 {MAX_DEPTH}，将强制基于现有信息回答"
+            )
+            force_final_answer = True
+            # 添加系统指令，要求 LLM 基于现有信息回答
+            self.dialogue.put(
+                Message(
+                    role="user",
+                    content="[系统提示] 已达到最大工具调用次数限制，请你基于目前已经获取的所有信息，直接给出最终答案。不要再尝试调用任何工具。",
+                )
+            )
+
         # Define intent functions
         functions = None
-        if self.intent_type == "function_call" and hasattr(self, "func_handler"):
+        # 达到最大深度时，禁用工具调用，强制 LLM 直接回答
+        if (
+            self.intent_type == "function_call"
+            and hasattr(self, "func_handler")
+            and not force_final_answer
+        ):
             functions = self.func_handler.get_functions()
         response_message = []
 
@@ -775,9 +833,8 @@ class ConnectionHandler:
 
         # 处理流式响应
         tool_call_flag = False
-        function_name = None
-        function_id = None
-        function_arguments = ""
+        # 支持多个并行工具调用 - 使用列表存储
+        tool_calls_list = []  # 格式: [{"id": "", "name": "", "arguments": ""}]
         content_arguments = ""
         self.client_abort = False
         emotion_flag = True
@@ -798,12 +855,7 @@ class ConnectionHandler:
 
                 if tools_call is not None and len(tools_call) > 0:
                     tool_call_flag = True
-                    if tools_call[0].id is not None and tools_call[0].id != "":
-                        function_id = tools_call[0].id
-                    if tools_call[0].function.name is not None and tools_call[0].function.name != "":
-                        function_name = tools_call[0].function.name
-                    if tools_call[0].function.arguments is not None:
-                        function_arguments += tools_call[0].function.arguments
+                    self._merge_tool_calls(tool_calls_list, tools_call)
             else:
                 content = response
 
@@ -829,16 +881,22 @@ class ConnectionHandler:
         # 处理function call
         if tool_call_flag:
             bHasError = False
-            if function_id is None:
+            # 处理基于文本的工具调用格式
+            if len(tool_calls_list) == 0 and content_arguments:
                 a = extract_json_from_string(content_arguments)
                 if a is not None:
                     try:
                         content_arguments_json = json.loads(a)
-                        function_name = content_arguments_json["name"]
-                        function_arguments = json.dumps(
-                            content_arguments_json["arguments"], ensure_ascii=False
+                        tool_calls_list.append(
+                            {
+                                "id": str(uuid.uuid4().hex),
+                                "name": content_arguments_json["name"],
+                                "arguments": json.dumps(
+                                    content_arguments_json["arguments"],
+                                    ensure_ascii=False,
+                                ),
+                            }
                         )
-                        function_id = str(uuid.uuid4().hex)
                     except Exception as e:
                         bHasError = True
                         response_message.append(a)
@@ -849,30 +907,43 @@ class ConnectionHandler:
                     self.logger.bind(tag=TAG).error(
                         f"function call error: {content_arguments}"
                     )
-            if not bHasError:
+
+            if not bHasError and len(tool_calls_list) > 0:
                 # 如需要大模型先处理一轮，添加相关处理后的日志情况
                 if len(response_message) > 0:
                     text_buff = "".join(response_message)
                     self.tts_MessageText = text_buff
                     self.dialogue.put(Message(role="assistant", content=text_buff))
                 response_message.clear()
-                self.logger.bind(tag=TAG).debug(
-                    f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}"
-                )
-                function_call_data = {
-                    "name": function_name,
-                    "id": function_id,
-                    "arguments": function_arguments,
-                }
 
-                # 使用统一工具处理器处理所有工具调用
-                result = asyncio.run_coroutine_threadsafe(
-                    self.func_handler.handle_llm_function_call(
-                        self, function_call_data
-                    ),
-                    self.loop,
-                ).result()
-                self._handle_function_result(result, function_call_data, depth=depth)
+                self.logger.bind(tag=TAG).debug(
+                    f"检测到 {len(tool_calls_list)} 个工具调用"
+                )
+
+                # 收集所有工具调用的 Future
+                futures_with_data = []
+                for tool_call_data in tool_calls_list:
+                    self.logger.bind(tag=TAG).debug(
+                        f"function_name={tool_call_data['name']}, function_id={tool_call_data['id']}, function_arguments={tool_call_data['arguments']}"
+                    )
+
+                    future = asyncio.run_coroutine_threadsafe(
+                        self.func_handler.handle_llm_function_call(
+                            self, tool_call_data
+                        ),
+                        self.loop,
+                    )
+                    futures_with_data.append((future, tool_call_data))
+
+                # 等待协程结束（实际等待时长为最慢的那个）
+                tool_results = []
+                for future, tool_call_data in futures_with_data:
+                    result = future.result()
+                    tool_results.append((result, tool_call_data))
+
+                # 统一处理所有工具调用结果
+                if tool_results:
+                    self._handle_function_result(tool_results, depth=depth)
 
         # 存储对话内容
         if len(response_message) > 0:
@@ -887,64 +958,69 @@ class ConnectionHandler:
                     content_type=ContentType.ACTION,
                 )
             )
-        self.llm_finish_task = True
-        # 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
-        self.logger.bind(tag=TAG).debug(
-            lambda: json.dumps(
-                self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
+            self.llm_finish_task = True
+            # 使用lambda延迟计算，只有在DEBUG级别时才执行get_llm_dialogue()
+            self.logger.bind(tag=TAG).debug(
+                lambda: json.dumps(
+                    self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False
+                )
             )
-        )
 
         return True
 
-    def _handle_function_result(self, result, function_call_data, depth):
-        if result.action == Action.RESPONSE:  # 直接回复前端
-            text = result.response
-            self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
-            self.dialogue.put(Message(role="assistant", content=text))
-        elif result.action == Action.REQLLM:  # 调用函数后再请求llm生成回复
-            text = result.result
-            if text is not None and len(text) > 0:
-                function_id = function_call_data["id"]
-                function_name = function_call_data["name"]
-                function_arguments = function_call_data["arguments"]
-                self.dialogue.put(
-                    Message(
-                        role="assistant",
-                        tool_calls=[
-                            {
-                                "id": function_id,
-                                "function": {
-                                    "arguments": (
-                                        "{}"
-                                        if function_arguments == ""
-                                        else function_arguments
-                                    ),
-                                    "name": function_name,
-                                },
-                                "type": "function",
-                                "index": 0,
-                            }
-                        ],
-                    )
-                )
+    def _handle_function_result(self, tool_results, depth):
+        need_llm_tools = []
 
-                self.dialogue.put(
-                    Message(
-                        role="tool",
-                        tool_call_id=(
-                            str(uuid.uuid4()) if function_id is None else function_id
+        for result, tool_call_data in tool_results:
+            if result.action in [
+                Action.RESPONSE,
+                Action.NOTFOUND,
+                Action.ERROR,
+            ]:  # 直接回复前端
+                text = result.response if result.response else result.result
+                self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
+                self.dialogue.put(Message(role="assistant", content=text))
+            elif result.action == Action.REQLLM:
+                # 收集需要 LLM 处理的工具
+                need_llm_tools.append((result, tool_call_data))
+            else:
+                pass
+
+        if need_llm_tools:
+            all_tool_calls = [
+                {
+                    "id": tool_call_data["id"],
+                    "function": {
+                        "arguments": (
+                            "{}"
+                            if tool_call_data["arguments"] == ""
+                            else tool_call_data["arguments"]
                         ),
-                        content=text,
+                        "name": tool_call_data["name"],
+                    },
+                    "type": "function",
+                    "index": idx,
+                }
+                for idx, (_, tool_call_data) in enumerate(need_llm_tools)
+            ]
+            self.dialogue.put(Message(role="assistant", tool_calls=all_tool_calls))
+
+            for result, tool_call_data in need_llm_tools:
+                text = result.result
+                if text is not None and len(text) > 0:
+                    self.dialogue.put(
+                        Message(
+                            role="tool",
+                            tool_call_id=(
+                                str(uuid.uuid4())
+                                if tool_call_data["id"] is None
+                                else tool_call_data["id"]
+                            ),
+                            content=text,
+                        )
                     )
-                )
-                self.chat(text, depth=depth + 1)
-        elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
-            text = result.response if result.response else result.result
-            self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text)
-            self.dialogue.put(Message(role="assistant", content=text))
-        else:
-            pass
+
+            self.chat(None, depth=depth + 1)
 
     def _report_worker(self):
         """聊天记录上报工作线程"""
@@ -972,8 +1048,8 @@ class ConnectionHandler:
     def _process_report(self, type, text, audio_data, report_time):
         """处理上报任务"""
         try:
-            # 执行上报（传入二进制数据）
-            report(self, type, text, audio_data, report_time)
+            # 执行异步上报（在事件循环中运行）
+            asyncio.run(report(self, type, text, audio_data, report_time))
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"上报处理异常: {e}")
         finally:
@@ -1064,7 +1140,6 @@ class ConnectionHandler:
                         f"关闭线程池时出错: {executor_error}"
                     )
                 self.executor = None
-
             self.logger.bind(tag=TAG).info("连接资源已释放")
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"关闭连接时出错: {e}")
@@ -1119,13 +1194,14 @@ class ConnectionHandler:
         """检查连接超时"""
         try:
             while not self.stop_event.is_set():
+                last_activity_time = self.last_activity_time
+                if self.need_bind:
+                    last_activity_time = self.first_activity_time
+
                 # 检查是否超时（只有在时间戳已初始化的情况下）
-                if self.last_activity_time > 0.0:
+                if last_activity_time > 0.0:
                     current_time = time.time() * 1000
-                    if (
-                        current_time - self.last_activity_time
-                        > self.timeout_seconds * 1000
-                    ):
+                    if current_time - last_activity_time > self.timeout_seconds * 1000:
                         if not self.stop_event.is_set():
                             self.logger.bind(tag=TAG).info("连接超时，准备关闭")
                             # 设置停止事件，防止重复处理
@@ -1144,3 +1220,31 @@ class ConnectionHandler:
             self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")
         finally:
             self.logger.bind(tag=TAG).info("超时检查任务已退出")
+
+    def _merge_tool_calls(self, tool_calls_list, tools_call):
+        """合并工具调用列表
+
+        Args:
+            tool_calls_list: 已收集的工具调用列表
+            tools_call: 新的工具调用
+        """
+        for tool_call in tools_call:
+            tool_index = getattr(tool_call, "index", None)
+            if tool_index is None:
+                if tool_call.function.name:
+                    # 有 function_name，说明是新的工具调用
+                    tool_index = len(tool_calls_list)
+                else:
+                    tool_index = len(tool_calls_list) - 1 if tool_calls_list else 0
+
+            # 确保列表有足够的位置
+            if tool_index >= len(tool_calls_list):
+                tool_calls_list.append({"id": "", "name": "", "arguments": ""})
+
+            # 更新工具调用信息
+            if tool_call.id:
+                tool_calls_list[tool_index]["id"] = tool_call.id
+            if tool_call.function.name:
+                tool_calls_list[tool_index]["name"] = tool_call.function.name
+            if tool_call.function.arguments:
+                tool_calls_list[tool_index]["arguments"] += tool_call.function.arguments

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

@@ -10,7 +10,7 @@ TTS上报功能已集成到ConnectionHandler类中。
 """
 
 import time
-
+import gc
 import opuslib_next
 
 from config.manage_api_client import report as manage_report
@@ -18,7 +18,7 @@ from config.manage_api_client import report as manage_report
 TAG = __name__
 
 
-def report(conn, type, text, opus_data, report_time):
+async def report(conn, type, text, opus_data, report_time):
     """执行聊天记录上报操作
 
     Args:
@@ -33,8 +33,8 @@ def report(conn, type, text, opus_data, report_time):
             audio_data = opus_to_wav(conn, opus_data)
         else:
             audio_data = None
-        # 执行上报
-        manage_report(
+        # 执行异步上报
+        await manage_report(
             mac_address=conn.device_id,
             session_id=conn.session_id,
             chat_type=type,
@@ -56,41 +56,49 @@ def opus_to_wav(conn, opus_data):
     Returns:
         bytes: WAV格式的音频数据
     """
-    decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
-    pcm_data = []
+    decoder = None
+    try:
+        decoder = opuslib_next.Decoder(16000, 1)  # 16kHz, 单声道
+        pcm_data = []
 
-    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:
-            conn.logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
+        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:
+                conn.logger.bind(tag=TAG).error(f"Opus解码错误: {e}", exc_info=True)
 
-    if not pcm_data:
-        raise ValueError("没有有效的PCM数据")
+        if not pcm_data:
+            raise ValueError("没有有效的PCM数据")
 
-    # 创建WAV文件头
-    pcm_data_bytes = b"".join(pcm_data)
-    num_samples = len(pcm_data_bytes) // 2  # 16-bit samples
+        # 创建WAV文件头
+        pcm_data_bytes = b"".join(pcm_data)
+        num_samples = len(pcm_data_bytes) // 2  # 16-bit samples
 
-    # WAV文件头
-    wav_header = bytearray()
-    wav_header.extend(b"RIFF")  # ChunkID
-    wav_header.extend((36 + len(pcm_data_bytes)).to_bytes(4, "little"))  # ChunkSize
-    wav_header.extend(b"WAVE")  # Format
-    wav_header.extend(b"fmt ")  # Subchunk1ID
-    wav_header.extend((16).to_bytes(4, "little"))  # Subchunk1Size
-    wav_header.extend((1).to_bytes(2, "little"))  # AudioFormat (PCM)
-    wav_header.extend((1).to_bytes(2, "little"))  # NumChannels
-    wav_header.extend((16000).to_bytes(4, "little"))  # SampleRate
-    wav_header.extend((32000).to_bytes(4, "little"))  # ByteRate
-    wav_header.extend((2).to_bytes(2, "little"))  # BlockAlign
-    wav_header.extend((16).to_bytes(2, "little"))  # BitsPerSample
-    wav_header.extend(b"data")  # Subchunk2ID
-    wav_header.extend(len(pcm_data_bytes).to_bytes(4, "little"))  # Subchunk2Size
+        # WAV文件头
+        wav_header = bytearray()
+        wav_header.extend(b"RIFF")  # ChunkID
+        wav_header.extend((36 + len(pcm_data_bytes)).to_bytes(4, "little"))  # ChunkSize
+        wav_header.extend(b"WAVE")  # Format
+        wav_header.extend(b"fmt ")  # Subchunk1ID
+        wav_header.extend((16).to_bytes(4, "little"))  # Subchunk1Size
+        wav_header.extend((1).to_bytes(2, "little"))  # AudioFormat (PCM)
+        wav_header.extend((1).to_bytes(2, "little"))  # NumChannels
+        wav_header.extend((16000).to_bytes(4, "little"))  # SampleRate
+        wav_header.extend((32000).to_bytes(4, "little"))  # ByteRate
+        wav_header.extend((2).to_bytes(2, "little"))  # BlockAlign
+        wav_header.extend((16).to_bytes(2, "little"))  # BitsPerSample
+        wav_header.extend(b"data")  # Subchunk2ID
+        wav_header.extend(len(pcm_data_bytes).to_bytes(4, "little"))  # Subchunk2Size
 
-    # 返回完整的WAV数据
-    return bytes(wav_header) + pcm_data_bytes
+        # 返回完整的WAV数据
+        return bytes(wav_header) + pcm_data_bytes
+    finally:
+        if decoder is not None:
+            try:
+                del decoder
+            except Exception as e:
+                conn.logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")
 
 
 def enqueue_tts_report(conn, text, opus_data):

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

@@ -4,6 +4,7 @@ import asyncio
 from core.utils import textUtils
 from core.utils.util import audio_to_data
 from core.providers.tts.dto.dto import SentenceType
+from core.utils.audioRateController import AudioRateController
 
 TAG = __name__
 
@@ -23,38 +24,13 @@ async def sendAudioMessage(conn, sentenceType, audios, text):
         conn.logger.bind(tag=TAG).info(f"发送音频消息: {sentenceType}, {text}")
 
     # 发送结束消息（如果是最后一个文本）
-    if conn.llm_finish_task and sentenceType == SentenceType.LAST:
+    if sentenceType == SentenceType.LAST:
         await send_tts_message(conn, "stop", None)
         conn.client_is_speaking = False
         if conn.close_after_chat:
             await conn.close()
 
 
-def calculate_timestamp_and_sequence(conn, start_time, packet_index, frame_duration=60):
-    """
-    计算音频数据包的时间戳和序列号
-    Args:
-        conn: 连接对象
-        start_time: 起始时间（性能计数器值）
-        packet_index: 数据包索引
-        frame_duration: 帧时长（毫秒），匹配 Opus 编码
-    Returns:
-        tuple: (timestamp, sequence)
-    """
-    # 计算时间戳（使用播放位置计算）
-    timestamp = int((start_time + packet_index * frame_duration / 1000) * 1000) % (
-        2**32
-    )
-
-    # 计算序列号
-    if hasattr(conn, "audio_flow_control"):
-        sequence = conn.audio_flow_control["sequence"]
-    else:
-        sequence = packet_index  # 如果没有流控状态，直接使用索引
-
-    return timestamp, sequence
-
-
 async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     """
     发送带16字节头部的opus数据包给mqtt_gateway
@@ -77,15 +53,20 @@ async def _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence):
     await conn.websocket.send(complete_packet)
 
 
-# 播放音频
+# 播放音频 - 使用 AudioRateController 进行精确流控
 async def sendAudio(conn, audios, frame_duration=60):
     """
-    发送单个opus包，支持流控
+    发送音频包，使用 AudioRateController 进行精确的流量控制
+
     Args:
         conn: 连接对象
-        opus_packet: 单个opus数据包
-        pre_buffer: 快速发送音频
-        frame_duration: 帧时长（毫秒），匹配 Opus 编码
+        audios: 单个opus包(bytes) 或 opus包列表
+        frame_duration: 帧时长（毫秒），默认60ms
+
+    改进点：
+    1. 使用单一时间基准，避免累积误差
+    2. 每次检查队列时重新计算 elapsed_ms，更精准
+    3. 支持高并发而不产生时间偏差
     """
     if audios is None or len(audios) == 0:
         return
@@ -94,114 +75,133 @@ async def sendAudio(conn, audios, frame_duration=60):
     send_delay = conn.config.get("tts_audio_send_delay", -1) / 1000.0
 
     if isinstance(audios, bytes):
-        # 重置流控状态,第一次读取和会话发生转变时
-        if not hasattr(conn, "audio_flow_control") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id:
-            conn.audio_flow_control = {
-                "last_send_time": 0,
-                "packet_count": 0,
-                "start_time": time.perf_counter(),
-                "sequence": 0,  # 添加序列号
-                "sentence_id": conn.sentence_id,
-            }
+        # 单个 opus 包处理
+        await _sendAudio_single(conn, audios, send_delay, frame_duration)
+    else:
+        # 音频列表处理（如文件型音频）
+        await _sendAudio_list(conn, audios, send_delay, frame_duration)
 
+
+async def _sendAudio_single(conn, opus_packet, send_delay, frame_duration=60):
+    """
+    发送单个 opus 包
+    使用 AudioRateController 进行流控
+    """
+    # 重置流控状态，第一次读取和会话发生转变时
+    if not hasattr(conn, "audio_rate_controller") or conn.audio_flow_control.get("sentence_id") != conn.sentence_id:
+        if hasattr(conn, "audio_rate_controller"):
+            conn.audio_rate_controller.reset()
+        else:
+            conn.audio_rate_controller = AudioRateController(frame_duration)
+            conn.audio_rate_controller.reset()
+
+        conn.audio_flow_control = {
+            "packet_count": 0,
+            "sequence": 0,
+            "sentence_id": conn.sentence_id,
+        }
+
+    if conn.client_abort:
+        return
+
+    conn.last_activity_time = time.time() * 1000
+
+    rate_controller = conn.audio_rate_controller
+    flow_control = conn.audio_flow_control
+    packet_count = flow_control["packet_count"]
+
+    # 预缓冲：前5个包直接发送，不做延迟
+    pre_buffer_count = 5
+
+    if packet_count < pre_buffer_count or send_delay > 0:
+        # 预缓冲阶段或固定延迟模式，直接发送
+        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
+
+        if send_delay > 0 and packet_count >= pre_buffer_count:
+            await asyncio.sleep(send_delay)
+    else:
+        # 使用流控器进行精确的速率控制
+        rate_controller.add_audio(opus_packet)
+
+        async def send_callback(packet):
+            await _do_send_audio(conn, packet, flow_control, frame_duration)
+
+        await rate_controller.check_queue(send_callback)
+
+    # 更新流控状态
+    flow_control["packet_count"] += 1
+    flow_control["sequence"] += 1
+
+
+async def _sendAudio_list(conn, audios, send_delay, frame_duration=60):
+    """
+    发送音频列表（如文件型音频）
+    """
+    if not audios:
+        return
+
+    rate_controller = AudioRateController(frame_duration)
+    rate_controller.reset()
+
+    flow_control = {
+        "packet_count": 0,
+        "sequence": 0,
+    }
+
+    # 预缓冲：前5个包直接发送
+    pre_buffer_frames = min(5, len(audios))
+    for i in range(pre_buffer_frames):
         if conn.client_abort:
             return
+        await _do_send_audio(conn, audios[i], flow_control, frame_duration)
+        conn.client_is_speaking = True
+
+    remaining_audios = audios[pre_buffer_frames:]
+
+    # 处理剩余音频帧
+    for i, opus_packet in enumerate(remaining_audios):
+        if conn.client_abort:
+            break
 
         conn.last_activity_time = time.time() * 1000
 
-        # 预缓冲：前5个包直接发送，不做延迟
-        pre_buffer_count = 5
-        flow_control = conn.audio_flow_control
-        current_time = time.perf_counter()
-
-        if flow_control["packet_count"] < pre_buffer_count:
-            # 预缓冲阶段，直接发送不延迟
-            pass
-        elif send_delay > 0:
-            # 使用固定延迟
+        if send_delay > 0:
+            # 固定延迟模式
             await asyncio.sleep(send_delay)
         else:
-            effective_packet = flow_control["packet_count"] - pre_buffer_count
-            expected_time = flow_control["start_time"] + (
-                effective_packet * frame_duration / 1000
-            )
-            delay = expected_time - current_time
-            if delay > 0:
-                await asyncio.sleep(delay)
-            else:
-                # 纠正误差
-                flow_control["start_time"] += abs(delay)
+            # 使用流控器进行精确延迟
+            rate_controller.add_audio(opus_packet)
 
-        if conn.conn_from_mqtt_gateway:
-            # 计算时间戳和序列号
-            timestamp, sequence = calculate_timestamp_and_sequence(
-                conn,
-                flow_control["start_time"],
-                flow_control["packet_count"],
-                frame_duration,
-            )
-            # 调用通用函数发送带头部的数据包
-            await _send_to_mqtt_gateway(conn, audios, timestamp, sequence)
-        else:
-            # 直接发送opus数据包，不添加头部
-            await conn.websocket.send(audios)
+            async def send_callback(packet):
+                await _do_send_audio(conn, packet, flow_control, frame_duration)
 
-        # 更新流控状态
-        flow_control["packet_count"] += 1
-        flow_control["sequence"] += 1
-        flow_control["last_send_time"] = time.perf_counter()
+            await rate_controller.check_queue(send_callback)
+            conn.client_is_speaking = True
+            continue
+
+        await _do_send_audio(conn, opus_packet, flow_control, frame_duration)
+        conn.client_is_speaking = True
+
+
+async def _do_send_audio(conn, opus_packet, flow_control, frame_duration=60):
+    """
+    执行实际的音频发送
+    """
+    packet_index = flow_control.get("packet_count", 0)
+    sequence = flow_control.get("sequence", 0)
+
+    if conn.conn_from_mqtt_gateway:
+        # 计算时间戳（基于播放位置）
+        start_time = time.time()
+        timestamp = int(start_time * 1000) % (2**32)
+        await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence)
     else:
-        # 文件型音频走普通播放
-        start_time = time.perf_counter()
-        play_position = 0
+        # 直接发送opus数据包
+        await conn.websocket.send(opus_packet)
 
-        # 执行预缓冲
-        pre_buffer_frames = min(5, len(audios))
-        for i in range(pre_buffer_frames):
-            if conn.conn_from_mqtt_gateway:
-                # 计算时间戳和序列号
-                timestamp, sequence = calculate_timestamp_and_sequence(
-                    conn, start_time, i, frame_duration
-                )
-                # 调用通用函数发送带头部的数据包
-                await _send_to_mqtt_gateway(conn, audios[i], timestamp, sequence)
-            else:
-                # 直接发送预缓冲包，不添加头部
-                await conn.websocket.send(audios[i])
-        remaining_audios = audios[pre_buffer_frames:]
-
-        # 播放剩余音频帧
-        for i, opus_packet in enumerate(remaining_audios):
-            if conn.client_abort:
-                break
-
-            # 重置没有声音的状态
-            conn.last_activity_time = time.time() * 1000
-
-            if send_delay > 0:
-                # 固定延迟模式
-                await asyncio.sleep(send_delay)
-            else:
-                 # 计算预期发送时间
-                expected_time = start_time + (play_position / 1000)
-                current_time = time.perf_counter()
-                delay = expected_time - current_time
-                if delay > 0:
-                    await asyncio.sleep(delay)
-
-            if conn.conn_from_mqtt_gateway:
-                # 计算时间戳和序列号（使用当前的数据包索引确保连续性）
-                packet_index = pre_buffer_frames + i
-                timestamp, sequence = calculate_timestamp_and_sequence(
-                    conn, start_time, packet_index, frame_duration
-                )
-                # 调用通用函数发送带头部的数据包
-                await _send_to_mqtt_gateway(conn, opus_packet, timestamp, sequence)
-            else:
-                # 直接发送opus数据包，不添加头部
-                await conn.websocket.send(opus_packet)
-
-            play_position += frame_duration
+    # 更新流控状态
+    flow_control["packet_count"] = packet_index + 1
+    flow_control["sequence"] = sequence + 1
 
 
 async def send_tts_message(conn, state, text=None):
@@ -255,5 +255,4 @@ async def send_stt_message(conn, text):
     await conn.websocket.send(
         json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id})
     )
-    conn.client_is_speaking = True
     await send_tts_message(conn, "start")

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

@@ -5,6 +5,7 @@ import hmac
 import base64
 import hashlib
 import asyncio
+import gc
 import requests
 import websockets
 import opuslib_next
@@ -347,4 +348,11 @@ class ASRProvider(ASRProviderBase):
 
     async def close(self):
         """关闭资源"""
-        await self._cleanup()
+        await self._cleanup(None)
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+                self.decoder = None
+                logger.bind(tag=TAG).debug("Aliyun decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Aliyun decoder资源时出错: {e}")

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

@@ -10,6 +10,7 @@ import traceback
 import threading
 import opuslib_next
 import concurrent.futures
+import gc
 from abc import ABC, abstractmethod
 from config.logger import setup_logging
 from typing import Optional, Tuple, List
@@ -68,7 +69,7 @@ class ASRProviderBase(ABC):
             conn.asr_audio.clear()
             conn.reset_vad_states()
 
-            if len(asr_audio_task) > 15 or conn.client_listen_mode == "manual":
+            if len(asr_audio_task) > 15:
                 await self.handle_voice_stop(conn, asr_audio_task)
 
     # 处理语音停止
@@ -241,6 +242,7 @@ class ASRProviderBase(ABC):
     @staticmethod
     def decode_opus(opus_data: List[bytes]) -> List[bytes]:
         """将Opus音频数据解码为PCM数据"""
+        decoder = None
         try:
             decoder = opuslib_next.Decoder(16000, 1)
             pcm_data = []
@@ -265,3 +267,9 @@ class ASRProviderBase(ABC):
         except Exception as e:
             logger.bind(tag=TAG).error(f"音频解码过程发生错误: {e}")
             return []
+        finally:
+            if decoder is not None:
+                try:
+                    del decoder
+                except Exception as e:
+                    logger.bind(tag=TAG).debug(f"释放decoder资源时出错: {e}")

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

@@ -4,6 +4,7 @@ import uuid
 import asyncio
 import websockets
 import opuslib_next
+import gc
 from core.providers.asr.base import ASRProviderBase
 from config.logger import setup_logging
 from core.providers.asr.dto.dto import InterfaceType
@@ -370,6 +371,16 @@ class ASRProvider(ASRProviderBase):
                 pass
             self.forward_task = None
         self.is_processing = False
+        
+        # 显式释放decoder资源
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+                self.decoder = None
+                logger.bind(tag=TAG).debug("Doubao decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Doubao decoder资源时出错: {e}")
+
         # 清理所有连接的音频缓冲区
         if hasattr(self, '_connections'):
             for conn in self._connections.values():

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

@@ -5,6 +5,7 @@ import hashlib
 import asyncio
 import websockets
 import opuslib_next
+import gc
 from time import mktime
 from datetime import datetime
 from urllib.parse import urlencode
@@ -512,6 +513,16 @@ class ASRProvider(ASRProviderBase):
                 pass
             self.forward_task = None
         self.is_processing = False
+        
+        # 显式释放decoder资源
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+                self.decoder = None
+                logger.bind(tag=TAG).debug("Xunfei decoder resources released")
+            except Exception as e:
+                logger.bind(tag=TAG).debug(f"释放Xunfei decoder资源时出错: {e}")
+
         # 清理所有连接的音频缓冲区
         if hasattr(self, "_connections"):
             for conn in self._connections.values():

main/xiaozhi-server/core/providers/memory/mem_local_short/mem_local_short.py

@@ -5,6 +5,7 @@ import os
 import yaml
 from config.config_loader import get_project_dir
 from config.manage_api_client import save_mem_local_short
+import asyncio
 from core.utils.util import check_model_key
 
 
@@ -193,7 +194,13 @@ class MemoryProvider(MemoryProviderBase):
                 max_tokens=2000,
                 temperature=0.2,
             )
-            save_mem_local_short(self.role_id, result)
+            # 使用异步版本，需要在事件循环中运行
+            try:
+                loop = asyncio.get_running_loop()
+                loop.create_task(save_mem_local_short(self.role_id, result))
+            except RuntimeError:
+                # 如果没有运行中的事件循环，创建一个新的
+                asyncio.run(save_mem_local_short(self.role_id, result))
         logger.bind(tag=TAG).info(f"Save memory successful - Role: {self.role_id}")
 
         return self.short_memory

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_client.py

@@ -10,10 +10,13 @@ import concurrent.futures
 from contextlib import AsyncExitStack
 from typing import Optional, List, Dict, Any
 
-from mcp import ClientSession, StdioServerParameters
+from mcp import ClientSession, StdioServerParameters, Implementation
+from mcp.client.session import SamplingFnT, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT
 from mcp.client.stdio import stdio_client
 from mcp.client.sse import sse_client
 from mcp.client.streamable_http import streamablehttp_client
+from mcp.shared.session import ProgressFnT
+
 from config.logger import setup_logging
 from core.utils.util import sanitize_tool_name
 
@@ -41,13 +44,25 @@ class ServerMCPClient:
         self.tools_dict: Dict[str, Any] = {}
         self.name_mapping: Dict[str, str] = {}
 
-    async def initialize(self):
+    async def initialize(self, read_timeout_seconds: timedelta | None = None,
+             sampling_callback: SamplingFnT | None = None,
+             elicitation_callback: ElicitationFnT | None = None,
+             list_roots_callback: ListRootsFnT | None = None,
+             logging_callback: LoggingFnT | None = None,
+             message_handler: MessageHandlerFnT | None = None,
+             client_info: Implementation | None = None):
         """初始化MCP客户端连接"""
         if self._worker_task:
             return
 
         self._worker_task = asyncio.create_task(
-            self._worker(), name="ServerMCPClientWorker"
+            self._worker(read_timeout_seconds=read_timeout_seconds,
+                        sampling_callback=sampling_callback,
+                        elicitation_callback=elicitation_callback,
+                        list_roots_callback=list_roots_callback,
+                        logging_callback=logging_callback,
+                        message_handler=message_handler,
+                        client_info=client_info), name="ServerMCPClientWorker"
         )
         await self._ready_evt.wait()
 
@@ -97,12 +112,15 @@ class ServerMCPClient:
             for name, tool in self.tools_dict.items()
         ]
 
-    async def call_tool(self, name: str, args: dict) -> Any:
+    async def call_tool(self, name: str, arguments: dict, read_timeout_seconds: timedelta | None = None, progress_callback: ProgressFnT | None = None, *, meta: dict[str, Any] | None = None) -> Any:
         """调用指定工具
 
         Args:
             name: 工具名称
-            args: 工具参数
+            arguments: 工具参数
+            read_timeout_seconds:
+            progress_callback: 进度回调函数
+            meta:
 
         Returns:
             Any: 工具执行结果
@@ -115,7 +133,7 @@ class ServerMCPClient:
 
         real_name = self.name_mapping.get(name, name)
         loop = self._worker_task.get_loop()
-        coro = self.session.call_tool(real_name, args)
+        coro = self.session.call_tool(real_name, arguments=arguments, read_timeout_seconds=read_timeout_seconds, progress_callback=progress_callback, meta=meta)
 
         if loop is asyncio.get_running_loop():
             return await coro
@@ -144,7 +162,13 @@ class ServerMCPClient:
         # 所有检查都通过，连接正常
         return True
 
-    async def _worker(self):
+    async def _worker(self, read_timeout_seconds: timedelta | None = None,
+             sampling_callback: SamplingFnT | None = None,
+             elicitation_callback: ElicitationFnT | None = None,
+             list_roots_callback: ListRootsFnT | None = None,
+             logging_callback: LoggingFnT | None = None,
+             message_handler: MessageHandlerFnT | None = None,
+             client_info: Implementation | None = None):
         """MCP客户端工作协程"""
         async with AsyncExitStack() as stack:
             try:
@@ -208,7 +232,13 @@ class ServerMCPClient:
                     ClientSession(
                         read_stream=read_stream,
                         write_stream=write_stream,
-                        read_timeout_seconds=timedelta(seconds=15),
+                        read_timeout_seconds=read_timeout_seconds,
+                        sampling_callback=sampling_callback,
+                        elicitation_callback=elicitation_callback,
+                        list_roots_callback=list_roots_callback,
+                        logging_callback=logging_callback,
+                        message_handler=message_handler,
+                        client_info=client_info
                     )
                 )
                 await self.session.initialize()

main/xiaozhi-server/core/providers/tools/server_mcp/mcp_manager.py

@@ -3,7 +3,14 @@
 import asyncio
 import os
 import json
+from datetime import timedelta
 from typing import Dict, Any, List
+
+from mcp import Implementation
+from mcp.client.session import SamplingFnT, ElicitationFnT, ListRootsFnT, LoggingFnT, MessageHandlerFnT
+from mcp.shared.session import ProgressFnT
+from mcp.types import LoggingMessageNotificationParams
+
 from config.config_loader import get_project_dir
 from config.logger import setup_logging
 from .mcp_client import ServerMCPClient
@@ -56,7 +63,7 @@ class ServerMCPManager:
                 # 初始化服务端MCP客户端
                 logger.bind(tag=TAG).info(f"初始化服务端MCP客户端: {name}")
                 client = ServerMCPClient(srv_config)
-                await client.initialize()
+                await client.initialize(logging_callback=self.logging_callback)
                 self.clients[name] = client
                 client_tools = client.get_available_tools()
                 self.tools.extend(client_tools)
@@ -109,7 +116,7 @@ class ServerMCPManager:
         # 带重试机制的工具调用
         for attempt in range(max_retries):
             try:
-                return await target_client.call_tool(tool_name, arguments)
+                return await target_client.call_tool(tool_name, arguments, progress_callback=self.progress_callback)
             except Exception as e:
                 # 最后一次尝试失败时直接抛出异常
                 if attempt == max_retries - 1:
@@ -131,7 +138,7 @@ class ServerMCPManager:
                     config = self.load_config()
                     if client_name in config:
                         client = ServerMCPClient(config[client_name])
-                        await client.initialize()
+                        await client.initialize(logging_callback=self.logging_callback)
                         self.clients[client_name] = client
                         target_client = client
                         logger.bind(tag=TAG).info(
@@ -159,3 +166,11 @@ class ServerMCPManager:
             except (asyncio.TimeoutError, Exception) as e:
                 logger.bind(tag=TAG).error(f"关闭服务端MCP客户端 {name} 时出错: {e}")
         self.clients.clear()
+
+    # 可选回调方法
+
+    async def logging_callback(self, params: LoggingMessageNotificationParams):
+        logger.bind(tag=TAG).info(f"[Server Log - {params.level.upper()}] {params.data}")
+
+    async def progress_callback(self, progress: float, total: float | None, message: str | None) -> None:
+        logger.bind(tag=TAG).info(f"[Progress {progress}/{total}]: {message}")

main/xiaozhi-server/core/providers/vad/silero.py

@@ -2,6 +2,7 @@ import time
 import numpy as np
 import torch
 import opuslib_next
+import gc
 from config.logger import setup_logging
 from core.providers.vad.base import VADProviderBase
 
@@ -36,6 +37,13 @@ class VADProvider(VADProviderBase):
         # 至少要多少帧才算有语音
         self.frame_window_threshold = 3
 
+    def __del__(self):
+        if hasattr(self, 'decoder') and self.decoder is not None:
+            try:
+                del self.decoder
+            except Exception:
+                pass
+
     def is_vad(self, conn, opus_packet):
         try:
             pcm_frame = self.decoder.decode(opus_packet, 960)

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

@@ -0,0 +1,132 @@
+import time
+import asyncio
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class AudioRateController:
+    """
+    音频速率控制器 - 按照60ms帧时长精确控制音频发送
+    解决高并发下的时间累积误差问题
+
+    关键改进：
+    1. 单一时间基准（start_timestamp 只初始化一次）
+    2. 每次检查队列时重新计算 elapsed_ms，避免累积误差
+    3. 分离"检查时间"和"发送"两个操作
+    4. 支持高并发而不产生延迟
+    """
+
+    def __init__(self, frame_duration=60):
+        """
+        Args:
+            frame_duration: 单个音频帧时长（毫秒），默认60ms
+        """
+        self.frame_duration = frame_duration
+        self.queue = []
+        self.play_position = 0  # 虚拟播放位置（毫秒）
+        self.start_timestamp = None  # 开始时间戳（只读，不修改）
+        self.pending_send_task = None
+        self.logger = logger
+
+    def reset(self):
+        """重置控制器状态"""
+        if self.pending_send_task and not self.pending_send_task.done():
+            self.pending_send_task.cancel()
+            try:
+                # 等待任务取消完成
+                asyncio.get_event_loop().run_until_complete(self.pending_send_task)
+            except asyncio.CancelledError:
+                pass
+
+        self.queue.clear()
+        self.play_position = 0
+        self.start_timestamp = time.time()
+
+    def add_audio(self, opus_packet):
+        """添加音频包到队列"""
+        self.queue.append(("audio", opus_packet))
+
+    def _get_elapsed_ms(self):
+        """获取已经过的时间（毫秒）"""
+        if self.start_timestamp is None:
+            return 0
+        return (time.time() - self.start_timestamp) * 1000
+
+    async def check_queue(self, send_audio_callback):
+        """
+        检查队列并按时发送音频/消息
+
+        Args:
+            send_audio_callback: 发送音频的回调函数 async def(opus_packet)
+        """
+        if self.start_timestamp is None:
+            self.reset()
+
+        while self.queue:
+            item = self.queue[0]
+            item_type = item[0]
+
+            if item_type == "audio":
+                _, opus_packet = item
+
+                # 计算时间差
+                elapsed_ms = self._get_elapsed_ms()
+                output_ms = self.play_position
+
+                if elapsed_ms < output_ms:
+                    # 还不到发送时间，计算等待时长
+                    wait_ms = output_ms - elapsed_ms
+
+                    # 等待后继续检查（允许被中断）
+                    try:
+                        await asyncio.sleep(wait_ms / 1000)
+                    except asyncio.CancelledError:
+                        self.logger.bind(tag=TAG).debug("音频发送任务被取消")
+                        raise
+
+                    # 继续循环检查（时间可能已到）
+                    continue
+
+                # 时间已到，发送音频
+                self.queue.pop(0)
+                self.play_position += self.frame_duration
+
+                try:
+                    await send_audio_callback(opus_packet)
+                except Exception as e:
+                    self.logger.bind(tag=TAG).error(f"发送音频失败: {e}")
+                    raise
+
+
+    async def start_sending(self, send_audio_callback, send_message_callback=None):
+        """
+        启动异步发送任务
+
+        Args:
+            send_audio_callback: 发送音频的回调函数
+            send_message_callback: 发送消息的回调函数
+
+        Returns:
+            asyncio.Task: 发送任务
+        """
+        async def _send_loop():
+            try:
+                while True:
+                    await self.check_queue(send_audio_callback, send_message_callback)
+                    # 如果队列空了，短暂等待后再检查（避免 busy loop）
+                    await asyncio.sleep(0.01)
+            except asyncio.CancelledError:
+                self.logger.bind(tag=TAG).info("音频发送循环已停止")
+            except Exception as e:
+                self.logger.bind(tag=TAG).error(f"音频发送循环异常: {e}")
+
+        self.pending_send_task = asyncio.create_task(_send_loop())
+        return self.pending_send_task
+
+    def stop_sending(self):
+        """停止发送任务"""
+        if self.pending_send_task and not self.pending_send_task.done():
+            self.pending_send_task.cancel()
+            self.logger.bind(tag=TAG).debug("已取消音频发送任务")

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

@@ -0,0 +1,122 @@
+"""
+全局GC管理模块
+定期执行垃圾回收，避免频繁触发GC导致的GIL锁问题
+"""
+
+import gc
+import asyncio
+import threading
+from config.logger import setup_logging
+
+TAG = __name__
+logger = setup_logging()
+
+
+class GlobalGCManager:
+    """全局垃圾回收管理器"""
+
+    def __init__(self, interval_seconds=300):
+        """
+        初始化GC管理器
+
+        Args:
+            interval_seconds: GC执行间隔（秒），默认300秒（5分钟）
+        """
+        self.interval_seconds = interval_seconds
+        self._task = None
+        self._stop_event = asyncio.Event()
+        self._lock = threading.Lock()
+
+    async def start(self):
+        """启动定时GC任务"""
+        if self._task is not None:
+            logger.bind(tag=TAG).warning("GC管理器已经在运行")
+            return
+
+        logger.bind(tag=TAG).info(f"启动全局GC管理器，间隔{self.interval_seconds}秒")
+        self._stop_event.clear()
+        self._task = asyncio.create_task(self._gc_loop())
+
+    async def stop(self):
+        """停止定时GC任务"""
+        if self._task is None:
+            return
+
+        logger.bind(tag=TAG).info("停止全局GC管理器")
+        self._stop_event.set()
+
+        if self._task and not self._task.done():
+            self._task.cancel()
+            try:
+                await self._task
+            except asyncio.CancelledError:
+                pass
+
+        self._task = None
+
+    async def _gc_loop(self):
+        """GC循环任务"""
+        try:
+            while not self._stop_event.is_set():
+                # 等待指定间隔
+                try:
+                    await asyncio.wait_for(
+                        self._stop_event.wait(), timeout=self.interval_seconds
+                    )
+                    # 如果stop_event被设置，退出循环
+                    break
+                except asyncio.TimeoutError:
+                    # 超时表示到了执行GC的时间
+                    pass
+
+                # 执行GC
+                await self._run_gc()
+
+        except asyncio.CancelledError:
+            logger.bind(tag=TAG).info("GC循环任务被取消")
+            raise
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"GC循环任务异常: {e}")
+        finally:
+            logger.bind(tag=TAG).info("GC循环任务已退出")
+
+    async def _run_gc(self):
+        """执行垃圾回收"""
+        try:
+            # 在线程池中执行GC，避免阻塞事件循环
+            loop = asyncio.get_running_loop()
+
+            def do_gc():
+                with self._lock:
+                    before = len(gc.get_objects())
+                    collected = gc.collect()
+                    after = len(gc.get_objects())
+                    return before, collected, after
+
+            before, collected, after = await loop.run_in_executor(None, do_gc)
+            logger.bind(tag=TAG).debug(
+                f"全局GC执行完成 - 回收对象: {collected}, "
+                f"对象数量: {before} -> {after}"
+            )
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"执行GC时出错: {e}")
+
+
+# 全局单例
+_gc_manager_instance = None
+
+
+def get_gc_manager(interval_seconds=300):
+    """
+    获取全局GC管理器实例（单例模式）
+
+    Args:
+        interval_seconds: GC执行间隔（秒），默认300秒（5分钟）
+
+    Returns:
+        GlobalGCManager实例
+    """
+    global _gc_manager_instance
+    if _gc_manager_instance is None:
+        _gc_manager_instance = GlobalGCManager(interval_seconds)
+    return _gc_manager_instance

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

@@ -6,6 +6,7 @@ Opus编码工具类
 import logging
 import traceback
 import numpy as np
+import gc
 from opuslib_next import Encoder
 from opuslib_next import constants
 from typing import Optional, Callable, Any
@@ -128,5 +129,9 @@ class OpusEncoderUtils:
 
     def close(self):
         """关闭编码器并释放资源"""
-        # opuslib没有明确的关闭方法，Python的垃圾回收会处理
-        pass
+        if hasattr(self, 'encoder') and self.encoder:
+            try:
+                del self.encoder
+                self.encoder = None
+            except Exception as e:
+                logging.error(f"Error releasing Opus encoder: {e}")

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

@@ -8,6 +8,7 @@ import requests
 import subprocess
 import numpy as np
 import opuslib_next
+import gc
 from io import BytesIO
 from core.utils import p3
 from pydub import AudioSegment
@@ -372,26 +373,33 @@ def opus_datas_to_wav_bytes(opus_datas, sample_rate=16000, channels=1):
     将opus帧列表解码为wav字节流
     """
     decoder = opuslib_next.Decoder(sample_rate, channels)
-    pcm_datas = []
+    try:
+        pcm_datas = []
 
-    frame_duration = 60  # ms
-    frame_size = int(sample_rate * frame_duration / 1000)  # 960
+        frame_duration = 60  # ms
+        frame_size = int(sample_rate * frame_duration / 1000)  # 960
 
-    for opus_frame in opus_datas:
-        # 解码为PCM（返回bytes，2字节/采样点）
-        pcm = decoder.decode(opus_frame, frame_size)
-        pcm_datas.append(pcm)
+        for opus_frame in opus_datas:
+            # 解码为PCM（返回bytes，2字节/采样点）
+            pcm = decoder.decode(opus_frame, frame_size)
+            pcm_datas.append(pcm)
 
-    pcm_bytes = b"".join(pcm_datas)
+        pcm_bytes = b"".join(pcm_datas)
 
-    # 写入wav字节流
-    wav_buffer = BytesIO()
-    with wave.open(wav_buffer, "wb") as wf:
-        wf.setnchannels(channels)
-        wf.setsampwidth(2)  # 16bit
-        wf.setframerate(sample_rate)
-        wf.writeframes(pcm_bytes)
-    return wav_buffer.getvalue()
+        # 写入wav字节流
+        wav_buffer = BytesIO()
+        with wave.open(wav_buffer, "wb") as wf:
+            wf.setnchannels(channels)
+            wf.setsampwidth(2)  # 16bit
+            wf.setframerate(sample_rate)
+            wf.writeframes(pcm_bytes)
+        return wav_buffer.getvalue()
+    finally:
+        if decoder is not None:
+            try:
+                del decoder
+            except Exception:
+                pass
 
 
 def check_vad_update(before_config, new_config):

main/xiaozhi-server/core/websocket_server.py

@@ -4,7 +4,7 @@ import json
 import websockets
 from config.logger import setup_logging
 from core.connection import ConnectionHandler
-from config.config_loader import get_config_from_api
+from config.config_loader import get_config_from_api_async
 from core.auth import AuthManager, AuthenticationError
 from core.utils.modules_initialize import initialize_modules
 from core.utils.util import check_vad_update, check_asr_update
@@ -33,8 +33,6 @@ class WebSocketServer:
         self._intent = modules["intent"] if "intent" in modules else None
         self._memory = modules["memory"] if "memory" in modules else None
 
-        self.active_connections = set()
-
         auth_config = self.config["server"].get("auth", {})
         self.auth_enable = auth_config.get("enabled", False)
         # 设备白名单
@@ -98,14 +96,11 @@ class WebSocketServer:
             self._intent,
             self,  # 传入server实例
         )
-        self.active_connections.add(handler)
         try:
             await handler.handle_connection(websocket)
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"处理连接时出错: {e}")
         finally:
-            # 确保从活动连接集合中移除
-            self.active_connections.discard(handler)
             # 强制关闭连接（如果还没有关闭的话）
             try:
                 # 安全地检查WebSocket状态并关闭
@@ -138,8 +133,8 @@ class WebSocketServer:
         """
         try:
             async with self.config_lock:
-                # 重新获取配置
-                new_config = get_config_from_api(self.config)
+                # 重新获取配置（使用异步版本）
+                new_config = await get_config_from_api_async(self.config)
                 if new_config is None:
                     self.logger.bind(tag=TAG).error("获取新配置失败")
                     return False

main/xiaozhi-server/requirements.txt

@@ -1,3 +1,4 @@
+# -*- coding:utf-8 -*-
 #--------- 本项目推荐环境是python3.10，以下暂时不推荐升级的依赖
 torch==2.2.2
 torchaudio==2.2.2
@@ -10,7 +11,7 @@ silero_vad==6.1.0
 opuslib_next==1.1.5
 pydub==0.25.1
 funasr==1.2.7
-openai==2.7.1
+openai==2.8.1
 google-generativeai==0.8.5
 edge_tts==7.2.3
 httpx==0.28.1
@@ -24,11 +25,11 @@ cozepy==0.20.0
 mem0ai==1.0.0
 bs4==0.0.2
 modelscope==1.23.2
-sherpa_onnx==1.12.15
+sherpa_onnx==1.12.17
 mcp==1.20.0
 cnlunar==0.2.0
 PySocks==1.7.1
-dashscope==1.24.6
+dashscope==1.25.2
 baidu-aip==4.16.13
 chardet==5.2.0
 aioconsole==0.8.2
@@ -38,4 +39,4 @@ PyJWT==2.10.1
 psutil==7.0.0
 portalocker==3.2.0
 Jinja2==3.1.6
-vosk==0.3.44
+vosk==0.3.45