"""researcher_node：检索类型 `knowledge_base`（知识库检索）= 直连内网 ES 接口。 弱模型识别不了技能时用这条路：不编排任何技能、不跑通用 provider， 直接调 ai_writing.intranet_search_url 内网知识库 ES 接口检索。 """ from __future__ import annotations from types import SimpleNamespace import pytest from deerflow.agents.ai_writing.nodes import researcher as R from deerflow.agents.ai_writing.search.base import SearchResult class _FakeIntranet: instances: list = [] def __init__(self, **kw): self.errors = [] self.raw_responses = [] self.calls = [] self.kw = kw _FakeIntranet.instances.append(self) async def search(self, query, max_results=5, **_): self.calls.append({"query": query, "status": "ok", "ok": True, "count": 1, "elapsed_ms": 12}) return [SearchResult( title=f"内网-{query}", content="内网正文", source="ES", published_at="", url=f"http://es/{query}", source_type="intranet", )] class _EmptyIntranet(_FakeIntranet): async def search(self, query, max_results=5, **_): self.calls.append({"query": query, "status": "empty", "ok": True, "count": 0, "elapsed_ms": 9}) return [] class _FailingIntranet(_FakeIntranet): async def search(self, query, max_results=5, **_): msg = f"ConnectError: cannot connect for {query}" self.errors.append(msg) self.raw_responses.append("{\"success\": false, \"error\": \"cannot connect\"}") self.calls.append({"query": query, "status": "error", "ok": False, "count": 0, "elapsed_ms": 3, "error": msg}) return [] class _SkillProviderShouldNotRun: """knowledge_base 模式不应触碰通用 provider —— 调到就让测试爆。""" errors: list = [] raw_responses: list = [] async def search(self, query, max_results=5, **_): # pragma: no cover - 不应被调用 raise AssertionError("knowledge_base 模式不应调用通用 provider / 技能检索") def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet): _FakeIntranet.instances = [] cfg = { "max_materials": 20, "intranet_search_url": intranet_url, "intranet_knowledge_base": "samples", } monkeypatch.setattr(R, "get_ai_writing_config", lambda: cfg) monkeypatch.setattr(R, "use_builtin_agent", lambda *_: False) monkeypatch.setattr(R, "safe_get_stream_writer", lambda: (lambda *a, **k: None)) monkeypatch.setattr(R, "use_skill_search_agent", lambda: False) async def fake_llm_json(*a, **k): return {"keywords": ["特朗普竞选"]} monkeypatch.setattr(R, "llm_json", fake_llm_json) # knowledge_base 模式不该加载/编排技能，但仍打桩以防意外路径 monkeypatch.setattr(R, "_load_configured_skills", lambda: [SimpleNamespace(name="knowledge-search-v2", description="内网检索")]) monkeypatch.setattr(R, "get_search_provider", lambda _cfg: _SkillProviderShouldNotRun()) monkeypatch.setattr( "deerflow.agents.ai_writing.search.intranet_stub.IntranetSearchProvider", intranet_cls, ) def _state(material_source="knowledge_base"): return { "user_intent": "分析特朗普近期动向", "article_type": "政治分析", "target_audience": "普通读者", "keyword_count": 1, "material_source": material_source, } @pytest.mark.asyncio async def test_knowledge_base_hits_intranet_directly(monkeypatch): _patch(monkeypatch) out = await R.researcher_node(_state(), {"configurable": {}}) mats = out["material_package"]["materials"] assert mats, "知识库检索应返回内网素材" assert all(m["source_type"] == "intranet" for m in mats) assert mats[0]["title"] == "内网-特朗普竞选" assert _FakeIntranet.instances, "应实例化 IntranetSearchProvider 直连内网接口" # 走素材确认暂停点 assert out["status"] == "awaiting_material_confirm" dbg = out["material_package"].get("intranet_debug") assert dbg["mode"] == "knowledge_base" assert dbg["status"] == "ok" assert dbg["endpoint"] == "http://es/api" assert dbg["result_count"] == 1 assert dbg["queries"][0]["query"] == "特朗普竞选" assert dbg["queries"][0]["count"] == 1 assert dbg["calls"][0]["status"] == "ok" @pytest.mark.asyncio async def test_knowledge_base_no_url_falls_back_to_default_endpoint(monkeypatch): """未配置接口地址：仍实例化 IntranetSearchProvider（base_url 传空串，由 provider 内部 回退默认内网端点），内网直连依然可用——恢复旧版 search_provider=intranet 的行为。""" _patch(monkeypatch, intranet_url="") out = await R.researcher_node(_state(), {"configurable": {}}) mats = out["material_package"]["materials"] assert mats, "无地址时应回退默认内网端点，仍返回素材" assert all(m["source_type"] == "intranet" for m in mats) assert _FakeIntranet.instances, "应实例化 IntranetSearchProvider 直连内网接口" assert _FakeIntranet.instances[0].kw.get("base_url") == "", "空地址应原样下传，由 provider 回退默认端点" @pytest.mark.asyncio async def test_knowledge_base_does_not_orchestrate_skills(monkeypatch): """knowledge_base 不编排技能：通用 provider 一旦被调用就会 raise。""" _patch(monkeypatch) # 不抛异常即证明没走 _SkillProviderShouldNotRun out = await R.researcher_node(_state(), {"configurable": {}}) assert out["material_package"]["materials"] @pytest.mark.asyncio async def test_knowledge_base_empty_result_is_visible(monkeypatch): """接口成功但 0 条也要在返回里能看出来，而不是只显示"无素材"。""" _patch(monkeypatch, intranet_cls=_EmptyIntranet) out = await R.researcher_node(_state(), {"configurable": {}}) pkg = out["material_package"] assert pkg["materials"] == [] dbg = pkg.get("intranet_debug") assert dbg["status"] == "empty" assert dbg["result_count"] == 0 assert dbg["queries"][0]["count"] == 0 assert dbg["calls"][0]["status"] == "empty" steps = pkg["skill_search_steps"] assert any(step["type"] == "skill_result" and "0 条" in step["message"] for step in steps) @pytest.mark.asyncio async def test_knowledge_base_failure_is_visible(monkeypatch): """provider 吞掉异常返回 [] 时，前端仍应看到报错步骤与 notice。""" _patch(monkeypatch, intranet_cls=_FailingIntranet) out = await R.researcher_node(_state(), {"configurable": {}}) pkg = out["material_package"] dbg = pkg.get("intranet_debug") assert dbg["status"] == "error" assert dbg["errors"] assert dbg["raw_responses"] steps = pkg["skill_search_steps"] assert any(step["type"] == "skill_error" for step in steps) assert any(event["type"] == "notice" and event.get("raw") for event in out["progress_events"])