139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""Tests for pre-LLM scope grounding from D1/D3 catalog rows (no extra RAG layer)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
from app.core.agent.processes.v2 import V2IntentRouter
|
|
from app.core.agent.utils.process_v2.models import V2ScopeType
|
|
|
|
|
|
class FakeLlm:
|
|
def __init__(self, response: str) -> None:
|
|
self.response = response
|
|
|
|
def generate(self, prompt_name: str, user_input: str, **_kwargs) -> str:
|
|
del prompt_name, user_input
|
|
return self.response
|
|
|
|
|
|
def _llm_ok() -> str:
|
|
return json.dumps(
|
|
{
|
|
"routing_domain": "DOCS",
|
|
"intent": "DOC_EXPLAIN",
|
|
"subintent": "SUMMARY",
|
|
"confidence": 0.9,
|
|
"reason_short": "ok",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
|
|
def _fixture_rows() -> list[dict]:
|
|
return [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/billing/overview.md",
|
|
"title": "Billing",
|
|
"content": "",
|
|
"metadata": {"domain": "billing", "summary_text": "Billing domain overview"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/billing/invoices.md",
|
|
"title": "Invoices",
|
|
"content": "",
|
|
"metadata": {"domain": "billing", "subdomain": "invoice", "tags": ["invoice", "invoices"]},
|
|
},
|
|
{
|
|
"layer": "D3_ENTITY_CATALOG",
|
|
"path": "docs/domains/order.md",
|
|
"title": "Order",
|
|
"content": "",
|
|
"metadata": {"entity_name": "Order", "domain": "billing"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/invoices_post.md",
|
|
"title": "POST /api/v1/invoices",
|
|
"content": "",
|
|
"metadata": {
|
|
"doc_type": "api_method",
|
|
"domain": "billing",
|
|
"endpoint": "/api/v1/invoices",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/widgets/readme.md",
|
|
"title": "Widgets",
|
|
"content": "",
|
|
"metadata": {"domain": "widgets", "summary_text": "Unrelated domain for negative tests"},
|
|
},
|
|
]
|
|
|
|
|
|
def _router() -> V2IntentRouter:
|
|
return V2IntentRouter(llm=FakeLlm(_llm_ok()), scope_rows_provider=lambda _sid: _fixture_rows())
|
|
|
|
|
|
def test_scope_global_project_wide_enumeration() -> None:
|
|
r = _router().route("какие api методы есть в проекте", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.GLOBAL
|
|
|
|
|
|
def test_scope_domain_billing() -> None:
|
|
r = _router().route("какие api есть в billing", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.DOMAIN
|
|
assert r.anchors.process_domain == "billing"
|
|
assert any(c.value == "billing" for c in r.anchors.candidate_domains)
|
|
|
|
|
|
def test_scope_subdomain_billing_invoices() -> None:
|
|
r = _router().route("какие api есть в billing invoices", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.SUBDOMAIN
|
|
assert r.anchors.process_domain == "billing"
|
|
assert r.anchors.process_subdomain == "invoice"
|
|
|
|
|
|
def test_scope_entity_order_doc() -> None:
|
|
r = _router().route("дай доку по Order", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.ENTITY
|
|
assert "order" in [e.lower() for e in r.anchors.entity_names]
|
|
|
|
|
|
def test_scope_entity_endpoint_path() -> None:
|
|
r = _router().route("где описан POST /api/v1/invoices", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.ENTITY
|
|
assert "/api/v1/invoices" in r.anchors.endpoint_paths
|
|
|
|
|
|
def test_scope_vague_no_false_domain() -> None:
|
|
r = _router().route("что там с фывырапфыв", rag_session_id="sess-1")
|
|
assert r.scope_type == V2ScopeType.UNKNOWN
|
|
assert r.anchors.process_domain is None
|
|
|
|
|
|
def test_scope_russian_payments_phrase_matches_tag() -> None:
|
|
rows = [
|
|
*_fixture_rows(),
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/billing/payments_ru.md",
|
|
"title": "Платежи",
|
|
"content": "",
|
|
"metadata": {"domain": "billing", "tags": ["платежи"]},
|
|
},
|
|
]
|
|
router = V2IntentRouter(llm=FakeLlm(_llm_ok()), scope_rows_provider=lambda _sid: rows)
|
|
r = router.route("какие методы есть в платежи", rag_session_id="sess-1")
|
|
assert r.scope_type in {V2ScopeType.DOMAIN, V2ScopeType.ENTITY, V2ScopeType.SUBDOMAIN}
|
|
assert r.anchors.process_domain == "billing" or any("платеж" in c.value for c in r.anchors.candidate_entities)
|
|
|
|
|
|
def test_router_without_session_skips_db_and_keeps_target_terms() -> None:
|
|
r = V2IntentRouter(llm=FakeLlm(_llm_ok())).route("Покажи где описан RuntimeHealth и /health")
|
|
assert r.scope_type == V2ScopeType.UNKNOWN
|
|
assert "runtimehealth" in r.target_terms
|