diff --git a/.gitignore b/.gitignore index 476bea3c..60e88df4 100644 --- a/.gitignore +++ b/.gitignore @@ -132,6 +132,8 @@ offline-backend-20260512/backend/tests/* !offline-backend-20260512/backend/tests/test_writing_rewrite_prompt.py # LLM call timeout retry classification regression coverage. !offline-backend-20260512/backend/tests/test_llm_error_handling_middleware.py +# Writing-mode pipeline conversation-material regressions. +!offline-backend-20260512/backend/tests/test_deep_research_report_tool.py deploy .history diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py index c662961d..2d87627a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/deep_research/collection/thread_harvester.py @@ -72,9 +72,7 @@ async def harvest_thread_materials( visible text (its collection小结) and doubles as a fallback material when the thread produced no harvestable tool results. """ - checkpoint_tuple = await checkpointer.aget_tuple( - {"configurable": {"thread_id": runtime_thread_id, "checkpoint_ns": ""}} - ) + checkpoint_tuple = await checkpointer.aget_tuple({"configurable": {"thread_id": runtime_thread_id, "checkpoint_ns": ""}}) if checkpoint_tuple is None: logger.warning("deep-research harvest: no checkpoint for thread %s", runtime_thread_id) return [], "" @@ -121,9 +119,14 @@ def harvest_messages_materials( def _harvest_tool_messages(messages: list[Any], *, query: str) -> list[ResearchMaterial]: - """Convert every harvestable tool result in the thread into materials.""" + """Convert every harvestable tool result in the thread into materials. + + 每条材料在 ``metadata["message_index"]`` 记录其来源工具消息在对话中的 + 位置——写作管线据此区分「课题请求之后检索的（属于当前课题）」与 + 「课题切换前的旧检索」（后者才做词面相关度过滤）。 + """ materials: list[ResearchMaterial] = [] - for message in messages: + for message_index, message in enumerate(messages): if _message_type(message) != "tool": continue tool_name = str(_message_name(message) or "") @@ -140,15 +143,20 @@ def _harvest_tool_messages(messages: list[Any], *, query: str) -> list[ResearchM if isinstance(item, dict): material = result_to_material(item, query=envelope_query) material.metadata["retrieval_channel"] = "chat_web_search" + material.metadata["message_index"] = message_index materials.append(material) continue - materials.extend( - _harvest_generic_payload(payload, tool_name=tool_name, query=query) - ) + materials.extend(_harvest_generic_payload(payload, tool_name=tool_name, query=query, message_index=message_index)) return materials -def _harvest_generic_payload(payload: Any, *, tool_name: str, query: str) -> list[ResearchMaterial]: +def _harvest_generic_payload( + payload: Any, + *, + tool_name: str, + query: str, + message_index: int | None = None, +) -> list[ResearchMaterial]: """Normalize a non-web_search tool payload (skill/knowledge/bash/MCP).""" source_type = _TOOL_SOURCE_TYPES.get(tool_name, "other") materials: list[ResearchMaterial] = [] @@ -177,6 +185,7 @@ def _harvest_generic_payload(payload: Any, *, tool_name: str, query: str) -> lis metadata={ "retrieval_channel": "chat_tool", "tool_name": tool_name, + **({"message_index": message_index} if message_index is not None else {}), }, ) ) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py index 58d81783..8b26a7c0 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/deep_research_report_tool.py @@ -34,7 +34,7 @@ import re from typing import Annotated, Any from langchain.tools import InjectedToolCallId, ToolRuntime, tool -from langchain_core.messages import ToolMessage +from langchain_core.messages import HumanMessage, ToolMessage from langgraph.types import Command from langgraph.typing import ContextT @@ -42,7 +42,6 @@ from deerflow.agents.deep_research.adapters.artifacts import ThreadOutputsArtifa from deerflow.agents.deep_research.adapters.context import DeerFlowContextCompressor from deerflow.agents.deep_research.adapters.image import build_image_provider from deerflow.agents.deep_research.adapters.llm import DeerFlowCompletionBackend -from deerflow.agents.deep_research.adapters.material_provider import DeerFlowMaterialProvider from deerflow.agents.deep_research.collection.thread_harvester import ( StaticHarvestMaterialProvider, harvest_messages_materials, @@ -167,53 +166,6 @@ class StreamWriterEventSink: logger.debug("deep_research_report: custom stream frame dropped", exc_info=True) -class HarvestPlusSearchMaterialProvider: - """对话收割池优先 + 自检索补充的合并资料源（``retrieve=True`` 分支）。 - - 与 :class:`StaticHarvestMaterialProvider` 同协议：每个子查询返回合并池， - 交由 runner 的 dedupe / compress 阶段裁剪。自检索单项失败不致命—— - 退化为只用对话资料。 - """ - - def __init__(self, harvested: list[ResearchMaterial], search_provider: DeerFlowMaterialProvider) -> None: - self._harvested = list(harvested) - self._search = search_provider - - @property - def material_count(self) -> int: - return len(self._harvested) - - async def search( - self, - query: str, - *, - domains: list[str] | None = None, - limit: int = 8, - user_id: str = "", - session_id: str = "", - cancellation=None, - ) -> list[ResearchMaterial]: - from deerflow.agents.deep_research.adapters.material import dedupe_materials - - if cancellation is not None: - cancellation.raise_if_cancelled() - combined = list(self._harvested) - try: - combined.extend( - await self._search.search( - query, - domains=domains, - limit=limit, - user_id=user_id, - session_id=session_id, - cancellation=cancellation, - ) - ) - except Exception: # noqa: BLE001 - 自检索失败退化为对话资料 - logger.warning("deep_research_report: supplementary search failed", exc_info=True) - return dedupe_materials(combined) - - # 写作模式常在「不用搜索 / 对话资料很少」时开写。工作台默认的「材料不足就 # 写明资料不足」会把报告写成元分析；这里强制用领域知识补全正文。 _KNOWLEDGE_WRITE_INSTRUCTION = ( @@ -330,16 +282,91 @@ def _is_topic_relevant(material: ResearchMaterial, strong: frozenset[str], weak: def _filter_relevant_materials( - materials: list[ResearchMaterial], *, topic: str + materials: list[ResearchMaterial], + *, + topic: str, + anchor_index: int | None = None, ) -> tuple[list[ResearchMaterial], int]: - """按当前主题过滤收割池；返回 ``(相关池, 被剔除条数)``。""" + """按当前主题过滤收割池；返回 ``(相关池, 被剔除条数)``。 + + 两条保留通道（满足其一即入池）： + + 1. **位置**：材料来自课题请求消息（``anchor_index``）之后的工具消息—— + 那是模型为本课题检索的，与词面/语言无关，一律保留。线上实测中文课题 + + 英文检索结果会被纯词面匹配整批误杀（35 条只留 4 条）。 + 2. **词面**：材料词面命中课题（强词/标题/正文规则）——兜住课题切换前 + 的旧检索里仍与当前课题相关的部分。 + + ``anchor_index`` 之前的旧课题检索（词面不命中）继续被剔除（P1：旧课题 + 资料不得凑门槛、不得入报告）。``anchor_index=None`` 或负数（对话里找不 + 到课题请求消息，如纯工具消息的旧调用方）退化为纯词面过滤。 + """ strong, weak = _topic_terms(topic) if not strong and not weak: return list(materials), 0 - relevant = [m for m in materials if _is_topic_relevant(m, strong, weak)] + positional = anchor_index is not None and anchor_index >= 0 + relevant: list[ResearchMaterial] = [] + for material in materials: + message_index = material.metadata.get("message_index") + if positional and isinstance(message_index, int) and message_index > anchor_index: + relevant.append(material) + continue + if _is_topic_relevant(material, strong, weak): + relevant.append(material) return relevant, len(materials) - len(relevant) +def _anchor_visible_human(message: Any) -> bool: + if not isinstance(message, HumanMessage): + return False + kwargs = getattr(message, "additional_kwargs", None) or {} + return not kwargs.get("hide_from_ui") + + +def _anchor_is_clarification_result(message: Any) -> bool: + return isinstance(message, ToolMessage) and str(getattr(message, "name", "") or "") == "ask_clarification" + + +def _anchor_message_text(message: Any) -> str: + content = getattr(message, "content", "") + if isinstance(content, list): + parts: list[str] = [] + for part in content: + if isinstance(part, str): + parts.append(part) + elif isinstance(part, dict) and isinstance(part.get("text"), str): + parts.append(part["text"]) + return " ".join(parts).strip() + return str(content or "").strip() + + +def topic_anchor_index(messages: list[Any], *, topic: str = "") -> int: + """课题请求消息在对话中的位置（写作模式首报轮的资料过滤锚点）。 + + 规则与 ``writing_mode_report_middleware.extract_report_topic`` 一致：取 + 最新的可见人类消息、跳过对 ``ask_clarification`` 的直接回答（紧跟澄清 + ToolMessage 的人类消息不是新课题）。传入 ``topic`` 时优先返回与其文本 + 前缀一致的消息位置（模型原样复述课题的常见情形）。找不到返回 -1。 + """ + normalized_topic = " ".join(str(topic or "").split()) + prefix = normalized_topic[:80] + fallback = -1 + for index in range(len(messages) - 1, -1, -1): + message = messages[index] + if not _anchor_visible_human(message): + continue + if index > 0 and _anchor_is_clarification_result(messages[index - 1]): + continue + text = " ".join(_anchor_message_text(message).split()) + if not text: + continue + if prefix and text.startswith(prefix): + return index + if fallback < 0: + fallback = index + return fallback + + def _safe_stream_writer() -> Any: """获取当前 run 的 custom 事件写入器；非流式上下文返回 None。""" try: @@ -408,16 +435,18 @@ async def deep_research_report_tool( state = getattr(runtime, "state", None) or {} conversation_messages = state.get("messages") or [] harvested, _agent_summary = harvest_messages_materials(conversation_messages, query=query) - # 按当前主题强过滤：旧课题的检索结果不计门槛、不入报告（课题切换场景）。 - materials, dropped_count = _filter_relevant_materials(harvested, topic=query) + # 课题请求之后的检索结果整批保留（那是模型为本课题检索的）；只有课题 + # 切换前的旧检索才按词面强过滤（旧课题资料不得凑门槛、不得入报告）。 + anchor_index = topic_anchor_index(conversation_messages, topic=query) + materials, dropped_count = _filter_relevant_materials(harvested, topic=query, anchor_index=anchor_index) harvested_count = len(materials) logger.info( - "[deep_research_report] thread=%s harvested %d materials from conversation " - "(topic-relevant %d, off-topic dropped %d)", + "[deep_research_report] thread=%s harvested %d materials from conversation (topic-relevant %d, off-topic dropped %d, anchor_index=%d)", thread_id, len(harvested), harvested_count, dropped_count, + anchor_index, ) if harvested_count < _MIN_HARVESTED_MATERIALS: @@ -430,8 +459,7 @@ async def deep_research_report_tool( if retrieve is True: logger.info( - "[deep_research_report] thread=%s ignoring retrieve=true; " - "writing-mode pipeline writes from conversation materials only", + "[deep_research_report] thread=%s ignoring retrieve=true; writing-mode pipeline writes from conversation materials only", thread_id, ) materials_provider: Any = StaticHarvestMaterialProvider(materials) @@ -520,11 +548,7 @@ async def deep_research_report_tool( ) source_count = len(result.source_ids or []) - summary = ( - f"研究报告已完成：基于 {source_count} 条资料（{material_source}）按「通用分析结构」撰写，" - f"已写入 {REPORT_VIRTUAL_PATH} 并展示给用户。" - "请用 2-3 句话向用户简要总结报告的核心结论，不要复述全文。" - ) + summary = f"研究报告已完成：基于 {source_count} 条资料（{material_source}）按「通用分析结构」撰写，已写入 {REPORT_VIRTUAL_PATH} 并展示给用户。请用 2-3 句话向用户简要总结报告的核心结论，不要复述全文。" logger.info( "[deep_research_report] thread=%s completed (sources=%d, report_chars=%d)", thread_id, @@ -540,10 +564,10 @@ async def deep_research_report_tool( __all__ = [ - "HarvestPlusSearchMaterialProvider", "REPORT_VIRTUAL_PATH", "StreamWriterEventSink", "build_writing_research_config", "deep_research_report_tool", "is_writing_mode_report_path", + "topic_anchor_index", ] diff --git a/offline-backend-20260512/backend/tests/test_deep_research_report_tool.py b/offline-backend-20260512/backend/tests/test_deep_research_report_tool.py new file mode 100644 index 00000000..c2bcc819 --- /dev/null +++ b/offline-backend-20260512/backend/tests/test_deep_research_report_tool.py @@ -0,0 +1,649 @@ +"""写作模式内联深度研究报告工具的单元测试。 + +覆盖：默认配置组装（通用分析结构 / 继承聊天模型）、事件转发（custom 通道帧 +格式/过滤/容错）、工具主体（收割优先——资料足够直接写 / 不足也直接开写 / +retrieve 参数被忽略、按当前主题的相关度强过滤、线程解析、失败路径、 +artifacts 更新），以及 lead / agent-only 两条 prompt 模板的写作模式首轮合约。 +""" + +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from typing import Any + +from langchain_core.messages import AIMessage, HumanMessage, ToolMessage + +from deerflow.agents.deep_research.collection.thread_harvester import StaticHarvestMaterialProvider +from deerflow.agents.deep_research.types import DeepResearchRequest, DeepResearchResult +from deerflow.persistence.report_structures.defaults import GENERAL_OUTLINE +from deerflow.tools.builtins import deep_research_report_tool as tool_module +from deerflow.tools.builtins.deep_research_report_tool import ( + _KNOWLEDGE_WRITE_INSTRUCTION, + StreamWriterEventSink, + build_writing_research_config, + deep_research_report_tool, +) + +_DEFAULT_TOPIC = "新能源汽车市场分析" +_OFF_TOPIC = "养老产业发展研究" + +# ── 配置组装 ───────────────────────────────────────────────────────────────── + + +def test_writing_research_config_defaults_to_general_outline_basic() -> None: + cfg = build_writing_research_config() + + assert cfg.mode == "basic" + assert cfg.custom_outline == GENERAL_OUTLINE + assert cfg.structure_mode == "adaptive" + assert cfg.generate_images is False + assert cfg.report_instruction == _KNOWLEDGE_WRITE_INSTRUCTION + assert "必须基于模型自身的领域知识" in cfg.report_instruction + + +def test_writing_research_config_maps_focus_to_report_instruction() -> None: + cfg = build_writing_research_config(focus=" 侧重对比分析，3000 字左右 ") + + assert cfg.report_instruction is not None + assert cfg.report_instruction.startswith("侧重对比分析，3000 字左右") + assert _KNOWLEDGE_WRITE_INSTRUCTION in cfg.report_instruction + # 空白 focus 仍保留知识补全指令，不产生空行前缀 + empty_focus = build_writing_research_config(focus=" ") + assert empty_focus.report_instruction == _KNOWLEDGE_WRITE_INSTRUCTION + + +# ── 事件转发 sink ──────────────────────────────────────────────────────────── + + +def _capture_writer() -> tuple[list[dict], object]: + frames: list[dict] = [] + + def writer(frame: dict) -> None: + frames.append(frame) + + return frames, writer + + +def test_event_sink_wraps_forwardable_events() -> None: + frames, writer = _capture_writer() + sink = StreamWriterEventSink(writer, session_id="chat-t1", job_id="call-1") + + asyncio.run(sink.emit("phase_changed", phase="planning", payload={"to": "planning"})) + + assert len(frames) == 1 + frame = frames[0] + assert frame["type"] == "deep_research_report" + assert frame["event"]["type"] == "phase_changed" + assert frame["event"]["phase"] == "planning" + assert frame["event"]["payload"] == {"to": "planning"} + assert frame["event"]["jobId"] == "call-1" + + +def test_event_sink_drops_noise_events() -> None: + frames, writer = _capture_writer() + sink = StreamWriterEventSink(writer, session_id="s", job_id="j") + + asyncio.run(sink.emit("model_usage", payload={"tokens": 1})) + asyncio.run(sink.emit("heartbeat")) + + assert frames == [] + + +def test_event_sink_forwards_live_report_deltas_only() -> None: + frames, writer = _capture_writer() + sink = StreamWriterEventSink(writer, session_id="s", job_id="j") + + asyncio.run(sink.publish_live("report_delta", phase="writing", payload={"delta": "第一"})) + asyncio.run(sink.publish_live("summary_delta", phase="summarizing", payload={"delta": "x"})) + + assert len(frames) == 1 + assert frames[0]["event"]["type"] == "report_delta" + assert frames[0]["event"]["live"] is True + + +def test_event_sink_survives_writer_failure() -> None: + def broken_writer(frame: dict) -> None: + raise RuntimeError("stream closed") + + sink = StreamWriterEventSink(broken_writer, session_id="s", job_id="j") + + event = asyncio.run(sink.emit("report_completed", payload={"sourceCount": 3})) + + assert event.type == "report_completed" # 事件本体仍然返回 + asyncio.run(sink.publish_live("report_delta", payload={"delta": "x"})) # 不抛异常 + + +# ── 工具主体 ───────────────────────────────────────────────────────────────── + + +def _material_message(index: int, topic: str = _DEFAULT_TOPIC) -> ToolMessage: + """一条可被收割器识别、且与 ``topic`` 词面相关的工具结果消息。""" + return ToolMessage( + content=json.dumps( + { + "results": [ + { + "title": f"{topic}：资料{index}", + "content": f"第 {index} 条关于{topic}的对话检索资料，包含要点分析。", + "url": f"https://example.com/{index}", + } + ] + }, + ensure_ascii=False, + ), + name="weknora_search", + tool_call_id=f"call-m{index}", + ) + + +def _fake_runtime( + thread_id: str = "thread-1", + messages: list | None = None, +) -> SimpleNamespace: + return SimpleNamespace( + context={"thread_id": thread_id}, + config={"configurable": {"thread_id": thread_id}}, + state={"messages": messages if messages is not None else [_material_message(i) for i in range(1, 4)]}, + ) + + +class _FakeEngine: + """替代 DeepResearchEngine：记录请求与 adapters 并返回可控结果。""" + + instances: list[_FakeEngine] = [] + + def __init__(self) -> None: + self.requests: list[DeepResearchRequest] = [] + self.adapters: Any = None # noqa: RUF012 - 测试桩 + self.result = DeepResearchResult( + report_markdown="# 报告\n正文", + source_ids=["s1", "s2", "s3"], + ) + self.error: Exception | None = None + _FakeEngine.instances.append(self) + + async def run(self, request, *, adapters, cancellation=None) -> DeepResearchResult: + self.requests.append(request) + self.adapters = adapters + if self.error is not None: + raise self.error + return self.result + + +def _patch_tool_dependencies(monkeypatch, *, user_id: str = "u1", engine=None) -> list[dict]: + frames, writer = _capture_writer() + monkeypatch.setattr(tool_module, "DeepResearchEngine", lambda: engine or _FakeEngine()) + monkeypatch.setattr(tool_module, "resolve_path_user_id", lambda thread_id: user_id) + monkeypatch.setattr(tool_module, "_safe_stream_writer", lambda: writer) + monkeypatch.setattr( + tool_module, + "DeerFlowCompletionBackend", + lambda *args, **kwargs: SimpleNamespace(usage_totals={}), + ) + monkeypatch.setattr( + tool_module, + "DeerFlowContextCompressor", + lambda **kwargs: SimpleNamespace(captured=kwargs), + ) + monkeypatch.setattr(tool_module, "ThreadOutputsArtifactWriter", lambda *args, **kwargs: SimpleNamespace(captured=(args, kwargs))) + monkeypatch.setattr(tool_module, "build_image_provider", lambda: None) + return frames + + +def test_tool_writes_with_harvested_materials_when_sufficient(monkeypatch) -> None: + frames = _patch_tool_dependencies(monkeypatch) + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-9", messages=[_material_message(i) for i in range(1, 5)]), + "call-1", + _DEFAULT_TOPIC, + focus="侧重销量对比", + ) + ) + + engine = _FakeEngine.instances[-1] + assert len(engine.requests) == 1 + request = engine.requests[0] + assert request.query == "新能源汽车市场分析" + assert request.config.custom_outline == GENERAL_OUTLINE + assert request.config.report_instruction is not None + assert request.config.report_instruction.startswith("侧重销量对比") + assert _KNOWLEDGE_WRITE_INSTRUCTION in request.config.report_instruction + assert request.session_id == "chat-thread-9" + + # 资料足够 → 只用收割池，不触发自检索 + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + assert engine.adapters.materials.material_count == 4 + + update = command.update + assert update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + tool_message = update["messages"][0] + assert tool_message.tool_call_id == "call-1" + assert "对话资料" in tool_message.content + assert "3" in tool_message.content # 来源条数写入总结 + + # 终态帧已发出（success） + assert any(f.get("event", {}).get("type") == "tool_finished" and f["event"]["payload"]["status"] == "success" for f in frames) + + +def test_tool_writes_immediately_when_harvested_materials_insufficient(monkeypatch) -> None: + """资料不足也不再询问用户，直接用现有对话资料开写。""" + frames = _patch_tool_dependencies(monkeypatch) + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-a", messages=[_material_message(1, topic="课题")]), + "call-2", + "课题", + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + assert engine.adapters.materials.material_count == 1 + + update = command.update + assert update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + tool_message = update["messages"][0] + assert tool_message.tool_call_id == "call-2" + assert "对话资料" in tool_message.content + assert "ask_clarification" not in tool_message.content + assert any(f.get("event", {}).get("type") == "tool_finished" and f["event"]["payload"]["status"] == "success" for f in frames) + + +def test_tool_retrieve_true_still_writes_from_conversation_only(monkeypatch) -> None: + """写作管线忽略 retrieve=true：检索留给对话工具，这里只写文件。""" + _patch_tool_dependencies(monkeypatch) + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-b", messages=[_material_message(1, topic="课题")]), + "call-3", + "课题", + retrieve=True, + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + assert engine.adapters.materials.material_count == 1 + + update = command.update + assert update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + assert "对话资料" in update["messages"][0].content + assert "补充检索" not in update["messages"][0].content + + +def test_tool_drops_off_topic_materials_from_pool_and_threshold(monkeypatch) -> None: + """课题切换：旧课题检索结果既不计入门槛，也不进报告资料池。""" + _patch_tool_dependencies(monkeypatch) + messages = [ + *[_material_message(i) for i in range(1, 6)], # 旧课题（新能源）5 条 + *[_material_message(i, topic=_OFF_TOPIC) for i in range(11, 14)], # 当前课题 3 条 + ] + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-t", messages=messages), + "call-t", + _OFF_TOPIC, + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + # 只有当前课题的 3 条入池；旧课题 5 条被强过滤剔除 + assert engine.adapters.materials.material_count == 3 + assert command.update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + + +def test_tool_writes_immediately_when_all_materials_off_topic(monkeypatch) -> None: + """收割池非空但全部偏题 → 相关数 0，仍直接开写，不再问用户。""" + _patch_tool_dependencies(monkeypatch) + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-o", messages=[_material_message(i) for i in range(1, 6)]), + "call-o", + _OFF_TOPIC, + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + assert engine.adapters.materials.material_count == 0 + assert command.update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + assert "ask_clarification" not in command.update["messages"][0].content + + +def test_relevance_filter_rules() -> None: + """词面相关度判定的最小规则集（强词 / 标题 / 正文 / 短主题兜底）。""" + from deerflow.agents.deep_research.types import ResearchMaterial + + strong, weak = tool_module._topic_terms(_DEFAULT_TOPIC) + assert _DEFAULT_TOPIC in strong # 整段短语是强词 + assert "新能" in weak and "分析" in weak + + def _mat(title: str, content: str) -> ResearchMaterial: + return ResearchMaterial( + id=title, + query="q", + title=title, + url=None, + raw_content=content, + snippet=None, + source="t", + source_type="other", + content_hash=title, + ) + + # 强词命中标题 + assert tool_module._is_topic_relevant(_mat(f"{_DEFAULT_TOPIC}：综述", "无关键词"), strong, weak) is True + # 标题两个不同 bigram + assert tool_module._is_topic_relevant(_mat("新能源汽车销量榜", "……"), strong, weak) is True + # 完全偏题 + assert tool_module._is_topic_relevant(_mat("养老产业白皮书", "养老服务体系建设"), strong, weak) is False + # 正文密集命中（≥4 个不同 bigram） + assert tool_module._is_topic_relevant(_mat("行业观察", "新能源汽车市场分析"), strong, weak) is True + # 短主题兜底：唯一词命中即相关 + s2, w2 = tool_module._topic_terms("课题") + assert tool_module._is_topic_relevant(_mat("课题：资料1", "关于课题的要点"), s2, w2) is True + + +# ── 课题锚点：请求之后的对话检索整批保留（语言/词面无关） ──────────────────── + + +def _search_envelope_message(index: int, *, query: str, titles: list[str]) -> ToolMessage: + """一条 web_search 工具结果信封（收割器走最高保真映射分支）。""" + return ToolMessage( + content=json.dumps( + { + "query": query, + "results": [ + { + "title": title, + "content": f"{title} — key findings and technical details.", + "url": f"https://example.com/{index}-{i}", + } + for i, title in enumerate(titles) + ], + }, + ensure_ascii=False, + ), + name="web_search", + tool_call_id=f"call-s{index}", + ) + + +def test_topic_anchor_index_rules() -> None: + """锚点 = 最新的非澄清回答人类消息；澄清回答不充当新课题。""" + messages = [ + HumanMessage(content="写人工智能领域的论文md"), + AIMessage( + content="", + tool_calls=[{"id": "q1", "name": "ask_clarification", "args": {"question": "方向？"}, "type": "tool_call"}], + ), + ToolMessage(content="需要澄清", tool_call_id="q1", name="ask_clarification"), + HumanMessage(content="方向：AIGC；篇幅：3000字"), + HumanMessage(content="写人工智能领域的论文md"), + ] + # 文本前缀命中课题原文 + assert tool_module.topic_anchor_index(messages, topic="写人工智能领域的论文md") == 4 + # 无 topic 匹配时回落「最新的非回答人类消息」——回答被跳过 + messages[-1] = HumanMessage(content="补充：要英文文献") + assert tool_module.topic_anchor_index(messages) == 4 + assert tool_module.topic_anchor_index([]) == -1 + + +def test_tool_keeps_conversation_searches_after_topic_request(monkeypatch) -> None: + """课题请求之后的对话检索结果整批入池——中文课题 + 英文检索结果不再被 + 纯词面匹配误杀（线上 35 条只留 4 条的根因）。""" + _patch_tool_dependencies(monkeypatch) + messages = [ + HumanMessage(content="写一篇人工智能领域的论文md"), + _search_envelope_message( + 1, + query="AIGC generative AI survey 2024", + titles=[ + "Sora: A Review of Text-to-Video Generation Models", + "Diffusion Transformers for Image Synthesis", + "Challenges and Ethics of AI-Generated Content", + ], + ), + _search_envelope_message( + 2, + query="LLM alignment RLHF DPO", + titles=["Direct Preference Optimization: A Survey"], + ), + ] + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-keep", messages=messages), + "call-keep", + "写一篇人工智能领域的论文md", + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + # 4 条英文资料全部进入写作池（锚点=课题请求，之后的检索一律保留） + assert engine.adapters.materials.material_count == 4 + assert command.update["artifacts"] == ["/mnt/user-data/outputs/report.md"] + + +def test_tool_still_drops_old_topic_searches_before_new_request(monkeypatch) -> None: + """课题切换：新请求之前的旧课题检索仍被词面强过滤，不凑门槛不入报告。""" + _patch_tool_dependencies(monkeypatch) + messages = [ + HumanMessage(content="写养老产业发展研究报告"), + *[_material_message(i, topic=_OFF_TOPIC) for i in range(1, 4)], # 旧课题 3 条 + HumanMessage(content="改成写新能源汽车市场分析"), + *[_material_message(i, topic=_DEFAULT_TOPIC) for i in range(11, 14)], # 当前课题 3 条 + ] + + asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-switch", messages=messages), + "call-switch", + _DEFAULT_TOPIC, + ) + ) + + engine = _FakeEngine.instances[-1] + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + # 旧课题 3 条（锚点之前、词面不命中）被剔除；当前课题 3 条保留 + assert engine.adapters.materials.material_count == 3 + + +def test_tool_retrieve_false_writes_with_existing_materials(monkeypatch) -> None: + frames = _patch_tool_dependencies(monkeypatch) + engines_before = len(_FakeEngine.instances) + + command = asyncio.run( + deep_research_report_tool.coroutine( + _fake_runtime("thread-c", messages=[]), # 极端：一条资料也没有 + "call-4", + "课题", + retrieve=False, + ) + ) + + # 用户明确拒绝检索 → 即使收割为空也直接进入管线（runner 有空资料兜底） + engine = _FakeEngine.instances[-1] + assert len(_FakeEngine.instances) == engines_before + 1 + assert isinstance(engine.adapters.materials, StaticHarvestMaterialProvider) + assert engine.adapters.materials.material_count == 0 + assert "对话资料" in command.update["messages"][0].content + assert any(f.get("event", {}).get("type") == "tool_finished" and f["event"]["payload"]["status"] == "success" for f in frames) + + +def test_tool_reports_engine_failure_without_artifacts(monkeypatch) -> None: + failing = _FakeEngine() + failing.error = RuntimeError("检索服务不可用") + frames = _patch_tool_dependencies(monkeypatch, engine=failing) + + command = asyncio.run(deep_research_report_tool.coroutine(_fake_runtime("thread-x"), "call-5", _DEFAULT_TOPIC)) + + update = command.update + assert "artifacts" not in update + assert "检索服务不可用" in update["messages"][0].content + assert any(f.get("event", {}).get("type") == "tool_finished" and f["event"]["payload"]["status"] == "error" for f in frames) + + +def test_tool_errors_cleanly_without_thread_id(monkeypatch) -> None: + _patch_tool_dependencies(monkeypatch) + runtime = SimpleNamespace(context={}, config={}, state={"messages": []}) + engines_before = len(_FakeEngine.instances) + + command = asyncio.run(deep_research_report_tool.coroutine(runtime, "call-6", "课题")) + + update = command.update + assert "artifacts" not in update + assert "thread id" in update["messages"][0].content + # 未触达引擎 + assert len(_FakeEngine.instances) == engines_before + + +# ── 模型继承（P2：管线用聊天选的模型与思考开关） ───────────────────────────── + + +def test_writing_research_config_inherits_chat_model() -> None: + cfg = build_writing_research_config(thinking_enabled=True, model_name="qwen3.6") + + assert cfg.fast_model == "qwen3.6" + assert cfg.smart_model == "qwen3.6" + assert cfg.strategic_model == "qwen3.6" + assert cfg.thinking_enabled is True + # 缺省回落系统默认模型 + 关思考 + defaults = build_writing_research_config() + assert defaults.fast_model is None and defaults.smart_model is None and defaults.strategic_model is None + assert defaults.thinking_enabled is False + + +def _fake_app_config(monkeypatch, *, known: str = "qwen3.6", supports_thinking: bool = True) -> None: + import deerflow.config.app_config as app_config_module + + def get_model_config(name: str): + if name != known: + return None + return SimpleNamespace(supports_thinking=supports_thinking) + + monkeypatch.setattr( + app_config_module, + "get_app_config", + lambda: SimpleNamespace(get_model_config=get_model_config), + ) + + +def test_resolve_chat_model_passes_through_known_model(monkeypatch) -> None: + _fake_app_config(monkeypatch, known="qwen3.6", supports_thinking=True) + + name, thinking, force = tool_module._resolve_chat_model({"model_name": "qwen3.6", "thinking_enabled": True}) + + assert (name, thinking, force) == ("qwen3.6", True, False) + + +def test_resolve_chat_model_falls_back_on_unknown_or_unsupported(monkeypatch) -> None: + _fake_app_config(monkeypatch, known="qwen3.6", supports_thinking=False) + + # 不在 config.yaml 的模型 → 回落系统默认 + assert tool_module._resolve_chat_model({"model_name": "no-such-model"})[0] is None + # 模型不支持思考 → 关思考 + name, thinking, _ = tool_module._resolve_chat_model({"model_name": "qwen3.6", "thinking_enabled": True}) + assert (name, thinking) == ("qwen3.6", False) + # 未选模型 → None + 透传 force 开关 + assert tool_module._resolve_chat_model({"thinking_force_disabled": True}) == (None, True, True) + + +def test_tool_threads_chat_model_into_research_config(monkeypatch) -> None: + _patch_tool_dependencies(monkeypatch) + _fake_app_config(monkeypatch, known="qwen3.6", supports_thinking=True) + runtime = SimpleNamespace( + context={ + "thread_id": "thread-m", + "model_name": "qwen3.6", + "thinking_enabled": True, + "thinking_force_disabled": False, + }, + config={}, + state={"messages": [_material_message(i) for i in range(1, 4)]}, + ) + + asyncio.run(deep_research_report_tool.coroutine(runtime, "call-m", _DEFAULT_TOPIC)) + + config = _FakeEngine.instances[-1].requests[0].config + assert config.smart_model == "qwen3.6" + assert config.strategic_model == "qwen3.6" + assert config.thinking_enabled is True + + +# ── prompt 合约 ────────────────────────────────────────────────────────────── + + +def _patch_prompt_sections(monkeypatch, module) -> None: + monkeypatch.setattr(module, "_get_memory_context", lambda *a, **k: "") + monkeypatch.setattr(module, "get_skills_prompt_section", lambda *a, **k: "") + monkeypatch.setattr(module, "get_deferred_tools_prompt_section", lambda *a, **k: "") + monkeypatch.setattr(module, "_build_acp_section", lambda *a, **k: "") + monkeypatch.setattr(module, "_build_custom_mounts_section", lambda *a, **k: "") + monkeypatch.setattr(module, "_current_time_context", lambda: "") + + +def test_lead_prompt_first_turn_uses_research_tool_contract(monkeypatch) -> None: + from deerflow.agents.lead_agent import prompt as prompt_module + + _patch_prompt_sections(monkeypatch, prompt_module) + + prompt = prompt_module.apply_prompt_template(writing_mode=True) + + assert "deep_research_report" in prompt + assert "Work normally until delivery time" in prompt + assert "Do NOT ask the user whether to search for more" in prompt + assert "first report turn" in prompt + assert "do NOT call `read_file`" in prompt + assert "Current Markdown Artifact" not in prompt + + +def test_lead_prompt_follow_up_keeps_manual_edit_contract(monkeypatch) -> None: + from deerflow.agents.lead_agent import prompt as prompt_module + + _patch_prompt_sections(monkeypatch, prompt_module) + + prompt = prompt_module.apply_prompt_template( + writing_mode=True, + writing_artifact_path="/mnt/user-data/outputs/report.md", + ) + + # 续写轮保持原合约：编辑现有文件 + present_files，不再要求调用管线工具 + assert "Current Markdown Artifact" in prompt + assert "present_files" in prompt + assert "deep_research_report" not in prompt + + +def test_agent_only_prompt_mirrors_both_contracts(monkeypatch) -> None: + from deerflow.agents.lead_agent import prompt as prompt_module + + monkeypatch.setattr(prompt_module, "get_agent_soul", lambda *a, **k: "SOUL") + monkeypatch.setattr(prompt_module, "get_skills_prompt_section", lambda *a, **k: "") + monkeypatch.setattr(prompt_module, "get_deferred_tools_prompt_section", lambda *a, **k: "") + monkeypatch.setattr(prompt_module, "_build_acp_section", lambda *a, **k: "") + monkeypatch.setattr(prompt_module, "_build_custom_mounts_section", lambda *a, **k: "") + + first_turn = prompt_module.apply_agent_only_prompt_template(agent_id="a1", agent_name="A1", writing_mode=True) + follow_up = prompt_module.apply_agent_only_prompt_template( + agent_id="a1", + agent_name="A1", + writing_mode=True, + writing_artifact_path="/mnt/user-data/outputs/report.md", + ) + + assert "deep_research_report" in first_turn + assert "Work normally (chat, search, clarify) until the report is due" in first_turn + assert "Do NOT ask the user whether to search for more" in first_turn + assert "do NOT call `read_file`" in first_turn + assert "Current Markdown Artifact" in follow_up + assert "deep_research_report" not in follow_up