PMFlow/backend/tests/test_adaptive_rag_mainline_...

1393 lines
50 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import pytest
from types import SimpleNamespace
from app.config.ai_prompts import PROMPT_AI_PMO_QA
from app.services.adaptive_rag_service.mixin_answer_entry import AdaptiveRAGAnswerEntryMixin
from app.services.adaptive_rag_service.mixin_answer_mainline import AdaptiveRAGAnswerMainlineMixin
from app.services.adaptive_rag_service.mixin_answer_retrieval import AdaptiveRAGAnswerRetrievalMixin
from app.services.adaptive_rag_service.mixin_conversation_utils_a_p1 import AdaptiveRAGConversationUtilsAMixinP1
from app.services.adaptive_rag_service.mixin_deep_guard_answer import AdaptiveRAGDeepGuardAnswerMixin
from app.services.adaptive_rag_service.mixin_deep_guard_rules import AdaptiveRAGDeepGuardRulesMixin
from app.services.adaptive_rag_service.mixin_core_intent import AdaptiveRAGCoreIntentMixin
from app.services.adaptive_rag_service.mixin_deterministic_query_core import (
AdaptiveRAGDeterministicQueryCoreMixin,
)
from app.services.adaptive_rag_service.mixin_manifest_feature_p2 import AdaptiveRAGManifestFeatureMixinP2
from app.services.adaptive_rag_service.mixin_multistep_core import AdaptiveRAGMultiStepCoreMixin
from app.services.adaptive_rag_service.mixin_requirement_ack_core import AdaptiveRAGRequirementAckCoreMixin
from app.services.adaptive_rag_service.mixin_retrieval_router_db import AdaptiveRAGRetrievalRouterDBMixin
from app.services.adaptive_rag_service.mixin_retrieval_vector import AdaptiveRAGRetrievalVectorMixin
from app.services.adaptive_rag_service.shared_part_02 import RetrievedItem
from app.services.adaptive_rag_service.shared_part_02 import ACTION_MODE_READ, DEEP_RETRIEVAL_STEPS, DEEP_VECTOR_TOP_K
from app.services.ai_pmo.workflow_semantic_planner import WorkflowSemanticPlanner
class _FakeAdaptiveRAGService(
AdaptiveRAGAnswerEntryMixin,
AdaptiveRAGAnswerMainlineMixin,
):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 9
self.ack_calls = 0
self.base_calls = 0
self.base_kwargs: dict[str, object] = {}
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
return "deep" if str(thinking_mode or "").strip().lower() == "deep" else "normal"
async def _prepare_answer_runtime(self, **kwargs):
clean_q = str(kwargs.get("clean_q") or "").strip()
action_mode = str(kwargs.get("action_mode") or "").strip().lower()
return {
"session_id": 101,
"turn_index": 2,
"context_session_id": 101,
"conversation": [],
"state_row": None,
"resolved_question": clean_q,
"effective_q": clean_q,
"rewrite_meta": {"mode": "direct"},
"query_type": "explain",
"effective_action_mode": action_mode,
"anchor_resolution": None,
}
async def _handle_ack_first_requirement(self, **kwargs):
self.ack_calls += 1
_ = kwargs
return {"answer": "ack-required", "processing_stage": "ack_only"}
async def _execute_base_answer_mainline(self, **kwargs):
self.base_calls += 1
self.base_kwargs = dict(kwargs)
return {"answer": "base-answer", "processing_stage": "qa"}
@pytest.mark.asyncio
async def test_answer_entry_read_mainline_skips_requirement_ack() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="现在有哪些阻塞?",
asked_by=1,
asked_via="pmflow_manual",
action_mode="read",
sync_requirement=True,
)
assert result["processing_stage"] == "qa"
assert service.ack_calls == 0
assert service.base_calls == 1
@pytest.mark.asyncio
async def test_answer_entry_write_mode_uses_requirement_ack_gate() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="请同步写入一条风险",
asked_by=1,
asked_via="pmflow_manual",
action_mode="write",
sync_requirement=True,
)
assert result["processing_stage"] == "ack_only"
assert service.ack_calls == 1
assert service.base_calls == 0
@pytest.mark.asyncio
async def test_answer_entry_high_accuracy_channel_respects_requested_normal_mode() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="最近有哪些项目?",
asked_by=1,
asked_via="pmflow",
thinking_mode="normal",
sync_requirement=True,
)
assert result["processing_stage"] == "qa"
assert service.ack_calls == 0
assert service.base_calls == 1
assert service.base_kwargs["active_thinking_mode"] == "normal"
assert service.base_kwargs["deep_thinking"] is False
assert int(service.base_kwargs["deep_retrieval_steps"]) == DEEP_RETRIEVAL_STEPS
assert int(service.base_kwargs["deep_vector_top_k"]) == DEEP_VECTOR_TOP_K
@pytest.mark.asyncio
async def test_answer_entry_forwards_consultant_mode_preference_to_mainline() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="这个方案怎么做更稳妥?",
asked_by=1,
asked_via="pmflow",
thinking_mode="normal",
consultant_mode_preference="advise",
)
assert result["processing_stage"] == "qa"
assert service.base_calls == 1
assert service.base_kwargs["consultant_mode_preference"] == "advise"
@pytest.mark.asyncio
async def test_answer_entry_mixed_request_skips_requirement_ack_gate() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="现在进展如何,并且新增任务确认流",
asked_by=1,
asked_via="pmflow_manual",
action_mode="read_write",
sync_requirement=True,
message_intent="mixed",
)
assert result["processing_stage"] == "qa"
assert service.ack_calls == 0
assert service.base_calls == 1
assert service.base_kwargs["active_thinking_mode"] == "normal"
assert service.base_kwargs["deep_thinking"] is False
@pytest.mark.asyncio
async def test_answer_entry_feishu_write_mode_still_goes_base_mainline() -> None:
service = _FakeAdaptiveRAGService()
result = await service.answer_question(
question="请在飞书里补一条风险",
asked_by=1,
asked_via="feishu",
action_mode="write",
sync_requirement=True,
)
assert result["processing_stage"] == "qa"
assert service.ack_calls == 0
assert service.base_calls == 1
assert service.base_kwargs["active_thinking_mode"] == "normal"
assert service.base_kwargs["deep_thinking"] is False
assert int(service.base_kwargs["deep_retrieval_steps"]) == DEEP_RETRIEVAL_STEPS
assert int(service.base_kwargs["deep_vector_top_k"]) == DEEP_VECTOR_TOP_K
@pytest.mark.asyncio
async def test_requirement_ack_auto_mode_respects_structured_query_bundle(monkeypatch: pytest.MonkeyPatch) -> None:
class _ModeService(AdaptiveRAGRequirementAckCoreMixin):
pass
async def _fake_aplan_message(cls, *, message_text: str, source_system: str, metadata=None):
_ = (cls, message_text, source_system, metadata)
return WorkflowSemanticPlanner.deserialize_bundle(
{
"bundle_id": "semantic-test",
"semantic_version": "v1",
"source_system": "adaptive_rag_ack",
"primary_workflow": "query_flow",
"workflows": [
{
"workflow_type": "query_flow",
"workflow_action": "answer",
"target_scope": "global",
"operation_codes": ["query.answer"],
"requires_confirmation": False,
"confidence": 0.91,
"clarification_needed": False,
"metadata": {},
}
],
"message_gate": "auto",
"reply_contract": {"mode": "text", "template_key": "workflow_bundle", "summary": "query_flow"},
"diagnostics": {"message_intent": "question"},
}
)
monkeypatch.setattr(WorkflowSemanticPlanner, "aplan_message", classmethod(_fake_aplan_message))
mode = await _ModeService._resolve_action_mode_async(
requested_mode="auto",
question="先不要创建任务,告诉我当前进展和风险",
effective_question="先不要创建任务,告诉我当前进展和风险",
asked_via="pmflow_manual",
)
assert mode == ACTION_MODE_READ
def test_query_runtime_planner_prefers_fast_path_for_list_question_even_if_requested_deep() -> None:
class _FakePlannerService(AdaptiveRAGAnswerMainlineMixin):
pass
service = _FakePlannerService()
decision = service._plan_query_runtime(
question="现在有哪些项目?",
effective_question="现在有哪些项目?",
query_type="list",
requested_thinking_mode="deep",
enforce_scene_prompt_only=False,
)
assert decision.answer_mode == "fast_path"
assert decision.thinking_mode == "normal"
assert decision.allow_fast_path is True
def test_query_runtime_planner_bypasses_fast_path_when_consultant_mode_preference_present() -> None:
class _FakePlannerService(AdaptiveRAGAnswerMainlineMixin):
pass
service = _FakePlannerService()
decision = service._plan_query_runtime(
question="现在有哪些项目?",
effective_question="现在有哪些项目?",
query_type="list",
requested_thinking_mode="normal",
enforce_scene_prompt_only=False,
consultant_mode_preference="advise",
)
assert decision.answer_mode == "retrieval_normal"
assert decision.thinking_mode == "normal"
assert decision.allow_fast_path is False
@pytest.mark.asyncio
async def test_deterministic_query_prefers_project_created_at_answer() -> None:
calls: list[str] = []
class _FakeDeterministicService(AdaptiveRAGDeterministicQueryCoreMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
async def _try_fast_project_created_at_answer(self, *, clean_q: str, project_id):
calls.append("created_at")
_ = (clean_q, project_id)
return {"answer": "创建于 2026-03-01 09:30。", "sources": [], "missing_fields": []}
async def _try_fast_project_progress_answer(self, *, clean_q: str, project_id):
calls.append("progress")
_ = (clean_q, project_id)
return {"answer": "进度回答", "sources": []}
async def _try_fast_project_existence_answer(self, *, clean_q: str, project_id):
calls.append("existence")
_ = (clean_q, project_id)
return {"answer": "存在性回答", "sources": []}
service = _FakeDeterministicService()
result = await service._run_deterministic_query(
question="他是什么时候创建的?",
project_id=7,
query_type="status",
)
assert result is not None
assert "创建于 2026-03-01 09:30" in str(result["answer"])
assert calls[0] == "created_at"
@pytest.mark.asyncio
async def test_query_semantics_prefers_structured_model_output_over_hint_tables(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FakeLLM:
async def generate(self, prompt: str, **kwargs):
_ = (prompt, kwargs)
return (
'{"intent":"question","question_part":"我感觉很多重复的项目在系统,你能列一下吗?",'
'"statement_part":"","query_type":"list","retrieval_domains":["db","graph"],'
'"subject_focus":"project","scope_hint":"global","feature_focus":"none"}'
)
class _FakeService(AdaptiveRAGCoreIntentMixin, AdaptiveRAGDeterministicQueryCoreMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
async def _fake_aplan_message(cls, *, message_text: str, source_system: str, metadata=None):
_ = (cls, message_text, source_system, metadata)
return WorkflowSemanticPlanner.deserialize_bundle(
{
"bundle_id": "semantic-query-type",
"semantic_version": "v1",
"source_system": "adaptive_rag",
"primary_workflow": "query_flow",
"workflows": [
{
"workflow_type": "query_flow",
"workflow_action": "answer",
"target_scope": "global",
"operation_codes": ["query.answer"],
"requires_confirmation": False,
"confidence": 0.93,
"clarification_needed": False,
"metadata": {},
}
],
"message_gate": "auto",
"reply_contract": {"mode": "text", "template_key": "workflow_bundle", "summary": "query_flow"},
"diagnostics": {"message_intent": "question"},
}
)
monkeypatch.setattr(WorkflowSemanticPlanner, "aplan_message", classmethod(_fake_aplan_message))
service = _FakeService()
semantics = await service._resolve_query_semantics(
message="我感觉很多重复的项目在系统,你能列一下吗?",
source_system="pmflow",
)
assert semantics["query_type"] == "list"
assert semantics["retrieval_domains"] == ["db", "graph"]
assert semantics["intent"] == "question"
assert semantics["subject_focus"] == "project"
assert semantics["scope_hint"] == "global"
assert semantics["feature_focus"] == "none"
@pytest.mark.asyncio
async def test_classify_intent_prefers_structured_semantics_without_rule_split(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FakeLLM:
async def generate(self, prompt: str, **kwargs):
_ = (prompt, kwargs)
return (
'{"intent":"mixed","question_part":"现在进展如何?","statement_part":"并新增任务确认流",'
'"query_type":"status","retrieval_domains":["db","graph","pmflow"]}'
)
class _FakeService(AdaptiveRAGCoreIntentMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
async def _fake_aplan_message(cls, *, message_text: str, source_system: str, metadata=None):
_ = (cls, message_text, source_system, metadata)
return WorkflowSemanticPlanner.deserialize_bundle(
{
"bundle_id": "semantic-intent",
"semantic_version": "v1",
"source_system": "adaptive_rag",
"primary_workflow": "query_flow",
"workflows": [
{
"workflow_type": "query_flow",
"workflow_action": "answer",
"target_scope": "global",
"operation_codes": ["query.answer"],
"requires_confirmation": False,
"confidence": 0.88,
"clarification_needed": False,
"metadata": {},
}
],
"message_gate": "auto",
"reply_contract": {"mode": "text", "template_key": "workflow_bundle", "summary": "query_flow"},
"diagnostics": {"message_intent": "business_review"},
}
)
monkeypatch.setattr(WorkflowSemanticPlanner, "aplan_message", classmethod(_fake_aplan_message))
service = _FakeService()
result = await service.classify_intent("现在进展如何?并新增任务确认流")
assert result == {
"intent": "mixed",
"confidence": 0.88,
"question_part": "现在进展如何?",
"statement_part": "并新增任务确认流",
}
@pytest.mark.asyncio
async def test_vector_retrieval_skips_legacy_fallback_for_non_postgresql_session() -> None:
class _FakeLLM:
async def get_embedding_async(self, question: str, **kwargs):
_ = (question, kwargs)
return [0.1, 0.2, 0.3]
class _FakeService(AdaptiveRAGRetrievalVectorMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
self.legacy_called = False
def _map_domains_to_source_types(self, domains: list[str]) -> list[str]:
return list(domains)
def _is_postgresql_session(self) -> bool:
return False
async def _retrieve_vector_context_legacy(self, **kwargs):
_ = kwargs
self.legacy_called = True
return ["legacy-hit"]
service = _FakeService()
result = await service._retrieve_vector_context(
"查询项目文档",
domains=["doc"],
project_id=None,
source_chat_id=None,
)
assert result == []
assert service.legacy_called is False
@pytest.mark.asyncio
async def test_vector_retrieval_skips_legacy_fallback_when_pg_query_errors() -> None:
class _FakeLLM:
async def get_embedding_async(self, question: str, **kwargs):
_ = (question, kwargs)
return [0.1, 0.2, 0.3]
class _FakeService(AdaptiveRAGRetrievalVectorMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
self.legacy_called = False
def _map_domains_to_source_types(self, domains: list[str]) -> list[str]:
return list(domains)
def _is_postgresql_session(self) -> bool:
return True
async def _retrieve_vector_context_pg(self, **kwargs):
_ = kwargs
return None
async def _retrieve_vector_context_legacy(self, **kwargs):
_ = kwargs
self.legacy_called = True
return ["legacy-hit"]
service = _FakeService()
result = await service._retrieve_vector_context(
"查询项目文档",
domains=["doc"],
project_id=None,
source_chat_id=None,
)
assert result == []
assert service.legacy_called is False
@pytest.mark.asyncio
async def test_deep_guard_answer_raises_on_empty_llm_response() -> None:
class _FakeLLM:
async def generate(self, prompt: str, **kwargs):
_ = (prompt, kwargs)
return ""
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
return "normal" if str(thinking_mode or "").strip().lower() != "deep" else "deep"
async def get_project_manifest(self, project_id):
_ = project_id
return {}
async def _select_scene_prompt(self, **kwargs):
_ = kwargs
return "external_trend", "请回答问题"
def _enforce_feature_evidence_in_answer(self, **kwargs):
return str(kwargs.get("answer_text") or "")
service = _FakeService()
with pytest.raises(ValueError, match="adaptive_rag_answer_empty_response"):
await service._generate_answer(
question="最近AI行业有什么动态",
retrieved=[],
conversation=[],
project_id=None,
asked_via="feishu",
)
@pytest.mark.asyncio
async def test_deep_guard_answer_uses_low_first_routing_meta_for_deep_thinking() -> None:
class _FakeLLM:
def __init__(self) -> None:
self.kwargs: dict[str, object] = {}
async def generate(self, prompt: str, **kwargs):
_ = prompt
self.kwargs = dict(kwargs)
return "low-tier answer"
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
return "deep" if str(thinking_mode or "").strip().lower() == "deep" else "normal"
async def get_project_manifest(self, project_id):
_ = project_id
return {}
async def _select_scene_prompt(self, **kwargs):
_ = kwargs
return "external_trend", "请回答问题"
def _enforce_feature_evidence_in_answer(self, **kwargs):
return str(kwargs.get("answer_text") or "")
service = _FakeService()
result = await service._generate_answer(
question="最近AI行业有什么动态",
retrieved=[],
conversation=[],
project_id=None,
asked_via="feishu",
thinking_mode="deep",
)
assert result == "low-tier answer"
assert service.llm.kwargs["function_name"] == "adaptive_rag_answer_deep"
routing_meta = service.llm.kwargs["routing_meta"]
assert isinstance(routing_meta, dict)
assert routing_meta["tier_alias"] == "low"
assert routing_meta["planned_tier"] == "low"
assert routing_meta["response_contract"] == "complete"
assert routing_meta["fallback_policy"] == "low_to_high"
assert routing_meta["execution_mode"] == "query"
assert routing_meta["routing_decision"] == "force_low_first"
@pytest.mark.asyncio
async def test_deep_guard_answer_prefers_member_workload_direct_answer() -> None:
class _FakeResult:
def __init__(self, rows=None, scalar=None):
self._rows = rows or []
self._scalar = scalar
def all(self):
return self._rows
def first(self):
return self._rows[0] if self._rows else None
def scalar_one_or_none(self):
return self._scalar
class _FakeDB:
def __init__(self) -> None:
self.calls = 0
async def execute(self, stmt, params=None): # noqa: ANN001
self.calls += 1
if self.calls == 1:
return _FakeResult(rows=[SimpleNamespace(id=24, name="杨宇", username="yangyu")])
return _FakeResult(rows=[(265.56, 95.6, 60.0)])
class _FakeLLM:
async def generate(self, prompt: str, **kwargs): # pragma: no cover - should not be called
raise AssertionError("LLM should not be called for direct member workload answer")
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
self.llm = _FakeLLM()
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
return "normal"
async def get_project_manifest(self, project_id):
_ = project_id
return {}
async def _resolve_answer_scene_semantics(self, **kwargs):
_ = kwargs
return {
"scene_key": "member_workload",
"snapshot_focus": "none",
"deep_intent": "none",
"confidence": 0.96,
"reason": "structured-scene",
}
async def _select_scene_prompt(self, **kwargs):
_ = kwargs
return "member_workload", "member workload prompt"
def _enforce_feature_evidence_in_answer(self, **kwargs):
return str(kwargs.get("answer_text") or "")
service = _FakeService()
result = await service._generate_answer(
question="杨宇现在负载情况如何?",
retrieved=[],
conversation=[],
project_id=None,
asked_via="feishu",
)
assert "杨宇当前负载265.56%" in result
assert "严重超载" in result
@pytest.mark.asyncio
async def test_deep_guard_answer_prefers_structured_member_workload_scene_without_keyword_hints() -> None:
class _FakeResult:
def __init__(self, rows=None):
self._rows = rows or []
def all(self):
return self._rows
def first(self):
return self._rows[0] if self._rows else None
class _FakeDB:
def __init__(self) -> None:
self.calls = 0
async def execute(self, stmt, params=None): # noqa: ANN001
_ = (stmt, params)
self.calls += 1
if self.calls == 1:
return _FakeResult(rows=[SimpleNamespace(id=24, name="杨宇", username="yangyu")])
return _FakeResult(rows=[(82.0, 41.0, 50.0)])
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
_ = thinking_mode
return "normal"
async def get_project_manifest(self, project_id):
_ = project_id
return {}
async def _resolve_answer_scene_semantics(self, **kwargs):
_ = kwargs
return {
"scene_key": "member_workload",
"snapshot_focus": "none",
"deep_intent": "none",
"confidence": 0.95,
"reason": "structured-scene",
}
async def _select_scene_prompt(self, **kwargs): # pragma: no cover - should not be called
raise AssertionError("scene prompt should not be rendered for direct workload answer")
def _enforce_feature_evidence_in_answer(self, **kwargs):
return str(kwargs.get("answer_text") or "")
service = _FakeService()
result = await service._generate_answer(
question="杨宇现在忙不忙?",
retrieved=[],
conversation=[],
project_id=None,
asked_via="feishu",
)
assert "杨宇当前负载82.00%" in result
assert "负载偏高" in result
@pytest.mark.asyncio
async def test_structured_db_retrieval_prefers_structured_people_focus_without_keyword_hints() -> None:
class _FakeScalarResult:
def __init__(self, rows):
self._rows = rows
def scalars(self):
return self
def all(self):
return self._rows
class _FakeDB:
async def execute(self, stmt, params=None): # noqa: ANN001
_ = (stmt, params)
return _FakeScalarResult(
[
SimpleNamespace(id=7, name="杨宇", role="backend", updated_at=None),
SimpleNamespace(id=8, name="林宇航", role="pm", updated_at=None),
]
)
class _FakeService(AdaptiveRAGRetrievalRouterDBMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
def _query_subject_focus(self, query_semantics): # noqa: ANN001
return str((query_semantics or {}).get("subject_focus") or "unknown")
def _query_scope_hint(self, query_semantics): # noqa: ANN001
return str((query_semantics or {}).get("scope_hint") or "none")
def _normalize_source_registry_metadata(self, *, source_type: str, metadata: dict, explicit_updated_at=None):
_ = (source_type, explicit_updated_at)
return dict(metadata)
service = _FakeService()
rows = await service._retrieve_structured_db(
"他们都有哪些人?",
project_id=None,
query_semantics={
"subject_focus": "people",
"scope_hint": "global",
"feature_focus": "none",
},
)
assert len(rows) == 1
assert "成员清单" in rows[0].content
assert "杨宇" in rows[0].content
def test_feature_evidence_prefers_structured_scene_without_keyword_hints() -> None:
class _FakeService(AdaptiveRAGManifestFeatureMixinP2):
def _query_subject_focus(self, query_semantics): # noqa: ANN001
return str((query_semantics or {}).get("subject_focus") or "unknown")
def _query_feature_focus(self, query_semantics): # noqa: ANN001
return str((query_semantics or {}).get("feature_focus") or "none")
def _collect_feature_evidence(self, *, question: str, retrieved, system_snapshot):
_ = (question, retrieved, system_snapshot)
return [
{
"feature_name": "智能配置",
"route": "/ai-config",
"page_file": "frontend/src/pages/AIConfig.tsx",
"apis": ["/api/v1/ai-config"],
}
]
service = _FakeService()
text = service._enforce_feature_evidence_in_answer(
question="这个模块怎么进?",
answer_text="先看说明。",
retrieved=[],
system_snapshot={},
scene_semantics={"snapshot_focus": "feature"},
)
assert "证据补充(Feature Catalog)" in text
assert "/ai-config" in text
@pytest.mark.asyncio
async def test_deep_guard_answer_feishu_direct_prefers_manifest_and_compact_focus_context() -> None:
class _FakeLLM:
def __init__(self) -> None:
self.prompt = ""
async def generate(self, prompt: str, **kwargs):
_ = kwargs
self.prompt = prompt
return "信息不足,缺少:目标对象"
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
self.llm = _FakeLLM()
def _normalize_thinking_mode(self, thinking_mode: str) -> str:
return "normal"
async def get_project_manifest(self, project_id):
_ = project_id
return {
"manifest_text": "项目#7 Alpha(进行中);功能 18 个;KG 实体 12,关系 19。"
}
async def _render_scene_prompt(self, *, prompt_key: str, fallback_template: str, **kwargs):
_ = prompt_key
return str(fallback_template).format(**kwargs).strip()
def _enforce_feature_evidence_in_answer(self, **kwargs):
return str(kwargs.get("answer_text") or "")
service = _FakeService()
system_snapshot = {
"assistant_profile": {"name": "PMFlow", "identity": "项目管理顾问"},
"platform_profile": {
"backend": {"module_counts": {"projects": 28, "tasks": 146, "signals": 41}},
"feature_catalog": [
{"name": "项目管理", "route": "/projects"},
{"name": "信号接入", "route": "/signals"},
{"name": "任务看板", "route": "/board"},
{"name": "知识图谱", "route": "/kg"},
{"name": "训练中心", "route": "/training"},
],
},
}
user_snapshot = {
"current_user": {"name": "林宇航"},
"projects": [{"id": idx, "title": f"项目{idx}"} for idx in range(1, 15)],
"tasks": [{"id": idx, "title": f"任务{idx}"} for idx in range(1, 10)],
}
result = await service._generate_answer(
question="信息不全,继续提供",
retrieved=[],
conversation=[
{"question": "上一次你缺什么", "answer": "缺目标对象和验收口径"},
{"question": "现在呢", "answer": "还缺时间边界"},
],
project_id=7,
system_snapshot=system_snapshot,
user_snapshot=user_snapshot,
asked_via="feishu",
)
assert result == "信息不足,缺少:目标对象"
assert "项目画像:" in service.llm.prompt
assert "项目#7 Alpha(进行中)" in service.llm.prompt
assert "系统摘要:" in service.llm.prompt
assert "用户焦点:" in service.llm.prompt
assert "林宇航" in service.llm.prompt
assert '"projects"' not in service.llm.prompt
assert "项目14" not in service.llm.prompt
@pytest.mark.asyncio
async def test_select_scene_prompt_applies_consultant_mode_preference_wrapper() -> None:
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
async def _render_scene_prompt(self, *, prompt_key: str, fallback_template: str, **kwargs):
_ = (fallback_template, kwargs)
return f"PROMPT::{prompt_key}"
service = _FakeService()
scene_key, prompt = await service._select_scene_prompt(
question="这个方案怎么做更稳妥?",
project_id=7,
asked_via="pmflow",
message_intent=None,
mixed_signal_part=None,
conversation_text="无",
context_blocks_text="无",
system_snapshot={},
user_snapshot={},
system_snapshot_text="{}",
user_snapshot_text="{}",
relevance_max=0.61,
query_type="explain",
consultant_mode_preference="advise",
)
assert scene_key == "default_consultant_advise"
assert "PROMPT::PROMPT_RAG_ANSWER" in prompt
assert "稳妥建议" in prompt
def test_ai_pmo_qa_prompt_includes_five_lenses_and_runtime_state_guard() -> None:
assert "事实源在哪" in PROMPT_AI_PMO_QA
assert "输出模板在哪" in PROMPT_AI_PMO_QA
assert "检索范围在哪" in PROMPT_AI_PMO_QA
assert "执行器在哪" in PROMPT_AI_PMO_QA
assert "反馈回流在哪" in PROMPT_AI_PMO_QA
assert "静态资料" in PROMPT_AI_PMO_QA
assert "运行态配置" in PROMPT_AI_PMO_QA
@pytest.mark.asyncio
async def test_select_scene_prompt_routes_business_review_to_ai_pmo_prompt() -> None:
class _FakeService(AdaptiveRAGDeepGuardAnswerMixin):
def __init__(self) -> None:
self.db = object()
self.tenant_id = 1
async def _render_scene_prompt(self, *, prompt_key: str, fallback_template: str, **kwargs):
_ = fallback_template
return (
f"PROMPT::{prompt_key}|"
f"business_review={kwargs.get('business_review_mode')}|"
f"framework={kwargs.get('review_framework')}"
)
service = _FakeService()
scene_key, prompt = await service._select_scene_prompt(
question="我要做一个大模型 API 聚合平台,帮我分析下要怎么做",
project_id=None,
asked_via="feishu",
message_intent="business_review",
mixed_signal_part=None,
conversation_text="用户上一轮提到要做 API 聚合平台",
context_blocks_text="无",
system_snapshot={},
user_snapshot={},
system_snapshot_text="{}",
user_snapshot_text="{}",
relevance_max=0.51,
query_type="explain",
consultant_mode_preference="clarify",
)
assert scene_key == "ai_pmo_business_review_consultant_clarify"
assert "PROMPT::PROMPT_AI_PMO_QA" in prompt
assert "business_review=true" in prompt
assert "framework=five_lenses" in prompt
def test_runtime_state_query_guard_blocks_static_catalog_evidence() -> None:
class _FakeService(AdaptiveRAGDeepGuardRulesMixin):
pass
service = _FakeService()
decision = service._evaluate_post_retrieval_guard(
deep_intent="runtime_state_query",
sources=[
{"source": "feature_catalog", "label": "Feature Catalog", "metadata": {}},
{"source": "graph", "label": "Knowledge Graph", "metadata": {"path": "1->uses->2"}},
],
relevance_max=0.96,
)
assert decision["should_block_with_guard"] is True
assert decision["guard_focus"] == "runtime_state"
assert "无法仅凭静态资料确认" in service._build_evidence_guard_answer(
"高级模型路由是否启用?",
guard_focus=str(decision["guard_focus"]),
)
def test_runtime_state_query_guard_allows_routing_config_evidence() -> None:
class _FakeService(AdaptiveRAGDeepGuardRulesMixin):
pass
service = _FakeService()
decision = service._evaluate_post_retrieval_guard(
deep_intent="runtime_state_query",
sources=[
{
"source": "doc",
"label": "Routing Config",
"metadata": {"source_id": "backend/.control_plane/pmflow-routing-rules.json"},
}
],
relevance_max=0.74,
)
assert decision["should_block_with_guard"] is False
def test_engine_process_guard_requests_auto_retry_on_snapshot_only_sources() -> None:
class _FakeService(AdaptiveRAGDeepGuardRulesMixin):
pass
service = _FakeService()
decision = service._evaluate_post_retrieval_guard(
deep_intent="engine_process_query",
sources=[
{"source": "system", "label": "System Snapshot", "metadata": {}},
{"source": "system", "label": "User Snapshot", "metadata": {}},
],
relevance_max=0.41,
)
assert decision["auto_retry_reason"] == "snapshot_only"
def test_chat_external_fact_scene_policy_keeps_fact_challenge_in_chat() -> None:
class _FakeService(AdaptiveRAGAnswerMainlineMixin):
pass
service = _FakeService()
normalized = service._apply_chat_external_fact_scene_policy(
question="不对吧,截止当前日期都已经有 codex5.4 了",
project_id=None,
scene_semantics={
"scene_key": "ai_pmo_correction",
"snapshot_focus": "feature",
"deep_intent": "correction_turn",
"confidence": 0.95,
},
anchor_prompt_variant="external_trend",
)
assert normalized["scene_key"] == "industry_trend"
assert normalized["deep_intent"] == "none"
assert normalized["external_fact_chat"] is True
assert normalized["requires_freshness"] is True
def test_chat_external_fact_scene_policy_promotes_external_model_brand_queries() -> None:
class _FakeService(AdaptiveRAGAnswerMainlineMixin):
pass
service = _FakeService()
normalized = service._apply_chat_external_fact_scene_policy(
question="当前codex最新的模型用的是什么?",
project_id=None,
scene_semantics={
"scene_key": "ai_pmo",
"snapshot_focus": "identity",
"deep_intent": "none",
"confidence": 0.92,
},
anchor_prompt_variant=None,
)
assert normalized["scene_key"] == "industry_trend"
assert normalized["snapshot_focus"] == "none"
assert normalized["external_fact_chat"] is True
assert normalized["requires_freshness"] is True
def test_chat_external_fact_scene_policy_keeps_explicit_internal_model_config_question() -> None:
class _FakeService(AdaptiveRAGAnswerMainlineMixin):
pass
service = _FakeService()
normalized = service._apply_chat_external_fact_scene_policy(
question="我们系统当前接的是哪个 codex 模型配置?",
project_id=None,
scene_semantics={
"scene_key": "ai_pmo",
"snapshot_focus": "identity",
"deep_intent": "none",
"confidence": 0.92,
},
anchor_prompt_variant=None,
)
assert normalized["scene_key"] == "ai_pmo"
assert normalized["snapshot_focus"] == "identity"
assert normalized.get("external_fact_chat") is not True
def test_external_fact_freshness_guard_rejects_internal_only_sources() -> None:
class _FakeService(AdaptiveRAGDeepGuardRulesMixin):
pass
service = _FakeService()
assert service._requires_external_freshness(
question="codex 目前最高模型出到什么了",
scene_semantics={"scene_key": "industry_trend", "requires_freshness": True},
) is True
assert service._has_external_grounding_sources(
[
{"source": "graph", "label": "Knowledge Graph", "metadata": {"path": "1->belongs_to->2"}},
{"source": "feature_catalog", "label": "Feature Catalog", "metadata": {}},
]
) is False
answer = service._build_external_freshness_guard_answer("codex 目前最高模型出到什么了")
assert "当前未接入实时外部数据" in answer
assert "backend/app/services" not in answer
@pytest.mark.asyncio
async def test_second_pass_query_skips_internal_hints_for_external_freshness_only() -> None:
class _FakeService(AdaptiveRAGMultiStepCoreMixin):
async def entity_resolution(self, **kwargs): # pragma: no cover - should not be called
raise AssertionError(f"entity_resolution should be skipped: {kwargs}")
async def get_project_manifest(self, **kwargs): # pragma: no cover - should not be called
raise AssertionError(f"get_project_manifest should be skipped: {kwargs}")
service = _FakeService()
revised, meta = await service._build_second_pass_query(
question="codex 目前最高模型出到什么了",
project_id=None,
query_semantics={"external_freshness_only": True},
)
assert revised == "codex 目前最高模型出到什么了"
assert meta["mode"] == "freshness_guard_skip_internal_hint"
@pytest.mark.asyncio
async def test_live_external_freshness_context_fetches_url_rows_before_guard() -> None:
class _FakeService(AdaptiveRAGAnswerRetrievalMixin, AdaptiveRAGDeepGuardRulesMixin):
async def _retrieve_live_external_context(self, *, question: str, max_results: int = 3, max_pages: int = 3):
_ = (question, max_results, max_pages)
return (
[
RetrievedItem(
source="url",
content="OpenAI 在官方文章中说明 GPT-5.4 正在 rollout 到 Codex。",
score=0.93,
source_label="URL[Introducing GPT-5.4]",
metadata={
"url": "https://openai.com/index/introducing-gpt-5-4/",
"target_url": "https://openai.com/index/introducing-gpt-5-4/",
"title": "Introducing GPT-5.4",
},
)
],
{"query": question, "result_count": 1, "fetched_pages": 1},
)
service = _FakeService()
rows, meta = await service._maybe_fetch_live_external_freshness_context(
question="当前codex最新的模型用的是什么?",
effective_question="当前codex最新的模型用的是什么?",
query_semantics={},
scene_semantics={"scene_key": "industry_trend", "requires_freshness": True},
)
assert len(rows) == 1
assert rows[0].source == "url"
assert rows[0].metadata["target_url"] == "https://openai.com/index/introducing-gpt-5-4/"
assert meta["mode"] == "live_external_context"
@pytest.mark.asyncio
async def test_live_external_search_uses_proxy_aware_http_client(monkeypatch) -> None:
captured: dict[str, object] = {}
class _FakeResponse:
def __init__(self, text: str) -> None:
self.text = text
def raise_for_status(self) -> None:
return None
class _FakeAsyncClient:
def __init__(self, *args, **kwargs) -> None:
_ = args
captured.update(kwargs)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
_ = (exc_type, exc, tb)
return None
async def get(self, url: str, headers: dict[str, str]):
captured["url"] = url
captured["headers"] = headers
return _FakeResponse('<a class="result__a" href="https://example.com/post">OpenAI Codex latest model</a>')
monkeypatch.setattr(
"app.services.adaptive_rag_service.mixin_conversation_utils_a_p1.httpx.AsyncClient",
_FakeAsyncClient,
)
class _FakeService(AdaptiveRAGConversationUtilsAMixinP1):
pass
rows = await _FakeService()._search_live_external_results(
query="OpenAI Codex latest model",
max_results=1,
)
assert captured["trust_env"] is True
assert rows == [{"url": "https://example.com/post", "title": "OpenAI Codex latest model"}]
@pytest.mark.asyncio
async def test_live_external_fetch_uses_proxy_aware_http_client(monkeypatch) -> None:
captured: dict[str, object] = {}
class _FakeResponse:
def __init__(self) -> None:
self.headers = {"content-type": "text/html"}
self.text = "<html><body><h1>Fresh external fact</h1></body></html>"
def raise_for_status(self) -> None:
return None
class _FakeAsyncClient:
def __init__(self, *args, **kwargs) -> None:
_ = args
captured.update(kwargs)
async def __aenter__(self):
return self
async def __aexit__(self, exc_type, exc, tb) -> None:
_ = (exc_type, exc, tb)
return None
async def get(self, url: str, headers: dict[str, str]):
captured["url"] = url
captured["headers"] = headers
return _FakeResponse()
monkeypatch.setattr(
"app.services.adaptive_rag_service.mixin_conversation_utils_a_p1.httpx.AsyncClient",
_FakeAsyncClient,
)
class _FakeService(AdaptiveRAGConversationUtilsAMixinP1):
pass
text = await _FakeService()._fetch_url_text("https://example.com/post")
assert captured["trust_env"] is True
assert captured["url"] == "https://example.com/post"
assert "Fresh external fact" in text
def test_live_external_search_result_score_prefers_official_domain_for_model_brand_query() -> None:
official = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_search_result(
question="当前 codex 最新的模型用的是什么?",
title="Codex 全面升级 - OpenAI",
url="https://openai.com/zh-Hans-CN/index/introducing-upgrades-to-codex/",
)
third_party = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_search_result(
question="当前 codex 最新的模型用的是什么?",
title="Codex 模型选择 - 菜鸟教程",
url="https://www.runoob.com/codex/codex-models.html",
)
assert official > third_party
def test_live_external_search_result_score_prefers_codex_models_reference_page_for_latest_model_question() -> None:
reference_page = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_search_result(
question="当前 codex 最新的模型用的是什么?",
title="Models - Codex | OpenAI Developers",
url="https://developers.openai.com/codex/models",
)
announcement_page = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_search_result(
question="当前 codex 最新的模型用的是什么?",
title="Introducing GPT-5.3-Codex - OpenAI",
url="https://openai.com/index/introducing-gpt-5-3-codex/",
)
assert reference_page > announcement_page
def test_live_external_search_queries_prefer_official_site_for_model_brand_question() -> None:
queries = AdaptiveRAGConversationUtilsAMixinP1._build_live_external_search_queries(
"当前 codex 最新的模型用的是什么?"
)
assert queries
assert queries[0] == "site:openai.com codex latest model"
assert any(query == "当前 codex 最新的模型用的是什么" for query in queries)
def test_live_external_search_queries_strip_internal_runtime_tokens_for_model_brand_question() -> None:
queries = AdaptiveRAGConversationUtilsAMixinP1._build_live_external_search_queries(
"围绕“_sync_pmflow_rag_async->backend/app/api/sstc.py”回答:当前codex最新的模型用的是什么?"
)
assert queries
assert queries[0] == "site:openai.com codex latest model"
assert all("backend/app/api/sstc.py" not in query for query in queries)
assert all("_sync_pmflow_rag_async" not in query for query in queries)
def test_live_external_chunk_score_prefers_latest_fact_chunk_for_chinese_model_question() -> None:
latest_chunk = (
"Models – Codex | OpenAI Developers Latest: GPT-5.4 "
"Prompt guidance Core concepts Text generation Code generation"
)
nav_chunk = (
"Codex Docs Guides, concepts, and product docs for Codex "
"Use cases Example workflows and tasks teams hand to Codex"
)
latest_score = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_chunk(
question="当前codex最新的模型用的是什么?",
title="Models - Codex | OpenAI Developers",
content=latest_chunk,
url="https://developers.openai.com/codex/models",
)
nav_score = AdaptiveRAGConversationUtilsAMixinP1._score_live_external_chunk(
question="当前codex最新的模型用的是什么?",
title="Models - Codex | OpenAI Developers",
content=nav_chunk,
url="https://developers.openai.com/codex/models",
)
assert latest_score > nav_score
def test_external_model_fact_direct_answer_prefers_explicit_latest_evidence() -> None:
retrieved = [
RetrievedItem(
source="url",
content=(
"Models – Codex | OpenAI Developers Latest: GPT-5.4 "
"Prompt guidance Core concepts Text generation"
),
score=0.94,
source_label="URL[Models - Codex | OpenAI Developers]",
metadata={
"url": "https://developers.openai.com/codex/models",
"target_url": "https://developers.openai.com/codex/models",
},
)
]
answer = AdaptiveRAGDeepGuardAnswerMixin._build_external_model_fact_direct_answer(
question="当前codex最新的模型用的是什么?",
retrieved=retrieved,
)
assert answer is not None
assert "GPT-5.4" in answer
assert "OpenAI" in answer
@pytest.mark.asyncio
async def test_live_external_context_prefers_official_candidate_beyond_base_page_window() -> None:
class _FakeService(AdaptiveRAGConversationUtilsAMixinP1):
async def _search_live_external_results(self, *, query: str, max_results: int = 5):
_ = (query, max_results)
return [
{"url": "https://example.com/codex-guide", "title": "Codex 全面上手指南"},
{"url": "https://example.com/codex-models", "title": "Codex 模型选择"},
{"url": "https://example.com/codex-news", "title": "Codex 最新动态"},
{
"url": "https://openai.com/zh-Hans-CN/index/introducing-upgrades-to-codex/",
"title": "Codex 全面升级 - OpenAI",
},
]
async def _fetch_url_text(self, url: str) -> str:
if "openai.com" in url:
return "OpenAI 官方说明 Codex 当前使用 GPT-5.4。"
return "第三方教程,只介绍 Codex 的一般用法。"
rows, meta = await _FakeService()._retrieve_live_external_context(
question="当前 codex 最新的模型用的是什么?",
max_results=1,
max_pages=1,
)
assert rows
assert rows[0].metadata["url"] == "https://openai.com/zh-Hans-CN/index/introducing-upgrades-to-codex/"
assert meta["fetched_pages"] >= 1