ai写作

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

@@ -1,4 +1,4 @@
-"""编辑节点：三项独立 LLM 审查（事实/逻辑/语言）。"""
+"""编辑节点：三项独立 LLM 审查（事实/逻辑/语言），顺序执行以支持流式推送。"""
 from __future__ import annotations
 
 import asyncio
@@ -14,8 +14,6 @@ from ._utils import llm_json
 
 logger = logging.getLogger(__name__)
 
-# 弱模型产出质量有限，阈值过高会导致编辑反复打回直至触发修改上限。
-# 默认 75，可通过 ai_writing_config.pass_threshold 按实测调整。
 _DEFAULT_PASS_THRESHOLD = 75
 
 
@@ -51,34 +49,54 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
     pkg = state.get("material_package") or {}
     mats = _materials_summary(pkg)
 
-    ai_writing_config = config.get("configurable", {}).get("ai_writing_config", {}) or {}
+    cfg = config.get("configurable", {})
+    sse_queue: asyncio.Queue | None = cfg.get("sse_queue")
+    session_id: str = cfg.get("sse_session_id", "")
+    ai_writing_config = cfg.get("ai_writing_config", {}) or {}
     pass_threshold = int(ai_writing_config.get("pass_threshold", _DEFAULT_PASS_THRESHOLD))
 
-    # 三项独立审查并发执行
-    fact_task = llm_json(EDITOR_SYSTEM, FACT_REVIEW_USER.format(materials_summary=mats, article=article), {"score": 80, "issues": []}, config)
-    logic_task = llm_json(EDITOR_SYSTEM, LOGIC_REVIEW_USER.format(article=article), {"score": 80, "issues": []}, config)
-    lang_task = llm_json(EDITOR_SYSTEM, LANGUAGE_REVIEW_USER.format(article=article), {"score": 80, "issues": []}, config)
+    REVIEW_TASKS = [
+        ("fact",     FACT_REVIEW_USER.format(materials_summary=mats, article=article),  {"score": 80, "issues": []}),
+        ("logic",    LOGIC_REVIEW_USER.format(article=article),                          {"score": 80, "issues": []}),
+        ("language", LANGUAGE_REVIEW_USER.format(article=article),                       {"score": 80, "issues": []}),
+    ]
 
-    fact_res, logic_res, lang_res = await asyncio.gather(fact_task, logic_task, lang_task, return_exceptions=True)
+    CATEGORY_LABEL = {"fact": "事实准确性", "logic": "逻辑结构", "language": "语言文字"}
 
-    def safe(res, default):
-        return default if isinstance(res, Exception) else res
+    results: dict[str, dict] = {}
 
-    fact_res = safe(fact_res, {"score": 80, "issues": []})
-    logic_res = safe(logic_res, {"score": 80, "issues": []})
-    lang_res = safe(lang_res, {"score": 80, "issues": []})
+    for category, user_prompt, fallback in REVIEW_TASKS:
+        try:
+            res = await llm_json(EDITOR_SYSTEM, user_prompt, fallback, config)
+        except Exception as e:
+            logger.warning("editor review [%s] failed: %s", category, e)
+            res = fallback
 
-    fact_score = int(fact_res.get("score", 80))
-    logic_score = int(logic_res.get("score", 80))
-    lang_score = int(lang_res.get("score", 80))
+        results[category] = res
+
+        # 实时推送这一项审查结果
+        if sse_queue is not None:
+            issues = _parse_issues(res.get("issues", []), category)
+            await sse_queue.put(("review_item_done", {
+                "session_id": session_id,
+                "category": category,
+                "score": int(res.get("score", 80)),
+                "issues": [dict(iss) for iss in issues],
+            }))
+        else:
+            logger.debug("editor [%s] done (no sse_queue)", category)
+
+    # 汇总
+    fact_score    = int(results["fact"].get("score", 80))
+    logic_score   = int(results["logic"].get("score", 80))
+    lang_score    = int(results["language"].get("score", 80))
     overall_score = (fact_score + logic_score + lang_score) // 3
 
     all_issues: list[ReviewIssue] = (
-        _parse_issues(fact_res.get("issues", []), "fact") +
-        _parse_issues(logic_res.get("issues", []), "logic") +
-        _parse_issues(lang_res.get("issues", []), "language")
+        _parse_issues(results["fact"].get("issues", []),     "fact") +
+        _parse_issues(results["logic"].get("issues", []),    "logic") +
+        _parse_issues(results["language"].get("issues", []), "language")
     )
-    # major issues first
     all_issues.sort(key=lambda x: (0 if x["severity"] == "major" else 1))
 
     verdict = "pass" if overall_score >= pass_threshold else "reject"
@@ -101,7 +119,6 @@ async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
                 "type": "review_ready",
                 "agent_name": "编辑",
                 "message": f"审核完成，综合得分 {overall_score}，结论：{'通过' if verdict == 'pass' else '打回修改'}",
-                "review_result": review,
             },
         ],
     }