1611 lines
67 KiB
Python
1611 lines
67 KiB
Python
import asyncio
|
||
from types import SimpleNamespace
|
||
from datetime import datetime, timezone
|
||
|
||
import pytest
|
||
|
||
from app.services.external_query_service.gateway import ExternalQueryGateway
|
||
from app.services.external_query_service.research_runtime import ExternalResearchRuntime
|
||
from app.services.external_query_service.registry import ExternalProviderRegistry
|
||
from app.services.external_query_service.adapters import DashScopeResponsesSearchAdapter
|
||
from app.services.external_query_service.adapters.live_search import LiveSearchFallbackAdapter
|
||
from app.services.external_query_service.planner import ExternalQueryPlanner
|
||
from app.services.external_query_service.presenter import ExternalAnswerPresenter
|
||
from app.services.external_query_service.schemas import ExternalAdapterResult, ExternalEvidenceItem, ExternalQueryPlan
|
||
from app.services.mcp_config.runtime import DuckDuckGoSearchRuntime
|
||
|
||
|
||
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
|
||
|
||
|
||
def test_duckduckgo_backend_direct_news_search_uses_progressive_time_windows() -> None:
|
||
attempts = DuckDuckGoSearchRuntime._search_attempts(search_kind="news")
|
||
|
||
assert attempts[:3] == [("news", "m"), ("news", "w"), ("news", "d")]
|
||
assert ("text", None) in attempts
|
||
|
||
|
||
def test_external_gateway_blocks_off_topic_results_without_subject_entities() -> None:
|
||
plan = ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="国家医保局 近半年 政策动态",
|
||
search_queries=["国家医保局 2026 政策动态"],
|
||
freshness_required=True,
|
||
)
|
||
|
||
verifier = ExternalQueryGateway._verify_evidence_bundle(
|
||
plan=plan,
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
title="2026 NFL Draft is underway",
|
||
target_url="https://example.com/2026-nfl-draft",
|
||
snippet="The 2026 NFL Draft started in Pittsburgh.",
|
||
)
|
||
],
|
||
)
|
||
|
||
assert verifier["ok"] is False
|
||
assert verifier["reason_code"] == "external_evidence_off_topic"
|
||
|
||
|
||
def test_external_gateway_accepts_query_relevant_results_without_subject_entities() -> None:
|
||
plan = ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="国家医保局 近半年 政策动态",
|
||
search_queries=["国家医保局 2026 政策动态"],
|
||
freshness_required=True,
|
||
)
|
||
|
||
verifier = ExternalQueryGateway._verify_evidence_bundle(
|
||
plan=plan,
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
title="国家医保局发布 2026 年医保政策动态",
|
||
target_url="https://example.gov.cn/2026-medical-security-policy",
|
||
snippet="2026 年国家医保局持续推进医保支付方式改革和医保目录管理。",
|
||
)
|
||
],
|
||
)
|
||
|
||
assert verifier["ok"] is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_live_search_multi_source_executes_and_diversifies_planner_queries() -> None:
|
||
called_queries: list[str] = []
|
||
|
||
async def search_results_callable(*, query: str, max_results: int) -> list[dict[str, str]]:
|
||
called_queries.append(query)
|
||
prefix = query.split()[0].lower()
|
||
return [
|
||
{"title": f"{prefix} exact result", "url": f"https://example.com/{prefix}-1"},
|
||
{"title": f"{prefix} secondary result", "url": f"https://example.com/{prefix}-2"},
|
||
][:max_results]
|
||
|
||
async def fetch_text_callable(url: str) -> str:
|
||
return f"{url} official product update with adoption signal and user feedback."
|
||
|
||
def chunk_text_callable(text: str, **kwargs) -> list[str]:
|
||
_ = kwargs
|
||
return [text]
|
||
|
||
def score_search_result_callable(**kwargs) -> float:
|
||
title = str(kwargs.get("title") or "")
|
||
return 0.9 if "exact" in title else 0.4
|
||
|
||
def score_chunk_callable(**kwargs) -> float:
|
||
content = str(kwargs.get("content") or "")
|
||
return 0.95 if "adoption signal" in content else 0.1
|
||
|
||
plan = ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="AI meeting assistant market",
|
||
search_queries=[
|
||
"vendor updates",
|
||
"analyst summaries",
|
||
"app store reviews",
|
||
],
|
||
result_limit=3,
|
||
requires_multi_source=True,
|
||
)
|
||
|
||
result = await LiveSearchFallbackAdapter(
|
||
search_results_callable=search_results_callable,
|
||
fetch_text_callable=fetch_text_callable,
|
||
chunk_text_callable=chunk_text_callable,
|
||
score_search_result_callable=score_search_result_callable,
|
||
score_chunk_callable=score_chunk_callable,
|
||
max_pages=3,
|
||
).execute(plan)
|
||
|
||
assert called_queries == ["vendor updates", "analyst summaries", "app store reviews"]
|
||
assert [item.raw_metadata["query"] for item in result.evidence] == [
|
||
"vendor updates",
|
||
"analyst summaries",
|
||
"app store reviews",
|
||
]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_external_gateway_times_out_one_adapter_and_continues_to_next() -> None:
|
||
plan = ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="国家医保局 近半年 政策动态",
|
||
search_queries=["国家医保局 近半年 政策动态"],
|
||
freshness_required=True,
|
||
result_limit=3,
|
||
requires_multi_source=True,
|
||
)
|
||
|
||
class _Planner:
|
||
async def build_plan(self, *, question: str, query_semantics: dict | None = None):
|
||
_ = (question, query_semantics)
|
||
return plan
|
||
|
||
class _SlowAdapter:
|
||
adapter_name = "slow_external_provider"
|
||
supported_query_kinds = {"generic_web"}
|
||
|
||
async def execute(self, current_plan: ExternalQueryPlan) -> ExternalAdapterResult:
|
||
_ = current_plan
|
||
await asyncio.sleep(1.1)
|
||
return ExternalAdapterResult(adapter_name=self.adapter_name, status="empty", evidence=[])
|
||
|
||
class _FallbackAdapter:
|
||
adapter_name = "bounded_live_search"
|
||
supported_query_kinds = {"generic_web"}
|
||
|
||
async def execute(self, current_plan: ExternalQueryPlan) -> ExternalAdapterResult:
|
||
_ = current_plan
|
||
return ExternalAdapterResult(
|
||
adapter_name=self.adapter_name,
|
||
status="success",
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
source_label="国家医保局",
|
||
title="2026 年国家医保局政策动态",
|
||
target_url="https://example.gov.cn/2026-medical-security-policy",
|
||
snippet="2026 年国家医保局发布近半年政策动态。",
|
||
retrieved_at="2026-05-17T00:00:00+00:00",
|
||
freshness_score=0.9,
|
||
authority_score=0.9,
|
||
grounded_fields=["snippet"],
|
||
)
|
||
],
|
||
)
|
||
|
||
execution = await ExternalQueryGateway(
|
||
planner=_Planner(),
|
||
registry=ExternalProviderRegistry(adapters=[_SlowAdapter(), _FallbackAdapter()]),
|
||
adapter_timeout_seconds=1.0,
|
||
).execute(question="国家医保局 近半年 政策动态")
|
||
|
||
assert execution.adapter_name == "bounded_live_search"
|
||
assert execution.meta["adapter_attempts"][0]["status"] == "timeout"
|
||
assert execution.meta["fallback_used"] is True
|
||
assert execution.verifier["ok"] is True
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_external_research_runtime_fans_out_and_returns_before_slow_provider() -> None:
|
||
plan = ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="AI briefing assistant public evidence",
|
||
search_queries=["AI briefing assistant public evidence"],
|
||
result_limit=2,
|
||
requires_multi_source=False,
|
||
)
|
||
|
||
class _Planner:
|
||
async def build_plan(self, *, question: str, query_semantics: dict | None = None, fit_brief: dict | None = None):
|
||
_ = (question, query_semantics, fit_brief)
|
||
return plan, {"planner": "test"}
|
||
|
||
class _SlowAdapter:
|
||
adapter_name = "slow_dashscope_like"
|
||
supported_query_kinds = {"generic_web"}
|
||
|
||
async def execute(self, current_plan: ExternalQueryPlan) -> ExternalAdapterResult:
|
||
_ = current_plan
|
||
await asyncio.sleep(0.5)
|
||
return ExternalAdapterResult(adapter_name=self.adapter_name, status="empty", evidence=[])
|
||
|
||
class _FastAdapter:
|
||
adapter_name = "fast_search"
|
||
supported_query_kinds = {"generic_web"}
|
||
|
||
async def execute(self, current_plan: ExternalQueryPlan) -> ExternalAdapterResult:
|
||
_ = current_plan
|
||
return ExternalAdapterResult(
|
||
adapter_name=self.adapter_name,
|
||
status="success",
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
source_label="Example",
|
||
title="AI briefing assistant public evidence",
|
||
target_url="https://example.com/ai-briefing-assistant",
|
||
snippet="AI briefing assistant public evidence and product workflow notes.",
|
||
freshness_score=0.8,
|
||
authority_score=0.7,
|
||
)
|
||
],
|
||
)
|
||
|
||
started = asyncio.get_running_loop().time()
|
||
execution = await ExternalResearchRuntime(
|
||
planner=_Planner(),
|
||
registry=ExternalProviderRegistry(adapters=[_SlowAdapter(), _FastAdapter()]),
|
||
adapter_timeout_seconds=1.0,
|
||
min_verified_sources=1,
|
||
).execute(question="AI briefing assistant public evidence")
|
||
elapsed = asyncio.get_running_loop().time() - started
|
||
|
||
assert elapsed < 0.45
|
||
assert execution.adapter_name == "fast_search"
|
||
assert execution.verifier["ok"] is True
|
||
assert execution.meta["research_runtime"] == "parallel_external_research.v1"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dashscope_responses_adapter_maps_web_search_results_to_external_evidence(monkeypatch) -> None:
|
||
calls: list[dict[str, object]] = []
|
||
|
||
class FakeResponse:
|
||
status_code = 200
|
||
text = "{}"
|
||
|
||
def json(self):
|
||
return {
|
||
"id": "resp-test",
|
||
"model": "qwen3.6-plus",
|
||
"output_text": "2026 年 AI Agent 市场仍在增长。",
|
||
"search_info": {
|
||
"search_results": [
|
||
{
|
||
"title": "2026 AI Agent market report",
|
||
"url": "https://example.com/2026-agent-market",
|
||
"snippet": "2026 report says agent adoption increased.",
|
||
"date": "2026-05-01",
|
||
}
|
||
]
|
||
},
|
||
}
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
class FakeClient:
|
||
def __init__(self, *args, **kwargs):
|
||
_ = (args, kwargs)
|
||
|
||
async def __aenter__(self):
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
return False
|
||
|
||
async def post(self, *args, **kwargs):
|
||
_ = args
|
||
calls.append(dict(kwargs.get("json") or {}))
|
||
return FakeResponse()
|
||
|
||
async def fake_log(*args, **kwargs):
|
||
_ = (args, kwargs)
|
||
|
||
adapter = DashScopeResponsesSearchAdapter(
|
||
api_key="sk-test",
|
||
model="qwen3.6-plus",
|
||
client_factory=lambda: FakeClient(),
|
||
)
|
||
monkeypatch.setattr(adapter, "_log_ai_call", fake_log)
|
||
plan = ExternalQueryPlan(
|
||
query_kind="market",
|
||
topic="AI Agent market 2026",
|
||
search_queries=["AI Agent market 2026 report"],
|
||
freshness_required=True,
|
||
result_limit=3,
|
||
)
|
||
|
||
result = await adapter.execute(plan)
|
||
|
||
assert result.status == "success"
|
||
assert calls[0]["model"] == "qwen3.6-plus"
|
||
assert calls[0]["tools"] == [{"type": "web_search"}]
|
||
assert calls[0]["search_options"] == {"forced_search": True, "freshness": 30}
|
||
assert calls[0]["enable_thinking"] is True
|
||
assert result.evidence[0].target_url == "https://example.com/2026-agent-market"
|
||
assert result.evidence[0].raw_metadata["provider"] == "dashscope_responses"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dashscope_responses_adapter_maps_web_search_call_sources(monkeypatch) -> None:
|
||
class FakeResponse:
|
||
status_code = 200
|
||
text = "{}"
|
||
|
||
def json(self):
|
||
return {
|
||
"id": "resp-test",
|
||
"model": "qwen3.6-plus",
|
||
"output": [
|
||
{
|
||
"type": "web_search_call",
|
||
"action": {
|
||
"query": "Web search",
|
||
"sources": [
|
||
{"type": "url", "url": "https://www.elysee.fr/en/"},
|
||
],
|
||
},
|
||
},
|
||
{
|
||
"type": "message",
|
||
"content": [
|
||
{
|
||
"type": "output_text",
|
||
"text": "法国现任总统是 Emmanuel Macron。",
|
||
}
|
||
],
|
||
},
|
||
],
|
||
"usage": {"x_tools": {"web_search": {"count": 1}}},
|
||
}
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self):
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
return False
|
||
|
||
async def post(self, *args, **kwargs):
|
||
_ = (args, kwargs)
|
||
return FakeResponse()
|
||
|
||
async def fake_log(*args, **kwargs):
|
||
_ = (args, kwargs)
|
||
|
||
adapter = DashScopeResponsesSearchAdapter(
|
||
api_key="sk-test",
|
||
model="qwen3.6-plus",
|
||
client_factory=lambda: FakeClient(),
|
||
)
|
||
monkeypatch.setattr(adapter, "_log_ai_call", fake_log)
|
||
|
||
result = await adapter.execute(
|
||
ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="法国现任总统是谁？",
|
||
freshness_required=True,
|
||
result_limit=2,
|
||
)
|
||
)
|
||
|
||
assert result.status == "success"
|
||
assert result.meta["tool_usage"] == {"web_search": {"count": 1}}
|
||
assert result.evidence[0].target_url == "https://www.elysee.fr/en/"
|
||
assert "Emmanuel Macron" in result.evidence[0].snippet
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_dashscope_responses_adapter_does_not_copy_synthesized_answer_to_multiple_sources(monkeypatch) -> None:
|
||
class FakeResponse:
|
||
status_code = 200
|
||
text = "{}"
|
||
|
||
def json(self):
|
||
return {
|
||
"id": "resp-test",
|
||
"model": "qwen3.6-plus",
|
||
"output": [
|
||
{
|
||
"type": "web_search_call",
|
||
"action": {
|
||
"query": "Web search",
|
||
"sources": [
|
||
{"type": "url", "url": "https://example.com/a"},
|
||
{"type": "url", "url": "https://example.com/b"},
|
||
],
|
||
},
|
||
},
|
||
{
|
||
"type": "message",
|
||
"content": [
|
||
{
|
||
"type": "output_text",
|
||
"text": "这是模型综合答案，不是任一单独 URL 的网页摘要。",
|
||
}
|
||
],
|
||
},
|
||
],
|
||
"usage": {"x_tools": {"web_search": {"count": 1}}},
|
||
}
|
||
|
||
def raise_for_status(self):
|
||
return None
|
||
|
||
class FakeClient:
|
||
async def __aenter__(self):
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type, exc, tb):
|
||
return False
|
||
|
||
async def post(self, *args, **kwargs):
|
||
_ = (args, kwargs)
|
||
return FakeResponse()
|
||
|
||
adapter = DashScopeResponsesSearchAdapter(
|
||
api_key="sk-test",
|
||
model="qwen3.6-plus",
|
||
client_factory=lambda: FakeClient(),
|
||
)
|
||
async def fake_log(*args, **kwargs):
|
||
_ = (args, kwargs)
|
||
|
||
monkeypatch.setattr(adapter, "_log_ai_call", fake_log)
|
||
|
||
result = await adapter.execute(
|
||
ExternalQueryPlan(
|
||
query_kind="generic_web",
|
||
topic="行业趋势",
|
||
freshness_required=True,
|
||
result_limit=2,
|
||
)
|
||
)
|
||
|
||
assert result.status == "success"
|
||
assert len(result.evidence) == 2
|
||
assert all("模型综合答案" not in item.snippet for item in result.evidence)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_plan_prepends_freshness_queries_for_market_research() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
return SimpleNamespace(
|
||
payload={
|
||
"query_kind": "market",
|
||
"freshness_required": True,
|
||
"topic": "短剧行业趋势",
|
||
"subject_entities": ["短剧"],
|
||
"search_queries": ["中国短剧行业2024年市场规模与增长趋势"],
|
||
}
|
||
)
|
||
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
current_year = datetime.now(timezone.utc).year
|
||
previous_year = current_year - 1
|
||
|
||
plan, meta = await planner.build_plan(
|
||
question="你帮我分析下短剧行业的行业趋势",
|
||
query_semantics={
|
||
"query_profile": {"needs_freshness": True},
|
||
"object_mentions": ["短剧"],
|
||
},
|
||
)
|
||
|
||
assert meta["planner"] == "llm_structured"
|
||
assert plan.freshness_required is True
|
||
assert str(current_year) in plan.search_queries[0]
|
||
assert str(previous_year) in plan.search_queries[0]
|
||
assert "中国短剧行业2024年市场规模与增长趋势" in plan.search_queries
|
||
|
||
|
||
def test_external_query_gateway_blocks_stale_market_evidence_when_freshness_required() -> None:
|
||
plan = ExternalQueryPlan(
|
||
query_kind="market",
|
||
answer_contract="summary_with_sources",
|
||
freshness_required=True,
|
||
topic="短剧行业趋势",
|
||
subject_entities=["短剧"],
|
||
search_queries=["短剧行业趋势 2026 2025 最新 数据 报告"],
|
||
)
|
||
|
||
verifier = ExternalQueryGateway._verify_evidence_bundle(
|
||
plan=plan,
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
source_label="2024年微短剧行业白皮书",
|
||
title="2024年微短剧行业白皮书",
|
||
target_url="https://example.com/short-drama-2024",
|
||
snippet="2024年微短剧行业白皮书显示，短剧用户规模增长。",
|
||
retrieved_at=f"{datetime.now(timezone.utc).year}-05-09T00:00:00+00:00",
|
||
)
|
||
],
|
||
)
|
||
|
||
assert verifier["ok"] is False
|
||
assert verifier["reason_code"] == "external_evidence_stale"
|
||
assert verifier["missing_fields"] == ["fresh_external_evidence"]
|
||
|
||
|
||
def test_external_query_gateway_accepts_recent_market_evidence_when_freshness_required() -> None:
|
||
current_year = datetime.now(timezone.utc).year
|
||
plan = ExternalQueryPlan(
|
||
query_kind="market",
|
||
answer_contract="summary_with_sources",
|
||
freshness_required=True,
|
||
topic="短剧行业趋势",
|
||
subject_entities=["短剧"],
|
||
search_queries=[f"短剧行业趋势 {current_year} 最新 数据 报告"],
|
||
)
|
||
|
||
verifier = ExternalQueryGateway._verify_evidence_bundle(
|
||
plan=plan,
|
||
evidence=[
|
||
ExternalEvidenceItem(
|
||
source_type="url",
|
||
source_label=f"{current_year}年微短剧行业数据报告",
|
||
title=f"{current_year}年微短剧行业数据报告",
|
||
target_url=f"https://example.com/short-drama-{current_year}",
|
||
snippet=f"{current_year}年短剧行业数据报告显示，行业进入精细化运营阶段。",
|
||
)
|
||
],
|
||
)
|
||
|
||
assert verifier["ok"] is True
|
||
assert verifier["fresh_years"] == [current_year]
|
||
|
||
|
||
@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
|
||
|
||
|
||
def test_fallback_plan_builds_concrete_signal_source_recommendation_queries() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
query_semantics={
|
||
"query_profile": {"needs_freshness": True},
|
||
},
|
||
fit_brief={
|
||
"project_summary": "当前项目：#81 测试产能警告 (status=active)",
|
||
"recommendation_focus": ["给测试产能警告推荐最值得先看的外部信号源。"],
|
||
"signal_source_objectives": [
|
||
"设计产能监控逻辑与数据采集：编写脚本收集测试队列与资源使用数据",
|
||
"实现产能警告触发服务：实时比对数据与阈值",
|
||
],
|
||
},
|
||
)
|
||
|
||
joined_queries = " ".join(plan.search_queries[:2])
|
||
assert plan.search_queries[0].startswith("测试产能警告 设计产能监控逻辑")
|
||
assert "测试队列与资源使用数据" in joined_queries
|
||
assert "实时比对数据与阈值" in joined_queries
|
||
assert "外部信号源" not in joined_queries
|
||
assert "发布变更" not in joined_queries
|
||
assert "工单" not in joined_queries
|
||
|
||
|
||
def test_fallback_plan_preserves_structured_interaction_level() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="这个项目我想先补外部东西，你先帮我列几个最值得看的。",
|
||
query_semantics={
|
||
"query_profile": {
|
||
"needs_freshness": True,
|
||
"interaction_level": "recommendation",
|
||
},
|
||
"interaction_level": "recommendation",
|
||
},
|
||
fit_brief={
|
||
"project_summary": "当前项目：自动开票系统，目标是在收到开票请求后自动开票并发送发票图片。",
|
||
"recommendation_focus": ["给自动开票系统推荐最值得先看的外部信号源。"],
|
||
"signal_source_objectives": [
|
||
"处理开票请求：收到接口请求后自动开票",
|
||
"发送发票图片：把结果发给客户",
|
||
],
|
||
},
|
||
)
|
||
|
||
assert plan.interaction_level == "recommendation"
|
||
|
||
|
||
def test_fallback_plan_extracts_explicit_project_binding_from_fit_brief() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="把我们的项目产品和宝宝树对比下",
|
||
query_semantics={
|
||
"query_type": "compare",
|
||
"object_mentions": ["童析—学龄前儿童行为特质AI评估助手", "宝宝树"],
|
||
},
|
||
fit_brief={
|
||
"project_summary": "当前项目：#141 童析—学龄前儿童行为特质AI评估助手 (status=active)",
|
||
},
|
||
)
|
||
|
||
assert plan.project_binding == 141
|
||
assert plan.project_title == "童析—学龄前儿童行为特质AI评估助手"
|
||
|
||
|
||
def test_fallback_plan_builds_store_inventory_recommendation_queries_from_project_objectives() -> None:
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=object(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan = planner._build_fallback_plan(
|
||
question="门店库存异常日报助手如果想更早发现异常，先看哪几类外部信号最值？",
|
||
query_semantics={
|
||
"query_profile": {
|
||
"needs_freshness": True,
|
||
"interaction_level": "recommendation",
|
||
},
|
||
"interaction_level": "recommendation",
|
||
},
|
||
fit_brief={
|
||
"project_summary": "当前项目：门店库存异常日报助手，目标是每日自动汇总门店库存异常并发给区域经理。",
|
||
"recommendation_focus": ["给门店库存异常日报助手推荐最值得先看的外部信号源。"],
|
||
"signal_source_objectives": [
|
||
"汇总门店库存异常：识别缺货、库存跳变和异常门店分布",
|
||
"推送区域经理日报：把高风险异常按区域维度发给负责人",
|
||
],
|
||
},
|
||
)
|
||
|
||
joined_queries = " ".join(plan.search_queries[:3])
|
||
assert plan.interaction_level == "recommendation"
|
||
assert plan.search_queries[0].startswith("门店库存异常日报助手 汇总门店库存异常")
|
||
assert "缺货" in joined_queries
|
||
assert "库存跳变" in joined_queries
|
||
assert "区域经理" in joined_queries
|
||
assert "外部信号源" not in joined_queries
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_build_plan_signal_source_recommendation_keeps_concrete_seed_queries_ahead_of_generic_model_query() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
return SimpleNamespace(
|
||
payload={
|
||
"topic": "测试产能警告外部信号源推荐",
|
||
"search_queries": ["production capacity warning system external data sources industry best practices"],
|
||
}
|
||
)
|
||
|
||
planner = ExternalQueryPlanner(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
plan, meta = await planner.build_plan(
|
||
question="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
query_semantics={
|
||
"query_profile": {"needs_freshness": True},
|
||
},
|
||
fit_brief={
|
||
"project_summary": "当前项目：#81 测试产能警告 (status=active)",
|
||
"recommendation_focus": ["给测试产能警告推荐最值得先看的外部信号源。"],
|
||
"signal_source_objectives": [
|
||
"设计产能监控逻辑与数据采集：编写脚本收集测试队列与资源使用数据",
|
||
"实现产能警告触发服务：实时比对数据与阈值",
|
||
],
|
||
},
|
||
)
|
||
|
||
assert meta["planner"] == "llm_structured"
|
||
joined_queries = " ".join(plan.search_queries[:3])
|
||
assert plan.search_queries[0].startswith("测试产能警告 设计产能监控逻辑")
|
||
assert "测试队列与资源使用数据" in joined_queries
|
||
assert "实时比对数据与阈值" in joined_queries
|
||
assert "production capacity warning system external data sources industry best practices" in plan.search_queries
|
||
|
||
|
||
@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 适配性",
|
||
"fit_brief_summary": "当前项目：#141 PMFlow AI 管家 (status=active)",
|
||
"subject_entities": ["PMFlow AI 管家", "OpenAI Responses API", "Anthropic MCP"],
|
||
},
|
||
"adapter_attempts": [{"adapter": "live_search", "status": "success"}],
|
||
"verified_evidence": [
|
||
{
|
||
"source_type": "url",
|
||
"source_label": "URL[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 "当前可以继续围绕这些候选对象往下查" in result["answer_text"]
|
||
assert "URL[" not in result["answer_text"]
|
||
assert "OpenAI Docs" in result["answer_text"]
|
||
assert result["research_workspace"]["primary_object_ref"] == "project:141"
|
||
assert result["research_workspace"]["candidate_set_binding"].startswith("cset_")
|
||
assert len(result["research_workspace"]["candidate_items"]) >= 1
|
||
assert fake_llm.calls == 0
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_compare_with_sources_keeps_candidate_level_compare_when_indirect_only() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.calls = 0
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = kwargs
|
||
self.calls += 1
|
||
assert "当前仍是候选级比较" in prompt
|
||
assert "当前项目背景" in prompt
|
||
return "我先按候选级比较展开，明确哪些是高相关，哪些还需要进一步映射验证。"
|
||
|
||
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": "compare_with_sources",
|
||
"topic": "竞品详细对比",
|
||
"fit_brief_summary": "当前项目：#141 童析-学龄前儿童行为特质AI评估助手 (status=active)",
|
||
"project_binding": 141,
|
||
"project_title": "童析—学龄前儿童行为特质AI评估助手",
|
||
"subject_entities": ["童析—学龄前儿童行为特质AI评估助手", "Goally", "ABCmouse", "Cognoa"],
|
||
},
|
||
"adapter_attempts": [{"adapter": "live_search", "status": "success"}],
|
||
"verified_evidence": [
|
||
{
|
||
"source_type": "url",
|
||
"source_label": "Goally",
|
||
"title": "The Best Tablet for Kids | Goally",
|
||
"target_url": "https://example.com/goally",
|
||
"snippet": "Goally provides tools for kids routines and behavior support.",
|
||
},
|
||
{
|
||
"source_type": "url",
|
||
"source_label": "ABCmouse",
|
||
"title": "About the Assessment Center in ABCmouse",
|
||
"target_url": "https://example.com/abcmouse",
|
||
"snippet": "ABCmouse includes assessment-oriented learning flows.",
|
||
},
|
||
],
|
||
"verifier": {"ok": True},
|
||
"uncertainties": [],
|
||
"retrieval_status": "evidence_found",
|
||
"impact_status": "indirect_only",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == "external_compare_with_sources"
|
||
assert "候选级比较" in result["answer_text"]
|
||
assert result["research_workspace"]["primary_object_ref"] == "project:141"
|
||
assert result["research_workspace"]["candidate_set_binding"].startswith("cset_")
|
||
assert len(result["research_workspace"]["candidate_items"]) >= 2
|
||
candidate_names = [item["display_name"] for item in result["research_workspace"]["candidate_items"]]
|
||
assert "童析—学龄前儿童行为特质AI评估助手" not in candidate_names
|
||
assert fake_llm.calls == 1
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_prefers_recommendation_answer_for_signal_source_recommendation_question() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.calls = 0
|
||
self.prompt = ""
|
||
self.structured_calls = 0
|
||
|
||
async def generate_structured_object(self, **kwargs):
|
||
self.structured_calls += 1
|
||
self.prompt = str(kwargs.get("prompt") or "")
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "测试执行失败和阻塞记录",
|
||
"reason": "它最早反映测试队列是不是已经开始堆积，适合先盯住产能吃紧的信号。",
|
||
},
|
||
{
|
||
"name": "构建或发布变更记录",
|
||
"reason": "它能帮你判断产能波动是不是由最近的版本改动带出来的。",
|
||
},
|
||
{
|
||
"name": "工单或异常反馈",
|
||
"reason": "它能看出问题有没有开始影响到实际交付，不会只停在内部指标波动。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = kwargs
|
||
self.calls += 1
|
||
self.prompt = prompt
|
||
return (
|
||
"如果只按当前项目任务来排优先级，我会先补齐“设计产能监控逻辑与数据采集”需要的外部输入，"
|
||
"再看“实现产能警告触发服务”依赖的实时变化信号。"
|
||
)
|
||
|
||
fake_llm = _FakeLLM()
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=fake_llm,
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
context_packet={
|
||
"question": "测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "测试产能警告外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：测试产能警告，聚焦识别产能异常和告警触发。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给测试产能警告推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["实现产能警告触发服务", "告警接收链路"],
|
||
"signal_source_objectives": [
|
||
"设计产能监控逻辑与数据采集：编写脚本收集测试队列与资源使用数据",
|
||
"实现产能警告触发服务：实时比对数据与阈值",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == "external_signal_source_recommendation"
|
||
assert "1. 测试执行失败和阻塞记录" in result["answer_text"]
|
||
assert "2. 构建或发布变更记录" in result["answer_text"]
|
||
assert "3. 工单或异常反馈" in result["answer_text"]
|
||
assert "最早反映测试队列是不是已经开始堆积" in result["answer_text"]
|
||
assert "这些先算建议，不代表已经接入或验证完成。" in result["answer_text"]
|
||
assert "主链" not in result["answer_text"]
|
||
assert "这一层" not in result["answer_text"]
|
||
assert result["recommendation_packet"]["resource_role"] == "signal_source"
|
||
assert result["recommendation_packet"]["source_kind"] == "external"
|
||
assert result["recommendation_packet"]["evidence_mode"] == "advisory"
|
||
assert result["recommendation_packet"]["candidate_set_id"]
|
||
assert result["recommendation_packet"]["candidates"][0]["candidate_ref"]
|
||
assert result["recommendation_packet"]["candidates"][0]["display_name"] == "测试执行失败和阻塞记录"
|
||
assert (
|
||
result["recommendation_packet"]["candidates"][0]["materialization_payload"]["display_name"]
|
||
== "测试执行失败和阻塞记录"
|
||
)
|
||
assert fake_llm.structured_calls == 1
|
||
assert "给测试产能警告推荐最值得先看的外部信号源" in fake_llm.prompt
|
||
assert "测试队列与资源使用数据" in fake_llm.prompt
|
||
assert "实时比对数据与阈值" in fake_llm.prompt
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_varies_by_project_context() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.structured_calls = 0
|
||
|
||
async def generate_structured_object(self, **kwargs):
|
||
self.structured_calls += 1
|
||
prompt = str(kwargs.get("prompt") or "")
|
||
if "自动开票系统" in prompt:
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "开票失败或重试回执",
|
||
"reason": "它最早告诉你票是不是卡在开票动作本身。",
|
||
},
|
||
{
|
||
"name": "税控或发票平台状态变化",
|
||
"reason": "它能解释失败到底是业务问题还是外部平台波动。",
|
||
},
|
||
{
|
||
"name": "发票发送失败反馈",
|
||
"reason": "它能看出问题有没有真正影响到客户收票。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "用户投诉和差评变化",
|
||
"reason": "它最容易最先暴露客服回复质量是不是已经开始出问题。",
|
||
},
|
||
{
|
||
"name": "社媒互动异常波动",
|
||
"reason": "它能帮你看出问题是不是已经从单点反馈扩到公开场域。",
|
||
},
|
||
{
|
||
"name": "竞品或行业FAQ变化",
|
||
"reason": "它能提示当前话术和知识点是不是已经落后。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return "如果只看外部信号源，我会先看最关键的外部输入。"
|
||
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
invoice_result = await presenter.present(
|
||
question="自动开票系统这个项目，除了系统里现有的数据，你觉得还应该盯哪些外部信号？直接说几个最有用的，不要给我出方案。",
|
||
context_packet={
|
||
"question": "自动开票系统这个项目，除了系统里现有的数据，你觉得还应该盯哪些外部信号？直接说几个最有用的，不要给我出方案。",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "自动开票系统外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：自动开票系统，目标是在收到开票请求后自动开票并发送发票图片。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给自动开票系统推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["开票请求处理", "发票发送结果"],
|
||
"signal_source_objectives": [
|
||
"处理开票请求：收到接口请求后自动开票",
|
||
"发送发票图片：把结果发给客户",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
ai_customer_result = await presenter.present(
|
||
question="AI客服这个项目，除了现在这些数据，你觉得最值得再补哪几类外部线索？直接推荐几个。",
|
||
context_packet={
|
||
"question": "AI客服这个项目，除了现在这些数据，你觉得最值得再补哪几类外部线索？直接推荐几个。",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "AI客服外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：AI客服，核心痛点是内容本地化，当前能力围绕社媒运营和用户洞察。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给 AI 客服推荐最值得先看的外部线索。",
|
||
"project_object_anchors": ["用户洞察数据源", "社媒运营数据源"],
|
||
"signal_source_objectives": [
|
||
"用户洞察：更快发现用户反馈与投诉变化",
|
||
"社媒运营：跟踪公开互动和话题波动",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert "开票失败或重试回执" in invoice_result["answer_text"]
|
||
assert "发票发送失败反馈" in invoice_result["answer_text"]
|
||
assert "用户投诉和差评变化" not in invoice_result["answer_text"]
|
||
|
||
assert "用户投诉和差评变化" in ai_customer_result["answer_text"]
|
||
assert "社媒互动异常波动" in ai_customer_result["answer_text"]
|
||
assert "开票失败或重试回执" not in ai_customer_result["answer_text"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_varies_for_cashflow_and_knowledge_base_projects() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, *, prompt: str, **kwargs):
|
||
_ = kwargs
|
||
if "智能现金流管家" in prompt:
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "银行回单或到账异常反馈",
|
||
"reason": "它最早告诉你资金是不是已经在到账环节卡住了。",
|
||
},
|
||
{
|
||
"name": "税费或社保扣缴提醒变化",
|
||
"reason": "它能帮助区分现金流压力是不是已经开始影响刚性支出。",
|
||
},
|
||
{
|
||
"name": "上游平台结算延迟通知",
|
||
"reason": "它能提前暴露回款节奏是不是被外部平台拖慢了。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "外部知识源更新通知",
|
||
"reason": "它能让知识库问答先知道哪些权威资料已经变化。",
|
||
},
|
||
{
|
||
"name": "第三方模型服务可用性波动",
|
||
"reason": "它最容易直接影响问答响应质量和稳定性。",
|
||
},
|
||
{
|
||
"name": "企业 SSO 或目录服务异常公告",
|
||
"reason": "它能帮助判断问答不可用是不是卡在登录或权限链路。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return "关于外部信号，我现在还不能直接推荐。"
|
||
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
cashflow_result = await presenter.present(
|
||
question="智能现金流管家除了银行流水，还值得补哪些外部线索？",
|
||
context_packet={
|
||
"question": "智能现金流管家除了银行流水，还值得补哪些外部线索？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "智能现金流管家外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：智能现金流管家，目标是更早识别回款异常和资金压力。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给智能现金流管家推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["回款节奏监控", "资金压力预警"],
|
||
"signal_source_objectives": [
|
||
"识别到账异常：更早发现回款延迟和结算波动",
|
||
"识别刚性支出压力：尽快发现税费或社保扣缴变化",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
knowledge_base_result = await presenter.present(
|
||
question="企业知识库问答AI项目如果只补一个外部信号，你觉得先补哪个最值？",
|
||
context_packet={
|
||
"question": "企业知识库问答AI项目如果只补一个外部信号，你觉得先补哪个最值？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "企业知识库问答AI项目外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：企业知识库问答AI项目，目标是提升企业知识问答的准确性和可用性。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给企业知识库问答AI项目推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["知识更新同步", "问答服务稳定性"],
|
||
"signal_source_objectives": [
|
||
"识别知识更新：及时发现外部知识源变化",
|
||
"识别服务波动：尽快定位问答质量和可用性异常",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert "银行回单或到账异常反馈" in cashflow_result["answer_text"]
|
||
assert "税费或社保扣缴提醒变化" in cashflow_result["answer_text"]
|
||
assert "外部知识源更新通知" not in cashflow_result["answer_text"]
|
||
|
||
assert "外部知识源更新通知" in knowledge_base_result["answer_text"]
|
||
assert "第三方模型服务可用性波动" in knowledge_base_result["answer_text"]
|
||
assert "银行回单或到账异常反馈" not in knowledge_base_result["answer_text"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_varies_for_medical_industry_research_project() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "医药政策与监管口径更新",
|
||
"reason": "它最容易先改变研究假设和布局判断。",
|
||
},
|
||
{
|
||
"name": "重点药企投融资和并购动态",
|
||
"reason": "它能更快暴露产业链方向是不是正在变化。",
|
||
},
|
||
{
|
||
"name": "临床试验与注册进展公告",
|
||
"reason": "它能帮助判断哪些赛道是真的在加速，哪些还停在预期里。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return "我现在还不能直接推荐。"
|
||
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="医药行业十五五规划布局研究现在最值得优先盯的外部变化有哪些？",
|
||
context_packet={
|
||
"question": "医药行业十五五规划布局研究现在最值得优先盯的外部变化有哪些？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "医药行业十五五规划布局研究外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：医药行业十五五规划布局研究，目标是研判政策、产业和重点赛道变化。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给医药行业十五五规划布局研究推荐最值得优先看的外部变化。",
|
||
"project_object_anchors": ["政策口径变化", "重点赛道布局判断"],
|
||
"signal_source_objectives": [
|
||
"识别政策变化：尽快发现监管和支持方向的变化",
|
||
"识别产业变化：尽快发现资本和临床进展带来的赛道变化",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert "1. 医药政策与监管口径更新" in result["answer_text"]
|
||
assert "2. 重点药企投融资和并购动态" in result["answer_text"]
|
||
assert "3. 临床试验与注册进展公告" in result["answer_text"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_accepts_natural_business_phrasing() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "门店缺货和补货异常反馈",
|
||
"reason": "它最早反映异常是不是已经在门店端出现。",
|
||
},
|
||
{
|
||
"name": "供应商到货延迟变化",
|
||
"reason": "它能帮助区分库存异常到底是销售波动还是供给变慢。",
|
||
},
|
||
{
|
||
"name": "竞品价格或促销波动",
|
||
"reason": "它能解释销量突然变化是不是被外部活动带出来的。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return "关于外部信号，我现在还不能直接推荐。"
|
||
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="门店库存异常日报助手这个项目，如果想更早发现异常，你会先多看哪几类外部信号？",
|
||
context_packet={
|
||
"question": "门店库存异常日报助手这个项目，如果想更早发现异常，你会先多看哪几类外部信号？",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "门店库存异常日报助手外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：门店库存异常日报助手，目标是更早发现缺货、滞销和补货异常。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给门店库存异常日报助手推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["门店异常日报", "缺货异常识别"],
|
||
"signal_source_objectives": [
|
||
"识别门店异常：更早发现缺货和补货异常",
|
||
"日报推送：让运营尽快看到高风险门店",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert "1. 门店缺货和补货异常反馈" in result["answer_text"]
|
||
assert "2. 供应商到货延迟变化" in result["answer_text"]
|
||
assert "3. 竞品价格或促销波动" in result["answer_text"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_uses_plan_interaction_level_not_question_cues() -> None:
|
||
class _FakeLLM:
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "开票失败或重试回执",
|
||
"reason": "它最早告诉你票是不是卡在开票动作本身。",
|
||
},
|
||
{
|
||
"name": "税控或发票平台状态变化",
|
||
"reason": "它能解释失败到底是业务问题还是外部平台波动。",
|
||
},
|
||
{
|
||
"name": "发票发送失败反馈",
|
||
"reason": "它能看出问题有没有真正影响到客户收票。",
|
||
},
|
||
],
|
||
"caveat": "这些先算建议，不代表已经接入或验证完成。",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return "不该走到这里。"
|
||
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=_FakeLLM(),
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="自动开票系统这个项目，我想先把票开不出去这块外部情况摸清楚。",
|
||
context_packet={
|
||
"question": "自动开票系统这个项目，我想先把票开不出去这块外部情况摸清楚。",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "自动开票系统外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：自动开票系统，目标是在收到开票请求后自动开票并发送发票图片。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给自动开票系统推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["开票请求处理", "发票发送结果"],
|
||
"signal_source_objectives": [
|
||
"处理开票请求：收到接口请求后自动开票",
|
||
"发送发票图片：把结果发给客户",
|
||
],
|
||
},
|
||
"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": "verifier_blocked",
|
||
"impact_status": "no_direct_impact",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == "external_signal_source_recommendation"
|
||
assert "1. 开票失败或重试回执" in result["answer_text"]
|
||
assert "2. 税控或发票平台状态变化" in result["answer_text"]
|
||
assert "3. 发票发送失败反馈" in result["answer_text"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_presenter_signal_source_recommendation_falls_back_when_llm_overclaims_unverified_details() -> None:
|
||
class _FakeLLM:
|
||
def __init__(self) -> None:
|
||
self.structured_calls = 0
|
||
|
||
async def generate_structured_object(self, **kwargs):
|
||
_ = kwargs
|
||
self.structured_calls += 1
|
||
return SimpleNamespace(
|
||
payload={
|
||
"candidates": [
|
||
{
|
||
"name": "工厂级设备运行日志",
|
||
"reason": "近一年案例显示其与突发性产能下滑相关性达73%。",
|
||
},
|
||
{
|
||
"name": "测试任务调度系统记录",
|
||
"reason": "当前已确认无需额外对接验证。",
|
||
},
|
||
],
|
||
"caveat": "",
|
||
}
|
||
)
|
||
|
||
async def generate(self, prompt: str, **kwargs):
|
||
_ = (prompt, kwargs)
|
||
return (
|
||
"如果只看外部信号源，我会先看工厂级设备运行日志、测试任务调度系统记录、产线温湿度与电力波动数据。"
|
||
)
|
||
|
||
fake_llm = _FakeLLM()
|
||
presenter = ExternalAnswerPresenter(
|
||
llm_service=fake_llm,
|
||
db=object(),
|
||
tenant_id=1,
|
||
)
|
||
|
||
result = await presenter.present(
|
||
question="测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
context_packet={
|
||
"question": "测试产能警告这个项目，如果只考虑外部信号源，你会优先推荐哪 3 类？直接给推荐结果和理由。",
|
||
"plan": {
|
||
"answer_contract": "summary_with_sources",
|
||
"topic": "测试产能警告外部信号源推荐",
|
||
"interaction_level": "recommendation",
|
||
"fit_brief_summary": "当前项目：测试产能警告，聚焦识别产能异常和告警触发。",
|
||
"project_scope_hint": "current_project_and_repo",
|
||
"recommendation_goal": "给测试产能警告推荐最值得先看的外部信号源。",
|
||
"project_object_anchors": ["实现产能警告触发服务", "告警接收链路"],
|
||
"signal_source_objectives": [
|
||
"设计产能监控逻辑与数据采集：编写脚本收集测试队列与资源使用数据",
|
||
"实现产能警告触发服务：实时比对数据与阈值",
|
||
],
|
||
},
|
||
"adapter_attempts": [{"adapter": "live_search", "status": "success"}],
|
||
"verified_evidence": [
|
||
{
|
||
"source_type": "url",
|
||
"source_label": "FineReport",
|
||
"title": "MES 实时监控",
|
||
"target_url": "https://example.com/mes",
|
||
"snippet": "MES 可用于实时监控设备与产线状态。",
|
||
}
|
||
],
|
||
"verifier": {"ok": True},
|
||
"uncertainties": [],
|
||
"retrieval_status": "evidence_found",
|
||
"impact_status": "indirect_only",
|
||
},
|
||
thinking_mode="normal",
|
||
)
|
||
|
||
assert result["prompt_variant_id"] == "external_signal_source_recommendation"
|
||
assert "设计产能监控逻辑与数据采集" in result["answer_text"]
|
||
assert "实现产能警告触发服务" in result["answer_text"]
|
||
assert "73%" not in result["answer_text"]
|
||
assert "无需额外对接验证" not in result["answer_text"]
|
||
assert "测试执行结果或失败告警" not in result["answer_text"]
|
||
assert fake_llm.structured_calls == 1
|