Add intranet retrieval diagnostics for AI writing

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py

@@ -115,6 +115,31 @@ def _fallback_knowledge_skill() -> SimpleNamespace:
     return SimpleNamespace(name=KNOWLEDGE_SEARCH_SKILL_NAME, description=_KNOWLEDGE_SKILL_FALLBACK_DESC)
 
 
+def _unique_limited(values: list, limit: int) -> list:
+    """Stable de-dup for short diagnostic lists."""
+    return list(dict.fromkeys(values or []))[:limit]
+
+
+def _intranet_query_summary(rows: list[dict], limit: int = 5) -> str:
+    """Compact per-query stats for the visible skill step message."""
+    parts: list[str] = []
+    for row in rows[:limit]:
+        query = str(row.get("query") or "")[:24] or "(空)"
+        elapsed_ms = row.get("elapsed_ms")
+        try:
+            elapsed = f"{float(elapsed_ms) / 1000:.1f}s" if elapsed_ms is not None else ""
+        except (TypeError, ValueError):
+            elapsed = ""
+        if row.get("error"):
+            reason = str(row.get("error"))[:80]
+            parts.append(f"{query}: 失败{f'/{elapsed}' if elapsed else ''}（{reason}）")
+        else:
+            parts.append(f"{query}: {int(row.get('count') or 0)} 条{f'/{elapsed}' if elapsed else ''}")
+    if len(rows) > limit:
+        parts.append(f"另 {len(rows) - limit} 个方向")
+    return "；".join(parts)
+
+
 def _load_configured_skills() -> list:
     """解析素材收集专家 agent（ai-writing-researcher）配置的技能列表。
 
@@ -594,6 +619,20 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             timeout=_intranet_timeout_seconds(ai_writing_config),
         )
         endpoint_label = intranet_url or getattr(provider, "endpoint", "内置默认内网端点")
+        intranet_debug = {
+            "mode": "knowledge_base",
+            "configured": bool(intranet_url),
+            "endpoint": endpoint_label,
+            "knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
+            "timeout": _intranet_timeout_seconds(ai_writing_config),
+            "status": "running",
+            "queries": [],
+            "calls": [],
+            "errors": [],
+            "raw_responses": [],
+            "result_count": 0,
+            "used_as_fallback": False,
+        }
         # 用户输入框追加（is_user_query_search）时 keywords 已是 [用户原话]，全部检索；
         # 首次检索按 keyword_count 截断检索词。
         search_keywords = keywords if is_user_query_search else keywords[:keyword_count]
@@ -605,20 +644,65 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             "message": f"使用「{_KNOWLEDGE_SKILL_DISPLAY}」直连内网接口（{endpoint_label}）检索 {total} 个方向：{('、'.join(search_keywords[:3]))[:60]}",
             "skill_name": KNOWLEDGE_SEARCH_SKILL_NAME,
         })
+
+        async def _search_one_keyword(kw: str) -> tuple[str, list[SearchResult], str, int]:
+            started = time.monotonic()
+            try:
+                results = await provider.search(kw, max_results=per_keyword)
+                return kw, list(results or []), "", int((time.monotonic() - started) * 1000)
+            except Exception as e:  # noqa: BLE001 — provider 理论上会吞异常，这里兜住意外冒泡
+                logger.warning("[knowledge-base] keyword=%r search failed: %s", kw, e)
+                return kw, [], f"{type(e).__name__}: {e}", int((time.monotonic() - started) * 1000)
+
         gathered = await asyncio.gather(
-            *[provider.search(kw, max_results=per_keyword) for kw in search_keywords],
+            *[_search_one_keyword(kw) for kw in search_keywords],
             return_exceptions=True,
         )
-        for results in gathered:
-            if isinstance(results, Exception):
-                logger.warning("[knowledge-base] search failed: %s", results)
-                search_errors.append(f"{type(results).__name__}: {results}")
+        for item in gathered:
+            if isinstance(item, Exception):
+                logger.warning("[knowledge-base] search task failed: %s", item)
+                message = f"{type(item).__name__}: {item}"
+                search_errors.append(message)
+                intranet_debug["queries"].append({"query": "(unknown)", "count": 0, "error": message})
                 continue
+            kw, results, error, elapsed_ms = item
+            row = {"query": kw, "count": len(results or []), "elapsed_ms": elapsed_ms}
+            if error:
+                row["error"] = error
+                search_errors.append(error)
+            intranet_debug["queries"].append(row)
             new_results.extend(results or [])
-        search_errors.extend(getattr(provider, "errors", []))
+        provider_errors = _unique_limited(getattr(provider, "errors", []) or [], 5)
+        search_errors.extend(provider_errors)
+        intranet_debug["errors"] = provider_errors
+        intranet_debug["raw_responses"] = _unique_limited(getattr(provider, "raw_responses", []) or [], 3)
+        intranet_debug["calls"] = list(getattr(provider, "calls", []) or [])[:20]
+        intranet_debug["result_count"] = len(new_results)
+        if search_errors and new_results:
+            intranet_debug["status"] = "partial"
+        elif search_errors:
+            intranet_debug["status"] = "error"
+        elif new_results:
+            intranet_debug["status"] = "ok"
+        else:
+            intranet_debug["status"] = "empty"
+
+        query_summary = _intranet_query_summary(intranet_debug["queries"])
+        if search_errors:
+            uniq_errors = _unique_limited(search_errors, 3)
+            _emit_skill_step({
+                "type": "skill_error",
+                "message": f"「{_KNOWLEDGE_SKILL_DISPLAY}」内网接口调用异常：{'；'.join(uniq_errors)}",
+                "skill_name": KNOWLEDGE_SEARCH_SKILL_NAME,
+                "reason": "；".join(uniq_errors),
+            })
+        result_message = (
+            f"「{_KNOWLEDGE_SKILL_DISPLAY}」（内网接口）调用完成，检索到 {len(new_results)} 条素材"
+            + (f"（{query_summary}）" if query_summary else "")
+        )
         _emit_skill_step({
             "type": "skill_result",
-            "message": f"「{_KNOWLEDGE_SKILL_DISPLAY}」（内网接口）检索到 {len(new_results)} 条素材",
+            "message": result_message,
             "skill_name": KNOWLEDGE_SEARCH_SKILL_NAME,
             "matched_count": len(new_results),
         })
@@ -846,6 +930,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             "debug_force": intranet_debug_force,
             "status": "pending" if intranet_url else "not_configured",
             "queries": [],          # 每个检索词：{query, count, error?}
+            "calls": [],            # provider 每次真实 HTTP 调用，含简化词重试
             "errors": [],           # provider 聚合的报错
             "raw_responses": [],    # 解析失败时的原始响应片段
             "result_count": 0,      # 内网总命中条数
@@ -880,8 +965,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                         continue
                     intranet_debug["queries"].append({"query": kw, "count": len(res or [])})
                     collected.extend(res or [])
-                intranet_debug["errors"] = list(dict.fromkeys(getattr(intranet, "errors", [])))[:5]
-                intranet_debug["raw_responses"] = list(dict.fromkeys(getattr(intranet, "raw_responses", [])))[:3]
+                intranet_debug["errors"] = _unique_limited(getattr(intranet, "errors", []) or [], 5)
+                intranet_debug["raw_responses"] = _unique_limited(getattr(intranet, "raw_responses", []) or [], 3)
+                intranet_debug["calls"] = list(getattr(intranet, "calls", []) or [])[:20]
                 intranet_debug["result_count"] = len(collected)
                 intranet_debug["status"] = "ran"
                 return collected

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py

@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+import time
 from typing import List
 
 from .base import BaseSearchProvider, SearchResult, simplify_query
@@ -38,6 +39,16 @@ def _pick(d: dict, keys: tuple) -> str:
     return ""
 
 
+def _raw_snippet(body: str) -> str:
+    """Return a bounded raw response snippet for diagnostics."""
+    body = (body or "").strip()
+    if not body:
+        return ""
+    if len(body) > _RAW_SNIPPET_LIMIT:
+        return body[:_RAW_SNIPPET_LIMIT] + "…（响应体过长，已截断）"
+    return body
+
+
 class IntranetSearchProvider(BaseSearchProvider):
     """内网知识库搜索接口适配器。
 
@@ -61,15 +72,15 @@ class IntranetSearchProvider(BaseSearchProvider):
         self.errors: list[str] = []
         # 解析失败时收集的原始响应体，供前端「错误返回结果」折叠展示
         self.raw_responses: list[str] = []
+        # 每一次真实 HTTP 调用的诊断明细。search() 可能会因 0 结果触发简化词重试，
+        # 因此一个业务关键词可能对应多条 calls。
+        self.calls: list[dict] = []
 
     def _record_raw(self, body: str) -> None:
         """收集解析失败 / 结构异常时的原始响应体，供前端折叠查看。"""
-        body = (body or "").strip()
-        if not body:
-            return
-        if len(body) > _RAW_SNIPPET_LIMIT:
-            body = body[:_RAW_SNIPPET_LIMIT] + "…（响应体过长，已截断）"
-        self.raw_responses.append(body)
+        snippet = _raw_snippet(body)
+        if snippet:
+            self.raw_responses.append(snippet)
 
     async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
         results = await self._search_once(query, max_results)
@@ -92,40 +103,87 @@ class IntranetSearchProvider(BaseSearchProvider):
         }
         # 先拿到原始响应文本，JSON 解析失败时把它原样回传前端便于定位接口返回了什么
         raw_text = ""
+        started = time.monotonic()
+        call: dict = {
+            "query": query,
+            "endpoint": self.endpoint,
+            "knowledge_base_name": self.knowledge_base_name,
+            "max_results": max_results,
+            "status": "pending",
+            "ok": False,
+            "count": 0,
+        }
         try:
             async with httpx.AsyncClient(timeout=self.timeout) as client:
                 resp = await client.post(self.endpoint, json=payload)
+                call["http_status"] = getattr(resp, "status_code", None)
                 resp.raise_for_status()
                 raw_text = resp.text
                 data = resp.json()
         except httpx.HTTPStatusError as e:
             # 非 2xx：接口有响应但状态码异常，把响应体一并暴露给前端
             logger.warning("intranet search failed for query=%r: %s", query, e)
-            self.errors.append(f"{type(e).__name__}: {e}")
-            self._record_raw(e.response.text if e.response is not None else "")
+            message = f"{type(e).__name__}: {e}"
+            self.errors.append(message)
+            body = e.response.text if e.response is not None else ""
+            call.update({
+                "status": "http_error",
+                "error": message,
+                "http_status": getattr(e.response, "status_code", None),
+            })
+            snippet = _raw_snippet(body)
+            if snippet:
+                call["raw_response"] = snippet
+            self._record_raw(body)
             return []
         except Exception as e:
             # 连接失败 / 超时 / JSON 解析失败等；JSON 解析失败时 raw_text 已是接口响应体
             logger.warning("intranet search failed for query=%r: %s", query, e)
-            self.errors.append(f"{type(e).__name__}: {e}")
+            message = f"{type(e).__name__}: {e}"
+            self.errors.append(message)
+            call.update({"status": "error", "error": message})
+            snippet = _raw_snippet(raw_text)
+            if snippet:
+                call["raw_response"] = snippet
             self._record_raw(raw_text)
             return []
+        finally:
+            call["elapsed_ms"] = int((time.monotonic() - started) * 1000)
+            # 非 pending 表示已经在异常分支写好；成功分支会在 return 前覆盖并追加。
+            if call.get("status") != "pending":
+                self.calls.append(call)
 
         # 正常情况下接口直接返回数组；兼容包了一层 data/results 的结构（含工作正常的
         # knowledge-search-v2：{"success":true,"results":[...]}）。
         if isinstance(data, list):
             items = data
         elif isinstance(data, dict):
+            if "success" in data:
+                call["response_success"] = bool(data.get("success"))
+            if "count" in data:
+                call["response_count"] = data.get("count")
             # 接口显式标失败时把整体回传前端便于定位
             if data.get("success") is False:
-                self.errors.append(f"接口返回 success=false：{str(data.get('error') or '')[:200]}")
+                message = f"接口返回 success=false：{str(data.get('error') or '')[:200]}"
+                self.errors.append(message)
+                call.update({"status": "success_false", "error": message})
+                snippet = _raw_snippet(raw_text)
+                if snippet:
+                    call["raw_response"] = snippet
+                self.calls.append(call)
                 self._record_raw(raw_text)
                 return []
             items = data.get("data") or data.get("results") or data.get("documents") or []
         else:
             # 返回结构既不是数组也不是已知 dict 包装：视为异常，把原始响应回传前端
             logger.warning("intranet search unexpected response type for query=%r: %s", query, type(data).__name__)
-            self.errors.append(f"返回结构异常：期望数组，实际为 {type(data).__name__}")
+            message = f"返回结构异常：期望数组，实际为 {type(data).__name__}"
+            self.errors.append(message)
+            call.update({"status": "bad_shape", "error": message})
+            snippet = _raw_snippet(raw_text)
+            if snippet:
+                call["raw_response"] = snippet
+            self.calls.append(call)
             self._record_raw(raw_text)
             return []
 
@@ -158,6 +216,17 @@ class IntranetSearchProvider(BaseSearchProvider):
                 source_type="intranet",
                 recUuid=_pick(merged, ("recUuid", "rec_uuid", "id")),
             ))
+        call.update({
+            "status": "ok" if results else "empty",
+            "ok": True,
+            "count": len(results),
+            "item_count": len(items) if isinstance(items, list) else 0,
+        })
+        if not results:
+            snippet = _raw_snippet(raw_text)
+            if snippet:
+                call["raw_response"] = snippet
+        self.calls.append(call)
         return results
 
     async def health_check(self) -> bool:

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/state.py

@@ -40,6 +40,9 @@ class MaterialPackage(TypedDict, total=False):
     skill_search_used: bool
     # 技能检索的步骤记录（前端展示"使用了 X 技能"卡片）
     skill_search_steps: List[SkillStep]
+    # 内网知识库接口诊断：端点、每个检索词的命中数/耗时/错误、原始响应片段等。
+    # 用于排查 "检索很快返回 0 条" 时到底是否真正调用了内网接口。
+    intranet_debug: dict
 
 # ── 大纲 ──────────────────────────────────────────────────────────────────────
 # 三级结构:章节(##) → 小节(###) → 子要点(bullet -)。

offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py

@@ -19,16 +19,33 @@ class _FakeIntranet:
     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 = []
@@ -38,7 +55,7 @@ class _SkillProviderShouldNotRun:
         raise AssertionError("knowledge_base 模式不应调用通用 provider / 技能检索")
 
 
-def _patch(monkeypatch, intranet_url="http://es/api"):
+def _patch(monkeypatch, intranet_url="http://es/api", intranet_cls=_FakeIntranet):
     _FakeIntranet.instances = []
     cfg = {
         "max_materials": 20,
@@ -60,7 +77,7 @@ def _patch(monkeypatch, intranet_url="http://es/api"):
     monkeypatch.setattr(R, "get_search_provider", lambda _cfg: _SkillProviderShouldNotRun())
     monkeypatch.setattr(
         "deerflow.agents.ai_writing.search.intranet_stub.IntranetSearchProvider",
-        _FakeIntranet,
+        intranet_cls,
     )
 
 
@@ -85,6 +102,14 @@ async def test_knowledge_base_hits_intranet_directly(monkeypatch):
     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
@@ -107,3 +132,34 @@ async def test_knowledge_base_does_not_orchestrate_skills(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"])