159 lines
6.6 KiB
Python
159 lines
6.6 KiB
Python
"""RC-BE-005: pre-plan intent router and requirement resolver."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from app.report_collaboration.requirement.extractor import extract_topic, wants_direct_start
|
|
from app.report_collaboration.requirement.intent_router import route_pre_plan_turn
|
|
from app.report_collaboration.requirement.resolver import RequirementResolver
|
|
from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore
|
|
|
|
|
|
def test_topic_only_asks_plan_affecting_gaps() -> None:
|
|
decision = route_pre_plan_turn(
|
|
"写一份新能源汽车主要品牌市场趋势分析报告",
|
|
previous_snapshot=None,
|
|
previous_revision=0,
|
|
clarification_round=0,
|
|
max_clarification_rounds=3,
|
|
)
|
|
assert decision.action == "clarify"
|
|
assert "新能源" in decision.requirement.topic
|
|
assert decision.requirement.revision == 1
|
|
assert len(decision.questions) >= 2
|
|
assert any("读者" in item for item in decision.questions)
|
|
assert any("角度" in item for item in decision.questions)
|
|
|
|
|
|
def test_direct_start_records_default_assumptions() -> None:
|
|
first = route_pre_plan_turn(
|
|
"撰写新能源汽车趋势报告，直接开始",
|
|
previous_snapshot=None,
|
|
previous_revision=0,
|
|
clarification_round=0,
|
|
max_clarification_rounds=3,
|
|
)
|
|
assert first.action == "ready"
|
|
assert first.used_defaults is True
|
|
assert first.requirement.audience == "企业战略决策人员"
|
|
assert "市场份额" in first.requirement.required_angles
|
|
assert first.assumptions
|
|
|
|
|
|
def test_direct_start_without_topic_still_asks_topic() -> None:
|
|
decision = route_pre_plan_turn(
|
|
"直接开始",
|
|
previous_snapshot=None,
|
|
previous_revision=0,
|
|
clarification_round=0,
|
|
max_clarification_rounds=3,
|
|
)
|
|
assert decision.action == "clarify"
|
|
assert any("主题" in item for item in decision.questions)
|
|
|
|
|
|
def test_clarification_cap_uses_explicit_assumptions() -> None:
|
|
first = route_pre_plan_turn(
|
|
"写一份行业报告",
|
|
previous_snapshot=None,
|
|
previous_revision=0,
|
|
clarification_round=0,
|
|
max_clarification_rounds=1,
|
|
)
|
|
assert first.action == "clarify"
|
|
capped = route_pre_plan_turn(
|
|
"先这样吧",
|
|
previous_snapshot=first.requirement.model_dump(),
|
|
previous_revision=first.requirement.revision,
|
|
clarification_round=1,
|
|
max_clarification_rounds=1,
|
|
)
|
|
assert capped.action == "ready"
|
|
assert capped.used_defaults is True
|
|
|
|
|
|
def test_extract_helpers() -> None:
|
|
assert wants_direct_start("请按默认假设继续")
|
|
assert "新能源" in extract_topic("请帮我撰写一份新能源汽车报告")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolver_writes_clarification_without_planning() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s1")
|
|
user = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m1")
|
|
resolver = RequirementResolver(store)
|
|
decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3)
|
|
assert decision is not None
|
|
assert decision.action == "clarify"
|
|
|
|
snapshot = await store.get_snapshot(session["id"])
|
|
assert snapshot["session"]["status"] == "clarifying"
|
|
assert snapshot["session"]["requirement_revision"] == 1
|
|
assert snapshot["plans"] == []
|
|
ai = [item for item in snapshot["messages"] if item["role"] == "ai"]
|
|
assert len(ai) == 1
|
|
assert ai[0]["completed_at"]
|
|
assert ai[0]["name"] == "协调者"
|
|
record = await store.get_session_record(session["id"])
|
|
assert record["requirement_json"]["ready_to_plan"] is False
|
|
assert record["requirement_json"]["clarifications"][0]["resolved"] is False
|
|
|
|
replay = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3)
|
|
assert replay is None
|
|
snapshot2 = await store.get_snapshot(session["id"])
|
|
assert len([item for item in snapshot2["messages"] if item["role"] == "ai"]) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolver_ready_is_not_plan_ready() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s2")
|
|
user = await store.create_message(
|
|
session["id"],
|
|
role="human",
|
|
content="写一份新能源汽车趋势报告，面向战略决策层，必须覆盖市场、政策、海外和风险，直接开始",
|
|
idempotency_key="m2",
|
|
)
|
|
resolver = RequirementResolver(store)
|
|
decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m2")
|
|
assert decision is not None
|
|
assert decision.action == "ready"
|
|
|
|
snapshot = await store.get_snapshot(session["id"])
|
|
assert snapshot["session"]["status"] == "clarifying"
|
|
assert snapshot["plans"] == []
|
|
record = await store.get_session_record(session["id"])
|
|
assert record["requirement_json"]["ready_to_plan"] is True
|
|
assert all(item["resolved"] for item in record["requirement_json"]["clarifications"])
|
|
assert "不等于已经生成" in snapshot["messages"][-1]["content"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_answer_merges_and_resolves_clarification() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s3")
|
|
first = await store.create_message(session["id"], role="human", content="写一份新能源汽车趋势报告", idempotency_key="m3a")
|
|
resolver = RequirementResolver(store)
|
|
await resolver.handle_user_message(session["id"], user_message=first, idempotency_key="m3a")
|
|
|
|
second = await store.create_message(
|
|
session["id"],
|
|
role="human",
|
|
content="读者是投资机构，必须覆盖市场份额、价格策略和政策环境",
|
|
idempotency_key="m3b",
|
|
)
|
|
decision = await resolver.handle_user_message(session["id"], user_message=second, idempotency_key="m3b")
|
|
assert decision is not None
|
|
assert decision.action == "ready"
|
|
assert decision.used_defaults is False
|
|
assert decision.requirement.audience == "投资机构"
|
|
assert decision.requirement.revision == 2
|
|
record = await store.get_session_record(session["id"])
|
|
assert record["requirement_revision"] == 2
|
|
assert all(item["resolved"] for item in record["requirement_json"]["clarifications"])
|
|
snapshot = await store.get_snapshot(session["id"])
|
|
assert snapshot["plans"] == []
|
|
assert snapshot["session"]["status"] == "clarifying"
|