{{ getFunctionDisplayChar(func.name) }}
@@ -457,6 +463,13 @@
:checked-replacement-word-ids="checkedReplacementWordIds"
@save="handleTtsSettingsSave"
/>
+
@@ -470,6 +483,7 @@ import RequestService from "@/apis/httpRequest";
import FunctionDialog from "@/components/FunctionDialog.vue";
import ContextProviderDialog from "@/components/ContextProviderDialog.vue";
import TtsAdvancedSettings from "@/components/TtsAdvancedSettings.vue";
+import AgentSnapshotDialog from "@/components/AgentSnapshotDialog.vue";
import HeaderBar from "@/components/HeaderBar.vue";
import i18n from "@/i18n";
import featureManager from "@/utils/featureManager";
@@ -477,11 +491,12 @@ import VersionFooter from "@/components/VersionFooter.vue";
export default {
name: "RoleConfigPage",
- components: { HeaderBar, FunctionDialog, ContextProviderDialog, TtsAdvancedSettings, VersionFooter },
+ components: { HeaderBar, FunctionDialog, ContextProviderDialog, TtsAdvancedSettings, AgentSnapshotDialog, VersionFooter },
data() {
return {
showContextProviderDialog: false,
showTtsAdvancedDialog: false,
+ showSnapshotDialog: false,
ttsSettings: {
volume: 0,
speed: 0,
@@ -529,6 +544,7 @@ export default {
voiceOptions: [],
voiceDetails: {}, // 保存完整的音色信息
showFunctionDialog: false,
+ currentVersionNo: null,
currentFunctions: [],
currentContextProviders: [],
allFunctions: [],
@@ -546,6 +562,7 @@ export default {
asr: false, // 语音识别功能状态
},
dynamicTags: [],
+ originalTagNames: [],
inputVisible: false,
inputValue: '',
checkedReplacementWordIds: []
@@ -555,14 +572,26 @@ export default {
goToHome() {
this.$router.push("/home");
},
- async saveConfig() {
- try {
- await this.handleSaveAgentTags(this.$route.query.agentId);
- } catch (error) {
- console.error('保存标签失败:', error);
- return;
+ normalizeFunctionParams(params, fallback = {}) {
+ if (params === null || params === undefined || params === '') {
+ return { ...fallback };
}
-
+ if (typeof params === 'string') {
+ try {
+ const parsed = JSON.parse(params);
+ return parsed && typeof parsed === 'object' && !Array.isArray(parsed)
+ ? parsed
+ : { ...fallback };
+ } catch (error) {
+ return { ...fallback };
+ }
+ }
+ if (typeof params === 'object' && !Array.isArray(params)) {
+ return { ...params };
+ }
+ return { ...fallback };
+ },
+ async saveConfig() {
const configData = {
agentCode: this.form.agentCode,
agentName: this.form.agentName,
@@ -585,12 +614,17 @@ export default {
functions: this.currentFunctions.map((item) => {
return {
pluginId: item.id,
- paramInfo: item.params,
+ paramInfo: this.normalizeFunctionParams(item.params),
};
}),
contextProviders: this.currentContextProviders,
correctWordFileIds: this.checkedReplacementWordIds,
};
+ const tagNames = this.dynamicTags.map(tag => tag.tagName);
+ const tagsChanged = !this.isSameStringList(tagNames, this.originalTagNames);
+ if (tagsChanged) {
+ configData.tagNames = tagNames;
+ }
// 只在用户设置了TTS参数时才传递(不为null/undefined)
if (this.form.ttsVolume !== null && this.form.ttsVolume !== undefined) {
@@ -602,12 +636,20 @@ export default {
if (this.form.ttsPitch !== null && this.form.ttsPitch !== undefined) {
configData.ttsPitch = this.form.ttsPitch;
}
- Api.agent.updateAgentConfig(this.$route.query.agentId, configData, ({ data }) => {
+ const agentId = this.$route.query.agentId;
+ Api.agent.updateAgentConfig(agentId, configData, ({ data }) => {
if (data.code === 0) {
- this.$message.success({
- message: i18n.t("roleConfig.saveSuccess"),
- showClose: true,
- });
+ const afterSave = () => {
+ if (tagsChanged) {
+ this.originalTagNames = [...tagNames];
+ }
+ this.$message.success({
+ message: i18n.t("roleConfig.saveSuccess"),
+ showClose: true,
+ });
+ this.fetchCurrentVersion(agentId);
+ };
+ afterSave();
} else {
this.$message.error({
message: data.msg || i18n.t("roleConfig.saveFailed"),
@@ -617,6 +659,26 @@ export default {
});
},
+ handleSnapshotRestored() {
+ const agentId = this.$route.query.agentId;
+ if (agentId) {
+ this.fetchAgentConfig(agentId);
+ this.getAgentTags(agentId);
+ this.fetchCurrentVersion(agentId);
+ }
+ },
+ fetchCurrentVersion(agentId) {
+ if (!agentId) {
+ this.currentVersionNo = null;
+ return;
+ }
+
+ Api.agent.getDeviceConfig(agentId, ({ data }) => {
+ if (data.code === 0) {
+ this.currentVersionNo = data.data?.currentVersionNo || null;
+ }
+ });
+ },
resetConfig() {
this.$confirm(i18n.t("roleConfig.confirmReset"), i18n.t("message.info"), {
confirmButtonText: i18n.t("button.ok"),
@@ -754,7 +816,7 @@ export default {
id: mapping.pluginId,
name: meta.name,
// 后端如果还有 paramInfo 字段就用 mapping.paramInfo,否则用 meta.params 默认值
- params: mapping.paramInfo || { ...meta.params },
+ params: this.normalizeFunctionParams(mapping.paramInfo, meta.params),
fieldsMeta: meta.fieldsMeta, // 保留以便对话框渲染 tooltip
};
});
@@ -806,7 +868,7 @@ export default {
});
this.$set(this.modelOptions, model.type, LLMdata);
} else {
- this.$message.error(data.msg || "获取LLM模型列表失败");
+ this.$message.error(data.msg || i18n.t("roleConfig.fetchModelsFailed"));
}
});
}
@@ -1329,7 +1391,7 @@ export default {
handleInputConfirm() {
let inputValue = this.inputValue;
if (inputValue) {
- const tag = { id: new Date().getTime(), tagName: inputValue };
+ const tag = { id: `tmp-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`, tagName: inputValue };
this.dynamicTags.push(tag);
}
this.inputVisible = false;
@@ -1339,14 +1401,21 @@ export default {
Api.agent.getAgentTags(agentId, ({ data }) => {
if (data.code === 0) {
this.dynamicTags = data.data || [];
+ this.originalTagNames = this.dynamicTags.map(tag => tag.tagName);
}
});
},
- handleSaveAgentTags(agentId) {
+ isSameStringList(left, right) {
+ if (!Array.isArray(left) || !Array.isArray(right) || left.length !== right.length) {
+ return false;
+ }
+ return left.every((value, index) => value === right[index]);
+ },
+ handleSaveAgentTags(agentId, tagNames = this.dynamicTags.map(tag => tag.tagName)) {
return new Promise((resolve, reject) => {
- const tagNames = this.dynamicTags.map(tag => tag.tagName);
Api.agent.saveAgentTags(agentId, { tagNames }, ({ data }) => {
if (data.code === 0) {
+ this.originalTagNames = [...tagNames];
resolve();
} else {
reject(data.msg);
@@ -1382,6 +1451,7 @@ export default {
this.fetchAgentConfig(agentId);
this.getAgentTags(agentId);
this.fetchAllFunctions();
+ this.fetchCurrentVersion(agentId);
}
this.fetchModelOptions();
this.fetchTemplates();
@@ -1507,6 +1577,18 @@ export default {
flex-shrink: 0;
}
+.current-version-tag {
+ flex-shrink: 0;
+ padding: 3px 9px;
+ border: 1px solid #dfe7ff;
+ border-radius: 999px;
+ background: #f4f7ff;
+ color: #5778ff;
+ font-size: 12px;
+ font-weight: 500;
+ line-height: 1.5;
+}
+
.more-tag {
cursor: pointer;
flex-shrink: 0;
@@ -1785,6 +1867,16 @@ export default {
font-size: 14px;
}
+.header-actions .history-btn {
+ background: #ffffff;
+ color: #4d5b7c;
+ border: 1px solid #d8dce8;
+ border-radius: 18px;
+ padding: 8px 16px;
+ height: 32px;
+ font-size: 14px;
+}
+
.header-actions .reset-btn {
background: #e6ebff;
color: #5778ff;
diff --git a/main/xiaozhi-server/app.py b/main/xiaozhi-server/app.py
index bdefe42d..9fb3a9ea 100644
--- a/main/xiaozhi-server/app.py
+++ b/main/xiaozhi-server/app.py
@@ -45,7 +45,7 @@ async def monitor_stdin():
async def main():
check_ffmpeg_installed()
- config = load_config()
+ config = await load_config()
# auth_key优先级:配置文件server.auth_key > manager-api.secret > 自动生成
# auth_key用于jwt认证,比如视觉分析接口的jwt认证、ota接口的token生成与websocket认证
diff --git a/main/xiaozhi-server/config/assets/wakeup_words/ed76d459636c2481aec828516c1b4f54.wav b/main/xiaozhi-server/config/assets/wakeup_words/ed76d459636c2481aec828516c1b4f54.wav
deleted file mode 100644
index 1e7246be..00000000
Binary files a/main/xiaozhi-server/config/assets/wakeup_words/ed76d459636c2481aec828516c1b4f54.wav and /dev/null differ
diff --git a/main/xiaozhi-server/config/config_loader.py b/main/xiaozhi-server/config/config_loader.py
index 33a57e72..9e7e1c11 100644
--- a/main/xiaozhi-server/config/config_loader.py
+++ b/main/xiaozhi-server/config/config_loader.py
@@ -7,7 +7,6 @@ from config.manage_api_client import (
get_server_config,
get_agent_models,
get_correct_words,
- lookup_address_book,
DeviceNotFoundException,
DeviceBindException,
)
@@ -24,7 +23,7 @@ def read_config(config_path):
return config
-def load_config():
+async def load_config():
"""加载配置文件"""
from core.utils.cache.manager import cache_manager, CacheType
@@ -41,16 +40,7 @@ def load_config():
custom_config = read_config(custom_config_path)
if custom_config.get("manager-api", {}).get("url"):
- 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))
+ config = await get_config_from_api_async(custom_config)
else:
# 合并配置
config = merge_configs(default_config, custom_config)
@@ -139,7 +129,7 @@ def ensure_directories(config):
selected_provider = selected_modules.get(module_type)
if not selected_provider:
continue
- if config.get(module) is None:
+ if config.get(module_type) is None:
continue
if config.get(selected_provider) is None:
continue
diff --git a/main/xiaozhi-server/config/logger.py b/main/xiaozhi-server/config/logger.py
index ee1b4c36..34737247 100644
--- a/main/xiaozhi-server/config/logger.py
+++ b/main/xiaozhi-server/config/logger.py
@@ -1,9 +1,10 @@
import os
import sys
+import asyncio
from loguru import logger
from config.config_loader import load_config
from config.settings import check_config_file
-from datetime import datetime
+from core.utils.cache.manager import cache_manager, CacheType
SERVER_VERSION = "0.9.5"
_logger_initialized = False
@@ -45,10 +46,15 @@ def formatter(record):
return record["message"]
-def setup_logging():
- check_config_file()
+def setup_logging(config=None):
"""从配置文件中读取日志配置,并设置日志输出格式和级别"""
- config = load_config()
+ if config is None:
+ check_config_file()
+ # 先查缓存,避免在 async 上下文中重复 await load_config
+ config = cache_manager.get(CacheType.CONFIG, "main_config")
+ if config is None:
+ # 缓存也没有(理论上不该发生),才走 asyncio.run
+ config = asyncio.run(load_config())
log_config = config["log"]
global _logger_initialized
diff --git a/main/xiaozhi-server/config/manage_api_client.py b/main/xiaozhi-server/config/manage_api_client.py
index 7cc4c361..8eb2da20 100644
--- a/main/xiaozhi-server/config/manage_api_client.py
+++ b/main/xiaozhi-server/config/manage_api_client.py
@@ -73,6 +73,7 @@ class ManageApiClient:
},
timeout=cls.config.get("timeout", 30),
limits=limits, # 使用限制
+ trust_env=False,
)
return cls._async_clients[loop_id]
except RuntimeError:
diff --git a/main/xiaozhi-server/config/settings.py b/main/xiaozhi-server/config/settings.py
index fb8868b4..cfca89b6 100644
--- a/main/xiaozhi-server/config/settings.py
+++ b/main/xiaozhi-server/config/settings.py
@@ -1,4 +1,5 @@
import os
+import asyncio
from config.config_loader import read_config, get_project_dir, load_config
@@ -20,7 +21,7 @@ def check_config_file():
)
# 检查是否从API读取配置
- config = load_config()
+ config = asyncio.run(load_config())
if config.get("read_config_from_api", False):
print("从API读取配置")
old_config_origin = read_config(custom_config_file)
diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py
index 8772180c..fcf5183b 100644
--- a/main/xiaozhi-server/core/connection.py
+++ b/main/xiaozhi-server/core/connection.py
@@ -11,6 +11,8 @@ import threading
import traceback
import subprocess
import websockets
+import opuslib_next
+import numpy as np
from core.utils.util import (
extract_json_from_string,
@@ -114,6 +116,7 @@ class ConnectionHandler:
self.client_abort = False
self.client_is_speaking = False
self.client_listen_mode = "auto"
+ self.client_aec = False # 是否启用了服务端AEC
# 线程任务相关
self.loop = None # 在 handle_connection 中获取运行中的事件循环
@@ -153,7 +156,7 @@ class ConnectionHandler:
# asr相关变量
# 因为实际部署时可能会用到公共的本地ASR,不能把变量暴露给公共ASR
# 所以涉及到ASR的变量,需要在这里定义,属于connection的私有变量
- self.asr_audio = []
+ self.asr_audio = [] # 存储PCM帧列表,供VAD和ASR共享
self.asr_audio_queue = queue.Queue()
self.current_speaker = None # 存储当前说话人
@@ -231,6 +234,9 @@ class ConnectionHandler:
# 启动超时检查任务
self.timeout_task = asyncio.create_task(self._check_timeout())
+ # 启动AEC缓存清理任务
+ self._aec_cache_cleanup_task = asyncio.create_task(self._check_aec_cache_expiry())
+
self.welcome_msg = self.config["xiaozhi"]
self.welcome_msg["session_id"] = self.session_id
@@ -366,12 +372,14 @@ class ConnectionHandler:
if handled:
return
- # 不需要头部处理或没有头部时,直接处理原始消息
- self.asr_audio_queue.put(message)
+ # 入口处直接解码PCM,避免VAD和ASR重复解码
+ pcm_frame = self._decode_opus_packet(message)
+ if pcm_frame:
+ self.asr_audio_queue.put(pcm_frame)
async def _process_mqtt_audio_message(self, message):
"""
- 处理来自MQTT网关的音频消息,解析16字节头部并提取音频数据
+ 处理来自MQTT网关的音频消息,解析16字节头部并提取音频数据,在入队前进行AEC处理
Args:
message: 包含头部的音频消息
@@ -380,58 +388,192 @@ class ConnectionHandler:
bool: 是否成功处理了消息
"""
try:
- # 提取头部信息
+ # 解析timestamp
timestamp = int.from_bytes(message[8:12], "big")
- audio_length = int.from_bytes(message[12:16], "big")
- # 提取音频数据
- if audio_length > 0 and len(message) >= 16 + audio_length:
- # 有指定长度,提取精确的音频数据
- audio_data = message[16 : 16 + audio_length]
- # 基于时间戳进行排序处理
- self._process_websocket_audio(audio_data, timestamp)
- return True
- elif len(message) > 16:
- # 没有指定长度或长度无效,去掉头部后处理剩余数据
- audio_data = message[16:]
- self.asr_audio_queue.put(audio_data)
+ audio_data = message[16:]
+ # 入口直接解码PCM
+ pcm_frame = self._decode_opus_packet(audio_data)
+ if not pcm_frame:
return True
+
+ # AEC处理:如果timestamp>0且启用了AEC
+ if timestamp > 0 and self.client_aec:
+ pcm_frame = self._apply_aec(timestamp, pcm_frame)
+
+ self.asr_audio_queue.put(pcm_frame)
+ return True
except Exception as e:
self.logger.bind(tag=TAG).error(f"解析WebSocket音频包失败: {e}")
# 处理失败,返回False表示需要继续处理
return False
- def _process_websocket_audio(self, audio_data, timestamp):
- """处理WebSocket格式的音频包"""
- # 初始化时间戳序列管理
- if not hasattr(self, "audio_timestamp_buffer"):
- self.audio_timestamp_buffer = {}
- self.last_processed_timestamp = 0
- self.max_timestamp_buffer_size = 20
+ def _apply_aec(self, timestamp: int, pcm_frame: bytes) -> bytes:
+ """应用AEC处理 - 综合算法:互相关延迟估计 + Wiener滤波 + 频谱减法"""
+ try:
+ import numpy as np
- # 如果时间戳是递增的,直接处理
- if timestamp >= self.last_processed_timestamp:
- self.asr_audio_queue.put(audio_data)
- self.last_processed_timestamp = timestamp
+ if not pcm_frame or len(pcm_frame) == 0:
+ return pcm_frame
- # 处理缓冲区中的后续包
- processed_any = True
- while processed_any:
- processed_any = False
- for ts in sorted(self.audio_timestamp_buffer.keys()):
- if ts > self.last_processed_timestamp:
- buffered_audio = self.audio_timestamp_buffer.pop(ts)
- self.asr_audio_queue.put(buffered_audio)
- self.last_processed_timestamp = ts
- processed_any = True
- break
- else:
- # 乱序包,暂存
- if len(self.audio_timestamp_buffer) < self.max_timestamp_buffer_size:
- self.audio_timestamp_buffer[timestamp] = audio_data
+ if not hasattr(self, "aec_audio_cache") or not self.aec_audio_cache:
+ return pcm_frame
+
+ mic_audio = np.frombuffer(pcm_frame, dtype=np.int16).astype(np.float32)
+ mic_rms = np.sqrt(np.mean(mic_audio ** 2))
+
+ if mic_rms < 100:
+ return pcm_frame
+
+ # 初始化
+ if not hasattr(self, "_aec_filter_state"):
+ self._aec_filter_state = np.zeros(512) # 滤波器状态
+ self._aec_delay_ms = 200
+ self._aec_last_delay = 200
+ self._aec_coherence_history = []
+
+ sorted_timestamps = sorted(self.aec_audio_cache.keys())
+ if len(sorted_timestamps) < 2:
+ return pcm_frame
+
+ # ========== 匹配参考帧(对数功率谱匹配) ==========
+ n = len(mic_audio)
+ sorted_timestamps = sorted(self.aec_audio_cache.keys())
+
+ # 找最接近的timestamp作为起点
+ closest_idx = min(range(len(sorted_timestamps)), key=lambda i: abs(sorted_timestamps[i] - timestamp))
+
+ # 对数功率谱相关性计算
+ def log_powerSpectrum_corr(audio1, audio2):
+ window = np.hanning(len(audio1))
+ fft1 = np.fft.rfft(audio1 * window)
+ fft2 = np.fft.rfft(audio2 * window)
+ psd1 = np.abs(fft1) ** 2
+ psd2 = np.abs(fft2) ** 2
+ log1 = 10 * np.log10(psd1 + 1e-8)
+ log2 = 10 * np.log10(psd2 + 1e-8)
+ P_xy = np.dot(log1, log2)
+ P_xx = np.dot(log1, log1)
+ P_yy = np.dot(log2, log2)
+ return abs(P_xy) / (np.sqrt(P_xx) * np.sqrt(P_yy) + 1e-8)
+
+ # 用对数功率谱匹配找最佳帧:前后各找2帧
+ best_corr = -1
+ best_ref_idx = closest_idx
+
+ for offset in range(-2, 3): # T-2, T-1, T, T+1, T+2
+ test_idx = closest_idx + offset
+ if test_idx < 0 or test_idx >= len(sorted_timestamps):
+ continue
+ test_ts = sorted_timestamps[test_idx]
+ test_ref = np.frombuffer(self.aec_audio_cache[test_ts], dtype=np.int16).astype(np.float32)
+ test_ref_rms = np.sqrt(np.mean(test_ref ** 2))
+ if test_ref_rms < 50:
+ continue
+
+ # 对数功率谱相关性
+ corr = log_powerSpectrum_corr(mic_audio, test_ref)
+
+ if corr > best_corr:
+ best_corr = corr
+ best_ref_idx = test_idx
+
+ best_ts = sorted_timestamps[best_ref_idx]
+
+ best_ref = np.frombuffer(self.aec_audio_cache[best_ts], dtype=np.int16).astype(np.float32)
+ ref_rms = np.sqrt(np.mean(best_ref ** 2))
+
+ if ref_rms < 50:
+ return pcm_frame
+
+ # 对齐参考信号(直接截取相同长度)
+ aligned_ref = best_ref[:n]
+ if len(aligned_ref) < n:
+ aligned_ref = np.pad(aligned_ref, (0, n - len(aligned_ref)))
+
+ # ========== 频域 AEC 处理(谱减法) ==========
+ # 时域信号经过声学路径后相位失真,导致时域相关性低且P_xy正负不定
+ # 频域幅度谱不受相位影响,对数功率谱相关性稳定在0.97+
+ # 公式:result_mag = max(|mic_fft| - |ref_fft| * scale * coef, 0)
+
+ window = np.hanning(n)
+ mic_fft = np.fft.rfft(mic_audio * window)
+ ref_fft = np.fft.rfft(aligned_ref * window)
+
+ mic_mag = np.abs(mic_fft)
+ ref_mag = np.abs(ref_fft)
+ mic_phase = np.angle(mic_fft)
+
+ # 频域计算回声比例 scale
+ scale = np.sum(mic_mag * ref_mag) / (np.dot(ref_mag, ref_mag) + 1e-8)
+
+ # 使用对数功率谱相关性作为coh
+ coh = best_corr
+ ref_power = ref_rms
+
+ # 自适应系数:根据scale和coh动态调整
+ # scale大(回声强)-> coef大;coh高(匹配准)-> coef大
+ raw_coef = 1.0 + scale * 3 + (coh - 0.97) * 30
+ coef = max(0.5, min(3.0, raw_coef))
+
+ # 谱减法(过减 + 半波整流)
+ if ref_power < 50:
+ output = mic_audio
+ coef = 0
+ gain = 0
else:
- self.asr_audio_queue.put(audio_data)
+ echo_mag = ref_mag * scale * coef
+ # 过减:使用较大系数抑制回声
+ over_subtract = 1.5
+ result_mag = np.maximum(mic_mag - echo_mag * over_subtract, mic_mag * 0.1)
+
+ # 保留相位重建信号
+ result_fft = result_mag * np.exp(1j * mic_phase)
+ output = np.fft.irfft(result_fft, n)
+ gain = scale * coef
+
+ # 高置信度是纯回声时,再压一下确保VAD检测不到
+ if coh >= 0.97 and ref_power > 500:
+ output = output * 0.3
+
+ # 后处理:限幅
+ output = np.clip(output, -32768, 32767)
+
+ # 转换为bytes
+ result = output.astype(np.int16).tobytes()
+
+ return result
+
+ except Exception as e:
+ self.logger.bind(tag=TAG).warning(f"[AEC] 处理失败: {e}")
+ return pcm_frame
+
+ def _decode_opus_packet(self, opus_packet: bytes) -> bytes:
+ """
+ 解码Opus数据包为PCM数据
+
+ Args:
+ opus_packet: Opus编码的音频数据
+
+ Returns:
+ bytes: 解码后的PCM数据,失败返回None
+ """
+ try:
+ if not opus_packet or len(opus_packet) == 0:
+ return None
+
+ self._init_connection_state(self)
+ pcm_frame = self._connection_opus_decoder.decode(opus_packet, 960)
+ return pcm_frame
+ except Exception as e:
+ self.logger.bind(tag=TAG).debug(f"Opus解码失败: {e}")
+ return None
+
+ def _init_connection_state(self, conn):
+ """为连接初始化独立的Opus解码器"""
+ if not hasattr(conn, "_connection_opus_decoder"):
+ conn._connection_opus_decoder = opuslib_next.Decoder(16000, 1)
async def handle_restart(self, message):
"""处理服务器重启请求"""
@@ -1403,6 +1545,13 @@ class ConnectionHandler:
):
self.vad.release_conn_resources(self)
+ # 清理opus解码器
+ if hasattr(self, "_connection_opus_decoder"):
+ try:
+ delattr(self, "_connection_opus_decoder")
+ except Exception:
+ pass
+
# 清理音频缓冲区
if hasattr(self, "audio_buffer"):
self.audio_buffer.clear()
@@ -1416,6 +1565,20 @@ class ConnectionHandler:
pass
self.timeout_task = None
+ # 取消AEC缓存清理任务
+ if hasattr(self, "_aec_cache_cleanup_task") and self._aec_cache_cleanup_task and not self._aec_cache_cleanup_task.done():
+ self._aec_cache_cleanup_task.cancel()
+ try:
+ await self._aec_cache_cleanup_task
+ except asyncio.CancelledError:
+ pass
+ self._aec_cache_cleanup_task = None
+
+ # 清理AEC缓存
+ if hasattr(self, "aec_audio_cache"):
+ self.aec_audio_cache.clear()
+ self.aec_audio_cache_time.clear()
+
# 清理工具处理器资源
if hasattr(self, "func_handler") and self.func_handler:
try:
@@ -1579,6 +1742,26 @@ class ConnectionHandler:
finally:
self.logger.bind(tag=TAG).info("超时检查任务已退出")
+ async def _check_aec_cache_expiry(self):
+ """定期清理过期的AEC缓存"""
+ try:
+ while not self.stop_event.is_set():
+ if hasattr(self, "aec_audio_cache") and self.aec_audio_cache:
+ current_time = time.time()
+ expired_keys = [
+ ts for ts, cache_time in list(self.aec_audio_cache_time.items())
+ if current_time - cache_time > 120 # 2分钟过期
+ ]
+ for ts in expired_keys:
+ self.aec_audio_cache.pop(ts, None)
+ self.aec_audio_cache_time.pop(ts, None)
+ if expired_keys:
+ self.logger.bind(tag=TAG).debug(f"[AEC] 清理过期缓存 {len(expired_keys)} 条")
+ # 每30秒检查一次
+ await asyncio.sleep(30)
+ except Exception as e:
+ self.logger.bind(tag=TAG).error(f"AEC缓存清理任务出错: {e}")
+
@staticmethod
def _extract_direct_answer_response(arguments_str):
"""从 direct_answer 的参数中提取 response 值。
diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py
index 6bad8380..2f84a7b1 100644
--- a/main/xiaozhi-server/core/handle/helloHandle.py
+++ b/main/xiaozhi-server/core/handle/helloHandle.py
@@ -56,6 +56,9 @@ async def handleHelloMessage(conn: "ConnectionHandler", msg_json):
conn.mcp_client = MCPClient()
# 发送初始化
asyncio.create_task(send_mcp_initialize_message(conn))
+ if features.get("aec"):
+ conn.logger.bind(tag=TAG).debug("客户端启用了服务端AEC")
+ conn.client_aec = True
await conn.websocket.send(json.dumps(conn.welcome_msg))
diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py
index ae4673a0..63a9f610 100644
--- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py
+++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py
@@ -14,9 +14,9 @@ from core.handle.sendAudioHandle import send_stt_message, SentenceType
TAG = __name__
-async def handleAudioMessage(conn: "ConnectionHandler", audio):
+async def handleAudioMessage(conn: "ConnectionHandler", pcm_frame):
# 当前片段是否有人说话
- have_voice = conn.vad.is_vad(conn, audio)
+ have_voice = conn.vad.is_vad(conn, pcm_frame)
# 如果设备刚刚被唤醒,短暂忽略VAD检测
if hasattr(conn, "just_woken_up") and conn.just_woken_up:
have_voice = False
@@ -24,10 +24,14 @@ async def handleAudioMessage(conn: "ConnectionHandler", audio):
if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done():
conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn))
return
+ # 服务端AEC功能需要实时触发打断
+ if conn.client_aec and have_voice:
+ if conn.client_is_speaking and conn.client_listen_mode != "manual":
+ await handleAbortMessage(conn)
# 设备长时间空闲检测,用于say goodbye
await no_voice_close_connect(conn, have_voice)
# 接收音频
- await conn.asr.receive_audio(conn, audio, have_voice)
+ await conn.asr.receive_audio(conn, pcm_frame, have_voice)
async def resume_vad_detection(conn: "ConnectionHandler"):
diff --git a/main/xiaozhi-server/core/handle/reportHandle.py b/main/xiaozhi-server/core/handle/reportHandle.py
index 97fe8b47..ea621f74 100644
--- a/main/xiaozhi-server/core/handle/reportHandle.py
+++ b/main/xiaozhi-server/core/handle/reportHandle.py
@@ -50,33 +50,27 @@ async def report(conn: "ConnectionHandler", type, text, opus_data, report_time):
conn.logger.bind(tag=TAG).error(f"聊天记录上报失败: {e}")
-def opus_to_wav(conn: "ConnectionHandler", opus_data):
- """将Opus数据转换为WAV格式的字节流
+def opus_to_wav(conn: "ConnectionHandler", pcm_data):
+ """将PCM数据转换为WAV格式的字节流
Args:
output_dir: 输出目录(保留参数以保持接口兼容)
- opus_data: opus音频数据
+ pcm_data: PCM音频数据(可能是列表或bytes)
Returns:
bytes: WAV格式的音频数据
"""
- decoder = None
try:
- decoder = opuslib_next.Decoder(16000, 1) # 16kHz, 单声道
- pcm_data = []
+ # 处理可能是列表或bytes的PCM数据
+ if isinstance(pcm_data, list):
+ pcm_data_bytes = b"".join(pcm_data)
+ else:
+ pcm_data_bytes = 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)
-
- if not pcm_data:
+ if not pcm_data_bytes:
raise ValueError("没有有效的PCM数据")
# 创建WAV文件头
- pcm_data_bytes = b"".join(pcm_data)
num_samples = len(pcm_data_bytes) // 2 # 16-bit samples
# WAV文件头
@@ -97,12 +91,9 @@ def opus_to_wav(conn: "ConnectionHandler", opus_data):
# 返回完整的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}")
+ except Exception as e:
+ conn.logger.bind(tag=TAG).error(f"PCM转WAV失败: {e}", exc_info=True)
+ raise
def enqueue_tts_report(conn: "ConnectionHandler", text, opus_data):
diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py
index 17de11ae..421554d6 100644
--- a/main/xiaozhi-server/core/handle/sendAudioHandle.py
+++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py
@@ -1,6 +1,7 @@
import json
import time
import asyncio
+import opuslib_next
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -81,13 +82,24 @@ async def _send_to_mqtt_gateway(
conn: "ConnectionHandler", opus_packet, timestamp, sequence
):
"""
- 发送带16字节头部的opus数据包给mqtt_gateway
+ 发送带16字节头部的opus数据包给mqtt_gateway,同时缓存音频用于AEC处理
Args:
conn: 连接对象
opus_packet: opus数据包
timestamp: 时间戳
sequence: 序列号
"""
+ # 如果启用了服务端AEC,缓存PCM数据用于后续AEC处理
+ if conn.client_aec and timestamp > 0:
+ if not hasattr(conn, "aec_audio_cache"):
+ conn.aec_audio_cache = {}
+ conn.aec_audio_cache_time = {}
+ conn._send_opus_decoder = opuslib_next.Decoder(16000, 1)
+ # 解码opus为PCM后缓存
+ pcm_data = conn._send_opus_decoder.decode(bytes(opus_packet), 960)
+ conn.aec_audio_cache[timestamp] = bytes(pcm_data)
+ conn.aec_audio_cache_time[timestamp] = time.time()
+
# 为opus数据包添加16字节头部
header = bytearray(16)
header[0] = 1 # type
diff --git a/main/xiaozhi-server/core/providers/asr/aliyun.py b/main/xiaozhi-server/core/providers/asr/aliyun.py
index 8824d17b..ce9f4d15 100644
--- a/main/xiaozhi-server/core/providers/asr/aliyun.py
+++ b/main/xiaozhi-server/core/providers/asr/aliyun.py
@@ -213,7 +213,7 @@ class ASRProvider(ASRProviderBase):
return None
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if self._is_token_expired():
diff --git a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py
index 6c0bf093..74bf6137 100644
--- a/main/xiaozhi-server/core/providers/asr/aliyun_stream.py
+++ b/main/xiaozhi-server/core/providers/asr/aliyun_stream.py
@@ -7,7 +7,6 @@ import hashlib
import asyncio
import requests
import websockets
-import opuslib_next
from urllib import parse
from datetime import datetime
from config.logger import setup_logging
@@ -74,7 +73,6 @@ class ASRProvider(ASRProviderBase):
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
- self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False
@@ -129,9 +127,9 @@ class ASRProvider(ASRProviderBase):
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
- async def receive_audio(self, conn, audio, audio_have_voice):
+ async def receive_audio(self, conn, pcm_frame, audio_have_voice):
# 先调用父类方法处理基础逻辑
- await super().receive_audio(conn, audio, audio_have_voice)
+ await super().receive_audio(conn, pcm_frame, audio_have_voice)
# 只在有声音且没有连接时建立连接(排除正在停止的情况)
if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -144,7 +142,6 @@ class ASRProvider(ASRProviderBase):
if self.asr_ws and self.is_processing and self.server_ready:
try:
- pcm_frame = self.decoder.decode(audio, 960)
await self.asr_ws.send(pcm_frame)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
@@ -232,10 +229,9 @@ class ASRProvider(ASRProviderBase):
# 发送缓存音频
if conn.asr_audio:
- for cached_audio in conn.asr_audio[-10:]:
+ for cached_pcm in conn.asr_audio[-10:]:
try:
- pcm_frame = self.decoder.decode(cached_audio, 960)
- await self.asr_ws.send(pcm_frame)
+ await self.asr_ws.send(cached_pcm)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
break
@@ -328,7 +324,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
- async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
+ async def speech_to_text(self, opus_data, session_id, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
@@ -337,10 +333,3 @@ class ASRProvider(ASRProviderBase):
async def close(self):
"""关闭资源"""
await self._cleanup()
- 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}")
diff --git a/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py
index 83b79655..a7c2953a 100644
--- a/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py
+++ b/main/xiaozhi-server/core/providers/asr/aliyunbl_stream.py
@@ -2,7 +2,6 @@ import json
import uuid
import asyncio
import websockets
-import opuslib_next
from typing import List, TYPE_CHECKING
if TYPE_CHECKING:
@@ -22,7 +21,6 @@ class ASRProvider(ASRProviderBase):
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
- self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False
@@ -55,9 +53,9 @@ class ASRProvider(ASRProviderBase):
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
- async def receive_audio(self, conn, audio, audio_have_voice):
+ async def receive_audio(self, conn, pcm_frame, audio_have_voice):
# 先调用父类方法处理基础逻辑
- await super().receive_audio(conn, audio, audio_have_voice)
+ await super().receive_audio(conn, pcm_frame, audio_have_voice)
# 只在有声音且没有连接时建立连接
if audio_have_voice and not self.is_processing and not self.asr_ws:
@@ -71,8 +69,6 @@ class ASRProvider(ASRProviderBase):
# 发送音频数据
if self.asr_ws and self.is_processing and self.server_ready:
try:
- pcm_frame = self.decoder.decode(audio, 960)
- # 直接发送PCM音频数据(二进制)
await self.asr_ws.send(pcm_frame)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送音频失败: {str(e)}")
@@ -179,10 +175,9 @@ class ASRProvider(ASRProviderBase):
# 发送缓存音频
if conn.asr_audio:
- for cached_audio in conn.asr_audio[-10:]:
+ for cached_pcm in conn.asr_audio[-10:]:
try:
- pcm_frame = self.decoder.decode(cached_audio, 960)
- await self.asr_ws.send(pcm_frame)
+ await self.asr_ws.send(cached_pcm)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送缓存音频失败: {e}")
break
@@ -312,7 +307,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
- async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
+ async def speech_to_text(self, opus_data, session_id, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
@@ -320,11 +315,4 @@ class ASRProvider(ASRProviderBase):
async def close(self):
"""关闭资源"""
- await self._cleanup()
- if hasattr(self, 'decoder') and self.decoder is not None:
- try:
- del self.decoder
- self.decoder = None
- logger.bind(tag=TAG).debug("Aliyun BL decoder resources released")
- except Exception as e:
- logger.bind(tag=TAG).debug(f"释放Aliyun BL decoder资源时出错: {e}")
\ No newline at end of file
+ await self._cleanup()
\ No newline at end of file
diff --git a/main/xiaozhi-server/core/providers/asr/baidu.py b/main/xiaozhi-server/core/providers/asr/baidu.py
index aa14f443..541bb010 100644
--- a/main/xiaozhi-server/core/providers/asr/baidu.py
+++ b/main/xiaozhi-server/core/providers/asr/baidu.py
@@ -30,7 +30,7 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py
index 986dd275..dcfaf3a7 100644
--- a/main/xiaozhi-server/core/providers/asr/base.py
+++ b/main/xiaozhi-server/core/providers/asr/base.py
@@ -10,7 +10,6 @@ import asyncio
import tempfile
import traceback
import threading
-import opuslib_next
from abc import ABC, abstractmethod
from config.logger import setup_logging
@@ -59,13 +58,13 @@ class ASRProviderBase(ABC):
continue
# 接收音频
- async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
+ async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice):
if conn.client_listen_mode == "manual":
# 手动模式:缓存音频用于ASR识别
- conn.asr_audio.append(audio)
+ conn.asr_audio.append(pcm_frame)
else:
# 自动/实时模式:使用VAD检测
- conn.asr_audio.append(audio)
+ conn.asr_audio.append(pcm_frame)
# 如果没有语音,且之前也没有声音,缓存部分音频
if not audio_have_voice and not conn.client_have_voice:
@@ -74,24 +73,21 @@ class ASRProviderBase(ABC):
# 自动模式下通过VAD检测到语音停止时触发识别
if conn.asr.interface_type != InterfaceType.STREAM and conn.client_voice_stop:
- asr_audio_task = conn.asr_audio.copy()
+ # 直接使用asr_audio中的PCM数据
+ pcm_bytes = b"".join(conn.asr_audio)
+ # 检查是否有足够的音频数据(每帧1920字节,15帧约28800字节)
+ if len(pcm_bytes) > 1920 * 15:
+ await self.handle_voice_stop(conn, [pcm_bytes])
conn.reset_audio_states()
- if len(asr_audio_task) > 15:
- await self.handle_voice_stop(conn, asr_audio_task)
-
# 处理语音停止
async def handle_voice_stop(self, conn: "ConnectionHandler", asr_audio_task: List[bytes]):
"""并行处理ASR和声纹识别"""
try:
total_start_time = time.monotonic()
- # 准备音频数据
- if conn.audio_format == "pcm":
- pcm_data = asr_audio_task
- else:
- pcm_data = self.decode_opus(asr_audio_task)
-
+ # 数据已经是PCM直接使用
+ pcm_data = asr_audio_task
combined_pcm_data = b"".join(pcm_data)
# 预先准备WAV数据
@@ -101,7 +97,7 @@ class ASRProviderBase(ABC):
# 定义ASR任务
asr_task = self.speech_to_text_wrapper(
- asr_audio_task, conn.session_id, conn.audio_format
+ asr_audio_task, conn.session_id
)
if conn.voiceprint_provider and wav_data:
@@ -270,15 +266,11 @@ class ASRProviderBase(ABC):
return file_path
async def speech_to_text_wrapper(
- self, opus_data: List[bytes], session_id: str, audio_format="opus"
+ self, pcm_data: List[bytes], session_id: str
) -> Tuple[Optional[str], Optional[str]]:
file_path = None
temp_path = None
try:
- if audio_format == "pcm":
- pcm_data = opus_data
- else:
- pcm_data = self.decode_opus(opus_data)
combined_pcm_data = b"".join(pcm_data)
free_space = shutil.disk_usage(self.output_dir).free
@@ -304,7 +296,7 @@ class ASRProviderBase(ABC):
)
text, _ = await self.speech_to_text(
- opus_data, session_id, audio_format, artifacts
+ pcm_data, session_id, artifacts
)
return text, file_path
except OSError as e:
@@ -332,50 +324,13 @@ class ASRProviderBase(ABC):
self,
opus_data: List[bytes],
session_id: str,
- audio_format="opus",
artifacts: Optional[AudioArtifacts] = None,
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本
:param opus_data: 输入的Opus音频数据
:param session_id: 会话ID
- :param audio_format: 音频格式,默认"opus"
:param artifacts: 音频工件,包含PCM数据、文件路径等
:return: 识别结果文本和文件路径(如果有)
"""
pass
-
- @staticmethod
- def decode_opus(opus_data: List[bytes]) -> List[bytes]:
- """将Opus音频数据解码为PCM数据"""
- decoder = None
- try:
- decoder = opuslib_next.Decoder(16000, 1)
- pcm_data = []
- buffer_size = 960 # 每次处理960个采样点 (60ms at 16kHz)
-
- for i, opus_packet in enumerate(opus_data):
- try:
- if not opus_packet or len(opus_packet) == 0:
- continue
-
- pcm_frame = decoder.decode(opus_packet, buffer_size)
- if pcm_frame and len(pcm_frame) > 0:
- pcm_data.append(pcm_frame)
-
- except opuslib_next.OpusError as e:
- logger.bind(tag=TAG).warning(f"Opus解码错误,跳过数据包 {i}: {e}")
- except Exception as e:
- logger.bind(tag=TAG).error(f"音频处理错误,数据包 {i}: {e}")
-
- return pcm_data
-
- 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}")
diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py
index 016ac0fd..ee8b7cc8 100644
--- a/main/xiaozhi-server/core/providers/asr/doubao.py
+++ b/main/xiaozhi-server/core/providers/asr/doubao.py
@@ -232,7 +232,7 @@ class ASRProvider(ASRProviderBase):
yield data[offset:data_len], True
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
diff --git a/main/xiaozhi-server/core/providers/asr/doubao_stream.py b/main/xiaozhi-server/core/providers/asr/doubao_stream.py
index 5c34aa77..130729ba 100644
--- a/main/xiaozhi-server/core/providers/asr/doubao_stream.py
+++ b/main/xiaozhi-server/core/providers/asr/doubao_stream.py
@@ -3,7 +3,6 @@ import gzip
import uuid
import asyncio
import websockets
-import opuslib_next
from core.providers.asr.base import ASRProviderBase
from config.logger import setup_logging
from core.providers.asr.dto.dto import InterfaceType
@@ -22,7 +21,6 @@ class ASRProvider(ASRProviderBase):
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
- self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False # 添加处理状态标志
@@ -68,10 +66,10 @@ class ASRProvider(ASRProviderBase):
async def open_audio_channels(self, conn):
await super().open_audio_channels(conn)
- async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
+ async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice):
# 先调用父类方法处理基础逻辑
- await super().receive_audio(conn, audio, audio_have_voice)
-
+ await super().receive_audio(conn, pcm_frame, audio_have_voice)
+
# 如果本次有声音,且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
try:
@@ -123,10 +121,9 @@ class ASRProvider(ASRProviderBase):
# 发送缓存的音频数据
if conn.asr_audio and len(conn.asr_audio) > 0:
- for cached_audio in conn.asr_audio[-10:]:
+ for cached_pcm in conn.asr_audio[-10:]:
try:
- pcm_frame = self.decoder.decode(cached_audio, 960)
- payload = gzip.compress(pcm_frame)
+ payload = gzip.compress(cached_pcm)
audio_request = bytearray(
self.generate_audio_default_header()
)
@@ -151,7 +148,6 @@ class ASRProvider(ASRProviderBase):
# 发送当前音频数据
if self.asr_ws and self.is_processing and not self._is_stopping:
try:
- pcm_frame = self.decoder.decode(audio, 960)
payload = gzip.compress(pcm_frame)
audio_request = bytearray(self.generate_audio_default_header())
audio_request.extend(len(payload).to_bytes(4, "big"))
@@ -414,7 +410,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).error(f"原始响应数据: {res.hex()}")
raise
- async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
+ async def speech_to_text(self, opus_data, session_id, artifacts=None):
result = self.text
self.text = "" # 清空text
return result, None
@@ -432,12 +428,3 @@ 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}")
diff --git a/main/xiaozhi-server/core/providers/asr/fun_local.py b/main/xiaozhi-server/core/providers/asr/fun_local.py
index fa428710..5547104d 100644
--- a/main/xiaozhi-server/core/providers/asr/fun_local.py
+++ b/main/xiaozhi-server/core/providers/asr/fun_local.py
@@ -43,9 +43,13 @@ class ASRProvider(ASRProviderBase):
# 内存检测,要求大于2G
min_mem_bytes = 2 * 1024 * 1024 * 1024
- total_mem = psutil.virtual_memory().total
- if total_mem < min_mem_bytes:
- logger.bind(tag=TAG).error(f"可用内存不足2G,当前仅有 {total_mem / (1024*1024):.2f} MB,可能无法启动FunASR")
+ try:
+ total_mem = psutil.virtual_memory().total
+ except RuntimeError as e:
+ logger.bind(tag=TAG).warning(f"获取系统内存信息失败,跳过FunASR内存检测: {e}")
+ else:
+ if total_mem < min_mem_bytes:
+ logger.bind(tag=TAG).error(f"可用内存不足2G,当前仅有 {total_mem / (1024*1024):.2f} MB,可能无法启动FunASR")
self.interface_type = InterfaceType.LOCAL
self.model_dir = config.get("model_dir")
@@ -65,7 +69,7 @@ class ASRProvider(ASRProviderBase):
)
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
retry_count = 0
diff --git a/main/xiaozhi-server/core/providers/asr/fun_server.py b/main/xiaozhi-server/core/providers/asr/fun_server.py
index 19b5022d..d577a18e 100644
--- a/main/xiaozhi-server/core/providers/asr/fun_server.py
+++ b/main/xiaozhi-server/core/providers/asr/fun_server.py
@@ -101,7 +101,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug(f"Sent end message: {end_message}")
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""
Convert speech data to text using FunASR.
diff --git a/main/xiaozhi-server/core/providers/asr/openai.py b/main/xiaozhi-server/core/providers/asr/openai.py
index 7b215589..c917cce5 100644
--- a/main/xiaozhi-server/core/providers/asr/openai.py
+++ b/main/xiaozhi-server/core/providers/asr/openai.py
@@ -24,7 +24,7 @@ class ASRProvider(ASRProviderBase):
def requires_file(self) -> bool:
return True
- async def speech_to_text(self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None) -> Tuple[Optional[str], Optional[str]]:
+ async def speech_to_text(self, opus_data: List[bytes], session_id: str, artifacts=None) -> Tuple[Optional[str], Optional[str]]:
file_path = None
try:
if artifacts is None:
diff --git a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py
index 21438c1c..6444a0a4 100644
--- a/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py
+++ b/main/xiaozhi-server/core/providers/asr/qwen3_asr_flash.py
@@ -41,7 +41,7 @@ class ASRProvider(ASRProviderBase):
return True
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
temp_file_path = None
diff --git a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py
index 742c3567..54c981a6 100644
--- a/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py
+++ b/main/xiaozhi-server/core/providers/asr/sherpa_onnx_local.py
@@ -124,7 +124,7 @@ class ASRProvider(ASRProviderBase):
return True
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""语音转文本主处理逻辑"""
file_path = None
diff --git a/main/xiaozhi-server/core/providers/asr/tencent.py b/main/xiaozhi-server/core/providers/asr/tencent.py
index d873bd19..76a8867c 100644
--- a/main/xiaozhi-server/core/providers/asr/tencent.py
+++ b/main/xiaozhi-server/core/providers/asr/tencent.py
@@ -32,7 +32,7 @@ class ASRProvider(ASRProviderBase):
os.makedirs(self.output_dir, exist_ok=True)
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
if not opus_data:
diff --git a/main/xiaozhi-server/core/providers/asr/vosk.py b/main/xiaozhi-server/core/providers/asr/vosk.py
index 77cbf986..0871c085 100644
--- a/main/xiaozhi-server/core/providers/asr/vosk.py
+++ b/main/xiaozhi-server/core/providers/asr/vosk.py
@@ -44,7 +44,7 @@ class ASRProvider(ASRProviderBase):
raise
async def speech_to_text(
- self, opus_data: List[bytes], session_id: str, audio_format="opus", artifacts=None
+ self, opus_data: List[bytes], session_id: str, artifacts=None
) -> Tuple[Optional[str], Optional[str]]:
"""将语音数据转换为文本"""
try:
diff --git a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py
index 5667106d..fa59ca34 100644
--- a/main/xiaozhi-server/core/providers/asr/xunfei_stream.py
+++ b/main/xiaozhi-server/core/providers/asr/xunfei_stream.py
@@ -4,7 +4,6 @@ import base64
import hashlib
import asyncio
import websockets
-import opuslib_next
import gc
from time import mktime
from datetime import datetime
@@ -33,7 +32,6 @@ class ASRProvider(ASRProviderBase):
self.interface_type = InterfaceType.STREAM
self.config = config
self.text = ""
- self.decoder = opuslib_next.Decoder(16000, 1)
self.asr_ws = None
self.forward_task = None
self.is_processing = False
@@ -100,9 +98,9 @@ class ASRProvider(ASRProviderBase):
async def open_audio_channels(self, conn: "ConnectionHandler"):
await super().open_audio_channels(conn)
- async def receive_audio(self, conn: "ConnectionHandler", audio, audio_have_voice):
+ async def receive_audio(self, conn: "ConnectionHandler", pcm_frame, audio_have_voice):
# 先调用父类方法处理基础逻辑
- await super().receive_audio(conn, audio, audio_have_voice)
+ await super().receive_audio(conn, pcm_frame, audio_have_voice)
# 如果本次有声音,且之前没有建立连接
if audio_have_voice and self.asr_ws is None and not self.is_processing:
@@ -116,7 +114,6 @@ class ASRProvider(ASRProviderBase):
# 发送当前音频数据
if self.asr_ws and self.is_processing and self.server_ready:
try:
- pcm_frame = self.decoder.decode(audio, 960)
await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME)
except Exception as e:
logger.bind(tag=TAG).warning(f"发送音频数据时发生错误: {e}")
@@ -148,19 +145,15 @@ class ASRProvider(ASRProviderBase):
# 发送首帧音频
if conn.asr_audio and len(conn.asr_audio) > 0:
- first_audio = conn.asr_audio[-1] if conn.asr_audio else b""
- pcm_frame = (
- self.decoder.decode(first_audio, 960) if first_audio else b""
- )
- await self._send_audio_frame(pcm_frame, STATUS_FIRST_FRAME)
+ first_pcm = conn.asr_audio[-1] if conn.asr_audio else b""
+ await self._send_audio_frame(first_pcm, STATUS_FIRST_FRAME)
self.server_ready = True
logger.bind(tag=TAG).info("已发送首帧,开始识别")
# 发送缓存的音频数据
- for cached_audio in conn.asr_audio[-10:]:
+ for cached_pcm in conn.asr_audio[-10:]:
try:
- pcm_frame = self.decoder.decode(cached_audio, 960)
- await self._send_audio_frame(pcm_frame, STATUS_CONTINUE_FRAME)
+ await self._send_audio_frame(cached_pcm, STATUS_CONTINUE_FRAME)
except Exception as e:
logger.bind(tag=TAG).info(f"发送缓存音频数据时发生错误: {e}")
break
@@ -322,7 +315,7 @@ class ASRProvider(ASRProviderBase):
logger.bind(tag=TAG).debug("ASR会话清理完成")
- async def speech_to_text(self, opus_data, session_id, audio_format, artifacts=None):
+ async def speech_to_text(self, opus_data, session_id, artifacts=None):
"""获取识别结果"""
result = self.text
self.text = ""
@@ -342,12 +335,3 @@ class ASRProvider(ASRProviderBase):
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}")
-
diff --git a/main/xiaozhi-server/core/providers/memory/mem0ai/mem0ai.py b/main/xiaozhi-server/core/providers/memory/mem0ai/mem0ai.py
index cbf410ae..2dc93358 100644
--- a/main/xiaozhi-server/core/providers/memory/mem0ai/mem0ai.py
+++ b/main/xiaozhi-server/core/providers/memory/mem0ai/mem0ai.py
@@ -1,4 +1,5 @@
import json
+import asyncio
import traceback
from ..base import MemoryProviderBase, logger
@@ -59,7 +60,9 @@ class MemoryProvider(MemoryProviderBase):
messages.append({"role": message.role, "content": content})
- result = self.client.add(messages, user_id=self.role_id)
+ result = await asyncio.to_thread(
+ self.client.add, messages, user_id=self.role_id
+ )
logger.bind(tag=TAG).debug(f"Save memory result: {result}")
except Exception as e:
logger.bind(tag=TAG).error(f"保存记忆失败: {str(e)}")
@@ -84,7 +87,9 @@ class MemoryProvider(MemoryProviderBase):
except (json.JSONDecodeError, KeyError):
pass
- results = self.client.search(search_query, filters=filters)
+ results = await asyncio.to_thread(
+ self.client.search, search_query, filters=filters
+ )
if not results or "results" not in results:
return ""
diff --git a/main/xiaozhi-server/core/providers/memory/powermem/powermem.py b/main/xiaozhi-server/core/providers/memory/powermem/powermem.py
index 3e96fdfd..72d6526f 100644
--- a/main/xiaozhi-server/core/providers/memory/powermem/powermem.py
+++ b/main/xiaozhi-server/core/providers/memory/powermem/powermem.py
@@ -186,13 +186,19 @@ class MemoryProvider(MemoryProviderBase):
messages.append({"role": message.role, "content": content})
# Add memory using PowerMem SDK
- result = self.memory_client.add(
- messages=messages,
- user_id=self.role_id
- )
- # Handle both sync and async returns
- if asyncio.iscoroutine(result):
- result = await result
+ if self.enable_user_profile:
+ # UserMemory uses sync add
+ result = await asyncio.to_thread(
+ self.memory_client.add,
+ messages=messages,
+ user_id=self.role_id
+ )
+ else:
+ # AsyncMemory uses async add
+ result = await self.memory_client.add(
+ messages=messages,
+ user_id=self.role_id
+ )
logger.bind(tag=TAG).debug(f"Save memory result: {result}")
diff --git a/main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py b/main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py
index 249e040c..37f4fb9d 100644
--- a/main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py
+++ b/main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py
@@ -1,5 +1,6 @@
"""服务端插件工具执行器"""
+import asyncio
from typing import Dict, Any, TYPE_CHECKING
if TYPE_CHECKING:
@@ -41,6 +42,10 @@ class ServerPluginExecutor(ToolExecutor):
# 默认不传conn参数
result = func_item.func(**arguments)
+ # 兼容 async def 工具函数
+ if asyncio.iscoroutine(result):
+ result = await result
+
return result
except Exception as e:
diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py
index 3682f3b8..d134f25d 100644
--- a/main/xiaozhi-server/core/providers/vad/silero.py
+++ b/main/xiaozhi-server/core/providers/vad/silero.py
@@ -1,7 +1,6 @@
import time
import os
import numpy as np
-import opuslib_next
import onnxruntime
from config.logger import setup_logging
from core.providers.vad.base import VADProviderBase
@@ -39,8 +38,6 @@ class VADProvider(VADProviderBase):
def _init_connection_state(self, conn):
"""为连接初始化独立的 VAD 状态"""
- if not hasattr(conn, "_vad_opus_decoder"):
- conn._vad_opus_decoder = opuslib_next.Decoder(16000, 1)
if not hasattr(conn, "_vad_state"):
conn._vad_state = np.zeros((2, 1, 128), dtype=np.float32)
if not hasattr(conn, "_vad_context"):
@@ -48,14 +45,14 @@ class VADProvider(VADProviderBase):
def release_conn_resources(self, conn):
"""释放连接的 VAD 资源(连接关闭时调用)"""
- for attr in ("_vad_opus_decoder", "_vad_state", "_vad_context"):
+ for attr in ("_vad_state", "_vad_context"):
if hasattr(conn, attr):
try:
delattr(conn, attr)
except Exception:
pass
- def is_vad(self, conn, opus_packet):
+ def is_vad(self, conn, pcm_frame):
# 手动模式:直接返回True,不进行实时VAD检测,所有音频都缓存
if conn.client_listen_mode == "manual":
return True
@@ -63,7 +60,7 @@ class VADProvider(VADProviderBase):
try:
self._init_connection_state(conn)
- pcm_frame = conn._vad_opus_decoder.decode(opus_packet, 960)
+ # pcm_frame已经是处理后的PCM数据
conn.client_audio_buffer.extend(pcm_frame)
client_have_voice = False
@@ -115,7 +112,5 @@ class VADProvider(VADProviderBase):
conn.vad_last_voice_time = time.time() * 1000
return client_have_voice
- except opuslib_next.OpusError as e:
- logger.bind(tag=TAG).info(f"解码错误: {e}")
except Exception as e:
logger.bind(tag=TAG).error(f"Error processing audio packet: {e}")
diff --git a/main/xiaozhi-server/core/utils/prompt_manager.py b/main/xiaozhi-server/core/utils/prompt_manager.py
index af422529..edae6e9a 100644
--- a/main/xiaozhi-server/core/utils/prompt_manager.py
+++ b/main/xiaozhi-server/core/utils/prompt_manager.py
@@ -4,6 +4,8 @@
"""
import os
+import asyncio
+import threading
from typing import Dict, Any, TYPE_CHECKING
if TYPE_CHECKING:
@@ -169,12 +171,33 @@ class PromptManager:
if cached_weather is not None:
return cached_weather
- # 缓存未命中,调用get_weather函数获取
+ # 缓存未命中,调用 async get_weather 函数
+ # Windows ProactorEventLoop 不支持 run_coroutine_threadsafe().result()
+ # 因此用 call_soon_threadsafe 提交任务 + threading.Event 等待结果
+ # 注意:Event.wait() 只阻塞当前线程池线程,不阻塞主事件循环
from plugins_func.functions.get_weather import get_weather
from plugins_func.register import ActionResponse
- # 调用get_weather函数
- result = get_weather(conn, location=location, lang="zh_CN")
+ result_holder = []
+ exception_holder = []
+
+ async def _call():
+ try:
+ result_holder.append(
+ await get_weather(conn, location=location, lang="zh_CN")
+ )
+ except Exception as e:
+ exception_holder.append(e)
+ finally:
+ event.set()
+
+ event = threading.Event()
+ conn.loop.call_soon_threadsafe(lambda: asyncio.ensure_future(_call()))
+ if not event.wait(timeout=10):
+ raise TimeoutError("获取天气信息超时")
+ if exception_holder:
+ raise exception_holder[0]
+ result = result_holder[0]
if isinstance(result, ActionResponse):
weather_report = result.result
self.cache_manager.set(self.CacheType.WEATHER, location, weather_report)
diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py
index 077f7f4d..2ef7cda7 100644
--- a/main/xiaozhi-server/core/websocket_server.py
+++ b/main/xiaozhi-server/core/websocket_server.py
@@ -42,7 +42,7 @@ TAG = __name__
class WebSocketServer:
def __init__(self, config: dict):
self.config = config
- self.logger = setup_logging()
+ self.logger = setup_logging(config)
self.config_lock = asyncio.Lock()
modules = initialize_modules(
self.logger,
diff --git a/main/xiaozhi-server/plugins_func/functions/call_device.py b/main/xiaozhi-server/plugins_func/functions/call_device.py
index cfca1f33..e2b08fb5 100644
--- a/main/xiaozhi-server/plugins_func/functions/call_device.py
+++ b/main/xiaozhi-server/plugins_func/functions/call_device.py
@@ -1,9 +1,8 @@
"""呼叫设备工具"""
-import requests
-from typing import TYPE_CHECKING
-
+import httpx
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType, ActionResponse, Action
+from typing import TYPE_CHECKING
if TYPE_CHECKING:
from core.connection import ConnectionHandler
@@ -35,8 +34,9 @@ call_device_function_desc = {
}
-def _request_api(url: str, params: dict, headers: dict) -> requests.Response:
- return requests.get(url, params=params, headers=headers, timeout=10)
+async def _request_api(url: str, params: dict, headers: dict):
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ return await client.get(url, params=params, headers=headers)
def _failed_reply(msg: str) -> ActionResponse:
@@ -49,7 +49,7 @@ def _is_answering(conn: "ConnectionHandler") -> bool:
@register_function("call_device", call_device_function_desc, ToolType.SYSTEM_CTL)
-def call_device(conn: "ConnectionHandler", nickname: str):
+async def call_device(conn: "ConnectionHandler", nickname: str):
caller_mac = conn.headers.get("device-id")
if not caller_mac:
return _failed_reply("无法获取本机MAC地址")
@@ -71,13 +71,13 @@ def call_device(conn: "ConnectionHandler", nickname: str):
# 查询通讯录并发起呼叫
try:
- resp = _request_api(
+ resp = await _request_api(
f"{api_url}/device/address-book/call",
params=params,
headers=headers,
)
result = resp.json()
- except requests.RequestException as e:
+ except httpx.HTTPError as e:
logger.bind(tag=TAG).error(f"呼叫请求失败: {e}")
return _failed_reply("呼叫失败,请稍后再试")
diff --git a/main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py b/main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py
index cf64825b..94cec718 100644
--- a/main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py
+++ b/main/xiaozhi-server/plugins_func/functions/get_news_from_chinanews.py
@@ -1,5 +1,5 @@
import random
-import requests
+import httpx
import xml.etree.ElementTree as ET
from bs4 import BeautifulSoup
from config.logger import setup_logging
@@ -44,11 +44,11 @@ GET_NEWS_FROM_CHINANEWS_FUNCTION_DESC = {
}
-def fetch_news_from_rss(rss_url):
+async def fetch_news_from_rss(rss_url):
"""从RSS源获取新闻列表"""
try:
- response = requests.get(rss_url)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
+ response = await client.get(rss_url)
# 解析XML
root = ET.fromstring(response.content)
@@ -86,11 +86,11 @@ def fetch_news_from_rss(rss_url):
return []
-def fetch_news_detail(url):
+async def fetch_news_detail(url):
"""获取新闻详情页内容并总结"""
try:
- response = requests.get(url)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ response = await client.get(url)
soup = BeautifulSoup(response.content, "html.parser")
@@ -148,7 +148,7 @@ def map_category(category_text):
GET_NEWS_FROM_CHINANEWS_FUNCTION_DESC,
ToolType.SYSTEM_CTL,
)
-def get_news_from_chinanews(
+async def get_news_from_chinanews(
conn: "ConnectionHandler",
category: str = None,
detail: bool = False,
@@ -180,7 +180,7 @@ def get_news_from_chinanews(
logger.bind(tag=TAG).debug(f"获取新闻详情: {title}, URL={link}")
# 获取新闻详情
- detail_content = fetch_news_detail(link)
+ detail_content = await fetch_news_detail(link)
if not detail_content or detail_content == "无法获取详细内容":
return ActionResponse(
@@ -220,7 +220,7 @@ def get_news_from_chinanews(
)
# 获取新闻列表
- news_items = fetch_news_from_rss(rss_url)
+ news_items = await fetch_news_from_rss(rss_url)
if not news_items:
return ActionResponse(
diff --git a/main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py b/main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py
index cbf48cf0..8f0df016 100644
--- a/main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py
+++ b/main/xiaozhi-server/plugins_func/functions/get_news_from_newsnow.py
@@ -1,9 +1,8 @@
import random
-import requests
-import json
+import httpx
+from markitdown import MarkItDown
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType, ActionResponse, Action
-from markitdown import MarkItDown
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -110,7 +109,7 @@ GET_NEWS_FROM_NEWSNOW_FUNCTION_DESC = {
}
-def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
+async def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
"""从API获取新闻列表"""
try:
api_url = f"https://newsnow.busiyi.world/api/s?id={source}"
@@ -120,8 +119,8 @@ def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
api_url = news_config["url"] + source
headers = {"User-Agent": "Mozilla/5.0"}
- response = requests.get(api_url, headers=headers, timeout=10)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ response = await client.get(api_url, headers=headers)
data = response.json()
@@ -136,12 +135,12 @@ def fetch_news_from_api(conn: "ConnectionHandler", source="thepaper"):
return []
-def fetch_news_detail(url):
+async def fetch_news_detail(url):
"""获取新闻详情页内容并使用MarkItDown清理HTML"""
try:
headers = {"User-Agent": "Mozilla/5.0"}
- response = requests.get(url, headers=headers, timeout=10)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ response = await client.get(url, headers=headers)
# 使用MarkItDown清理HTML内容
md = MarkItDown(enable_plugins=False)
@@ -166,7 +165,7 @@ def fetch_news_detail(url):
GET_NEWS_FROM_NEWSNOW_FUNCTION_DESC,
ToolType.SYSTEM_CTL,
)
-def get_news_from_newsnow(
+async def get_news_from_newsnow(
conn: "ConnectionHandler",
source: str = "澎湃新闻",
detail: bool = False,
@@ -206,7 +205,7 @@ def get_news_from_newsnow(
)
# 获取新闻详情
- detail_content = fetch_news_detail(url)
+ detail_content = await fetch_news_detail(url)
if not detail_content or detail_content == "无法获取详细内容":
return ActionResponse(
@@ -248,7 +247,7 @@ def get_news_from_newsnow(
logger.bind(tag=TAG).info(f"获取新闻: 新闻源={source}({english_source_id})")
# 获取新闻列表
- news_items = fetch_news_from_api(conn, english_source_id)
+ news_items = await fetch_news_from_api(conn, english_source_id)
if not news_items:
return ActionResponse(
diff --git a/main/xiaozhi-server/plugins_func/functions/get_weather.py b/main/xiaozhi-server/plugins_func/functions/get_weather.py
index 88870529..c0c4118e 100644
--- a/main/xiaozhi-server/plugins_func/functions/get_weather.py
+++ b/main/xiaozhi-server/plugins_func/functions/get_weather.py
@@ -1,4 +1,4 @@
-import requests
+import httpx
from bs4 import BeautifulSoup
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType, ActionResponse, Action
@@ -111,20 +111,23 @@ WEATHER_CODE_MAP = {
}
-def fetch_city_info(location, api_key, api_host):
+async def fetch_city_info(location, api_key, api_host):
url = f"https://{api_host}/geo/v2/city/lookup?key={api_key}&location={location}&lang=zh"
- response = requests.get(url, headers=HEADERS).json()
- if response.get("error") is not None:
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
+ response = await client.get(url, headers=HEADERS)
+ data = response.json()
+ if data.get("error") is not None:
logger.bind(tag=TAG).error(
- f"获取天气失败,原因:{response.get('error', {}).get('detail')}"
+ f"获取天气失败,原因:{data.get('error', {}).get('detail')}"
)
return None
- return response.get("location", [])[0] if response.get("location") else None
+ return data.get("location", [])[0] if data.get("location") else None
-def fetch_weather_page(url):
- response = requests.get(url, headers=HEADERS)
- return BeautifulSoup(response.text, "html.parser") if response.ok else None
+async def fetch_weather_page(url):
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ response = await client.get(url, headers=HEADERS)
+ return BeautifulSoup(response.text, "html.parser") if response.status_code == 200 else None
def parse_weather_info(soup):
@@ -159,7 +162,7 @@ def parse_weather_info(soup):
@register_function("get_weather", GET_WEATHER_FUNCTION_DESC, ToolType.SYSTEM_CTL)
-def get_weather(conn: "ConnectionHandler", location: str = None, lang: str = "zh_CN"):
+async def get_weather(conn: "ConnectionHandler", location: str = None, lang: str = "zh_CN"):
from core.utils.cache.manager import cache_manager, CacheType
weather_config = conn.config.get("plugins", {}).get("get_weather", {})
@@ -195,12 +198,12 @@ def get_weather(conn: "ConnectionHandler", location: str = None, lang: str = "zh
return ActionResponse(Action.REQLLM, cached_weather_report, None)
# 缓存未命中,获取实时天气数据
- city_info = fetch_city_info(location, api_key, api_host)
+ city_info = await fetch_city_info(location, api_key, api_host)
if not city_info:
return ActionResponse(
Action.REQLLM, f"未找到相关的城市: {location},请确认地点是否正确", None
)
- soup = fetch_weather_page(city_info["fxLink"])
+ soup = await fetch_weather_page(city_info["fxLink"])
if not soup:
return ActionResponse(Action.REQLLM, None, "请求失败")
city_name, current_abstract, current_basic, temps_list = parse_weather_info(soup)
diff --git a/main/xiaozhi-server/plugins_func/functions/hass_get_state.py b/main/xiaozhi-server/plugins_func/functions/hass_get_state.py
index 86e866c1..e56fe5a8 100644
--- a/main/xiaozhi-server/plugins_func/functions/hass_get_state.py
+++ b/main/xiaozhi-server/plugins_func/functions/hass_get_state.py
@@ -1,8 +1,7 @@
-from plugins_func.register import register_function, ToolType, ActionResponse, Action
-from plugins_func.functions.hass_init import initialize_hass_handler
+import httpx
from config.logger import setup_logging
-import asyncio
-import requests
+from plugins_func.functions.hass_init import initialize_hass_handler
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -31,11 +30,11 @@ hass_get_state_function_desc = {
@register_function("hass_get_state", hass_get_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_get_state(conn: "ConnectionHandler", entity_id=""):
+async def hass_get_state(conn: "ConnectionHandler", entity_id=""):
try:
- ha_response = handle_hass_get_state(conn, entity_id)
+ ha_response = await handle_hass_get_state(conn, entity_id)
return ActionResponse(Action.REQLLM, ha_response, None)
- except asyncio.TimeoutError:
+ except httpx.TimeoutException:
logger.bind(tag=TAG).error("获取Home Assistant状态超时")
return ActionResponse(Action.ERROR, "请求超时", None)
except Exception as e:
@@ -44,13 +43,16 @@ def hass_get_state(conn: "ConnectionHandler", entity_id=""):
return ActionResponse(Action.ERROR, error_msg, None)
-def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
+async def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
ha_config = initialize_hass_handler(conn)
api_key = ha_config.get("api_key")
base_url = ha_config.get("base_url")
url = f"{base_url}/api/states/{entity_id}"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
- response = requests.get(url, headers=headers, timeout=5)
+
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
+ response = await client.get(url, headers=headers)
+
if response.status_code == 200:
responsetext = "设备状态:" + response.json()["state"] + " "
logger.bind(tag=TAG).info(f"api返回内容: {response.json()}")
@@ -92,8 +94,5 @@ def handle_hass_get_state(conn: "ConnectionHandler", entity_id):
)
logger.bind(tag=TAG).info(f"查询返回内容: {responsetext}")
return responsetext
- # return response.json()['attributes']
- # response.attributes
-
else:
return f"切换失败,错误码: {response.status_code}"
diff --git a/main/xiaozhi-server/plugins_func/functions/hass_play_music.py b/main/xiaozhi-server/plugins_func/functions/hass_play_music.py
index 7bd9f2a1..0d0b3eb7 100644
--- a/main/xiaozhi-server/plugins_func/functions/hass_play_music.py
+++ b/main/xiaozhi-server/plugins_func/functions/hass_play_music.py
@@ -1,8 +1,7 @@
-from plugins_func.register import register_function, ToolType, ActionResponse, Action
-from plugins_func.functions.hass_init import initialize_hass_handler
+import httpx
from config.logger import setup_logging
-import asyncio
-import requests
+from plugins_func.functions.hass_init import initialize_hass_handler
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -37,18 +36,17 @@ hass_play_music_function_desc = {
@register_function(
"hass_play_music", hass_play_music_function_desc, ToolType.SYSTEM_CTL
)
-def hass_play_music(conn: "ConnectionHandler", entity_id="", media_content_id="random"):
+async def hass_play_music(conn: "ConnectionHandler", entity_id="", media_content_id="random"):
try:
- # 执行音乐播放命令
- future = asyncio.run_coroutine_threadsafe(
- handle_hass_play_music(conn, entity_id, media_content_id), conn.loop
- )
- ha_response = future.result()
+ result = await handle_hass_play_music(conn, entity_id, media_content_id)
return ActionResponse(
- action=Action.RESPONSE, result="退出意图已处理", response=ha_response
+ action=Action.RECORD, result="指令已接收", response=result
)
except Exception as e:
logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
+ return ActionResponse(
+ action=Action.RESPONSE, result=str(e), response="播放音乐时出错了"
+ )
async def handle_hass_play_music(
@@ -60,7 +58,10 @@ async def handle_hass_play_music(
url = f"{base_url}/api/services/music_assistant/play_media"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
data = {"entity_id": entity_id, "media_id": media_content_id}
- response = requests.post(url, headers=headers, json=data)
+
+ async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=3.0)) as client:
+ response = await client.post(url, headers=headers, json=data)
+
if response.status_code == 200:
return f"正在播放{media_content_id}的音乐"
else:
diff --git a/main/xiaozhi-server/plugins_func/functions/hass_set_state.py b/main/xiaozhi-server/plugins_func/functions/hass_set_state.py
index 62807f9b..293ca9e2 100644
--- a/main/xiaozhi-server/plugins_func/functions/hass_set_state.py
+++ b/main/xiaozhi-server/plugins_func/functions/hass_set_state.py
@@ -1,8 +1,7 @@
-from plugins_func.register import register_function, ToolType, ActionResponse, Action
-from plugins_func.functions.hass_init import initialize_hass_handler
+import httpx
from config.logger import setup_logging
-import asyncio
-import requests
+from plugins_func.functions.hass_init import initialize_hass_handler
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -54,13 +53,13 @@ hass_set_state_function_desc = {
@register_function("hass_set_state", hass_set_state_function_desc, ToolType.SYSTEM_CTL)
-def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
+async def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
if state is None:
state = {}
try:
- ha_response = handle_hass_set_state(conn, entity_id, state)
+ ha_response = await handle_hass_set_state(conn, entity_id, state)
return ActionResponse(Action.REQLLM, ha_response, None)
- except asyncio.TimeoutError:
+ except httpx.TimeoutException:
logger.bind(tag=TAG).error("设置Home Assistant状态超时")
return ActionResponse(Action.ERROR, "请求超时", None)
except Exception as e:
@@ -69,7 +68,7 @@ def hass_set_state(conn: "ConnectionHandler", entity_id="", state=None):
return ActionResponse(Action.ERROR, error_msg, None)
-def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
+async def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
ha_config = initialize_hass_handler(conn)
api_key = ha_config.get("api_key")
base_url = ha_config.get("base_url")
@@ -169,7 +168,10 @@ def handle_hass_set_state(conn: "ConnectionHandler", entity_id, state):
data = {"entity_id": entity_id, arg: value}
url = f"{base_url}/api/services/{domain}/{action}"
headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
- response = requests.post(url, headers=headers, json=data, timeout=5) # 设置5秒超时
+
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0)) as client:
+ response = await client.post(url, headers=headers, json=data)
+
logger.bind(tag=TAG).info(
f"设置状态:{description},url:{url},return_code:{response.status_code}"
)
diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py
index 18c605d2..fd21f9f8 100644
--- a/main/xiaozhi-server/plugins_func/functions/play_music.py
+++ b/main/xiaozhi-server/plugins_func/functions/play_music.py
@@ -5,10 +5,8 @@ import random
import difflib
import traceback
from pathlib import Path
-from core.handle.sendAudioHandle import send_stt_message
-from plugins_func.register import register_function, ToolType, ActionResponse, Action
-from core.utils.dialogue import Message
from core.providers.tts.dto.dto import TTSMessageDTO, SentenceType, ContentType
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -38,34 +36,12 @@ play_music_function_desc = {
@register_function("play_music", play_music_function_desc, ToolType.SYSTEM_CTL)
-def play_music(conn: "ConnectionHandler", song_name: str):
+async def play_music(conn: "ConnectionHandler", song_name: str):
try:
music_intent = (
f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
)
-
- # 检查事件循环状态
- if not conn.loop.is_running():
- conn.logger.bind(tag=TAG).error("事件循环未运行,无法提交任务")
- return ActionResponse(
- action=Action.RESPONSE, result="系统繁忙", response="请稍后再试"
- )
-
- # 提交异步任务
- task = conn.loop.create_task(
- handle_music_command(conn, music_intent) # 封装异步逻辑
- )
-
- # 非阻塞回调处理
- def handle_done(f):
- try:
- f.result() # 可在此处理成功逻辑
- conn.logger.bind(tag=TAG).info("播放完成")
- except Exception as e:
- conn.logger.bind(tag=TAG).error(f"播放失败: {e}")
-
- task.add_done_callback(handle_done)
-
+ await handle_music_command(conn, music_intent)
return ActionResponse(
action=Action.RECORD, result="指令已接收", response="正在为您播放音乐"
)
diff --git a/main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py b/main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py
index 79d546ee..40abf8c4 100644
--- a/main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py
+++ b/main/xiaozhi-server/plugins_func/functions/search_from_ragflow.py
@@ -1,5 +1,5 @@
-import requests
-import sys
+import json
+import httpx
from config.logger import setup_logging
from plugins_func.register import register_function, ToolType, ActionResponse, Action
from typing import TYPE_CHECKING
@@ -28,7 +28,7 @@ SEARCH_FROM_RAGFLOW_FUNCTION_DESC = {
@register_function(
"search_from_ragflow", SEARCH_FROM_RAGFLOW_FUNCTION_DESC, ToolType.SYSTEM_CTL
)
-def search_from_ragflow(conn: "ConnectionHandler", question=None):
+async def search_from_ragflow(conn: "ConnectionHandler", question=None):
# 确保字符串参数正确处理编码
if question and isinstance(question, str):
# 确保问题参数是UTF-8编码的字符串
@@ -49,13 +49,8 @@ def search_from_ragflow(conn: "ConnectionHandler", question=None):
try:
# 使用ensure_ascii=False确保JSON序列化时正确处理中文
- response = requests.post(
- url,
- json=payload,
- headers=headers,
- timeout=5,
- verify=False,
- )
+ async with httpx.AsyncClient(timeout=httpx.Timeout(5.0, connect=3.0), verify=False) as client:
+ response = await client.post(url, json=payload, headers=headers)
# 显式设置响应的编码为utf-8
response.encoding = "utf-8"
@@ -64,7 +59,6 @@ def search_from_ragflow(conn: "ConnectionHandler", question=None):
# 先获取文本内容,然后手动处理JSON解码
response_text = response.text
- import json
result = json.loads(response_text)
@@ -110,48 +104,30 @@ def search_from_ragflow(conn: "ConnectionHandler", question=None):
context_text = "根据知识库查询结果,没有相关信息。"
return ActionResponse(Action.REQLLM, context_text, None)
- except requests.exceptions.RequestException as e:
- # 网络请求异常
- error_type = type(e).__name__
- logger.bind(tag=TAG).error(
- f"RAGflow网络请求失败,异常类型:{error_type},详情:{str(e)}"
- )
-
- # 根据异常类型提供更详细的错误信息和解决方案
- if isinstance(e, requests.exceptions.ConnectTimeout):
- error_response = "RAG接口连接超时(5秒)"
- error_response += "\n可能原因:RAGflow服务未启动或网络连接问题"
- error_response += "\n解决方案:请检查RAGflow服务状态和网络连接"
-
- elif isinstance(e, requests.exceptions.ConnectionError):
- error_response = "无法连接到RAG接口"
- error_response += "\n可能原因:RAGflow服务地址错误或服务未运行"
- error_response += "\n解决方案:请检查RAGflow服务地址配置和服务状态"
-
- elif isinstance(e, requests.exceptions.Timeout):
- error_response = "RAG接口请求超时"
- error_response += "\n可能原因:RAGflow服务响应缓慢或网络延迟"
- error_response += "\n解决方案:请稍后重试或检查RAGflow服务性能"
-
- elif isinstance(e, requests.exceptions.HTTPError):
- # 处理HTTP错误状态码
- if hasattr(e.response, "status_code"):
- status_code = e.response.status_code
- error_response = f"RAG接口HTTP错误(状态码:{status_code})"
-
- # 尝试获取响应内容中的错误信息
- try:
- error_detail = e.response.json().get("error", {}).get("message", "")
- if error_detail:
- error_response += f"\n错误详情:{error_detail}"
- except:
- pass
- else:
- error_response = f"RAG接口HTTP异常:{str(e)}"
+ except httpx.TimeoutException as e:
+ error_response = "RAG接口请求超时"
+ error_response += "\n可能原因:RAGflow服务响应缓慢或网络延迟"
+ error_response += "\n解决方案:请稍后重试或检查RAGflow服务性能"
+ return ActionResponse(Action.RESPONSE, None, error_response)
+ except httpx.HTTPStatusError as e:
+ if hasattr(e.response, "status_code"):
+ status_code = e.response.status_code
+ error_response = f"RAG接口HTTP错误(状态码:{status_code})"
+ try:
+ error_detail = e.response.json().get("error", {}).get("message", "")
+ if error_detail:
+ error_response += f"\n错误详情:{error_detail}"
+ except:
+ pass
else:
- error_response = f"RAG接口网络异常({error_type}):{str(e)}"
+ error_response = f"RAG接口HTTP异常:{str(e)}"
+ return ActionResponse(Action.RESPONSE, None, error_response)
+ except httpx.HTTPError as e:
+ error_response = "无法连接到RAG接口"
+ error_response += "\n可能原因:RAGflow服务地址错误或服务未运行"
+ error_response += "\n解决方案:请检查RAGflow服务地址配置和服务状态"
return ActionResponse(Action.RESPONSE, None, error_response)
except Exception as e:
diff --git a/main/xiaozhi-server/plugins_func/functions/web_search.py b/main/xiaozhi-server/plugins_func/functions/web_search.py
index 6411a0ed..519f8b2b 100644
--- a/main/xiaozhi-server/plugins_func/functions/web_search.py
+++ b/main/xiaozhi-server/plugins_func/functions/web_search.py
@@ -1,4 +1,4 @@
-import requests
+import httpx
from config.logger import setup_logging
from plugins_func.register import (
register_function,
@@ -37,7 +37,7 @@ WEB_SEARCH_FUNCTION_DESC = {
}
-def _search_metaso(api_key: str, query: str, max_results: int) -> str:
+async def _search_metaso(api_key: str, query: str, max_results: int) -> str:
"""调用秘塔搜索API"""
url = "https://metaso.cn/api/v1/search"
headers = {
@@ -54,8 +54,8 @@ def _search_metaso(api_key: str, query: str, max_results: int) -> str:
"conciseSnippet": False,
}
logger.bind(tag=TAG).debug(f"秘塔搜索请求 | URL: {url} | payload: {payload}")
- response = requests.post(url, json=payload, headers=headers, timeout=15)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=3.0)) as client:
+ response = await client.post(url, json=payload, headers=headers)
data = response.json()
logger.bind(tag=TAG).debug(f"秘塔搜索响应 | status: {response.status_code}")
@@ -77,7 +77,7 @@ def _search_metaso(api_key: str, query: str, max_results: int) -> str:
return "\n".join(lines)
-def _search_tavily(api_key: str, query: str, max_results: int) -> str:
+async def _search_tavily(api_key: str, query: str, max_results: int) -> str:
"""调用Tavily搜索API"""
url = "https://api.tavily.com/search"
headers = {
@@ -91,8 +91,8 @@ def _search_tavily(api_key: str, query: str, max_results: int) -> str:
"include_answer": "advanced",
}
logger.bind(tag=TAG).debug(f"Tavily搜索请求 | URL: {url} | payload: {payload}")
- response = requests.post(url, json=payload, headers=headers, timeout=15)
- response.raise_for_status()
+ async with httpx.AsyncClient(timeout=httpx.Timeout(15.0, connect=3.0)) as client:
+ response = await client.post(url, json=payload, headers=headers)
data = response.json()
logger.bind(tag=TAG).debug(f"Tavily搜索响应 | status: {response.status_code} | data: {data}")
@@ -113,7 +113,7 @@ def _search_tavily(api_key: str, query: str, max_results: int) -> str:
@register_function("web_search", WEB_SEARCH_FUNCTION_DESC, ToolType.SYSTEM_CTL)
-def web_search(conn: "ConnectionHandler", query: str = None):
+async def web_search(conn: "ConnectionHandler", query: str = None):
logger.bind(tag=TAG).info(f"web_search 被调用 | query={query}")
if not query:
return ActionResponse(Action.REQLLM, "请提供搜索关键词。", None)
@@ -131,24 +131,22 @@ def web_search(conn: "ConnectionHandler", query: str = None):
None,
)
- if provider == "metaso":
- search_fn = lambda: _search_metaso(api_key, query, max_results)
- elif provider == "tavily":
- search_fn = lambda: _search_tavily(api_key, query, max_results)
- else:
- return ActionResponse(
- Action.REQLLM,
- f"联网搜索功能未配置或配置的搜索源无效(当前:{provider}),请检查配置。",
- None,
- )
-
try:
- result_text = search_fn()
+ if provider == "metaso":
+ result_text = await _search_metaso(api_key, query, max_results)
+ elif provider == "tavily":
+ result_text = await _search_tavily(api_key, query, max_results)
+ else:
+ return ActionResponse(
+ Action.REQLLM,
+ f"联网搜索功能未配置或配置的搜索源无效(当前:{provider}),请检查配置。",
+ None,
+ )
logger.bind(tag=TAG).info(f"搜索结果组装完成:\n{result_text}")
- except requests.exceptions.Timeout:
+ except httpx.TimeoutException:
logger.bind(tag=TAG).error("联网搜索请求超时")
result_text = "联网搜索请求超时,请稍后重试。"
- except requests.exceptions.RequestException as e:
+ except httpx.HTTPStatusError as e:
logger.bind(tag=TAG).error(f"联网搜索请求失败: {e}")
result_text = "联网搜索请求失败,请稍后重试。"
except Exception as e: