From 4fce6f9104a710e40ccbbf2a0636518606c0c66b Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Mon, 29 Jun 2026 10:21:10 +0800 Subject: [PATCH] Add intranet retrieval diagnostics for AI writing --- .../agents/ai_writing/nodes/researcher.py | 104 ++++++++++++++++-- .../agents/ai_writing/search/intranet_stub.py | 91 +++++++++++++-- .../deerflow/agents/ai_writing/state.py | 3 + .../tests/test_ai_writing_knowledge_base.py | 60 +++++++++- 4 files changed, 236 insertions(+), 22 deletions(-) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py index fea5a945..77b51ff5 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/researcher.py +++ b/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 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py index ae1ad4d7..ce3e6f2b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py +++ b/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: diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/state.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/state.py index ebceca9f..577bce95 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/state.py +++ b/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 -)。 diff --git a/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py b/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py index 02a064f5..01ed878f 100644 --- a/offline-backend-20260512/backend/tests/test_ai_writing_knowledge_base.py +++ b/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"])