440 lines
19 KiB
Python
440 lines
19 KiB
Python
from __future__ import annotations
|
|
|
|
from app.modules.agent.intent_router_v2 import IntentRouterV2
|
|
from app.modules.agent.runtime.docs_qa_pipeline import DocsQAPipelineRunner
|
|
from tests.docs_qa_eval.fixture_adapter import InMemoryDocsRetrievalAdapter
|
|
from tests.unit_tests.rag.intent_router_testkit import repo_context
|
|
|
|
|
|
def test_docs_pipeline_uses_synthesizer_for_explain() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/billing.md",
|
|
"title": "Billing facts",
|
|
"content": "Billing обновляет статус заказа после оплаты.",
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни модуль billing в документации", "docs-session")
|
|
|
|
assert result.router_result.intent == "DOCUMENTATION_EXPLAIN"
|
|
assert "Billing" in result.answer
|
|
assert result.openapi_result is None
|
|
assert result.diagnostics.facts_found == 1
|
|
|
|
|
|
def test_docs_pipeline_uses_openapi_branch() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders-create.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {
|
|
"endpoint": "/orders",
|
|
"http_method": "post",
|
|
"request_schema": {"type": "object", "properties": {"customer_id": {}}},
|
|
"response_schema": {"type": "object", "properties": {"order_id": {}}},
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi spec для создания заказа", "docs-session")
|
|
|
|
assert result.router_result.intent == "OPENAPI_GENERATION"
|
|
assert result.openapi_result is not None
|
|
assert result.openapi_result.path == "/orders"
|
|
assert result.openapi_result.method == "post"
|
|
assert result.answer_mode != "degraded"
|
|
assert result.diagnostics.gate_decision != "reject"
|
|
assert "paths:" in result.answer
|
|
assert "/orders" in result.answer
|
|
|
|
|
|
def test_docs_pipeline_keeps_query_candidates_for_not_found_entity() -> None:
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter([]), repo_context=repo_context())
|
|
|
|
result = runner.run("Что такое сущность PaymentTransaction?", "docs-session")
|
|
|
|
assert result.answer_mode == "degraded"
|
|
assert "PaymentTransaction" in result.diagnostics.query_entity_candidates
|
|
assert result.diagnostics.resolved_entity_candidates == []
|
|
assert result.diagnostics.degraded_reason == "not_found_exact_anchor"
|
|
|
|
|
|
def test_docs_pipeline_reports_canonical_doc_ids_and_layers() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/api/health.md",
|
|
"title": "Health relations",
|
|
"content": "Health -> runtime health",
|
|
"metadata": {"doc_id": "api.health_endpoint", "target_doc_id": "domain.runtime_health", "endpoint": "/health"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Runtime health",
|
|
"content": "Runtime health overview",
|
|
"metadata": {"doc_id": "domain.runtime_health"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session")
|
|
|
|
assert result.diagnostics.doc_ids[:2] == ["api.health_endpoint", "domain.runtime_health"]
|
|
assert result.diagnostics.planned_layers == ["D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.executed_layers == ["D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.non_empty_layers == ["D5_RELATION_GRAPH"]
|
|
assert result.diagnostics.relation_hits_count == 1
|
|
assert result.diagnostics.relation_targets == ["domain.runtime_health"]
|
|
assert result.diagnostics.answer_mode == "answered"
|
|
assert result.diagnostics.layer_diagnostics["D5_RELATION_GRAPH"]["hits"] == 1
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
|
|
|
|
def test_related_docs_uses_relation_graph_as_primary_layer() -> None:
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter([]), repo_context=repo_context())
|
|
|
|
result = runner.run("Что еще посмотреть по теме billing?", "docs-session")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "RELATED_DOCS_EXPLAIN"
|
|
assert result.diagnostics.planned_layers[0] == "D5_RELATION_GRAPH"
|
|
assert result.diagnostics.executed_layers[0] == "D5_RELATION_GRAPH"
|
|
|
|
|
|
def test_openapi_partial_contract_returns_partial_mode() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders-create.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {
|
|
"endpoint": "/orders",
|
|
"http_method": "post",
|
|
"request_schema": {"type": "object", "properties": {"customer_id": {}}},
|
|
"doc_id": "api.orders_create",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi для /orders", "docs-session")
|
|
|
|
assert result.answer_mode == "structured_spec_partial"
|
|
assert result.degraded_reason == "answered_with_gaps"
|
|
assert result.openapi_result is not None
|
|
assert result.openapi_result.path == "/orders"
|
|
assert result.diagnostics.gate_decision == "partial"
|
|
assert "paths:" in result.answer
|
|
assert "/orders" in result.answer
|
|
|
|
|
|
def test_pre_llm_mode_returns_diagnostic_only_without_answer_generation() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "RuntimeManager",
|
|
"content": "Coordinates runtime orchestration.",
|
|
"metadata": {"doc_id": "architecture.runtime_manager", "component": "RuntimeManager"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какую роль в системе играет RuntimeManager?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.mode == "pre_llm_only"
|
|
assert result.answer == ""
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
assert "RuntimeManager" in result.diagnostics.query_entity_candidates
|
|
assert "RuntimeManager" in result.diagnostics.resolved_entity_candidates
|
|
|
|
|
|
def test_pre_llm_mode_detects_path_anchor_candidates() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Health endpoint",
|
|
"metadata": {"doc_id": "api.health_endpoint", "endpoint": "/health", "http_method": "get"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни API метод /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert "/health" in result.diagnostics.query_anchor_candidates
|
|
assert "/health" in result.diagnostics.resolved_anchor_candidates
|
|
assert result.diagnostics.planned_layers == ["D2_FACT_INDEX", "D4_WORKFLOW_INDEX", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert set(result.diagnostics.executed_layers) == {"D1_DOCUMENT_CATALOG", "D2_FACT_INDEX", "D4_WORKFLOW_INDEX", "D0_DOC_CHUNKS"}
|
|
|
|
|
|
def test_pre_llm_openapi_gate_reports_missing_signals() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {"doc_id": "api.orders_create", "endpoint": "/orders", "http_method": "post"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi для /orders", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.gate_decision == "partial"
|
|
assert result.diagnostics.gate_decision_reason == "answered_with_gaps"
|
|
assert result.answer_mode == "ready_partial"
|
|
assert result.answer
|
|
assert "paths:" in result.answer
|
|
assert "path_found" in result.diagnostics.gate_satisfied_requirements
|
|
assert "http_method_found" in result.diagnostics.gate_satisfied_requirements
|
|
assert "contract_fields_found" in result.diagnostics.gate_missing_requirements
|
|
assert result.diagnostics.evidence_summary["openapi_signals"]["path_found"] is True
|
|
assert result.diagnostics.openapi_evidence["path_found"] is True
|
|
assert result.diagnostics.layer_diagnostics["D2_FACT_INDEX"]["hits"] == 1
|
|
|
|
|
|
def test_related_docs_keeps_relation_hits_when_exact_anchor_is_missing() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/relations/health-links.md",
|
|
"title": "Health relations",
|
|
"content": "Runtime health and service overview",
|
|
"metadata": {
|
|
"document_id": "api.health_endpoint",
|
|
"target_document_id": "domain.runtime_health",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "RELATED_DOCS_EXPLAIN"
|
|
assert result.diagnostics.doc_ids[:2] == ["api.health_endpoint", "domain.runtime_health"]
|
|
assert result.diagnostics.relation_hits_count == 1
|
|
assert result.diagnostics.exact_anchor_match is False
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
assert result.diagnostics.gate_decision_reason == "relation_evidence_available"
|
|
|
|
|
|
def test_related_docs_exact_path_outranks_neighboring_api_doc() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/documentation/api/control-actions-endpoint.md",
|
|
"title": "HTTP API /actions/{action}",
|
|
"content": "Nearby API relation",
|
|
"metadata": {
|
|
"document_id": "api.control_actions_endpoint",
|
|
"target_document_id": "architecture.telegram_notify_app",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/documentation/api/health-endpoint.md",
|
|
"title": "HTTP API /health",
|
|
"content": "Health relation",
|
|
"metadata": {
|
|
"document_id": "api.health_endpoint",
|
|
"target_document_id": "domain.runtime_health",
|
|
"endpoint": "/health",
|
|
},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.anchor_candidates[0] == "api.health_endpoint"
|
|
assert result.diagnostics.selected_anchor == "api.health_endpoint"
|
|
assert result.diagnostics.anchor_match_type == "exact_path"
|
|
assert result.diagnostics.anchor_selection_reason == "metadata.endpoint exact match"
|
|
assert result.diagnostics.doc_ids[0] == "api.health_endpoint"
|
|
|
|
|
|
def test_pre_llm_targeted_chunk_fallback_enriches_api_method_layers() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Health endpoint",
|
|
"metadata": {"doc_id": "api.health_endpoint"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:Scenario",
|
|
"content": "request -> check runtime -> return status",
|
|
"metadata": {"doc_id": "api.health_endpoint"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни API метод /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert "D0_DOC_CHUNKS" in result.diagnostics.executed_layers
|
|
assert "D0_DOC_CHUNKS" in result.diagnostics.non_empty_layers
|
|
assert result.diagnostics.layer_diagnostics["D0_DOC_CHUNKS"]["hits"] == 1
|
|
assert result.diagnostics.selected_chunk_ids == ["api.health_endpoint"]
|
|
|
|
|
|
def test_openapi_request_fragment_uses_fragment_aware_gate() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/send.md",
|
|
"title": "POST /send",
|
|
"content": "Request payload contains message and chat_id fields.",
|
|
"metadata": {
|
|
"document_id": "api.send_message_endpoint",
|
|
"endpoint": "/send",
|
|
"http_method": "post",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй request schema для endpoint /send", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "OPENAPI_FRAGMENT_GENERATE"
|
|
assert result.diagnostics.requested_fragment_type == "request_schema"
|
|
assert "path" in result.diagnostics.fragment_evidence_found
|
|
assert "payload_description" in result.diagnostics.fragment_evidence_found
|
|
assert result.diagnostics.gate_decision in {"allow", "partial"}
|
|
assert result.diagnostics.gate_decision_reason in {"evidence_sufficient", "fragment_payload_only"}
|
|
assert result.diagnostics.fragment_missing_requirements == []
|
|
assert result.answer_mode in {"ready", "ready_partial"}
|
|
assert result.answer
|
|
assert "type: object" in result.answer
|
|
assert "message:" in result.answer
|
|
assert "chat_id:" in result.answer
|
|
|
|
|
|
def test_openapi_method_with_only_path_is_rejected() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "/health",
|
|
"content": "",
|
|
"metadata": {"endpoint": "/health", "doc_id": "api.health_endpoint"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй OpenAPI для endpoint /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.gate_decision == "reject"
|
|
assert result.answer_mode == "degraded"
|
|
|
|
|
|
def test_openapi_response_fragment_does_not_require_request_payload() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Returns 200 with status and uptime fields.",
|
|
"metadata": {
|
|
"endpoint": "/health",
|
|
"http_method": "get",
|
|
"response_schema": {"type": "object", "properties": {"status": {}, "uptime": {}}},
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй response schema для endpoint /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.requested_fragment_type == "response_schema"
|
|
assert result.diagnostics.gate_decision != "reject"
|
|
assert result.answer_mode != "degraded"
|
|
assert "properties:" in result.answer
|
|
|
|
|
|
def test_system_flow_evidence_prefers_workflows_and_relations() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout overview",
|
|
"content": "Overview doc",
|
|
"metadata": {"doc_id": "workflow.checkout"},
|
|
},
|
|
{
|
|
"layer": "D4_WORKFLOW_INDEX",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout steps",
|
|
"content": "cart -> payment -> confirm",
|
|
"metadata": {"workflow_id": "workflow.checkout"},
|
|
},
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout relations",
|
|
"content": "Related to payment entity",
|
|
"metadata": {"doc_id": "workflow.checkout", "target_doc_id": "domain.payment"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "workflow.checkout:chunk",
|
|
"content": "payment step persists order",
|
|
"metadata": {"chunk_id": "chunk.checkout.1", "doc_id": "workflow.checkout"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни как работает checkout workflow", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.planned_layers == ["D4_WORKFLOW_INDEX", "D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.selected_workflow_ids == ["workflow.checkout"]
|
|
assert result.diagnostics.selected_relation_ids == ["domain.payment"]
|
|
assert result.diagnostics.selected_chunk_ids == ["chunk.checkout.1"]
|
|
|
|
|
|
def test_component_evidence_prefers_facts_and_targeted_chunks() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "Runtime manager",
|
|
"content": "Overview",
|
|
"metadata": {"doc_id": "architecture.runtime_manager"},
|
|
},
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "Runtime facts",
|
|
"content": "Coordinates runtime orchestration",
|
|
"metadata": {"fact_id": "fact.runtime_manager.role", "doc_id": "architecture.runtime_manager"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "RuntimeManager chunk",
|
|
"content": "Starts worker loops and supervises control channels",
|
|
"metadata": {"chunk_id": "chunk.runtime_manager.1", "doc_id": "architecture.runtime_manager"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какую роль в системе играет RuntimeManager?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.planned_layers == ["D2_FACT_INDEX", "D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.selected_fact_ids == ["fact.runtime_manager.role"]
|
|
assert result.diagnostics.selected_doc_ids == ["architecture.runtime_manager"]
|
|
assert result.diagnostics.selected_chunk_ids == ["chunk.runtime_manager.1"]
|