mrleo1nidandClaude Opus 5
84d8aaa848
- Факты о пользователе в SQLite, в системном промпте с номерами - Инструменты remember / update_memory / forget, автоматическое запоминание отключается - Цикл вызова инструментов со стримингом (до 5 кругов), проверка и приведение аргументов - Откат без инструментов для моделей, которые их не поддерживают, с одним предупреждением - Действия в журнале, вкладка «Память» в настройках - Фейковый OpenAI-совместимый сервер для тестов, тесты полного цикла через Assistant Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
142 lines
4.9 KiB
Python
142 lines
4.9 KiB
Python
"""Полный цикл запроса через Assistant: модель, инструменты памяти, откат без инструментов."""
|
||
|
||
import time
|
||
from collections.abc import Callable
|
||
from pathlib import Path
|
||
|
||
import numpy as np
|
||
import pytest
|
||
from PySide6.QtCore import QCoreApplication
|
||
from PySide6.QtWidgets import QApplication
|
||
|
||
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) -> 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, None, memory=memory, memory_auto_save=auto_save)
|
||
|
||
|
||
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"]
|