From 8f773d3f1bc1e3ac8cf77d59860504b0b429e98e Mon Sep 17 00:00:00 2001 From: fengxiangdebaozi <10845145+fengxiangdebaozi@user.noreply.gitee.com> Date: Thu, 25 Jun 2026 00:33:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(ai=E5=86=99=E4=BD=9C):=20=E6=8A=80=E8=83=BD?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E5=AE=9E=E6=97=B6=E6=B5=81=E5=BC=8F=E6=98=BE?= =?UTF-8?q?=E7=A4=BA=20+=20=E5=8E=BB=E6=8E=89=E5=A4=A7=E7=BA=B2=E6=8B=BC?= =?UTF-8?q?=E6=A3=80=E7=B4=A2=E8=AF=8D/=E5=A4=8D=E7=94=A8=E6=A3=80?= =?UTF-8?q?=E7=B4=A2=E8=AF=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 技能调用过程改为实时流式：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) --- .../agents/ai_writing/nodes/_skill_agent.py | 84 +++++++++++++++---- .../agents/ai_writing/nodes/_utils.py | 20 ----- .../agents/ai_writing/nodes/intent_parser.py | 30 ++----- .../agents/ai_writing/nodes/researcher.py | 84 ++++++++----------- .../ai_writing/prompts/researcher_prompts.py | 10 +-- .../deerflow/agents/ai_writing/state.py | 5 -- 6 files changed, 112 insertions(+), 121 deletions(-) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py index b0951756..617dbb4b 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_skill_agent.py @@ -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 [], "" diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py index 37b4a56e..6c023132 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/_utils.py @@ -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。 diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py index 73c0c773..30f325ed 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/nodes/intent_parser.py @@ -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 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 7f4afede..34043e04 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 @@ -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)) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/researcher_prompts.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/researcher_prompts.py index 015dd310..5776a764 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/researcher_prompts.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/researcher_prompts.py @@ -7,13 +7,9 @@ EXTRACT_KEYWORDS_SYSTEM = """你是一名专业情报研究员（素材收集专 你的工作背景：你在一家权威媒体机构任职，熟悉各类信息源的可信度评估。 你只能做以下事情： -1. 从写作意图（以及用户提供的文章大纲，若有）中按要求的数量提取搜索关键词 +1. 从写作意图中按要求的数量提取搜索关键词 你不能写文章，不能评价内容好坏，只做关键词提取。 -【结合大纲】 -- 若用户提供了文章大纲，关键词要结合写作意图，尽量覆盖大纲涉及的各章节主题， - 让检索到的素材能支撑每个章节，而不是只围绕一个笼统的意图。 - 【严禁思考】 - 这是一个简单的结构化抽取任务，禁止任何思考、分析、推理或规划过程。 - 禁止输出 标签及其中的任何内容，禁止输出"让我想想""首先"之类的过渡语。 @@ -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": ["...", "..."]}} 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 89b83a10..71156c0f 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 @@ -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]