"""Полный цикл запроса через Assistant: модель, инструменты памяти, откат без инструментов.""" import time from collections.abc import Callable from pathlib import Path import numpy as np import pytest from PySide6.QtCore import QCoreApplication, QObject, Signal from PySide6.QtWidgets import QApplication from agr_assistent.commands import actions from agr_assistent.commands.catalog import CommandCatalog from agr_assistent.config import LLMConfig, ProviderConfig from agr_assistent.core.assistant import Assistant from agr_assistent.core.memory import MemoryStore from agr_assistent.core.speech import Speaker from tests.fake_llm import FakeLLMServer, Reply class _SilentEngine: sample_rate = 24000 def load(self) -> None: pass def synthesize(self, text: str) -> np.ndarray: return np.zeros(1, dtype=np.float32) class _SilentPlayer: def play(self, *args: object) -> None: pass def finish(self) -> None: pass def abort(self) -> None: pass def _wait_until(condition: Callable[[], bool], timeout: float = 5.0) -> None: deadline = time.monotonic() + timeout while not condition(): assert time.monotonic() < deadline, "условие не выполнилось вовремя" QCoreApplication.processEvents() time.sleep(0.005) @pytest.fixture def memory(tmp_path: Path) -> MemoryStore: store = MemoryStore(tmp_path / "memory.sqlite3") yield store store.close() def _assistant( server: FakeLLMServer, memory: MemoryStore, *, auto_save: bool = True, commands: CommandCatalog | None = None, voice: object = None, ) -> Assistant: config = LLMConfig( provider="fake", providers={"fake": ProviderConfig("fake", server.base_url, "key", "fake-model")}, system_prompt="Будь краток.", temperature=0.5, follow_up_seconds=0, timeout_seconds=10, ) speaker = Speaker(_SilentEngine(), _SilentPlayer(), enabled=False) return Assistant( config, speaker, voice, # type: ignore[arg-type] memory=memory, memory_auto_save=auto_save, commands=commands, ) def test_remember_request_runs_tool_and_answers( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore ) -> None: memory.add("Живёт в Казани", "user") fake_llm.replies += [ Reply(tool_calls=[("remember", '{"text": "Машина — Škoda Octavia", "requested_by_user": true}')]), Reply(text="Запомнил."), ] assistant = _assistant(fake_llm, memory) tools: list[tuple[str, bool]] = [] finished: list[str] = [] assistant.tool_executed.connect(lambda display, ok: tools.append((display, ok))) assistant.reply_finished.connect(finished.append) assistant.send("Запомни, что у меня Škoda Octavia") _wait_until(lambda: bool(finished)) assert [fact.text for fact in memory.facts()] == ["Живёт в Казани", "Машина — Škoda Octavia"] assert tools == [("Запомнил: Машина — Škoda Octavia", True)] assert finished == ["Запомнил."] first, second = fake_llm.requests system = first["messages"][0]["content"] assert "Живёт в Казани" in system and "requested_by_user=false" in system assert {tool["function"]["name"] for tool in first["tools"]} == { "remember", "update_memory", "forget", } # Второй запрос несёт вызов инструмента и его результат assert second["messages"][-2]["tool_calls"][0]["function"]["name"] == "remember" assert second["messages"][-1] == { "role": "tool", "tool_call_id": "call_0", "content": "Сохранено под номером 2", } def test_model_without_tools_falls_back_once( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore ) -> None: fake_llm.replies += [ Reply(status=400, error="model does not support tools"), Reply(text="Привет!"), Reply(text="Снова привет!"), ] assistant = _assistant(fake_llm, memory) errors: list[str] = [] finished: list[str] = [] assistant.error_occurred.connect(errors.append) assistant.reply_finished.connect(finished.append) assistant.send("Привет") _wait_until(lambda: len(finished) == 1) assistant.send("Ещё раз привет") _wait_until(lambda: len(finished) == 2) assert finished == ["Привет!", "Снова привет!"] assert len(errors) == 1 and "не поддерживает инструменты" in errors[0] assert ["tools" in request for request in fake_llm.requests] == [True, False, False] def test_auto_save_can_be_disabled( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore ) -> None: fake_llm.replies.append(Reply(text="Ок")) assistant = _assistant(fake_llm, memory, auto_save=False) finished: list[str] = [] assistant.reply_finished.connect(finished.append) assistant.send("Меня зовут Лео") _wait_until(lambda: bool(finished)) assert "requested_by_user=false" not in fake_llm.requests[0]["messages"][0]["content"] # --- команды _COMMANDS_YAML = """ commands: - name: open_downloads description: Открыть папку «Загрузки» phrases: [открой загрузки] reply: Открываю. action: {type: open, target: "C:/Downloads"} - name: web_search description: Найти в интернете parameters: query: {type: string} action: {type: open, target: "https://example.com/?q={query}"} - name: shutdown_computer description: Выключить компьютер phrases: [выключи компьютер] confirm: true action: {type: open, target: "shutdown://now"} """ class _FakeVoice(QObject): listening_started = Signal() recognizing_started = Signal() finished = Signal() recognized = Signal(str) error_occurred = Signal(str) def __init__(self) -> None: super().__init__() self.started = 0 self.is_listening = False self.is_recognizing = False self.is_active = False def start(self) -> None: self.started += 1 def stop(self) -> None: pass def cancel(self) -> None: pass @pytest.fixture def opened(monkeypatch: pytest.MonkeyPatch) -> list[str]: targets: list[str] = [] monkeypatch.setattr(actions, "open_target", targets.append) return targets def _commands_assistant( server: FakeLLMServer, memory: MemoryStore, tmp_path: Path, voice: _FakeVoice | None = None ) -> tuple[Assistant, CommandCatalog]: path = tmp_path / "commands.yaml" path.write_text(_COMMANDS_YAML, encoding="utf-8") catalog = CommandCatalog(path) return _assistant(server, memory, commands=catalog, voice=voice), catalog def _collect(assistant: Assistant) -> dict[str, list]: events: dict[str, list] = {"tools": [], "replies": [], "errors": []} assistant.tool_executed.connect(lambda display, ok: events["tools"].append((display, ok))) assistant.reply_finished.connect(events["replies"].append) assistant.error_occurred.connect(events["errors"].append) return events def test_exact_phrase_runs_without_model( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) events = _collect(assistant) assistant.send("Открой загрузки!") _wait_until(lambda: bool(events["replies"])) assert opened == ["C:/Downloads"] assert events["replies"] == ["Открываю."] assert events["tools"] == [("Открыл: Открыть папку «Загрузки»", True)] assert fake_llm.requests == [] def test_model_calls_command_tool( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: fake_llm.replies += [ Reply(tool_calls=[("web_search", '{"query": "погода в Казани"}')]), Reply(text="Открыл поиск."), ] assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) events = _collect(assistant) assistant.send("Поищи погоду в Казани") _wait_until(lambda: bool(events["replies"])) assert opened == ["https://example.com/?q=%D0%BF%D0%BE%D0%B3%D0%BE%D0%B4%D0%B0%20%D0%B2%20%D0%9A%D0%B0%D0%B7%D0%B0%D0%BD%D0%B8"] tool_names = {tool["function"]["name"] for tool in fake_llm.requests[0]["tools"]} assert {"web_search", "open_downloads", "remember"} <= tool_names assert "команды для управления компьютером" in fake_llm.requests[0]["messages"][0]["content"] assert events["replies"] == ["Открыл поиск."] def test_model_command_with_confirmation_waits_for_yes( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: fake_llm.replies += [ Reply(tool_calls=[("shutdown_computer", "{}")]), Reply(text="Выключить компьютер?"), ] assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) events = _collect(assistant) assistant.send("Выключи комп, пожалуйста") _wait_until(lambda: len(events["replies"]) == 1) assert opened == [] assert events["tools"] == [("Ждёт подтверждения: Выключить компьютер", True)] assert "НЕ выполнена" in fake_llm.requests[1]["messages"][-1]["content"] assistant.send("Да") _wait_until(lambda: len(events["replies"]) == 2) assert opened == ["shutdown://now"] assert events["replies"][1] == "Готово." assert len(fake_llm.requests) == 2 # «да» обработано без модели def test_phrase_confirmation_can_be_declined_or_dropped( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: fake_llm.replies.append(Reply(text="Не понял.")) assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path) events = _collect(assistant) assistant.send("Выключи компьютер") _wait_until(lambda: len(events["replies"]) == 1) assert events["replies"][0] == "Выполнить: Выключить компьютер? Скажите «да» или «нет»." assistant.send("нет") _wait_until(lambda: len(events["replies"]) == 2) assert events["replies"][1] == "Хорошо, не выполняю." # После отказа «да» уже ничего не подтверждает и уходит модели assistant.send("да") _wait_until(lambda: len(events["replies"]) == 3) assert opened == [] assert len(fake_llm.requests) == 1 def test_voice_confirmation_starts_listening( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: voice = _FakeVoice() assistant, _catalog = _commands_assistant(fake_llm, memory, tmp_path, voice) events = _collect(assistant) assistant.send("выключи компьютер", by_voice=True) _wait_until(lambda: voice.started == 1) assistant.send("открой загрузки", by_voice=True) _wait_until(lambda: len(events["replies"]) == 2) for _ in range(20): QCoreApplication.processEvents() time.sleep(0.005) assert voice.started == 1 # без вопроса микрофон сам не включается def test_commands_file_changes_are_picked_up( qapp: QApplication, fake_llm: FakeLLMServer, memory: MemoryStore, tmp_path: Path, opened: list[str] ) -> None: assistant, catalog = _commands_assistant(fake_llm, memory, tmp_path) events = _collect(assistant) catalog.path.write_text( _COMMANDS_YAML.replace("[открой загрузки]", "[покажи загрузки]") + " - name: broken\n description: x\n action: {type: nope}\n", encoding="utf-8", ) catalog.reload() assistant.send("покажи загрузки") _wait_until(lambda: bool(events["replies"])) assert opened == ["C:/Downloads"] assert any("broken" in error for error in events["errors"])