From b77ef426e3b1392f6b4d0afe1e93899ee10db2b5 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Sat, 24 Oct 2026 07:57:49 +0000 Subject: [PATCH] =?UTF-8?q?=E5=8A=A0=E7=94=A8=E4=BE=8B=E7=9B=AF=E6=8F=90?= =?UTF-8?q?=E7=A4=BA=E8=AF=8D=E9=87=8C=E7=9A=84=E5=A4=A9=E6=B0=94=E4=B8=89?= =?UTF-8?q?=E7=A7=8D=E6=83=85=E5=86=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 没配凭据、缓存里留着空条目、凭据齐全这三种情况各有用例，空条目那种以前会挡住取天气，现在下一次就能拿到报告。 --- .../core/utils/test_prompt_manager_weather.py | 177 ++++++++++++++++++ 1 file changed, 177 insertions(+) create mode 100644 main/xiaozhi-server/tests/core/utils/test_prompt_manager_weather.py diff --git a/main/xiaozhi-server/tests/core/utils/test_prompt_manager_weather.py b/main/xiaozhi-server/tests/core/utils/test_prompt_manager_weather.py new file mode 100644 index 000000000..7a1ce786e --- /dev/null +++ b/main/xiaozhi-server/tests/core/utils/test_prompt_manager_weather.py @@ -0,0 +1,177 @@ +"""提示词里的天气内容 + +天气有两条路进模型：工具调用那条走插件，提示词那条走 {weather_info} 占位符。 +这条用例盯的是占位符那条：凭据没配好时它得给用户一句说明，不能留空让模型 +自己编天气；取到的报告也不能被空条目挡住。 +""" +from __future__ import annotations + +import asyncio +import threading +from pathlib import Path + +import pytest + +_PROJECT_ROOT = Path(__file__).resolve().parents[3] + +# 真插件没配凭据时给用户的提示语，里面带着给开发者看的文档路径。 +# 提示词里不能出现这个路径，模型会照着念给终端用户 +CREDENTIAL_HINT = "尚未配置和风天气 api_key，请参考 docs/weather-integration.md 申请后填入" +WEATHER_REPORT = "您查询的位置是：北京\n\n当前天气: 晴\n" + + +@pytest.fixture(autouse=True) +def _config_file(): + """import 链路要 data/.config.yaml 在场，缺了就现造一个空的""" + config_file = _PROJECT_ROOT / "data" / ".config.yaml" + if not config_file.exists(): + config_file.parent.mkdir(parents=True, exist_ok=True) + config_file.write_text("{}\n", encoding="utf-8") + yield + + +class _FakeLoop: + """给 _get_weather_info 用的假事件循环 + +

+ 那边用 call_soon_threadsafe 提交协程，这里起一个真循环在后台线程里跑， + 协程里的代码原样执行。 + """ + + def __init__(self): + self._loop = asyncio.new_event_loop() + self._thread = threading.Thread(target=self._loop.run_forever, daemon=True) + self._thread.start() + + def call_soon_threadsafe(self, callback): + self._loop.call_soon_threadsafe(callback) + + def close(self): + self._loop.call_soon_threadsafe(self._loop.stop) + self._thread.join(timeout=5) + self._loop.close() + + +class _FakeConn: + def __init__(self, loop, api_key: str): + self.loop = loop + self.config = { + "plugins": { + "get_weather": { + "api_key": api_key, + "api_host": "example-host", + } + } + } + self.client_ip = "1.2.3.4" + self.device_id = "test-device" + + +def _install_fake_plugin(monkeypatch): + """把真插件换成假的：按有没有 api_key 走两种返回""" + import plugins_func.functions.get_weather as weather_module + from plugins_func.register import Action, ActionResponse + + async def _fake_get_weather(conn, location=None, lang="zh_CN"): + api_key = str(conn.config.get("plugins", {}).get("get_weather", {}).get("api_key") or "") + if not api_key.strip(): + return ActionResponse(Action.RESPONSE, None, CREDENTIAL_HINT) + return ActionResponse(Action.REQLLM, WEATHER_REPORT, None) + + monkeypatch.setattr(weather_module, "get_weather", _fake_get_weather) + + +@pytest.fixture +def _manager(tmp_path): + """一个只带天气占位符的提示词管理器""" + from core.utils.prompt_manager import PromptManager + + template = tmp_path / "prompt.txt" + template.write_text("当前天气：{{weather_info}}", encoding="utf-8") + return PromptManager({"prompt_template": str(template)}) + + +@pytest.fixture +def _cache(): + """全局缓存，测试之间要清干净""" + from core.utils.cache.manager import cache_manager, CacheType + + cache_manager.clear(CacheType.WEATHER) + cache_manager.clear(CacheType.LOCATION) + yield cache_manager + cache_manager.clear(CacheType.WEATHER) + cache_manager.clear(CacheType.LOCATION) + + +def _build_prompt(manager, conn): + """走一遍真实的两步：先更新上下文，再拼提示词""" + manager.update_context_info(conn, conn.client_ip) + return manager.build_enhanced_prompt("你是助手", conn.device_id, conn.client_ip) + + +def test_unconfigured_weather_gives_notice_and_caches_nothing( + monkeypatch, _manager, _cache +): + """凭据没配好：提示词里要有说明，缓存里不能留下空条目""" + from core.utils.cache.manager import CacheType + + _install_fake_plugin(monkeypatch) + loop = _FakeLoop() + conn = _FakeConn(loop, api_key="") + _cache.set(CacheType.LOCATION, conn.client_ip, "北京") + + try: + prompt = _build_prompt(_manager, conn) + finally: + loop.close() + + assert "未配置" in prompt + # 给开发者的文档路径不能进提示词，模型会照着念给终端用户 + assert "docs/weather-integration.md" not in prompt + # 天气内容不能是空的，空着模型会自己编一个 + assert prompt.strip() != "当前天气：" + # 空结果不写缓存 + assert _cache.get(CacheType.WEATHER, "北京") is None + + +@pytest.mark.parametrize("stale", [None, ""]) +def test_empty_cached_entry_does_not_block_weather( + monkeypatch, _manager, _cache, stale +): + """缓存里留着空条目时，配好凭据后下一次就该出天气""" + from core.utils.cache.manager import CacheType + + _install_fake_plugin(monkeypatch) + loop = _FakeLoop() + conn = _FakeConn(loop, api_key="not-a-real-key") + _cache.set(CacheType.LOCATION, conn.client_ip, "北京") + _cache.set(CacheType.WEATHER, "北京", stale) + + try: + prompt = _build_prompt(_manager, conn) + finally: + loop.close() + + assert "当前天气: 晴" in prompt + assert _cache.get(CacheType.WEATHER, "北京") == WEATHER_REPORT + + +def test_configured_weather_report_goes_to_placeholder( + monkeypatch, _manager, _cache +): + """凭据齐全：报告照旧进占位符，也照旧进缓存""" + from core.utils.cache.manager import CacheType + + _install_fake_plugin(monkeypatch) + loop = _FakeLoop() + conn = _FakeConn(loop, api_key="not-a-real-key") + _cache.set(CacheType.LOCATION, conn.client_ip, "北京") + + try: + prompt = _build_prompt(_manager, conn) + finally: + loop.close() + + assert "当前天气: 晴" in prompt + assert "未配置" not in prompt + assert _cache.get(CacheType.WEATHER, "北京") == WEATHER_REPORT