From 59f6566cfb4dad1d29127792b55520c9d98629ae Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Tue, 7 Jul 2026 14:45:50 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8C=E6=AD=A5=E5=B7=A5=E5=85=B7=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E5=8C=96=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tools/server_plugins/plugin_executor.py | 5 ++ .../core/utils/prompt_manager.py | 29 ++++++- .../plugins_func/functions/call_device.py | 16 ++-- .../functions/get_news_from_chinanews.py | 20 ++--- .../functions/get_news_from_newsnow.py | 23 +++--- .../plugins_func/functions/get_weather.py | 27 ++++--- .../plugins_func/functions/hass_get_state.py | 23 +++--- .../plugins_func/functions/hass_play_music.py | 29 +++---- .../plugins_func/functions/hass_set_state.py | 20 ++--- .../plugins_func/functions/play_music.py | 30 +------- .../functions/search_from_ragflow.py | 76 +++++++------------ .../plugins_func/functions/web_search.py | 42 +++++----- 12 files changed, 156 insertions(+), 184 deletions(-) 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/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/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 a386493f..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,7 +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 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: @@ -36,23 +36,11 @@ 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: - task = conn.loop.create_task( - handle_hass_play_music(conn, entity_id, media_content_id) - ) - - def handle_done(f): - try: - f.result() - logger.bind(tag=TAG).info("音乐播放完成") - except Exception as e: - logger.bind(tag=TAG).error(f"音乐播放失败: {e}") - - task.add_done_callback(handle_done) - + result = await handle_hass_play_music(conn, entity_id, media_content_id) return ActionResponse( - action=Action.RECORD, result="指令已接收", response="正在为您播放音乐" + action=Action.RECORD, result="指令已接收", response=result ) except Exception as e: logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}") @@ -70,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: