检索结果过程添加

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

@@ -30,7 +30,21 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
     intervention = state.get("last_user_intervention") or {}
     extra_keywords: list[str] = intervention.get("_extra_keywords") or intervention.get("extra_keywords") or []
 
+    # SSE 实时进度推送（与 outline/draft 节点共用 sse_queue 机制）
+    cfg = config.get("configurable", {})
+    sse_queue = cfg.get("sse_queue")
+    sse_session_id: str = cfg.get("sse_session_id", "")
+
+    async def _emit(message: str) -> None:
+        if sse_queue is None:
+            return
+        await sse_queue.put(("research_progress", {
+            "session_id": sse_session_id,
+            "message": message,
+        }))
+
     # Step 1: 提取关键词
+    await _emit("正在生成检索词…")
     kw_result = await llm_json(
         EXTRACT_KEYWORDS_SYSTEM,
         EXTRACT_KEYWORDS_USER.format(
@@ -49,7 +63,25 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
     ai_writing_config = config.get("configurable", {}).get("ai_writing_config", {})
     provider = get_search_provider(ai_writing_config)
 
-    search_tasks = [provider.search(kw, max_results=3) for kw in keywords[:6]]
+    search_keywords = keywords[:6]
+    total = len(search_keywords)
+    await _emit(f"检索词已生成，开始检索 {total} 个方向…")
+
+    # 并发搜索，但每完成一个就推送进度（计数在单线程 asyncio 下无需加锁）
+    completed = 0
+
+    async def _search_one(kw: str) -> list[SearchResult]:
+        nonlocal completed
+        try:
+            return await provider.search(kw, max_results=3)
+        finally:
+            completed += 1
+            if completed < total:
+                await _emit(f"已完成 {completed}/{total} 次检索，正在继续检索相关内容…")
+            else:
+                await _emit(f"已完成全部 {total} 次检索")
+
+    search_tasks = [_search_one(kw) for kw in search_keywords]
     all_results_nested: list[list[SearchResult]] = await asyncio.gather(*search_tasks, return_exceptions=True)
 
     # 合并去重
@@ -82,6 +114,7 @@ async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict
             ))
 
     # Step 3: 相关度排序（如素材较多，让 LLM 排序）
+    await _emit("检索完成，正在筛选排序素材…")
     if len(raw_materials) > _MAX_MATERIALS:
         materials_list = "\n".join(
             f"- ID: {m['id']}  标题: {m['title']}  摘要: {m['content'][:100]}"