371 lines
12 KiB
Python
371 lines
12 KiB
Python
"""Failure-ordering regression coverage for final Deep Research report writes."""
|
|
|
|
import pytest
|
|
|
|
from app.gateway import deep_research_job_executor as executor_module
|
|
from app.gateway.deep_research_job_executor import DeepResearchJobExecutor
|
|
from deerflow.agents.deep_research.events import DeepResearchEvent
|
|
from deerflow.agents.deep_research.types import DeepResearchResult
|
|
|
|
|
|
class _JobStore:
|
|
def __init__(self, session_store) -> None:
|
|
self._session_store = session_store
|
|
self.progress_calls: list[dict] = []
|
|
|
|
async def update_progress(self, job_id: str, **fields): # noqa: ARG002
|
|
assert self._session_store.report_updates, "report must be persisted before job completion"
|
|
self.progress_calls.append(fields)
|
|
return {"id": job_id, **fields}
|
|
|
|
|
|
class _SessionStore:
|
|
def __init__(self, *, available: bool = True) -> None:
|
|
self.available = available
|
|
self.report_updates: list[dict] = []
|
|
|
|
async def update(self, session_id: str, **fields): # noqa: ARG002
|
|
if not self.available:
|
|
raise RuntimeError("temporary session database failure")
|
|
self.report_updates.append(fields)
|
|
return {"id": session_id, **fields}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_report_is_persisted_before_job_is_marked_completed():
|
|
sessions = _SessionStore()
|
|
jobs = _JobStore(sessions)
|
|
runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)
|
|
|
|
await runner._mark_completed(
|
|
"drj_1",
|
|
"drs_1",
|
|
"worker_1",
|
|
DeepResearchResult(report_markdown="# 报告", source_ids=["src_1"]),
|
|
user_id="user_1",
|
|
query="课题",
|
|
)
|
|
|
|
assert sessions.report_updates[0]["report_markdown"] == "# 报告"
|
|
assert jobs.progress_calls[0]["status"] == "completed"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_report_projection_failure_never_marks_job_completed(monkeypatch):
|
|
monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_ATTEMPTS", 2)
|
|
monkeypatch.setattr(executor_module, "_REPORT_PROJECTION_RETRY_DELAY_SECONDS", 0)
|
|
sessions = _SessionStore(available=False)
|
|
jobs = _JobStore(sessions)
|
|
runner = DeepResearchJobExecutor(jobs, sessions, event_store=None)
|
|
|
|
with pytest.raises(RuntimeError, match="研究报告未能保存"):
|
|
await runner._mark_completed(
|
|
"drj_1",
|
|
"drs_1",
|
|
"worker_1",
|
|
DeepResearchResult(report_markdown="# 报告"),
|
|
user_id="user_1",
|
|
query="课题",
|
|
)
|
|
|
|
assert jobs.progress_calls == []
|
|
|
|
|
|
def test_terminal_timeline_events_are_valid_durable_event_types():
|
|
assert DeepResearchEvent(type="job_completed", phase="done").type == "job_completed"
|
|
assert DeepResearchEvent(type="job_failed", phase="done").type == "job_failed"
|
|
|
|
|
|
def test_zero_source_fallback_is_a_valid_completed_report():
|
|
from app.gateway.deep_research_job_executor import _terminal_report_validation_error
|
|
from deerflow.agents.deep_research.runners.basic import _build_report_fallback
|
|
|
|
report = _build_report_fallback("没有命中的课题", "", 0)
|
|
assert _terminal_report_validation_error(report) is None
|
|
assert "没有命中的课题" in report
|
|
|
|
|
|
def test_fixed_outline_fallback_retains_every_selected_heading():
|
|
from deerflow.agents.deep_research.runners.basic import (
|
|
_build_report_fallback,
|
|
_outline_adherence_error,
|
|
)
|
|
|
|
outline = """# 报告标题
|
|
|
|
# 一、现状判断
|
|
## 1.1 事实基线
|
|
## 1.2 核心矛盾
|
|
# 二、实施方案
|
|
## 2.1 重点任务
|
|
# 参考资料"""
|
|
report = _build_report_fallback(
|
|
"园区升级方案",
|
|
"已确认材料 A\n已确认材料 B",
|
|
2,
|
|
custom_outline=outline,
|
|
structure_mode="fixed",
|
|
)
|
|
|
|
assert "# 园区升级方案" in report
|
|
for heading in (
|
|
"# 一、现状判断",
|
|
"## 1.1 事实基线",
|
|
"## 1.2 核心矛盾",
|
|
"# 二、实施方案",
|
|
"## 2.1 重点任务",
|
|
"# 参考资料",
|
|
):
|
|
assert heading in report
|
|
assert (
|
|
_outline_adherence_error(
|
|
report,
|
|
outline,
|
|
structure_mode="fixed",
|
|
)
|
|
is None
|
|
)
|
|
assert "# 一、现状判断" in (
|
|
_outline_adherence_error(
|
|
report.replace("# 一、现状判断", "## 一、现状判断"),
|
|
outline,
|
|
structure_mode="fixed",
|
|
)
|
|
or ""
|
|
)
|
|
|
|
|
|
def test_adaptive_outline_fallback_keeps_core_chapters_without_forcing_subsections():
|
|
from deerflow.agents.deep_research.runners.basic import _build_report_fallback
|
|
|
|
outline = """# 报告标题
|
|
## 一、事实基线
|
|
### 1.1 可选细节
|
|
## 二、行动建议
|
|
### 2.1 可选措施"""
|
|
report = _build_report_fallback(
|
|
"产业研究",
|
|
"",
|
|
0,
|
|
custom_outline=outline,
|
|
structure_mode="adaptive",
|
|
)
|
|
|
|
assert "# 产业研究" in report
|
|
assert "## 一、事实基线" in report
|
|
assert "## 二、行动建议" in report
|
|
assert "### 1.1 可选细节" not in report
|
|
assert "### 2.1 可选措施" not in report
|
|
|
|
|
|
def test_adaptive_outline_matching_ignores_generic_sections_and_chapter_prefixes():
|
|
from deerflow.agents.deep_research.runners.basic import _outline_adherence_error
|
|
|
|
outline = """# 报告标题
|
|
## 第一章 引言
|
|
## 第二章 市场态势
|
|
## 第三章 结论
|
|
## 参考资料"""
|
|
generic_report = """# 通用报告
|
|
## 引言
|
|
## 结论
|
|
## 参考资料"""
|
|
|
|
error = _outline_adherence_error(
|
|
generic_report,
|
|
outline,
|
|
structure_mode="adaptive",
|
|
)
|
|
|
|
assert error is not None
|
|
assert "第二章 市场态势" in error
|
|
assert (
|
|
_outline_adherence_error(
|
|
"# 市场研究\n\n## 二、市场态势\n\n已按材料展开。",
|
|
outline,
|
|
structure_mode="adaptive",
|
|
)
|
|
is None
|
|
)
|
|
assert (
|
|
_outline_adherence_error(
|
|
"# 简报\n\n## 引言\n\n正文。",
|
|
"# 报告标题\n\n## 第一章 引言\n\n## 第二章 结论",
|
|
structure_mode="adaptive",
|
|
)
|
|
is None
|
|
)
|
|
named_generic_outline = "# 市场态势报告\n\n## 第一章 引言\n\n## 第二章 结论"
|
|
assert (
|
|
_outline_adherence_error(
|
|
"# 市场态势报告\n\n## 完全不同的通用正文",
|
|
named_generic_outline,
|
|
structure_mode="adaptive",
|
|
)
|
|
is not None
|
|
)
|
|
assert (
|
|
_outline_adherence_error(
|
|
"# 改写后的题名\n\n## 引言\n\n正文。",
|
|
named_generic_outline,
|
|
structure_mode="adaptive",
|
|
)
|
|
is None
|
|
)
|
|
|
|
|
|
class _ReportEvents:
|
|
def __init__(self) -> None:
|
|
self.events: list[tuple[str, dict]] = []
|
|
|
|
async def emit(self, event_type: str, *, phase: str = "initializing", payload=None):
|
|
self.events.append((event_type, {"phase": phase, "payload": payload or {}}))
|
|
return None
|
|
|
|
async def publish_live(self, event_type: str, *, phase: str = "initializing", payload=None):
|
|
self.events.append((event_type, {"phase": phase, "payload": payload or {}}))
|
|
|
|
|
|
class _IgnoringOutlineLLM:
|
|
def __init__(self) -> None:
|
|
self.calls: list[list[dict[str, str]]] = []
|
|
|
|
async def stream_complete(self, *, messages, on_delta, **kwargs): # noqa: ARG002
|
|
from deerflow.agents.deep_research.types import CompletionResult
|
|
|
|
self.calls.append(messages)
|
|
text = "# 通用报告\n\n## 摘要\n\n这是未按所选模板生成的普通报告正文。"
|
|
await on_delta(text)
|
|
return CompletionResult(text=text)
|
|
|
|
async def complete_prose(self, *, messages, **kwargs): # noqa: ARG002
|
|
from deerflow.agents.deep_research.types import CompletionResult
|
|
|
|
self.calls.append(messages)
|
|
return CompletionResult(text="# 通用报告\n\n## 结论\n\n重试仍然忽略固定大纲。")
|
|
|
|
|
|
class _PartialAdaptiveOutlineLLM(_IgnoringOutlineLLM):
|
|
async def stream_complete(self, *, messages, on_delta, **kwargs): # noqa: ARG002
|
|
from deerflow.agents.deep_research.types import CompletionResult
|
|
|
|
self.calls.append(messages)
|
|
text = "# 年度增长方案\n\n## 一、目标\n\n保留了一个模板核心章节，其余内容按材料弹性组织。"
|
|
await on_delta(text)
|
|
return CompletionResult(text=text)
|
|
|
|
async def complete_prose(self, *, messages, **kwargs): # noqa: ARG002
|
|
raise AssertionError("adaptive outline with core overlap must not retry")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fixed_outline_mismatch_uses_one_retry_then_template_shaped_final_report():
|
|
from deerflow.agents.deep_research.config import DeepResearchConfig
|
|
from deerflow.agents.deep_research.runners.basic import BasicRunner
|
|
from deerflow.agents.deep_research.types import DeepResearchRequest
|
|
|
|
outline = "# 报告标题\n\n# 一、目标\n\n# 二、路径\n\n# 三、验收"
|
|
request = DeepResearchRequest(
|
|
session_id="session-outline",
|
|
user_id="alice",
|
|
query="年度增长方案",
|
|
config=DeepResearchConfig(
|
|
custom_outline=outline,
|
|
structure_mode="fixed",
|
|
),
|
|
)
|
|
llm = _IgnoringOutlineLLM()
|
|
events = _ReportEvents()
|
|
|
|
report = await BasicRunner()._generate_report(
|
|
llm,
|
|
request,
|
|
"已确认项目目标与约束",
|
|
events,
|
|
source_count=1,
|
|
)
|
|
|
|
assert len(llm.calls) == 2
|
|
assert "固定大纲结构校验失败" in llm.calls[1][-1]["content"]
|
|
assert report.index("# 一、目标") < report.index("# 二、路径") < report.index("# 三、验收")
|
|
assert any(
|
|
event_type == "warning"
|
|
and event["payload"].get("code") == "report_template_fallback"
|
|
for event_type, event in events.events
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_adaptive_outline_does_not_retry_a_material_driven_layout():
|
|
from deerflow.agents.deep_research.config import DeepResearchConfig
|
|
from deerflow.agents.deep_research.runners.basic import BasicRunner
|
|
from deerflow.agents.deep_research.types import DeepResearchRequest
|
|
|
|
request = DeepResearchRequest(
|
|
session_id="session-adaptive-outline",
|
|
user_id="alice",
|
|
query="年度增长方案",
|
|
config=DeepResearchConfig(
|
|
custom_outline="# 报告标题\n\n## 一、目标\n\n## 二、路径",
|
|
structure_mode="adaptive",
|
|
),
|
|
)
|
|
llm = _PartialAdaptiveOutlineLLM()
|
|
|
|
report = await BasicRunner()._generate_report(
|
|
llm,
|
|
request,
|
|
"已确认项目目标与约束",
|
|
_ReportEvents(),
|
|
source_count=1,
|
|
)
|
|
|
|
assert len(llm.calls) == 1
|
|
assert "## 一、目标" in report
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_adaptive_outline_with_zero_core_overlap_retries_then_keeps_core_shape():
|
|
from deerflow.agents.deep_research.config import DeepResearchConfig
|
|
from deerflow.agents.deep_research.runners.basic import BasicRunner
|
|
from deerflow.agents.deep_research.types import DeepResearchRequest
|
|
|
|
request = DeepResearchRequest(
|
|
session_id="session-adaptive-outline-missed",
|
|
user_id="alice",
|
|
query="年度增长方案",
|
|
config=DeepResearchConfig(
|
|
custom_outline="# 报告标题\n\n## 一、目标\n\n### 1.1 指标\n\n## 二、路径\n\n### 2.1 动作",
|
|
structure_mode="adaptive",
|
|
),
|
|
)
|
|
llm = _IgnoringOutlineLLM()
|
|
|
|
report = await BasicRunner()._generate_report(
|
|
llm,
|
|
request,
|
|
"已确认项目目标与约束",
|
|
_ReportEvents(),
|
|
source_count=1,
|
|
)
|
|
|
|
assert len(llm.calls) == 2
|
|
assert "弹性大纲结构校验失败" in llm.calls[1][-1]["content"]
|
|
assert "## 一、目标" in report
|
|
assert "## 二、路径" in report
|
|
assert "### 1.1 指标" not in report
|
|
|
|
|
|
def test_source_table_reload_does_not_become_session_error():
|
|
from app.gateway.deep_research_job_executor import _session_diagnostic_error
|
|
|
|
assert _session_diagnostic_error(
|
|
[
|
|
{
|
|
"code": "client_materials_reloaded_before_write",
|
|
"stage": "collecting",
|
|
"detail": "写作前再次从素材表读到 15 条",
|
|
}
|
|
]
|
|
) is None
|