Merge branch 'ji' into 'master'

fix(ai写作): 弱模型也能真正执行技能脚本(执行优先的强提示)

See merge request knowledge-base-bj/deerflow-server!152

offline-backend-20260512/backend/config.yaml

@@ -440,6 +440,10 @@ ai_writing:
   # 接口契约：POST {url} 体 {"query","knowledge_base_name","ts":"yes"} → [{"page_content","metadata":{m_title,content,recUuid,m_publish,source1}}]
   intranet_search_url: "" # 例如 http://10.106.57.180:7863/knowledge_base/search_docs_es
   intranet_knowledge_base: "samples" # 兜底检索用的知识库名
+  # 内网接口诊断开关：开启后即使技能已检索成功，也照样把内网接口跑完并把执行结果
+  # （端点/每个检索词命中数/报错/原始响应片段）记到 material_package.intranet_debug 里，
+  # 便于在接口返回(devtools)排查内网接口到底通不通。生产建议关掉（会多跑一次内网）。
+  intranet_search_debug: false
   # 四阶段「内置功能型智能体」开关（按阶段灰度）。true = 该阶段节点改为加载
   # .deer-flow/agents/ai-writing-<stage>/ 的 SOUL.md + config（admin 可在智能体
   # 管理页编辑人设/模型），false / 缺失 = 走旧的硬编码 prompt 路径。

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

@@ -58,15 +58,23 @@ _REAL_AGENT_MAX_TURNS = 16
 _REAL_AGENT_TIMEOUT_SECONDS = 300
 
 _MATERIAL_OUTPUT_CONTRACT = (
-    "你的任务：严格按上面挂载的检索技能（SKILL.md）的说明，用相应工具检索与「写作意图」相关的**真实素材**。\n"
+    "你的任务：用上面挂载的检索技能（<skill> 里的 SKILL.md）按它的说明**实际执行检索**，"
-    "- 必须按技能说明实际调用工具完成检索（读取知识库文件 / 调用内网检索接口 / MCP 等），**不得凭记忆编造**。\n"
+    "收集与「写作意图」相关的真实素材。\n"
-    "- **某个检索词返回 0 条或报「请求超时」时，不要就此放弃**：请把它换成更核心、更简短的检索词再试一次"
+    "\n"
-    "（去掉「竞选/诉讼/政策/动向」等话题后缀，只保留核心实体，例如「特朗普竞选」→「特朗普」、"
+    "【第一步 · 必须真正执行技能，绝不能跳过】\n"
-    "「特朗普法律诉讼」→「特朗普」；多词时只留最关键的那个词）。\n"
+    "- 你**必须先实际调用工具去检索**，再谈输出。技能通常自带脚本——按 SKILL.md 的说明用 **bash** 运行它，"
-    "- **只要有任何检索词成功拿到了数据，就必须把这些数据全部收下并输出**，绝不能因为别的检索词没出结果/超时就丢弃已经检索到的素材。\n"
+    "例如：`cd <技能目录> && python3 scripts/xxx.py \"检索词\"`（具体脚本名/参数以 SKILL.md 为准），"
-    "- 检索完成后，**只输出一个 JSON 对象**（不要解释、不要 markdown 代码块、用英文双引号），把检索到的真实素材列出：\n"
+    "或调用它说明里的内网接口 / MCP 工具。\n"
+    "- **严禁不调用任何工具就直接输出 JSON**，更不许凭记忆编造素材。哪怕你“觉得”知道答案，也必须真的去检索。\n"
+    "- 对每个检索方向各执行一次；可多次调用工具覆盖不同角度。\n"
+    "- 某个检索词返回 0 条或报「请求超时」时**不要放弃**：换成更核心、更短的检索词再试一次"
+    "（去掉「竞选/诉讼/政策/动向」等后缀，只留核心实体，如「特朗普竞选」→「特朗普」）。\n"
+    "\n"
+    "【第二步 · 检索完成后再整理输出】\n"
+    "- 把工具**实际返回**的真实素材整理成一个 JSON 对象（不要解释、不要 markdown 代码块、用英文双引号）：\n"
     '{"materials": [{"title": "标题", "content": "素材正文摘录(尽量保留原文, <=500字)", "source": "来源", "url": "原文链接(可空)", "published_at": "日期(可空)"}]}\n'
-    "- content 必须来自检索到的真实内容；只有当**所有**检索词（含简化后重试）都没拿到任何素材时，才输出 {\"materials\": []}。"
+    "- content 必须来自工具实际返回的内容；只要有任何检索词拿到数据就**全部收下**，不要因别的检索词没结果而丢弃。\n"
+    "- 只有当你**确实执行了检索、且所有检索词（含简化重试）都没有结果**时，才输出 {\"materials\": []}。"
 )
 
 
@@ -318,6 +326,12 @@ async def run_skill_via_real_agent(
     if keywords:
         task_lines.append(f"可参考的检索方向：{', '.join(keywords[:8])}")
     task_lines.append(f"目标素材数量约 {max_results} 条。")
+    # 弱模型强提示：明确要求「先动手执行技能脚本」，避免它直接吐空 JSON 不检索。
+    task_lines.append(
+        f"现在请**立刻开始执行检索**：按上面挂载技能的 SKILL.md，用 bash 运行它的脚本"
+        f"（如 `cd {skill_dir or '<技能目录>'} && python3 scripts/...py \"检索词\"`）实际检索，"
+        "先执行工具拿到真实数据，再整理输出 JSON；不要不检索就直接输出。"
+    )
     task = "\n".join(task_lines)
 
     sub_cfg = SubagentConfig(

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

@@ -554,6 +554,9 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
     # 本轮新检索到的原始结果（统一在下方合并段转 Material、与旧素材去重）
     new_results: list[SearchResult] = []
     provider = None
+    # 内网检索接口执行诊断（排查用，最终挂到 material_package["intranet_debug"]，前端忽略）。
+    # 记录：是否配置 / 端点 / 每个检索词命中数与报错 / 原始响应片段 / 是否被采用为兜底。
+    intranet_debug: dict | None = None
 
     if material_source == "notebook" and notebook_search_space_id:
         # ── 纯笔记模式 ──
@@ -772,12 +775,27 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
         # 预跑好的内网结果兜底（省去"技能全失败后再串行等一遍内网"的时间）。需在 config.yaml
         # 的 ai_writing.intranet_search_url 配地址才启用；技能拿到结果则取消这个任务、不浪费。
         intranet_url = (ai_writing_config.get("intranet_search_url") or "").strip()
+        # 诊断开关：开启后即使技能成功也照样跑完内网并记录完整结果（用于验证内网接口是否真能通）。
+        intranet_debug_force = bool(ai_writing_config.get("intranet_search_debug", False))
         intranet_task: "asyncio.Task | None" = None
         intranet_holder: dict = {}
+        intranet_debug = {
+            "configured": bool(intranet_url),
+            "endpoint": intranet_url or "(未配置 ai_writing.intranet_search_url)",
+            "knowledge_base": ai_writing_config.get("intranet_knowledge_base", "samples"),
+            "timeout": ai_writing_config.get("intranet_search_timeout", 600),
+            "debug_force": intranet_debug_force,
+            "status": "pending" if intranet_url else "not_configured",
+            "queries": [],          # 每个检索词：{query, count, error?}
+            "errors": [],           # provider 聚合的报错
+            "raw_responses": [],    # 解析失败时的原始响应片段
+            "result_count": 0,      # 内网总命中条数
+            "used_as_fallback": False,  # 内网结果是否被真正采用为素材
+        }
 
         async def _run_intranet_search() -> list[SearchResult]:
             # 全程吞异常：并行内网任务**绝不**能把主流程带挂（实测内网接口可能瞬间失败）。
-            # 失败只记 search_errors + 日志，返回空列表，交由技能结果或空结果文案兜底。
+            # 失败只记 search_errors + 诊断 + 日志，返回空列表，交由技能结果或空结果文案兜底。
             try:
                 from ..search.intranet_stub import IntranetSearchProvider
 
@@ -795,31 +813,56 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                     return_exceptions=True,
                 )
                 collected: list[SearchResult] = []
-                for res in gathered:
+                for kw, res in zip(fb_keywords, gathered):
                     if isinstance(res, Exception):
-                        logger.warning("[intranet-parallel] one keyword failed: %s", res)
+                        logger.warning("[intranet-parallel] keyword=%r failed: %s", kw, res)
                         search_errors.append(f"{type(res).__name__}: {res}")
+                        intranet_debug["queries"].append({"query": kw, "count": 0, "error": f"{type(res).__name__}: {res}"})
                         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["result_count"] = len(collected)
+                intranet_debug["status"] = "ran"
                 return collected
             except asyncio.CancelledError:
+                intranet_debug["status"] = "cancelled"
                 raise  # 取消要正常传播（技能已拿到结果时主动取消）
             except Exception as e:  # noqa: BLE001
                 logger.warning("[intranet-parallel] intranet fallback crashed: %s", e, exc_info=True)
                 search_errors.append(f"内网兜底异常：{type(e).__name__}: {e}")
+                intranet_debug["status"] = "crashed"
+                intranet_debug["errors"].append(f"{type(e).__name__}: {e}")
                 return []
 
         if intranet_url:
             intranet_task = asyncio.create_task(_run_intranet_search())
 
         async def _cancel_intranet_task() -> None:
-            """技能已拿到素材：取消并行的内网检索任务，避免无谓占用 / 未取回异常告警。"""
+            """技能已拿到素材：默认取消并行的内网检索任务，避免无谓占用。
-            if intranet_task is not None and not intranet_task.done():
+
-                intranet_task.cancel()
+            诊断模式（intranet_search_debug=true）下**不取消**，而是等它跑完拿到完整诊断，
+            这样即使技能成功也能在 material_package.intranet_debug 里看到内网接口的真实返回。
+            """
+            if intranet_task is None or intranet_task.done():
+                if intranet_task is not None:
+                    try:
+                        await intranet_task
+                    except BaseException:  # noqa: BLE001
+                        pass
+                return
+            if intranet_debug_force:
                 try:
                     await intranet_task
-                except BaseException:  # noqa: BLE001 — CancelledError / 任务内异常都吞掉
+                except BaseException:  # noqa: BLE001
                     pass
+                return
+            intranet_task.cancel()
+            try:
+                await intranet_task
+            except BaseException:  # noqa: BLE001 — CancelledError / 任务内异常都吞掉
+                pass
 
         for sk in ordered:
             added, error_detail = await _execute_skill(sk)
@@ -871,6 +914,8 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
                     fb_results = []
                     search_errors.append(f"{type(e).__name__}: {e}")
                 fb_added = _collect(fb_results)
+                if intranet_debug is not None:
+                    intranet_debug["used_as_fallback"] = True
                 intranet = intranet_holder.get("provider")
                 if intranet is not None:
                     search_errors.extend(getattr(intranet, "errors", []))
@@ -1004,6 +1049,11 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
         pkg["skill_search_used"] = existing_pkg.get("skill_search_used", False)
         pkg["skill_search_steps"] = existing_pkg.get("skill_search_steps", [])
 
+    # 内网检索接口执行诊断：挂进 material_package，随 values 事件回前端（前端忽略此字段，
+    # 仅供在浏览器 devtools / 接口返回里排查内网接口到底有没有被调、返回/报错是什么）。
+    if intranet_debug is not None:
+        pkg["intranet_debug"] = intranet_debug
+
     # 容错告警：检索接口 / 模型调用出错时以非致命 notice 提示用户，不打断写作流程
     notices: list[dict] = []
     if kw_result.get("_llm_error"):