/
Vibek
/
Agent_GG
Обзор
Документация
Войти
/
Vibek
/
Agent_GG
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
master
tests/test_graph.py
445 строк
16 KB
Vibek
нормализация протестирована, сценарий с парсингом без явного запроса работает
27 май 2026, 15:12
27 май 2026, 15:12
43e0908
Код
Авторство
О чём код?
""" Интеграционные проверки графов LangGraph. """ from __future__ import annotations import os import sys from pathlib import Path # Корень репозитория на path для `import agent` _ROOT = Path(__file__).resolve().parents[1] if str(_ROOT) not in sys.path: sys.path.insert(0, str(_ROOT)) # Без вызова GigaChat: ожидаем честное сообщение о недоступности for _k in ( "GIGACHAT_CLIENT_ID", "GIGACHAT_CLIENT_SECRET", "GIGACHAT_CREDENTIALS", "GIGACHAT_ACCESS_TOKEN", ): os.environ[_k] = "" # Тесты графа не должны требовать установленный psycopg. os.environ.setdefault("DATABASE_URL", "sqlite+pysqlite:///:memory:") os.environ.setdefault("METRICS_DATABASE_URL", "sqlite+pysqlite:///:memory:") _backend = _ROOT / "backend" if str(_backend) not in sys.path: sys.path.insert(0, str(_backend)) import src.core.config as _cfg # noqa: E402 _cfg.get_settings.cache_clear() from agent.evaluator.graph import evaluation_subgraph # noqa: E402 from agent.orchestrator.graph import _aggregate_step_decision # noqa: E402 from agent.orchestrator.graph import _build_more_results_nudge # noqa: E402 from agent.orchestrator.graph import _trim_messages_for_context # noqa: E402 from agent.orchestrator.graph import _response_from_envelopes # noqa: E402 from agent.orchestrator.graph import orchestrator_graph # noqa: E402 from agent.orchestrator.graph import tool_plan_normalize_node # noqa: E402 from agent.tools.contracts import normalize_tool_envelope, resolve_next_step_policy # noqa: E402 from src.modules.chat.agent_gateway import _extract_answer_from_graph_result # noqa: E402 def test_orchestrator_graph() -> None: result = orchestrator_graph.invoke( { "messages": [{"role": "user", "content": "найди офис в Москве"}], "user_id": "test_user", "session_id": "test_session", } ) assert "messages" in result assert result["messages"] assert result["messages"][-1].get("role") == "assistant" def test_evaluation_subgraph(monkeypatch) -> None: def _fake_rag_search_documents(vector_store, query, category, k=3): _ = vector_store _ = query _ = k return [ { "chunk_id": "r1#signals#1", "doc_id": "R1_price_market_ru", "criterion": category, "section": "Сигналы цены", "section_kind": "signals", "source_path": "docs/R1.md", "content": "Сравнивайте цену за м2 с рынком по городу и району.", } ] monkeypatch.setattr("agent.evaluator.nodes.rag_search_documents", _fake_rag_search_documents) subject_listing = { "id": 42, "title": "Офис на Тверской", "address": "Москва, Тверская, 15", "city": "moskva", "district": "tverskoy", "purpose": "office", "area_total": 100.0, "price_total": 1_000_000.0, "price_per_sqm": 10_000.0, "floor": 4, "full_description": "Офис с хорошим доступом, рядом с метро, есть базовые инженерные системы.", "updated_at": "2026-04-20T10:00:00+00:00", } candidate_listings = [ { "id": 101, "city": "moskva", "district": "tverskoy", "purpose": "office", "area_sqm": 98.0, "price_total": 980_000.0, "price_per_sqm": 10_050.0, "floor": 4, "updated_at": "2026-04-21T10:00:00+00:00", }, { "id": 102, "city": "moskva", "district": "tverskoy", "purpose": "office", "area_sqm": 107.0, "price_total": 1_060_000.0, "price_per_sqm": 9_900.0, "floor": 5, "updated_at": "2026-04-21T09:00:00+00:00", }, { "id": 103, "city": "moskva", "district": "tverskoy", "purpose": "office", "area_sqm": 111.0, "price_total": 1_080_000.0, "price_per_sqm": 10_120.0, "floor": 3, "updated_at": "2026-04-19T09:00:00+00:00", }, ] result = evaluation_subgraph.invoke( { "property_id": 42, "subject_listing": subject_listing, "candidate_listings": candidate_listings, } ) assert "final_report" in result assert isinstance(result["final_report"], str) report_text = result["final_report"].lower() assert report_text.startswith("#") assert len(report_text) > 50 assert result.get("status") == "done" assert result.get("progress") == 1.0 # cleanup_state: тяжелые поля должны быть очищены перед END assert result.get("doppel_listings") is None assert result.get("questions_pool") is None assert result.get("audit_results") is None def test_graph_without_intent_routing() -> None: result = orchestrator_graph.invoke( { "messages": [{"role": "user", "content": "оцени объект 42"}], "user_id": "test_user", "session_id": "test_session", } ) assert result.get("messages") assert result["messages"][-1]["role"] == "assistant" def test_orchestrator_graph_returns_top_level_answer() -> None: result = orchestrator_graph.invoke( { "messages": [{"role": "user", "content": "найди офис в Москве"}], "user_id": "test_user", "session_id": "test_session_top_level_answer", } ) assert isinstance(result.get("answer"), str) assert result["answer"] def test_gateway_extracts_answer_from_messages_when_top_level_field_missing() -> None: answer = _extract_answer_from_graph_result( { "messages": [ {"role": "user", "content": "Привет"}, {"role": "assistant", "content": "Готов помочь."}, ] } ) assert answer == "Готов помочь." def test_context_window_trimming() -> None: messages = [{"role": "user", "content": f"msg-{i}"} for i in range(20)] selected, trimmed = _trim_messages_for_context(messages) assert len(selected) == 12 assert trimmed == 8 assert selected[0]["content"] == "msg-8" assert selected[-1]["content"] == "msg-19" def test_more_results_nudge_without_narrowing_filters_prefers_parse() -> None: messages = [ {"role": "assistant", "content": "В базе 6 объявлений."}, {"role": "user", "content": "найди еще объявления, этих мало"}, ] ui_context = { "applied_filters": {"city_key": "ekaterinburg"}, "parser_source": "avito", } nudge = _build_more_results_nudge(messages, ui_context) assert isinstance(nudge, str) and nudge assert "Запусти parse" in nudge def test_more_results_nudge_with_narrowing_filters_requests_clarification() -> None: messages = [ {"role": "assistant", "content": "Вижу 4 объявления по району."}, {"role": "user", "content": "мне нужно еще"}, ] ui_context = { "applied_filters": {"city_key": "ekaterinburg", "district": "чкаловский"}, "parser_source": "avito", } nudge = _build_more_results_nudge(messages, ui_context) assert isinstance(nudge, str) and nudge assert "снять/ослабить фильтры" in nudge def test_more_results_nudge_affirmative_followup_detected() -> None: messages = [ {"role": "assistant", "content": "Хотите еще объявления или расширить выборку?"}, {"role": "user", "content": "да"}, ] ui_context = {"applied_filters": {"city_key": "ekaterinburg"}} nudge = _build_more_results_nudge(messages, ui_context) assert isinstance(nudge, str) and nudge assert "Интент 'больше объявлений'" in nudge def test_tool_plan_normalize_does_not_inject_missing_filters() -> None: state = { "last_ai_tool_calls": [{"name": "market_stats", "args": {"city_key": "ekaterinburg"}}], "diagnostics": {}, } result = tool_plan_normalize_node(state) planned = result["tool_calls_planned"][0]["args"] assert planned == {"city_key": "ekaterinburg"} def test_tool_plan_normalize_normalizes_only_explicit_filters() -> None: state = { "last_ai_tool_calls": [ { "name": "listings_count", "args": { "city_key": " EKATERINBURG ", "district": " Чкаловский ", "min_area": "100.5", "max_area": "220", "limit": "700", }, } ], "diagnostics": {}, } result = tool_plan_normalize_node(state) args = result["tool_calls_planned"][0]["args"] assert args["city_key"] == "ekaterinburg" assert args["district"] == "чкаловский" assert args["min_area"] == 100.5 assert args["max_area"] == 220.0 assert args["limit"] == 500 def test_tool_envelope_invalid_args_user_safe_message() -> None: envelope = normalize_tool_envelope( "market_stats", { "success": False, "status": "invalid_args", "message": "Укажите city_key: moskva, sankt_peterburg, kazan, ekaterinburg, novosibirsk.", }, ) assert envelope["status"] == "invalid_args" assert envelope["error"] is not None assert envelope["error"]["code"] == "INVALID_ARGUMENTS" assert envelope["next_step_policy"] == "retry_once" assert isinstance(envelope["message_user"], str) and envelope["message_user"] def test_district_catalog_payload_contract_validation() -> None: envelope = normalize_tool_envelope( "district_catalog", { "success": True, "status": "ok", "message": "Найдено районов: 1.", "city_key": "ekaterinburg", "query": "чкал", "districts": [{"district": "чкаловский", "count": 12}], "total": 1, }, ) assert envelope["status"] == "ok" assert envelope["error"] is None def test_policy_resolver_retry_once_for_failed() -> None: envelope = normalize_tool_envelope( "market_stats", {"success": False, "status": "failed", "message": "db timeout"}, ) decision_first = resolve_next_step_policy(envelope, retry_count=0, max_retry=1) decision_second = resolve_next_step_policy(envelope, retry_count=1, max_retry=1) assert decision_first["action"] == "retry" assert decision_second["action"] == "ask_clarification" def test_policy_resolver_retry_once_for_invalid_args() -> None: envelope = normalize_tool_envelope( "parse", {"success": False, "status": "invalid_category", "message": "Неверная категория"}, ) first = resolve_next_step_policy(envelope, retry_count=0, max_retry=1) second = resolve_next_step_policy(envelope, retry_count=1, max_retry=1) assert first["action"] == "retry" assert second["action"] == "ask_clarification" assert "Допустимые значения" in envelope["message_model"] def test_policy_resolver_invalid_args_unchanged_args() -> None: envelope = normalize_tool_envelope( "parse", {"success": False, "status": "invalid_args", "message": "city_key required"}, ) decision = resolve_next_step_policy(envelope, retry_count=0, max_retry=1, args_changed=False) assert decision["action"] == "ask_clarification" assert decision["reason_code"] == "invalid_args_unchanged" def test_policy_resolver_retries_transient_error() -> None: envelope = normalize_tool_envelope( "market_stats", { "success": False, "status": "failed", "message": "dependency timeout", "error": {"code": "TIMEOUT", "detail": "request timeout", "retryable": True, "field": None}, }, ) decision = resolve_next_step_policy(envelope, retry_count=0, max_retry=1) assert decision["action"] == "retry" assert decision["reason_code"] == "transient_retry_once" def test_contract_validation_error_for_invalid_status() -> None: envelope = normalize_tool_envelope( "market_stats", { "tool": "market_stats", "status": "something_weird", "success": True, "message": "ok", "message_user": "ok", "message_model": "ok", "next_step_policy": "continue", "data": {}, "meta": {}, "error": None, }, ) assert envelope["status"] == "failed" assert envelope["next_step_policy"] == "stop_and_report" assert envelope["error"] is not None assert envelope["error"]["code"] == "CONTRACT_VALIDATION_ERROR" def test_policy_aggregation_uses_highest_priority_action() -> None: env_continue = normalize_tool_envelope( "listings_count", { "tool": "listings_count", "status": "ok", "success": True, "message": "ok", "message_user": "ok", "message_model": "ok", "next_step_policy": "continue", "data": {}, "meta": {}, "error": None, }, ) env_stop = normalize_tool_envelope( "market_stats", { "tool": "market_stats", "status": "failed", "success": False, "message": "fatal failure", "message_user": "Не удалось", "message_model": "Остановись", "next_step_policy": "stop_and_report", "data": {}, "meta": {}, "error": {"code": "TOOL_EXECUTION_FAILED", "field": None, "retryable": False, "detail": "fatal"}, }, ) selected = _aggregate_step_decision( entries=[ { "envelope": env_continue, "tool": "listings_count", "args_hash": "a1", "tool_call_id": "c1", "context_unmatched": False, }, { "envelope": env_stop, "tool": "market_stats", "args_hash": "b2", "tool_call_id": "c2", "context_unmatched": False, }, ], retry_budget={}, last_args_by_tool={}, max_retry=1, ) assert selected["action"] == "stop_and_report" assert selected["tool"] == "market_stats" def test_response_from_envelopes_forces_tool_message_for_evaluate_listing() -> None: envelope = normalize_tool_envelope( "evaluate_listing", { "success": True, "status": "ok", "message": "Найдено похожих: 1 (JSON):\nID 1: площадь=100, цена_за_м2=1200, Δплощади=0.0%, Δцены_за_м2=0.0%", "comparables_count": 5, }, ) result = _response_from_envelopes( messages=[{"role": "user", "content": "Оцени объект 146"}], envelopes=[envelope], llm_answer="Итоговая оценка: средняя конкурентоспособность...", ) assert result["answer"].startswith("Найдено похожих: 1 (JSON):")