134 lines
7.1 KiB
Python
134 lines
7.1 KiB
Python
"""P1-R 需求确认：规划师智能体是唯一路径（无规则/模板兜底）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.report_collaboration.requirement.extractor import extract_topic, wants_direct_start
|
|
from app.report_collaboration.requirement.resolver import RequirementResolver
|
|
from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore, ReportCollaborationValidationError
|
|
|
|
|
|
class _ScriptedLLM:
|
|
"""按顺序吐出预置 JSON 决策（模拟规划师智能体多轮）。"""
|
|
|
|
def __init__(self, payloads: list[str]) -> None:
|
|
self._payloads = list(payloads)
|
|
|
|
async def ainvoke(self, messages):
|
|
payload = self._payloads.pop(0) if self._payloads else self._payloads[0]
|
|
return SimpleNamespace(content=payload)
|
|
|
|
|
|
_CLARIFY = """{"reply": "新能源汽车趋势是个好题目。先确认两个关键点：这份报告主要给谁看？希望覆盖哪些角度？",
|
|
"action": "clarify", "questions": ["这份报告主要给谁看？", "希望覆盖哪些角度？"],
|
|
"impact": ["读者影响研究线", "角度影响成员分工"],
|
|
"requirement": {"topic": "新能源汽车趋势报告"}}"""
|
|
|
|
_READY = """{"reply": "清楚了：给投资机构，覆盖市场份额、价格策略和政策环境。我来为你们设计工作流。",
|
|
"action": "ready", "assumptions": ["时间范围默认近两年"],
|
|
"requirement": {"topic": "新能源汽车趋势报告", "audience": "投资机构", "required_angles": ["市场份额", "价格策略", "政策环境"]}}"""
|
|
|
|
|
|
def test_extract_helpers() -> None:
|
|
assert wants_direct_start("请按默认假设继续")
|
|
assert "新能源" in extract_topic("请帮我撰写一份新能源汽车报告")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolver_clarify_round_persists_agent_reply_and_record() -> 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")
|
|
llm = _ScriptedLLM([_CLARIFY])
|
|
resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
|
|
decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m1", max_clarification_rounds=3)
|
|
assert decision is not None and decision.action == "clarify"
|
|
assert "先确认两个关键点" in (decision.reply or "")
|
|
|
|
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"] == "协调者"
|
|
# 智能体文案直接落库，模板开场不得出现
|
|
assert "先确认两个关键点" in ai[0]["content"]
|
|
assert "收到您的报告需求" not in ai[0]["content"]
|
|
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
|
|
|
|
# 幂等：同一消息重放不产生第二条 AI 回复
|
|
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_writes_requirement_metadata_for_summary_card() -> 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"
|
|
)
|
|
llm = _ScriptedLLM([_READY])
|
|
resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
|
|
decision = await resolver.handle_user_message(session["id"], user_message=user, idempotency_key="m2")
|
|
assert decision is not None and decision.action == "ready"
|
|
|
|
snapshot = await store.get_snapshot(session["id"])
|
|
assert snapshot["session"]["status"] == "clarifying"
|
|
assert snapshot["plans"] == [] # ready ≠ 已生成方案
|
|
ai = [item for item in snapshot["messages"] if item["role"] == "ai"][-1]
|
|
assert ai["metadata"]["ready_to_plan"] is True
|
|
assert ai["metadata"]["requirement"]["audience"] == "投资机构"
|
|
assert "设计工作流" in ai["content"]
|
|
record = await store.get_session_record(session["id"])
|
|
assert record["requirement_json"]["ready_to_plan"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolver_answer_merges_fields_across_rounds() -> 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")
|
|
llm = _ScriptedLLM([_CLARIFY, _READY])
|
|
resolver = RequirementResolver(store, planner_llm_factory=lambda: llm)
|
|
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 and decision.action == "ready"
|
|
assert decision.requirement.revision == 2
|
|
assert decision.requirement.audience == "投资机构"
|
|
assert set(decision.requirement.required_angles) == {"市场份额", "价格策略", "政策环境"}
|
|
record = await store.get_session_record(session["id"])
|
|
assert all(item["resolved"] for item in record["requirement_json"]["clarifications"])
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_resolver_raises_planner_unavailable_without_fallback() -> None:
|
|
store = MemoryReportCollaborationStore()
|
|
session = await store.create_session(owner_id="u1", title="", idempotency_key="s4")
|
|
user = await store.create_message(session["id"], role="human", content="写报告", idempotency_key="m4")
|
|
|
|
# 工厂显式禁用（模型不可用）→ 直接报错，无规则兜底
|
|
disabled = RequirementResolver(store, planner_llm_factory=lambda: None)
|
|
with pytest.raises(ReportCollaborationValidationError) as exc:
|
|
await disabled.handle_user_message(session["id"], user_message=user, idempotency_key="m4")
|
|
assert exc.value.code == "PLANNER_UNAVAILABLE"
|
|
|
|
# 模型输出不合法 → 同样报错
|
|
broken = RequirementResolver(store, planner_llm_factory=lambda: _ScriptedLLM(["我说人话不行吗"]))
|
|
with pytest.raises(ReportCollaborationValidationError) as exc2:
|
|
await broken.handle_user_message(session["id"], user_message=user, idempotency_key="m4")
|
|
assert exc2.value.code == "PLANNER_UNAVAILABLE"
|