207 lines
7.1 KiB
Python
207 lines
7.1 KiB
Python
from types import SimpleNamespace
|
||
|
||
import pytest
|
||
|
||
from app.services.external_query_service.planner import ExternalQueryPlanner
|
||
from app.services.external_query_service.presenter import ExternalAnswerPresenter
|
||
|
||
|
||
def test_fallback_plan_stays_generic_for_fresh_weather_semantics() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="帮我看看未来5天的广州天气如何",
|
||
query_semantics={
|
||
"query_type": "status",
|
||
"object_mentions": ["广州"],
|
||
"query_profile": {
|
||
"needs_freshness": True,
|
||
"referenced_objects": ["weather"],
|
||
},
|
||
},
|
||
)
|
||
|
||
assert plan.query_kind == "generic_web"
|
||
assert plan.answer_contract == "summary_with_sources"
|
||
assert plan.location_hint is None
|
||
assert plan.result_limit == 3
|
||
|
||
|
||
def test_fallback_plan_uses_compare_contract_without_object_kind_mapping() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="对比 OpenAI 和 Anthropic 最新模型能力差异",
|
||
query_semantics={
|
||
"query_type": "compare",
|
||
"object_mentions": ["OpenAI", "Anthropic"],
|
||
"query_profile": {
|
||
"needs_freshness": True,
|
||
"referenced_objects": ["public_model"],
|
||
},
|
||
},
|
||
)
|
||
|
||
assert plan.query_kind == "generic_web"
|
||
assert plan.answer_contract == "compare_with_sources"
|
||
assert plan.requires_multi_source is True
|
||
assert plan.result_limit == 3
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_plan_threads_fit_brief_guidance_into_prompt_and_plan() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.prompt = ""
|
||
|
||
async def generate_structured_object(self, **kwargs):
|
||
self.prompt = str(kwargs.get("prompt") or "")
|
||
return SimpleNamespace(
|
||
payload={
|
||
"topic": "Responses API 适配 PMFlow",
|
||
"search_queries": ["OpenAI Responses API project management workflow"],
|
||
}
|
||
)
|
||
|
||
fake_llm = _FakeLLM()
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=fake_llm,
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
fit_brief = {
|
||
"project_summary": "当前项目：PMFlow AI 管家，聚焦项目内问答与需求流转。",
|
||
"repo_summary": "当前仓库：FastAPI 后端 + React 前端。",
|
||
"current_capabilities": ["AI 管家", "智能配置"],
|
||
"constraints": ["只允许基于外部公开证据给出推荐。"],
|
||
"recommendation_focus": ["判断 Responses API 是否适合当前项目接入"],
|
||
}
|
||
plan, meta = await planner.build_plan(
|
||
question="这个项目适不适合接 OpenAI Responses API？",
|
||
query_semantics={
|
||
"query_profile": {"needs_freshness": True},
|
||
"object_mentions": ["OpenAI Responses API"],
|
||
},
|
||
fit_brief=fit_brief,
|
||
)
|
||
|
||
assert meta["planner"] == "llm_structured"
|
||
assert "PMFlow AI 管家" in fake_llm.prompt
|
||
assert "智能配置" in fake_llm.prompt
|
||
assert plan.fit_brief_summary
|
||
assert plan.project_scope_hint == "current_project_and_repo"
|
||
assert "适合当前项目接入" in str(plan.recommendation_goal)
|
||
assert "AI 管家" in plan.project_object_anchors
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
@pytest.mark.parametrize(
|
||
("retrieval_status", "expected_variant", "expected_phrase"),
|
||
[
|
||
("verifier_blocked", "external_verifier_blocked", "没有通过核验"),
|
||
("no_relevant_results", "external_no_relevant_results", "没有找到足够相关的外部公开证据"),
|
||
("retrieval_failed", "external_retrieval_failed", "外部检索这次执行失败"),
|
||
],
|
||
)
|
||
async def test_presenter_distinguishes_external_failure_states(
|
||
retrieval_status: str,
|
||
expected_variant: str,
|
||
expected_phrase: str,
|
||
) -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.calls = 0
|
||
|
||
async def generate(self, *args, **kwargs):
|
||
_ = (args, kwargs)
|
||
self.calls += 1
|
||
return "should not be used"
|
||
|
||
fake_llm = _FakeLLM()
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=fake_llm,
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="这个项目适不适合接某个外部能力？",
|
||
context_packet={
|
||
"question": "这个项目适不适合接某个外部能力？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "外部能力适配判断",
|
||
},
|
||
"adapter_attempts": [{"adapter": "live_search", "status": "empty"}],
|
||
"verified_evidence": [],
|
||
"verifier": {"ok": False, "reason_code": "external_evidence_off_topic"},
|
||
"uncertainties": ["external_evidence_off_topic"],
|
||
"retrieval_status": retrieval_status,
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == expected_variant
|
||
assert expected_phrase in result["answer_text"]
|
||
assert fake_llm.calls == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_distinguishes_indirect_only_without_overclaiming_direct_match() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.calls = 0
|
||
|
||
async def generate(self, *args, **kwargs):
|
||
_ = (args, kwargs)
|
||
self.calls += 1
|
||
return "should not be used"
|
||
|
||
fake_llm = _FakeLLM()
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=fake_llm,
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="这个项目要不要跟进 OpenAI Responses API？",
|
||
context_packet={
|
||
"question": "这个项目要不要跟进 OpenAI Responses API？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "Responses API 适配性",
|
||
},
|
||
"adapter_attempts": [{"adapter": "live_search", "status": "success"}],
|
||
"verified_evidence": [
|
||
{
|
||
"source_type": "url",
|
||
"source_label": "OpenAI Docs",
|
||
"title": "Responses API guide",
|
||
"target_url": "https://platform.openai.com/docs/guides/responses",
|
||
"snippet": "Responses API can orchestrate tool calls and structured outputs.",
|
||
}
|
||
],
|
||
"verifier": {"ok": True},
|
||
"uncertainties": [],
|
||
"retrieval_status": "evidence_found",
|
||
"impact_status": "indirect_only",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == "external_indirect_only"
|
||
assert "只能说明它和当前项目有背景相关性" in result["answer_text"]
|
||
assert "不能直接说明它已经命中当前项目对象" in result["answer_text"]
|
||
assert fake_llm.calls == 0
|