fix(ai写作): 技能调用实时流式显示 + 去掉大纲拼检索词/复用检索词

- 技能调用过程改为实时流式：run_skill_via_real_agent 不再 to_thread 阻塞跑 executor.execute()，而是复用其 agent/state 构建后在图事件循环里 astream，每个工具调用（读 SKILL.md / bash / 检索）实时推一条步骤到时间线——内网慢时用户能看到技能在执行，不会以为卡死。整段 asyncio.wait_for 卡 5 分钟超时。 - 回退检索词相关改动（实测无用、且结合大纲后检索词变长不利按词匹配）： - intent_parser 不再顺便生成 keywords，删 precomputed_keywords 状态字段； - researcher 不再复用，每次自己现生成检索词（回到原行为）； - 检索词提示词去掉「结合大纲/{outline_section}」，删 keyword_outline_section。保留：关思考、技能真实运行时、技能目录修复、内网兜底。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-25 00:33:57 +08:00
1 parent c0305c9269
commit 8f773d3f1b
@@ -66,6 +66,22 @@ _MATERIAL_OUTPUT_CONTRACT = (
 )


+def _summarize_tool_args(args: dict) -> str:
+    """把工具调用参数压成一句简短摘要，放进步骤条文案（取最能代表本次调用的字段）。"""
+    if isinstance(args, dict):
+        for key in ("command", "query", "path", "file_path", "pattern", "name", "input", "url"):
+            v = args.get(key)
+            if isinstance(v, str) and v.strip():
+                return v.strip()[:80]
+        try:
+            import json as _json
+
+            return _json.dumps(args, ensure_ascii=False)[:80]
+        except Exception:  # noqa: BLE001
+            return str(args)[:80]
+    return str(args)[:80]
+
+
 async def run_skill_via_real_agent(
    *,
    skill,
@@ -86,14 +102,16 @@ async def run_skill_via_real_agent(
    SearchResult。返回 ``(results, error_detail)``：results 为空且 error 非空 = 该技能失败
    （researcher 据此回退直连检索）。
    """
+    from langchain_core.messages import AIMessage
+
    from deerflow.config import get_app_config
    from deerflow.subagents.config import SubagentConfig
-    from deerflow.subagents.executor import SubagentExecutor, SubagentStatus
+    from deerflow.subagents.executor import SubagentExecutor
    from deerflow.tools import get_available_tools

    from ..search.base import SearchResult
    from ._agent_runner import RESEARCHER_AGENT_ID, WritingAgentUnavailable, load_writing_agent
-    from ._utils import parse_json
+    from ._utils import _content_to_text, parse_json

    skill_name = getattr(skill, "name", "") or "技能"

@@ -156,34 +174,68 @@ async def run_skill_via_real_agent(
        timeout_seconds=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS),
    )

+    # 复用 SubagentExecutor 的 agent/state 构建（模型 + 中间件 + 沙箱 + 技能注入），但**不**走
+    # 它阻塞式的 execute()——而是直接在本图的事件循环里 astream，边跑边把每个工具调用实时推到
+    # 步骤条。内网慢时用户能看到「技能正在调 bash/检索…」的进度，而不是干等一段冻住的空白。
    thread_id = (config.get("configurable") or {}).get("thread_id")
    try:
        executor = SubagentExecutor(
            config=sub_cfg, tools=tools, app_config=app_config,
            parent_model=model_name, thread_id=thread_id,
        )
+        agent = executor._create_agent()
+        init_state = await executor._build_initial_state(task)
    except Exception as e:  # noqa: BLE001
-        logger.warning("skill real-agent[%s]: executor init failed: %s", skill_name, e)
+        logger.warning("skill real-agent[%s]: init failed: %s", skill_name, e)
        return [], f"运行时初始化失败: {e}"

-    # 步骤条「轻量对齐」别的智能体：读取 SKILL.md → 执行检索 两段（结果段由 researcher
-    # 的 _emit_skill_outcome 以 skill_result 收尾）。这两步在图的事件循环里同步发出，
-    # 写到时间线没问题；技能真正执行在下面的线程里跑。
-    if emit_query:
-        emit_query(f"读取技能「{skill_name}」的 SKILL.md…")
-        emit_query(f"技能「{skill_name}」按 SKILL.md 说明执行检索…")
+    run_config: dict = {"recursion_limit": _REAL_AGENT_MAX_TURNS}
+    run_context: dict = {}
+    if thread_id:
+        run_config["configurable"] = {"thread_id": thread_id}
+        run_context["thread_id"] = thread_id
+    if app_config is not None:
+        run_context["app_config"] = app_config
+
+    # 首步：技能开始（SKILL.md 已注入，给一条起始进度）
+    if emit_query:
+        emit_query(f"读取技能「{skill_name}」的 SKILL.md，开始执行检索…")
+
+    final_text = ""
+    seen_msg_ids: set[str] = set()
+
+    async def _drive() -> None:
+        nonlocal final_text
+        async for chunk in agent.astream(
+            init_state, config=run_config, context=run_context, stream_mode="values"
+        ):
+            msgs = chunk.get("messages") or []
+            for m in msgs:
+                mid = getattr(m, "id", None)
+                if not mid or mid in seen_msg_ids:
+                    continue
+                seen_msg_ids.add(mid)
+                # 每个新的工具调用 → 实时推一条步骤（让用户看到技能在执行，而不是冻住）
+                for tc in (getattr(m, "tool_calls", None) or []):
+                    if not emit_query:
+                        break
+                    tname = (tc.get("name") or "").strip() or "工具"
+                    emit_query(f"技能「{skill_name}」调用 {tname}：{_summarize_tool_args(tc.get('args') or {})}")
+                # 记录最后一条 AIMessage 正文（含最终 JSON 素材）
+                if isinstance(m, AIMessage):
+                    text = _content_to_text(m.content)
+                    if text.strip():
+                        final_text = text

-    # execute() 内部用「沙箱/MCP 友好」的隔离事件循环跑；放进线程，避免阻塞图的事件循环。
    try:
-        result = await asyncio.to_thread(executor.execute, task)
+        await asyncio.wait_for(_drive(), timeout=max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS))
+    except asyncio.TimeoutError:
+        return [], f"技能检索超时（>{max(timeout_s, _REAL_AGENT_TIMEOUT_SECONDS)}s）"
    except Exception as e:  # noqa: BLE001
-        logger.warning("skill real-agent[%s]: execute failed: %s", skill_name, e, exc_info=True)
+        logger.warning("skill real-agent[%s]: astream failed: %s", skill_name, e, exc_info=True)
        return [], f"技能检索失败: {e}"

-    if result.status != SubagentStatus.COMPLETED:
-        return [], result.error or f"技能检索未完成（{result.status.value}）"
-
-    parsed = parse_json(result.result or "", {"materials": []})
+    parsed = parse_json(final_text or "", {"materials": []})
    raw_materials = parsed.get("materials")
    if not isinstance(raw_materials, list):
        return [], ""
@@ -40,26 +40,6 @@ def get_ai_writing_config() -> dict:
        return {}


-# 用户自带大纲时，检索词生成的大纲提示片段做长度截断，防止 token 溢出。
-OUTLINE_KEYWORD_SNIPPET_CHARS = 1000
-
-
-def keyword_outline_section(state: dict) -> str:
-    """用户自带大纲时，给检索词生成附上大纲原文，让关键词覆盖各章节主题。
-
-    意图解析（intent_parser）与素材收集（researcher）的检索词生成共用，无大纲返回
-    空串（提示词里该行变空行，行为与原来一致）。
-    """
-    raw = (state.get("user_outline_raw") or "").strip()
-    if not raw:
-        return ""
-    snippet = raw[:OUTLINE_KEYWORD_SNIPPET_CHARS]
-    return (
-        "用户已提供文章大纲，请结合写作意图，让检索词覆盖大纲涉及的各章节主题：\n"
-        f"{snippet}"
-    )
-
-
 def safe_get_stream_writer():
    """返回一个 stream writer callable；在没有 LangGraph runtime 上下文时退化成 no-op。

@@ -18,7 +18,6 @@ from ._model_fallback import (
 from ._utils import (
    ThinkTagParser,
    _content_to_text,
-    keyword_outline_section,
    parse_json,
    reasoning_delta,
    safe_get_stream_writer,
@@ -76,18 +75,17 @@ _DEFAULT_ARTICLE_TYPES: list[dict] = [
 ]

 _SYSTEM_TEMPLATE = """你是一名写作助手。
-你的任务：分析用户的写作意图，提取结构化的写作配置，并顺便提炼用于搜集素材的搜索关键词。
+你的任务：分析用户的写作意图，提取结构化的写作配置。

 【输出要求】
 - 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
 - 必须使用英文双引号 "。
 - article_type 只能是以下之一：{type_options}。
- keywords 为字符串数组：依据写作意图（及用户提供的大纲，若有）提炼的搜索关键词，要具体、可搜索，避免过于宽泛。

 【完整示例】
 输入意图：帮我写一篇面向行业专家的深度文章，分析今年新能源汽车出口为什么大涨，大概两千字
 正确输出：
-{{"article_type": "feature", "target_audience": "行业专家", "word_count_target": 2000, "refined_intent": "分析今年新能源汽车出口高速增长的原因", "keywords": ["新能源汽车出口", "汽车出口数据", "新能源汽车海外市场", "电动车出口增长"]}}"""
+{{"article_type": "feature", "target_audience": "行业专家", "word_count_target": 2000, "refined_intent": "分析今年新能源汽车出口高速增长的原因"}}"""

 _USER_TEMPLATE = """用户写作意图：{user_intent}

@@ -95,14 +93,13 @@ _USER_TEMPLATE = """用户写作意图：{user_intent}
 - 文章类型（article_type）：{preconfigured_article_type}
 - 目标读者（target_audience）：{preconfigured_audience}
 - 目标字数（word_count_target）：{preconfigured_word_count}
-{outline_section}
-请输出以下 JSON：article_type / target_audience / word_count_target 必须与上述一致；根据用户意图精炼 refined_intent；并提炼 {keyword_count} 个用于搜集素材的搜索关键词放入 keywords（数量不多不少、具体可搜索）：
+
+请输出以下 JSON，article_type / target_audience / word_count_target 必须与上述一致，只需根据用户意图精炼 refined_intent：
 {{{{
  "article_type": "{preconfigured_article_type}",
  "target_audience": "{preconfigured_audience}",
  "word_count_target": {preconfigured_word_count},
-  "refined_intent": "精炼后的写作意图（一句话）",
-  "keywords": ["检索词1", "检索词2"]
+  "refined_intent": "精炼后的写作意图（一句话）"
 }}}}"""


@@ -207,19 +204,11 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d
    if not isinstance(preconfigured_wc, int) or preconfigured_wc <= 0:
        preconfigured_wc = 800

-    # 检索词数量（用户表单指定，1~10，默认 6）；意图解析顺便按此数量生成检索词，
-    # 供 researcher 首轮直接复用，省掉素材收集阶段单独的关键词 LLM 调用。
-    raw_kc = state.get("keyword_count")
-    keyword_count = raw_kc if isinstance(raw_kc, int) and 1 <= raw_kc <= 10 else 6
-
    user = _USER_TEMPLATE.format(
        user_intent=state["user_intent"],
        preconfigured_article_type=fallback_type,
        preconfigured_audience=fallback_result["target_audience"],
        preconfigured_word_count=preconfigured_wc,
-        keyword_count=keyword_count,
-        # 用户自带大纲时附上大纲原文，让检索词覆盖各章节主题（与 researcher 共用同一逻辑）
-        outline_section=keyword_outline_section(state),
    )

    # 流式调用：与 writer_outline / writer_draft 同套机制 —— LLM 逐字吐出 JSON，
@@ -323,13 +312,4 @@ async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> d
    if final_refined_intent != state["user_intent"]:
        updates["user_intent"] = final_refined_intent

-    # 意图解析顺便生成的检索词：清洗后存入 state，researcher 首轮直接复用。
-    # 解析失败/为空时不写该字段（保持 None），researcher 回退到自己生成（优雅降级）。
-    raw_keywords = result.get("keywords")
-    if isinstance(raw_keywords, list):
-        precomputed = [s for kw in raw_keywords if (s := str(kw).strip())]
-        precomputed = list(dict.fromkeys(precomputed))
-        if precomputed:
-            updates["precomputed_keywords"] = precomputed
-
    return updates
@@ -41,7 +41,7 @@ from ._agent_runner import (
    use_builtin_agent,
 )
 from ._skill_agent import run_skill_via_real_agent, use_skill_search_agent
-from ._utils import get_ai_writing_config, keyword_outline_section, llm_json, safe_get_stream_writer
+from ._utils import get_ai_writing_config, llm_json, safe_get_stream_writer

 logger = logging.getLogger(__name__)

@@ -52,9 +52,9 @@ _AGENT_KEYWORDS_USER = """任务：提取检索词
 写作意图：{user_intent}
 文章类型：{article_type}
 目标读者：{target_audience}
-{outline_section}
+
 请提取 {keyword_count} 个搜索关键词，用于搜集相关素材。
-关键词数量必须为 {keyword_count} 个，不多不少；必须依据上述写作意图（及大纲，若有）提炼真实检索词。
+关键词数量必须为 {keyword_count} 个，不多不少；必须依据上述写作意图提炼真实检索词。
 严格按 SOUL 约定输出 [KEYWORDS_READY] + JSON。

 /no_think"""
@@ -479,52 +479,40 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
        keywords = _clean_keywords(extra_keywords, intent)
        _emit(f"追加检索 {len(keywords)} 个关键词…")
    else:
-        # 首轮检索（无任何 pause 干预）且意图解析阶段已顺便生成检索词 → 直接复用，
-        # 省掉素材收集阶段单独的「生成检索词」LLM 往返。re_search（用户要求重新生成
-        # 检索词、覆盖式重检）等再入路径 intervention 非空，仍走下方正常生成。
-        precomputed = state.get("precomputed_keywords") or []
-        if precomputed and not intervention:
-            keywords = _clean_keywords(precomputed, intent)
-            _emit(f"复用意图解析阶段生成的 {len(keywords)} 个检索词…")
+        _emit("正在生成检索词…")
+        if use_agent:
+            kw_result = await run_writing_agent(
+                agent_id=RESEARCHER_AGENT_ID,
+                user_payload=_AGENT_KEYWORDS_USER.format(
+                    user_intent=intent,
+                    article_type=article_type,
+                    target_audience=target_audience,
+                    keyword_count=keyword_count,
+                ),
+                marker=KEYWORDS_READY_MARKER,
+                fallback={"keywords": [intent]},
+                state=state,
+                config=config,
+                explicit_model_fallback=keyword_model,
+                strict_model=False,
+                disable_thinking=True,
+            )
        else:
-            _emit("正在生成检索词…")
-            # 用户自带大纲时，把大纲原文带进检索词生成，让关键词结合意图覆盖各章节主题
-            outline_section = keyword_outline_section(state)
-            if use_agent:
-                kw_result = await run_writing_agent(
-                    agent_id=RESEARCHER_AGENT_ID,
-                    user_payload=_AGENT_KEYWORDS_USER.format(
-                        user_intent=intent,
-                        article_type=article_type,
-                        target_audience=target_audience,
-                        keyword_count=keyword_count,
-                        outline_section=outline_section,
-                    ),
-                    marker=KEYWORDS_READY_MARKER,
-                    fallback={"keywords": [intent]},
-                    state=state,
-                    config=config,
-                    explicit_model_fallback=keyword_model,
-                    strict_model=False,
-                    disable_thinking=True,
-                )
-            else:
-                kw_result = await llm_json(
-                    EXTRACT_KEYWORDS_SYSTEM,
-                    EXTRACT_KEYWORDS_USER.format(
-                        user_intent=intent,
-                        article_type=article_type,
-                        target_audience=target_audience,
-                        keyword_count=keyword_count,
-                        outline_section=outline_section,
-                    ),
-                    fallback={"keywords": [intent]},
-                    config=config,
-                    model_name=keyword_model,
-                    disable_thinking=True,
-                )
-            # 清洗关键词：剔除占位符 / 空值，全空时回退写作意图
-            keywords = _clean_keywords(kw_result.get("keywords"), intent)
+            kw_result = await llm_json(
+                EXTRACT_KEYWORDS_SYSTEM,
+                EXTRACT_KEYWORDS_USER.format(
+                    user_intent=intent,
+                    article_type=article_type,
+                    target_audience=target_audience,
+                    keyword_count=keyword_count,
+                ),
+                fallback={"keywords": [intent]},
+                config=config,
+                model_name=keyword_model,
+                disable_thinking=True,
+            )
+        # 清洗关键词：剔除占位符 / 空值，全空时回退写作意图
+        keywords = _clean_keywords(kw_result.get("keywords"), intent)
        if extra_keywords:
            keywords = list(dict.fromkeys(keywords + extra_keywords))

@@ -7,13 +7,9 @@ EXTRACT_KEYWORDS_SYSTEM = """你是一名专业情报研究员（素材收集专
 你的工作背景：你在一家权威媒体机构任职，熟悉各类信息源的可信度评估。

 你只能做以下事情：
-1. 从写作意图（以及用户提供的文章大纲，若有）中按要求的数量提取搜索关键词
+1. 从写作意图中按要求的数量提取搜索关键词
 你不能写文章，不能评价内容好坏，只做关键词提取。

-【结合大纲】
- 若用户提供了文章大纲，关键词要结合写作意图，尽量覆盖大纲涉及的各章节主题，
-  让检索到的素材能支撑每个章节，而不是只围绕一个笼统的意图。
-
 【严禁思考】
 - 这是一个简单的结构化抽取任务，禁止任何思考、分析、推理或规划过程。
 - 禁止输出 <think></think> 标签及其中的任何内容，禁止输出"让我想想""首先"之类的过渡语。
@@ -32,10 +28,10 @@ EXTRACT_KEYWORDS_SYSTEM = """你是一名专业情报研究员（素材收集专
 EXTRACT_KEYWORDS_USER = """写作意图：{user_intent}
 文章类型：{article_type}
 目标读者：{target_audience}
-{outline_section}
+
 请提取 {keyword_count} 个搜索关键词，用于搜集相关素材。
 关键词数量必须为 {keyword_count} 个，不多不少。
-关键词必须是依据上述写作意图（及大纲，若有）提炼的真实检索词，不要照抄下方 JSON 模板里的占位符。
+关键词必须是依据上述写作意图提炼的真实检索词，不要照抄下方 JSON 模板里的占位符。
 只输出 JSON（下方 ... 是占位示意，不是真实关键词）：
 {{"keywords": ["...", "..."]}}

@@ -157,11 +157,6 @@ class AIWritingState(TypedDict):
    notebook_search_space_id: Optional[int]
    notebook_source_ids: Optional[List[int]]

-    # 意图解析阶段（intent_parser）顺便生成的检索词：与意图理解共用同一次 LLM 调用，
-    # researcher 首轮检索时直接复用，省掉素材收集阶段单独的「生成检索词」LLM 往返。
-    # 解析失败/为空时为 None，researcher 回退到自己生成检索词（优雅降级）。
-    precomputed_keywords: Optional[List[str]]
-
    # Agent A 输出
    material_package: Optional[MaterialPackage]