fix: 修复ha读取异常

This commit is contained in:
Sakura-RanChen committed 2026-11-05 15:36:04 +08:00
1 parent 70f1a7bfd8
commit 9f5f520576
3 files changed
+42 -32

No files matched your search

@@ -5,6 +5,7 @@ if TYPE_CHECKING:
from core.connection import ConnectionHandler from core.connection import ConnectionHandler
from ..base import IntentProviderBase from ..base import IntentProviderBase
from plugins_func.functions.play_music import initialize_music_handler from plugins_func.functions.play_music import initialize_music_handler
from plugins_func.functions.hass_init import _devices_text, _hass_plugin_config
from config.logger import setup_logging from config.logger import setup_logging
from core.utils.util import get_system_error_response from core.utils.util import get_system_error_response
import re import re
@@ -181,16 +182,11 @@ class IntentProvider(IntentProviderBase):
music_file_names = music_config["music_file_names"] music_file_names = music_config["music_file_names"]
prompt_music = f"{self.promot}\n<musicNames>{music_file_names}\n</musicNames>" prompt_music = f"{self.promot}\n<musicNames>{music_file_names}\n</musicNames>"
home_assistant_cfg = conn.config["plugins"].get("home_assistant") plugin_config = _hass_plugin_config(conn.config.get("plugins", {}))
if home_assistant_cfg: devices_text = _devices_text(plugin_config.get("devices", ""))
devices = home_assistant_cfg.get("devices", []) if devices_text:
else:
devices = []
if len(devices) > 0:
hass_prompt = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n" hass_prompt = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n"
for device in devices: prompt_music += hass_prompt + devices_text + "\n"
hass_prompt += device + "\n"
prompt_music += hass_prompt
logger.bind(tag=TAG).debug(f"User prompt: {prompt_music}") logger.bind(tag=TAG).debug(f"User prompt: {prompt_music}")
@@ -1,10 +1,19 @@
from config.logger import setup_logging from config.logger import setup_logging
from core.utils.util import check_model_key from core.utils.util import check_model_key
from plugins_func.register import module_func_map
TAG = __name__ TAG = __name__
logger = setup_logging() logger = setup_logging()
def _expand_module_names(funcs):
"""把 functions 列表里的模块名(如 hass_state)展开成具体函数名,与工具加载口径一致。"""
expanded = []
for name in funcs:
expanded.extend(module_func_map.get(name, [name]))
return expanded
def _devices_text(devices) -> str: def _devices_text(devices) -> str:
"""把 devices 配置转成拼进提示词的文字。 """把 devices 配置转成拼进提示词的文字。
@@ -12,29 +21,33 @@ def _devices_text(devices) -> str:
网页端智控台和 HA 集成文档是一段字符串,原样用;其它(如 None)当空。 网页端智控台和 HA 集成文档是一段字符串,原样用;其它(如 None)当空。
""" """
if isinstance(devices, (list, tuple)): if isinstance(devices, (list, tuple)):
return "\n".join(str(d).strip() for d in devices if str(d).strip()) return "\n".join(
str(d).strip() for d in devices if d is not None and str(d).strip()
)
return devices if isinstance(devices, str) else "" return devices if isinstance(devices, str) else ""
def _hass_plugin_config(plugins_config: dict) -> dict:
"""取 HA 插件配置:优先本地 config.yaml 的 home_assistant,其次智控台下发的 hass_state。"""
for key in ("home_assistant", "hass_state"):
cfg = plugins_config.get(key)
if cfg:
return cfg
return {}
def append_devices_to_prompt(conn): def append_devices_to_prompt(conn):
if conn.intent_type == "function_call": if conn.intent_type == "function_call":
funcs = conn.config["Intent"][conn.config["selected_module"]["Intent"]].get( funcs = _expand_module_names(
"functions", [] conn.config["Intent"][conn.config["selected_module"]["Intent"]].get(
) "functions", []
)
# 安全地获取插件配置
plugins_config = conn.config.get("plugins", {})
config_source = (
"home_assistant"
if plugins_config.get("home_assistant")
else "hass_state"
) )
if "hass_get_state" in funcs or "hass_set_state" in funcs: if "hass_get_state" in funcs or "hass_set_state" in funcs:
prompt = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n" prompt = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n"
deviceStr = _devices_text( plugin_config = _hass_plugin_config(conn.config.get("plugins", {}))
plugins_config.get(config_source, {}).get("devices", "") deviceStr = _devices_text(plugin_config.get("devices", ""))
)
conn.prompt += prompt + deviceStr + "\n" conn.prompt += prompt + deviceStr + "\n"
# 更新提示词 # 更新提示词
conn.dialogue.update_system_message(conn.prompt) conn.dialogue.update_system_message(conn.prompt)
@@ -45,17 +58,10 @@ def initialize_hass_handler(conn):
if not conn.load_function_plugin: if not conn.load_function_plugin:
return ha_config return ha_config
# 安全地获取插件配置 plugin_config = _hass_plugin_config(conn.config.get("plugins", {}))
plugins_config = conn.config.get("plugins", {}) if not plugin_config:
# 确定配置来源
config_source = (
"home_assistant" if plugins_config.get("home_assistant") else "hass_state"
)
if not plugins_config.get(config_source):
return ha_config return ha_config
# 统一获取配置
plugin_config = plugins_config[config_source]
ha_config["base_url"] = plugin_config.get("base_url") ha_config["base_url"] = plugin_config.get("base_url")
ha_config["api_key"] = plugin_config.get("api_key") ha_config["api_key"] = plugin_config.get("api_key")
@@ -5,6 +5,7 @@ can only concatenate str (not "list") to str,设备清单进不了提示词,
""" """
import pytest import pytest
import plugins_func.functions.hass_state # noqa: F401 注册 hass_state 模块,填充 module_func_map
from plugins_func.functions.hass_init import append_devices_to_prompt from plugins_func.functions.hass_init import append_devices_to_prompt
HEADER = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n" HEADER = "\n下面是我家智能设备列表(位置,设备名,entity_id),可以通过homeassistant控制\n"
@@ -62,6 +63,13 @@ def test_mobile_console_hass_state_list():
assert conn.prompt.endswith(HEADER + "客厅,玩具灯,switch.toy_lamp\n") assert conn.prompt.endswith(HEADER + "客厅,玩具灯,switch.toy_lamp\n")
def test_module_name_in_functions():
# functions 里写模块名 hass_state(config.yaml 注释的写法),展开后应能匹配
conn = FakeConn(_plugin(devices=["客厅,玩具灯,switch.toy_lamp"]), functions=("hass_state",))
append_devices_to_prompt(conn)
assert conn.prompt.endswith(HEADER + "客厅,玩具灯,switch.toy_lamp\n")
@pytest.mark.parametrize("plugin", [_plugin(), _plugin(devices=None), _plugin(devices=[])]) @pytest.mark.parametrize("plugin", [_plugin(), _plugin(devices=None), _plugin(devices=[])])
def test_missing_or_empty_devices_does_not_raise(plugin): def test_missing_or_empty_devices_does_not_raise(plugin):
conn = FakeConn(plugin) conn = FakeConn(plugin)