ai写作

offline-backend-20260512/backend/app/gateway/app.py

@@ -22,6 +22,7 @@ from app.gateway.config import get_gateway_config
 from app.gateway.csrf_middleware import CSRFMiddleware
 from app.gateway.deps import langgraph_runtime
 from app.gateway.routers import (
+    ai_writing,
     agents,
     artifacts,
     assistants_compat,
@@ -675,6 +676,9 @@ This gateway provides custom endpoints for models, MCP configuration, skills, an
     # Writing rewrite API is mounted at /api/writing
     app.include_router(writing.router)
 
+    # AI Writing multi-agent API is mounted at /api/ai-writing
+    app.include_router(ai_writing.router)
+
     # Scheduled task API is mounted at /api/scheduled-tasks
     app.include_router(scheduled_tasks.router)
 

offline-backend-20260512/backend/app/gateway/routers/__init__.py

@@ -1,3 +1,3 @@
-from . import artifacts, assistants_compat, mcp, models, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
+from . import ai_writing, artifacts, assistants_compat, mcp, models, recommended_questions, skills, suggestions, thread_runs, threads, uploads, writing
 
-__all__ = ["artifacts", "assistants_compat", "mcp", "models", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]
+__all__ = ["ai_writing", "artifacts", "assistants_compat", "mcp", "models", "recommended_questions", "skills", "suggestions", "threads", "thread_runs", "uploads", "writing"]

offline-backend-20260512/backend/app/gateway/routers/ai_writing.py

@@ -0,0 +1,268 @@
+"""AI 写作路由 —— 独立于原有路由，前缀 /api/ai-writing/
+
+POST /api/ai-writing/start          启动写作流程，返回 session_id
+GET  /api/ai-writing/{id}/stream    SSE 流（进度/干预/草稿事件）
+POST /api/ai-writing/{id}/resume    用户干预响应，恢复流程
+GET  /api/ai-writing/{id}/state     查询当前流程状态
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import uuid
+from collections.abc import AsyncGenerator
+from typing import Optional
+
+from fastapi import APIRouter, HTTPException
+from fastapi.responses import StreamingResponse
+from langgraph.types import Command
+from pydantic import BaseModel, Field
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/ai-writing", tags=["ai-writing"])
+
+# ── 请求/响应模型 ─────────────────────────────────────────────────────────────
+
+class StartWritingRequest(BaseModel):
+    user_intent: str
+    article_type: str = "analysis"
+    target_audience: str = "普通读者"
+    word_count_target: int = 800
+    model_name: str = ""
+
+
+class StartWritingResponse(BaseModel):
+    session_id: str
+    message: str = "写作流程已启动"
+
+
+class ResumeRequest(BaseModel):
+    action: str
+    approved_material_ids: list[str] = Field(default_factory=list)
+    extra_keywords: list[str] = Field(default_factory=list)
+    edited_outline: Optional[dict] = None
+    outline_feedback: str = ""
+    user_revision_notes: str = ""
+    override_verdict: bool = False
+
+
+# ── 内部状态 ──────────────────────────────────────────────────────────────────
+
+# session_id → asyncio.Queue
+_event_queues: dict[str, asyncio.Queue] = {}
+
+# session_id → model_name（resume 时复用）
+_session_model_names: dict[str, str] = {}
+
+
+# ── 工具函数 ──────────────────────────────────────────────────────────────────
+
+def _sse(event: str, data: dict) -> str:
+    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"
+
+
+def _make_config(session_id: str, model_name: str = "") -> dict:
+    return {
+        "configurable": {
+            "thread_id": session_id,
+            "model_name": model_name or _session_model_names.get(session_id, ""),
+            "ai_writing_config": {},
+        }
+    }
+
+
+# ── 路由 ──────────────────────────────────────────────────────────────────────
+
+@router.post("/start", response_model=StartWritingResponse)
+async def start_writing(body: StartWritingRequest):
+    """启动写作流程，返回 session_id。"""
+    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+
+    session_id = str(uuid.uuid4())
+    queue: asyncio.Queue = asyncio.Queue()
+    _event_queues[session_id] = queue
+    _session_model_names[session_id] = body.model_name
+
+    initial_state = {
+        "user_intent": body.user_intent,
+        "article_type": body.article_type,
+        "target_audience": body.target_audience,
+        "word_count_target": body.word_count_target,
+        "revision_count": 0,
+        "max_revisions": 3,
+        "status": "researching",
+        "drafts": [],
+        "progress_events": [],
+        "material_package": None,
+        "current_outline": None,
+        "current_draft": None,
+        "review_result": None,
+        "last_user_intervention": None,
+    }
+    config = _make_config(session_id, body.model_name)
+
+    asyncio.create_task(_stream_graph_segment(session_id, initial_state, config, AI_WRITING_GRAPH))
+    return StartWritingResponse(session_id=session_id)
+
+
+@router.get("/{session_id}/stream")
+async def stream_progress(session_id: str):
+    """SSE 流：推送进度事件、干预请求、草稿内容。"""
+    return StreamingResponse(
+        _sse_generator(session_id),
+        media_type="text/event-stream",
+        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
+    )
+
+
+@router.post("/{session_id}/resume")
+async def resume_writing(session_id: str, body: ResumeRequest):
+    """用户提交干预决策，恢复暂停的流程。"""
+    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+
+    if session_id not in _event_queues:
+        raise HTTPException(status_code=404, detail="Session not found")
+
+    config = _make_config(session_id)
+    intervention = {
+        "action": body.action,
+        "approvedMaterialIds": body.approved_material_ids,
+        "extraKeywords": body.extra_keywords,
+        "editedOutline": body.edited_outline,
+        "outlineFeedback": body.outline_feedback,
+        "userRevisionNotes": body.user_revision_notes,
+        "overrideVerdict": body.override_verdict,
+    }
+
+    asyncio.create_task(
+        _stream_graph_segment(session_id, Command(resume=intervention), config, AI_WRITING_GRAPH)
+    )
+    return {"message": "已提交，流程恢复中"}
+
+
+@router.get("/{session_id}/state")
+async def get_state(session_id: str):
+    """查询当前流程状态快照。"""
+    from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH
+
+    config = _make_config(session_id)
+    snapshot = await AI_WRITING_GRAPH.aget_state(config)
+    if not snapshot or not snapshot.values:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return snapshot.values
+
+
+# ── 内部辅助 ──────────────────────────────────────────────────────────────────
+
+async def _stream_graph_segment(
+    session_id: str,
+    input_or_command,
+    config: dict,
+    graph,
+):
+    """运行一个 Graph 片段（初始运行或 resume），将输出推入队列。
+    遇到 interrupt 时停止（不关闭队列），等待下次 resume。
+    Graph 结束时关闭队列（放入 None 哨兵）。
+    """
+    queue = _event_queues.get(session_id)
+    if queue is None:
+        return
+
+    reached_end = False
+    try:
+        async for chunk in graph.astream(input_or_command, config=config, stream_mode="updates"):
+            node_name = next(iter(chunk))
+            node_output = chunk[node_name]
+
+            # interrupt() 触发
+            if node_name == "__interrupt__":
+                # node_output 是 (Interrupt(...),) 元组
+                interrupt_list = node_output if isinstance(node_output, (list, tuple)) else [node_output]
+                for intr in interrupt_list:
+                    interrupt_data = intr.value if hasattr(intr, "value") else intr
+                    await queue.put(("await_user", interrupt_data))
+                return  # 等待 resume，不关闭队列
+
+            if not isinstance(node_output, dict):
+                continue
+
+            # 推送 progress_events
+            for evt in node_output.get("progress_events") or []:
+                await queue.put((_event_type(evt), {**evt, "session_id": session_id}))
+
+            # 素材就绪
+            if pkg := node_output.get("material_package"):
+                await queue.put(("materials_ready", {
+                    "session_id": session_id,
+                    "keywords": pkg.get("keywords", []),
+                    "materials": pkg.get("materials", []),
+                    "summary": pkg.get("summary", ""),
+                }))
+
+            # 大纲就绪
+            if outline := node_output.get("current_outline"):
+                await queue.put(("outline_ready", {"session_id": session_id, "outline": outline}))
+
+            # 草稿就绪
+            if draft := node_output.get("current_draft"):
+                await queue.put(("draft_ready", {
+                    "session_id": session_id,
+                    "title": draft.get("title", ""),
+                    "markdown": draft.get("full_markdown", ""),
+                    "revision_count": draft.get("revision_count", 0),
+                }))
+
+            # 编辑审核就绪
+            if review := node_output.get("review_result"):
+                await queue.put(("review_ready", {"session_id": session_id, "review_result": review}))
+
+            # 状态 done：Graph 结束
+            if node_output.get("status") == "done":
+                reached_end = True
+
+        # astream 正常结束 = Graph 执行到 END
+        reached_end = True
+        await queue.put(("done", {"session_id": session_id, "message": "写作完成"}))
+
+    except Exception as e:
+        logger.exception("ai_writing graph segment failed: session=%s", session_id)
+        await queue.put(("error", {"session_id": session_id, "message": str(e)}))
+        reached_end = True
+    finally:
+        if reached_end:
+            await queue.put(None)  # 哨兵，通知 SSE 生成器关闭
+            _event_queues.pop(session_id, None)
+            _session_model_names.pop(session_id, None)
+
+
+def _event_type(evt: dict) -> str:
+    t = evt.get("type", "progress")
+    # 过滤掉已在上方单独推送的事件类型，避免重复
+    if t in ("materials_ready", "outline_ready", "draft_ready", "review_ready", "await_user"):
+        return "step_done"
+    return t
+
+
+async def _sse_generator(session_id: str) -> AsyncGenerator[str, None]:
+    """从队列消费事件，推送为 SSE 帧。"""
+    # 等待队列创建（最多 5 秒）
+    for _ in range(50):
+        if session_id in _event_queues:
+            break
+        await asyncio.sleep(0.1)
+    else:
+        yield _sse("error", {"message": "Session not found"})
+        return
+
+    queue = _event_queues[session_id]
+    yield _sse("connected", {"session_id": session_id})
+
+    while True:
+        item = await queue.get()
+        if item is None:
+            break
+        event_type, data = item
+        yield _sse(event_type, data)
+        await asyncio.sleep(0)

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

@@ -0,0 +1,3 @@
+from .graph import AI_WRITING_GRAPH, build_ai_writing_graph
+
+__all__ = ["AI_WRITING_GRAPH", "build_ai_writing_graph"]

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

@@ -0,0 +1,231 @@
+"""AI 写作 LangGraph Graph，含 4 个用户干预点。"""
+from __future__ import annotations
+
+from langgraph.checkpoint.memory import MemorySaver
+from langgraph.graph import END, StateGraph
+from langgraph.types import interrupt
+
+from .state import AIWritingState
+from .nodes.intent_parser import intent_parser_node
+from .nodes.researcher import researcher_node
+from .nodes.writer_outline import writer_outline_node
+from .nodes.writer_draft import writer_draft_node
+from .nodes.editor import editor_node
+
+# ── 干预节点 ──────────────────────────────────────────────────────────────────
+
+async def pause_material_confirm(state: AIWritingState) -> dict:
+    """PAUSE-1：素材确认"""
+    pkg = state["material_package"]
+    user_input: dict = interrupt({
+        "pause_point": "material_confirm",
+        "message": "素材收集专家已完成素材搜集，请确认素材后继续。",
+        "materials": pkg["materials"],
+        "keywords": pkg["keywords"],
+        "summary": pkg["summary"],
+    })
+
+    action = user_input.get("action", "confirm")
+
+    if action == "confirm":
+        approved_ids = set(user_input.get("approvedMaterialIds") or [m["id"] for m in pkg["materials"]])
+        filtered = [m for m in pkg["materials"] if m["id"] in approved_ids]
+        updated_pkg = {**pkg, "materials": filtered}
+        return {
+            "material_package": updated_pkg,
+            "last_user_intervention": user_input,
+            "status": "writing_outline",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户已确认素材，开始规划大纲"}],
+        }
+
+    # re_search：重新搜索
+    extra_kws = user_input.get("extraKeywords") or []
+    return {
+        "last_user_intervention": {**user_input, "_extra_keywords": extra_kws},
+        "status": "researching",
+        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "重新搜索中..."}],
+    }
+
+
+def _normalize_outline(outline: dict) -> dict:
+    """将前端回传的 camelCase 大纲统一为后端 snake_case 格式。"""
+    if not outline:
+        return outline
+    sections = []
+    for s in outline.get("sections", []):
+        sections.append({
+            "section_title": s.get("section_title") or s.get("sectionTitle", ""),
+            "key_points":    s.get("key_points")    or s.get("keyPoints", []),
+            "material_ids":  s.get("material_ids")  or s.get("materialIds", []),
+        })
+    return {"title": outline.get("title", ""), "sections": sections}
+
+
+async def pause_outline_confirm(state: AIWritingState) -> dict:
+    """PAUSE-2：大纲确认"""
+    user_input: dict = interrupt({
+        "pause_point": "outline_confirm",
+        "message": "作家已完成大纲规划，请确认或编辑大纲后继续。",
+        "outline": state["current_outline"],
+    })
+
+    action = user_input.get("action", "confirm")
+
+    if action == "confirm":
+        raw_outline = user_input.get("editedOutline") or state["current_outline"]
+        edited = _normalize_outline(raw_outline) if raw_outline else state["current_outline"]
+        return {
+            "current_outline": edited,
+            "last_user_intervention": user_input,
+            "status": "writing_draft",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户已确认大纲，开始写作"}],
+        }
+
+    # re_outline
+    feedback = user_input.get("outlineFeedback", "")
+    return {
+        "last_user_intervention": {**user_input, "_feedback": feedback},
+        "status": "writing_outline",
+        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": f"重新规划大纲（{feedback[:30]}）"}],
+    }
+
+
+async def pause_draft_confirm(state: AIWritingState) -> dict:
+    """PAUSE-3：草稿确认"""
+    draft = state["current_draft"]
+    user_input: dict = interrupt({
+        "pause_point": "draft_confirm",
+        "message": "草稿已完成，请确认或给出修改意见。",
+        "draft_title": draft["title"],
+        "revision_count": draft["revision_count"],
+        "review_result": state.get("review_result"),
+    })
+
+    action = user_input.get("action", "to_editor")
+
+    if action == "finalize":
+        return {
+            "last_user_intervention": user_input,
+            "status": "done",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户确认定稿"}],
+        }
+
+    if action == "user_revise":
+        notes = user_input.get("userRevisionNotes", "")
+        return {
+            "last_user_intervention": {**user_input, "_user_notes": notes},
+            "revision_count": state.get("revision_count", 0) + 1,
+            "status": "revising",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": f"按用户意见重写（{notes[:30]}）"}],
+        }
+
+    # to_editor
+    return {
+        "last_user_intervention": user_input,
+        "status": "reviewing",
+        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "提交编辑审核"}],
+    }
+
+
+async def pause_review_confirm(state: AIWritingState) -> dict:
+    """PAUSE-4：编辑打回确认"""
+    user_input: dict = interrupt({
+        "pause_point": "review_confirm",
+        "message": "编辑认为文章需要修改，请决定下一步操作。",
+        "review_result": state["review_result"],
+        "revision_count": state.get("revision_count", 0),
+        "max_revisions": state.get("max_revisions", 3),
+    })
+
+    if user_input.get("overrideVerdict") or user_input.get("action") == "force_finalize":
+        return {
+            "last_user_intervention": user_input,
+            "status": "done",
+            "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "用户强制定稿"}],
+        }
+
+    action = user_input.get("action", "accept_review")
+    extra_notes = user_input.get("userRevisionNotes", "")
+    return {
+        "last_user_intervention": {**user_input, "_extra_notes": extra_notes, "action": action},
+        "revision_count": state.get("revision_count", 0) + 1,
+        "status": "revising",
+        "progress_events": [{"type": "resumed", "agent_name": "系统", "message": "按编辑意见重写"}],
+    }
+
+
+# ── 条件路由 ──────────────────────────────────────────────────────────────────
+
+def after_material_pause(state: AIWritingState) -> str:
+    return "researcher" if state["status"] == "researching" else "writer_outline"
+
+
+def after_outline_pause(state: AIWritingState) -> str:
+    return "writer_outline" if state["status"] == "writing_outline" else "writer_draft"
+
+
+def after_draft_pause(state: AIWritingState) -> str:
+    status = state["status"]
+    if status == "done":
+        return "end"
+    if status == "revising":
+        return "writer_draft"
+    return "editor"
+
+
+def after_review(state: AIWritingState) -> str:
+    review = state.get("review_result") or {}
+    if review.get("verdict") == "pass":
+        return "pause_draft"
+    rev_count = state.get("revision_count", 0)
+    max_rev = state.get("max_revisions", 3)
+    if rev_count >= max_rev:
+        return "pause_draft"   # 超出上限，强制到草稿确认
+    return "pause_review"
+
+
+def after_review_pause(state: AIWritingState) -> str:
+    return "end" if state["status"] == "done" else "writer_draft"
+
+
+# ── 组装 Graph ────────────────────────────────────────────────────────────────
+
+def build_ai_writing_graph():
+    g = StateGraph(AIWritingState)
+
+    g.add_node("intent_parser",   intent_parser_node)
+    g.add_node("researcher",      researcher_node)
+    g.add_node("pause_material",  pause_material_confirm)
+    g.add_node("writer_outline",  writer_outline_node)
+    g.add_node("pause_outline",   pause_outline_confirm)
+    g.add_node("writer_draft",    writer_draft_node)
+    g.add_node("pause_draft",     pause_draft_confirm)
+    g.add_node("editor",          editor_node)
+    g.add_node("pause_review",    pause_review_confirm)
+
+    g.set_entry_point("intent_parser")
+    g.add_edge("intent_parser",  "researcher")
+    g.add_edge("researcher",     "pause_material")
+
+    g.add_conditional_edges("pause_material", after_material_pause,
+                             {"researcher": "researcher", "writer_outline": "writer_outline"})
+
+    g.add_edge("writer_outline", "pause_outline")
+    g.add_conditional_edges("pause_outline", after_outline_pause,
+                             {"writer_outline": "writer_outline", "writer_draft": "writer_draft"})
+
+    g.add_edge("writer_draft",   "pause_draft")
+    g.add_conditional_edges("pause_draft", after_draft_pause,
+                             {"end": END, "writer_draft": "writer_draft", "editor": "editor"})
+
+    g.add_conditional_edges("editor", after_review,
+                             {"pause_draft": "pause_draft", "pause_review": "pause_review"})
+
+    g.add_conditional_edges("pause_review", after_review_pause,
+                             {"end": END, "writer_draft": "writer_draft"})
+
+    checkpointer = MemorySaver()
+    return g.compile(checkpointer=checkpointer)
+
+
+AI_WRITING_GRAPH = build_ai_writing_graph()

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

@@ -0,0 +1,146 @@
+"""节点共用工具：JSON 解析容错、LLM 创建。
+
+弱模型经常输出不规范 JSON（中文标点、结尾多余逗号、被 markdown 包裹、
+夹带解释文字、嵌套对象）。parse_json 用多级降级策略尽量抢救出结构化结果。
+"""
+from __future__ import annotations
+
+import json
+import logging
+import re
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import RunnableConfig
+
+logger = logging.getLogger(__name__)
+
+
+def get_model(config: RunnableConfig):
+    from deerflow.models import create_chat_model
+    model_name = config.get("configurable", {}).get("model_name", "") or None
+    return create_chat_model(name=model_name)
+
+
+# 弱模型常把结构性标点写成中文全角符号，仅在常规解析失败后作为修复手段
+_FULLWIDTH_FIX = {
+    "“": '"',   # 左双引号 “
+    "”": '"',   # 右双引号 ”
+    "‘": "'",   # 左单引号 ‘
+    "’": "'",   # 右单引号 ’
+    "：": ":",   # 全角冒号 ：
+    "，": ",",   # 全角逗号 ，
+    "（": "(",   # 全角左括号 （
+    "）": ")",   # 全角右括号 ）
+    "｛": "{",   # 全角左花括号 ｛
+    "｝": "}",   # 全角右花括号 ｝
+    "［": "[",   # 全角左方括号 ［
+    "］": "]",   # 全角右方括号 ］
+    "【": "[",   # 黑方括号 【
+    "】": "]",   # 黑方括号 】
+}
+
+
+def _extract_balanced(text: str) -> str | None:
+    """从文本中提取第一个括号平衡的 {...} 块（支持嵌套，忽略字符串内的括号）。"""
+    start = text.find("{")
+    if start == -1:
+        return None
+    depth = 0
+    in_str = False
+    escape = False
+    for i in range(start, len(text)):
+        ch = text[i]
+        if escape:
+            escape = False
+            continue
+        if ch == "\\":
+            escape = True
+            continue
+        if ch == '"':
+            in_str = not in_str
+            continue
+        if in_str:
+            continue
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+            if depth == 0:
+                return text[start:i + 1]
+    return None
+
+
+def _repair_json(text: str) -> str:
+    """尽力修复弱模型常见的 JSON 错误：全角标点、结尾多余逗号。"""
+    for bad, good in _FULLWIDTH_FIX.items():
+        text = text.replace(bad, good)
+    # 去掉对象/数组结尾多余的逗号： ,}  或  ,]
+    text = re.sub(r",(\s*[}\]])", r"\1", text)
+    return text
+
+
+def _try_load(text: str) -> dict | None:
+    try:
+        result = json.loads(text)
+    except Exception:
+        return None
+    return result if isinstance(result, dict) else None
+
+
+def parse_json(text: str, fallback: dict) -> dict:
+    """从 LLM 响应中提取 JSON 对象，多级降级，失败时返回 fallback。"""
+    if not text or not text.strip():
+        return fallback
+    text = text.strip()
+
+    # 1. 直接解析
+    if (r := _try_load(text)) is not None:
+        return r
+
+    # 2. markdown 代码块内容
+    m = re.search(r"```(?:json)?\s*(.+?)\s*```", text, re.DOTALL)
+    if m:
+        fenced = m.group(1).strip()
+        if (r := _try_load(fenced)) is not None:
+            return r
+        text = fenced  # 用代码块内容继续后续步骤
+
+    # 3. 平衡括号提取（处理夹带解释文字、嵌套对象的情况）
+    candidate = _extract_balanced(text)
+    if candidate:
+        if (r := _try_load(candidate)) is not None:
+            return r
+        # 4. 修复后再试
+        if (r := _try_load(_repair_json(candidate))) is not None:
+            return r
+
+    # 5. 整体修复后兜底
+    if (r := _try_load(_repair_json(text))) is not None:
+        return r
+
+    logger.warning("JSON parse failed, using fallback. raw=%r", text[:200])
+    return fallback
+
+
+def _content_to_text(content) -> str:
+    """将 LLM 响应内容统一为字符串（部分模型返回结构化 content 块列表）。"""
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return "".join(
+            part.get("text", "") if isinstance(part, dict) else str(part)
+            for part in content
+        )
+    return str(content)
+
+
+async def llm_json(system: str, user: str, fallback: dict, config: RunnableConfig) -> dict:
+    """调用 LLM，解析 JSON 输出，失败用 fallback。"""
+    model = get_model(config)
+    messages = [SystemMessage(content=system), HumanMessage(content=user)]
+    try:
+        resp = await model.ainvoke(messages)
+        return parse_json(_content_to_text(resp.content), fallback)
+    except Exception as e:
+        logger.error("llm_json failed: %s", e, exc_info=True)
+        return {**fallback, "_llm_error": str(e)}

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

@@ -0,0 +1,107 @@
+"""编辑节点：三项独立 LLM 审查（事实/逻辑/语言）。"""
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from langchain_core.runnables import RunnableConfig
+
+from ..state import AIWritingState, ReviewResult, ReviewIssue
+from ..prompts.editor_prompts import (
+    EDITOR_SYSTEM, FACT_REVIEW_USER, LOGIC_REVIEW_USER, LANGUAGE_REVIEW_USER,
+)
+from ._utils import llm_json
+
+logger = logging.getLogger(__name__)
+
+# 弱模型产出质量有限，阈值过高会导致编辑反复打回直至触发修改上限。
+# 默认 75，可通过 ai_writing_config.pass_threshold 按实测调整。
+_DEFAULT_PASS_THRESHOLD = 75
+
+
+def _materials_summary(pkg: dict) -> str:
+    lines = []
+    for m in (pkg.get("materials") or [])[:6]:
+        lines.append(f"[{m['id']}] {m['title']}: {m['content'][:200]}")
+    return "\n".join(lines) or "（无素材）"
+
+
+def _parse_issues(raw_issues: list, category: str) -> list[ReviewIssue]:
+    issues = []
+    for item in (raw_issues or []):
+        issues.append(ReviewIssue(
+            category=category,
+            severity=item.get("severity", "minor"),
+            location=item.get("location", ""),
+            issue=item.get("issue", ""),
+            suggestion=item.get("suggestion", ""),
+        ))
+    return issues
+
+
+async def editor_node(state: AIWritingState, config: RunnableConfig) -> dict:
+    draft = state.get("current_draft")
+    if not draft:
+        return {
+            "status": "error",
+            "progress_events": [{"type": "error", "agent_name": "编辑", "message": "草稿为空，无法审核"}],
+        }
+
+    article = draft["full_markdown"]
+    pkg = state.get("material_package") or {}
+    mats = _materials_summary(pkg)
+
+    ai_writing_config = config.get("configurable", {}).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)
+
+    fact_res, logic_res, lang_res = await asyncio.gather(fact_task, logic_task, lang_task, return_exceptions=True)
+
+    def safe(res, default):
+        return default if isinstance(res, Exception) else res
+
+    fact_res = safe(fact_res, {"score": 80, "issues": []})
+    logic_res = safe(logic_res, {"score": 80, "issues": []})
+    lang_res = safe(lang_res, {"score": 80, "issues": []})
+
+    fact_score = int(fact_res.get("score", 80))
+    logic_score = int(logic_res.get("score", 80))
+    lang_score = int(lang_res.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")
+    )
+    # 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"
+
+    review = ReviewResult(
+        verdict=verdict,
+        fact_score=fact_score,
+        logic_score=logic_score,
+        language_score=lang_score,
+        overall_score=overall_score,
+        pass_threshold=pass_threshold,
+        revision_notes=all_issues,
+    )
+
+    return {
+        "review_result": review,
+        "progress_events": [
+            {"type": "step_start", "agent_name": "编辑", "message": "开始审核"},
+            {
+                "type": "review_ready",
+                "agent_name": "编辑",
+                "message": f"审核完成，综合得分 {overall_score}，结论：{'通过' if verdict == 'pass' else '打回修改'}",
+                "review_result": review,
+            },
+        ],
+    }

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

@@ -0,0 +1,60 @@
+"""意图解析节点：将用户自然语言意图结构化为写作配置。"""
+from __future__ import annotations
+
+from langchain_core.runnables import RunnableConfig
+
+from ..state import AIWritingState
+from ._utils import llm_json
+
+_SYSTEM = """你是一名写作助手。
+你的任务：分析用户的写作意图，提取结构化的写作配置。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
+- 必须使用英文双引号 "。
+- article_type 只能是 news / analysis / feature / report 之一。
+
+【完整示例】
+输入意图：帮我写一篇面向行业专家的深度文章，分析今年新能源汽车出口为什么大涨，大概两千字
+正确输出：
+{"article_type": "feature", "target_audience": "行业专家", "word_count_target": 2000, "refined_intent": "分析今年新能源汽车出口高速增长的原因"}"""
+
+_USER = """用户写作意图：{user_intent}
+
+请分析并输出以下 JSON（如果意图中未明确某字段，给出合理默认值）：
+{{
+  "article_type": "news|analysis|feature|report 之一",
+  "target_audience": "目标读者描述",
+  "word_count_target": 800,
+  "refined_intent": "精炼后的写作意图（一句话）"
+}}"""
+
+
+async def intent_parser_node(state: AIWritingState, config: RunnableConfig) -> dict:
+    result = await llm_json(
+        _SYSTEM,
+        _USER.format(user_intent=state["user_intent"]),
+        fallback={
+            "article_type": state.get("article_type", "analysis"),
+            "target_audience": state.get("target_audience", "普通读者"),
+            "word_count_target": state.get("word_count_target", 800),
+            "refined_intent": state["user_intent"],
+        },
+        config=config,
+    )
+
+    updates: dict = {
+        "status": "researching",
+        "progress_events": [{"type": "step_done", "agent_name": "系统", "message": "意图解析完成"}],
+    }
+
+    if result.get("article_type"):
+        updates["article_type"] = result["article_type"]
+    if result.get("target_audience"):
+        updates["target_audience"] = result["target_audience"]
+    if result.get("word_count_target"):
+        updates["word_count_target"] = int(result["word_count_target"])
+    if result.get("refined_intent"):
+        updates["user_intent"] = result["refined_intent"]
+
+    return updates

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

@@ -0,0 +1,119 @@
+"""素材收集专家节点：关键词提取 → 并发搜索 → 相关度排序。"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import uuid
+from typing import List
+
+from langchain_core.runnables import RunnableConfig
+
+from ..search import get_search_provider, SearchResult
+from ..state import AIWritingState, Material, MaterialPackage
+from ..prompts.researcher_prompts import (
+    EXTRACT_KEYWORDS_SYSTEM, EXTRACT_KEYWORDS_USER,
+    RANK_MATERIALS_SYSTEM, RANK_MATERIALS_USER,
+)
+from ._utils import llm_json
+
+logger = logging.getLogger(__name__)
+
+_MAX_MATERIALS = 12
+
+
+async def researcher_node(state: AIWritingState, config: RunnableConfig) -> dict:
+    intent = state["user_intent"]
+    article_type = state.get("article_type", "analysis")
+    target_audience = state.get("target_audience", "普通读者")
+
+    # 检查是否是补充搜索（PAUSE-1 re_search action）
+    intervention = state.get("last_user_intervention") or {}
+    extra_keywords: list[str] = intervention.get("_extra_keywords") or intervention.get("extra_keywords") or []
+
+    # Step 1: 提取关键词
+    kw_result = await llm_json(
+        EXTRACT_KEYWORDS_SYSTEM,
+        EXTRACT_KEYWORDS_USER.format(
+            user_intent=intent,
+            article_type=article_type,
+            target_audience=target_audience,
+        ),
+        fallback={"keywords": [intent]},
+        config=config,
+    )
+    keywords: list[str] = kw_result.get("keywords", [intent])
+    if extra_keywords:
+        keywords = list(dict.fromkeys(keywords + extra_keywords))
+
+    # Step 2: 并发搜索
+    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]]
+    all_results_nested: list[list[SearchResult]] = await asyncio.gather(*search_tasks, return_exceptions=True)
+
+    # 合并去重
+    seen_urls: set[str] = set()
+    raw_materials: list[Material] = []
+    existing_pkg = state.get("material_package")
+    if existing_pkg and intervention.get("action") != "re_search":
+        for m in existing_pkg.get("materials", []):
+            seen_urls.add(m.get("url", ""))
+            raw_materials.append(m)
+
+    for results in all_results_nested:
+        if isinstance(results, Exception):
+            logger.warning("search failed: %s", results)
+            continue
+        for r in results:
+            if r.url and r.url in seen_urls:
+                continue
+            seen_urls.add(r.url)
+            raw_materials.append(Material(
+                id=f"m{len(raw_materials)+1:03d}",
+                title=r.title,
+                content=r.content[:500],
+                source=r.source,
+                published_at=r.published_at,
+                relevance_score=r.relevance_score,
+                source_type=r.source_type,
+                url=r.url,
+                recUuid=r.recUuid,
+            ))
+
+    # Step 3: 相关度排序（如素材较多，让 LLM 排序）
+    if len(raw_materials) > _MAX_MATERIALS:
+        materials_list = "\n".join(
+            f"- ID: {m['id']}  标题: {m['title']}  摘要: {m['content'][:100]}"
+            for m in raw_materials
+        )
+        rank_result = await llm_json(
+            RANK_MATERIALS_SYSTEM,
+            RANK_MATERIALS_USER.format(user_intent=intent, materials_list=materials_list),
+            fallback={"ranked_ids": [m["id"] for m in raw_materials[:_MAX_MATERIALS]], "summary": f"共搜索到{len(raw_materials)}条素材"},
+            config=config,
+        )
+        ranked_ids: list[str] = rank_result.get("ranked_ids", [m["id"] for m in raw_materials])
+        id_to_mat = {m["id"]: m for m in raw_materials}
+        sorted_materials = [id_to_mat[mid] for mid in ranked_ids if mid in id_to_mat]
+        # 加入未被排序的素材
+        sorted_ids_set = set(ranked_ids)
+        for m in raw_materials:
+            if m["id"] not in sorted_ids_set:
+                sorted_materials.append(m)
+        materials = sorted_materials[:_MAX_MATERIALS]
+        summary = rank_result.get("summary", f"共搜索到{len(raw_materials)}条素材，精选{len(materials)}条")
+    else:
+        materials = sorted(raw_materials, key=lambda m: m.get("relevance_score", 0), reverse=True)[:_MAX_MATERIALS]
+        summary = f"共搜索到{len(raw_materials)}条素材，精选{len(materials)}条"
+
+    pkg = MaterialPackage(keywords=keywords, materials=materials, summary=summary)
+
+    return {
+        "material_package": pkg,
+        "status": "awaiting_material_confirm",
+        "progress_events": [
+            {"type": "step_start", "agent_name": "素材收集专家", "message": f"搜索关键词：{', '.join(keywords[:3])}"},
+            {"type": "materials_ready", "agent_name": "素材收集专家", "message": summary},
+        ],
+    }

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

@@ -0,0 +1,123 @@
+"""作家节点 Phase 2：按大纲逐段写正文。"""
+from __future__ import annotations
+
+import logging
+
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables import RunnableConfig
+
+from ..state import AIWritingState, DraftArticle
+from ..prompts.writer_prompts import DRAFT_SECTION_SYSTEM, DRAFT_SECTION_USER
+from ._utils import get_model, _content_to_text
+
+logger = logging.getLogger(__name__)
+
+
+def _get_section_materials(material_ids: list[str], pkg: dict) -> str:
+    mat_map = {m["id"]: m for m in pkg.get("materials", [])}
+    lines = []
+    for mid in material_ids:
+        if mid in mat_map:
+            m = mat_map[mid]
+            lines.append(f"[{mid}] {m['title']}: {m['content'][:300]}")
+    if not lines:
+        # 没有指定素材时，使用前两条
+        for m in pkg.get("materials", [])[:2]:
+            lines.append(f"[{m['id']}] {m['title']}: {m['content'][:300]}")
+    return "\n".join(lines) if lines else "（无相关素材）"
+
+
+async def writer_draft_node(state: AIWritingState, config: RunnableConfig) -> dict:
+    outline = state.get("current_outline")
+    if not outline:
+        return {
+            "status": "error",
+            "progress_events": [{"type": "error", "agent_name": "作家", "message": "大纲为空，无法写作"}],
+        }
+
+    pkg = state.get("material_package") or {"keywords": [], "materials": [], "summary": ""}
+    intervention = state.get("last_user_intervention") or {}
+    revision_count = state.get("revision_count", 0)
+
+    # 修改意见（来自用户或编辑）
+    revision_notes = ""
+    if intervention.get("action") == "user_revise":
+        notes = intervention.get("_user_notes") or intervention.get("user_revision_notes", "")
+        if notes:
+            revision_notes = f"\n用户修改意见：{notes}"
+    elif intervention.get("action") in ("accept_review", "custom_revise"):
+        review = state.get("review_result") or {}
+        issues = review.get("revision_notes", [])
+        extra = intervention.get("_extra_notes") or intervention.get("user_revision_notes", "")
+        parts = []
+        for iss in issues[:5]:
+            parts.append(f"- [{iss.get('severity','minor')}] {iss.get('location','')}: {iss.get('suggestion','')}")
+        if extra:
+            parts.append(f"- 用户补充：{extra}")
+        if parts:
+            revision_notes = "\n修改要求：\n" + "\n".join(parts)
+
+    word_count_target = state.get("word_count_target", 800)
+    sections = outline.get("sections", [])
+    words_per_section = max(100, word_count_target // max(len(sections), 1))
+
+    model = get_model(config)
+    section_texts: list[str] = []
+
+    for i, section in enumerate(sections):
+        # 兼容后端 snake_case 和前端编辑后回传的 camelCase
+        sec_title  = section.get("section_title") or section.get("sectionTitle", f"章节{i+1}")
+        key_points = section.get("key_points") or section.get("keyPoints") or []
+        mat_ids    = section.get("material_ids") or section.get("materialIds") or []
+
+        section_materials = _get_section_materials(mat_ids, pkg)
+        key_points_str = "\n".join(f"- {p}" for p in key_points)
+
+        user_msg = DRAFT_SECTION_USER.format(
+            article_title=outline.get("title", "文章"),
+            section_title=sec_title,
+            key_points=key_points_str,
+            section_materials=section_materials,
+            revision_section=revision_notes,
+            target_words=words_per_section,
+        )
+
+        try:
+            resp = await model.ainvoke([
+                SystemMessage(content=DRAFT_SECTION_SYSTEM),
+                HumanMessage(content=user_msg),
+            ])
+            # 部分模型（如 DeepSeek thinking 模式）返回列表结构的 content，统一转为字符串
+            text = _content_to_text(resp.content).strip()
+            section_texts.append(f"## {sec_title}\n\n{text}")
+        except Exception as e:
+            logger.error("draft section [%s] failed: %s", sec_title, e, exc_info=True)
+            section_texts.append(f"## {sec_title}\n\n（内容生成失败，请重试）")
+
+    full_markdown = f"# {outline['title']}\n\n" + "\n\n".join(section_texts)
+    word_count = len([c for c in full_markdown if '一' <= c <= '鿿'])
+
+    draft = DraftArticle(
+        title=outline["title"],
+        outline=outline,
+        full_markdown=full_markdown,
+        word_count=word_count,
+        revision_count=revision_count,
+    )
+
+    return {
+        "current_draft": draft,
+        "drafts": [draft],
+        "status": "awaiting_draft_confirm",
+        "progress_events": [
+            {"type": "step_start", "agent_name": "作家", "message": f"开始写作（第{revision_count+1}稿）"},
+            {
+                "type": "draft_ready",
+                "agent_name": "作家",
+                "message": f"草稿完成，共 {word_count} 字",
+                "draft_title": outline["title"],
+                "draft_markdown": full_markdown,
+                "revision_count": revision_count,
+            },
+        ],
+    }

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

@@ -0,0 +1,83 @@
+"""作家节点 Phase 1：生成文章大纲。"""
+from __future__ import annotations
+
+from langchain_core.runnables import RunnableConfig
+
+from ..state import AIWritingState, Outline, OutlineSection
+from ..prompts.writer_prompts import OUTLINE_SYSTEM, OUTLINE_USER
+from ._utils import llm_json
+
+
+# 上下文压缩：传给大纲规划的素材摘要做条数和单条长度双重截断，防止 token 溢出。
+_MAX_OUTLINE_MATERIALS = 12
+_MATERIAL_SNIPPET_CHARS = 150
+
+
+def _materials_summary(pkg: dict) -> str:
+    lines = []
+    for m in pkg.get("materials", [])[:_MAX_OUTLINE_MATERIALS]:
+        lines.append(f"[{m['id']}] {m['title']} | {m['source']} | {m['content'][:_MATERIAL_SNIPPET_CHARS]}")
+    return "\n".join(lines) or "（无素材）"
+
+
+def _parse_outline(raw: dict) -> Outline:
+    sections = []
+    for s in raw.get("sections", []):
+        sections.append(OutlineSection(
+            section_title=s.get("section_title", ""),
+            key_points=s.get("key_points", []),
+            material_ids=s.get("material_ids", []),
+        ))
+    if not sections:
+        sections = [OutlineSection(section_title="正文", key_points=["主要内容"], material_ids=[])]
+    return Outline(title=raw.get("title", "未命名文章"), sections=sections)
+
+
+async def writer_outline_node(state: AIWritingState, config: RunnableConfig) -> dict:
+    pkg = state.get("material_package") or {"keywords": [], "materials": [], "summary": ""}
+    intervention = state.get("last_user_intervention") or {}
+
+    feedback_section = ""
+    if intervention.get("action") == "re_outline" and intervention.get("_feedback"):
+        feedback_section = f"\n重新规划要求：{intervention['_feedback']}"
+    elif intervention.get("outline_feedback"):
+        feedback_section = f"\n重新规划要求：{intervention['outline_feedback']}"
+
+    result = await llm_json(
+        OUTLINE_SYSTEM,
+        OUTLINE_USER.format(
+            user_intent=state["user_intent"],
+            article_type=state.get("article_type", "analysis"),
+            target_audience=state.get("target_audience", "普通读者"),
+            word_count_target=state.get("word_count_target", 800),
+            materials_summary=_materials_summary(pkg),
+            feedback_section=feedback_section,
+        ),
+        fallback={
+            "title": "未命名文章",
+            "sections": [
+                {"section_title": "引言", "key_points": ["背景介绍"], "material_ids": []},
+                {"section_title": "正文", "key_points": ["核心内容"], "material_ids": []},
+                {"section_title": "结论", "key_points": ["总结"], "material_ids": []},
+            ],
+        },
+        config=config,
+    )
+
+    if llm_err := result.pop("_llm_error", None):
+        return {
+            "status": "error",
+            "progress_events": [{"type": "error", "agent_name": "作家", "message": f"大纲生成失败：{llm_err}"}],
+        }
+
+    outline = _parse_outline(result)
+    section_count = len(outline["sections"])
+
+    return {
+        "current_outline": outline,
+        "status": "awaiting_outline_confirm",
+        "progress_events": [
+            {"type": "step_start", "agent_name": "作家", "message": "开始规划大纲"},
+            {"type": "outline_ready", "agent_name": "作家", "message": f"大纲已生成，共 {section_count} 个章节"},
+        ],
+    }

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/editor_prompts.py

@@ -0,0 +1,52 @@
+EDITOR_SYSTEM = """你是一名资深主编（编辑），标准极高，要求严格。
+你的目标：审查文章质量，给出客观评分和具体修改意见。
+你的工作背景：你在一家权威媒体担任主编十五年，见过各种质量的文章。
+
+你只能做以下事情：
+1. 从指定维度审查文章
+2. 给出该维度的评分（0-100 的整数）和具体问题列表
+你不能改写文章，只做当前指定维度的审查。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
+- 必须使用英文双引号 "。
+- score 是 0-100 的整数。
+- 每个问题的 severity 必须是 "major"（必须修改）或 "minor"（建议修改）。
+- 文章在该维度没有问题时，issues 填空数组 []。
+
+【完整示例】
+正确输出：
+{
+  "score": 82,
+  "issues": [
+    {"severity": "major", "location": "第二段", "issue": "引用的增长数据没有素材支撑", "suggestion": "补充素材 m002 中的具体数据，并注明来源"},
+    {"severity": "minor", "location": "结尾段", "issue": "结论略显仓促", "suggestion": "增加一句对未来行业趋势的判断"}
+  ]
+}"""
+
+FACT_REVIEW_USER = """请审查以下文章的【事实准确性】维度。
+判断标准：引用的数据/事实是否有素材支撑，是否有捏造或夸大。
+
+参考素材：
+{materials_summary}
+
+待审文章：
+{article}
+
+只输出 JSON（score: 0-100 整数，issues 列表，每条含 severity）："""
+
+LOGIC_REVIEW_USER = """请审查以下文章的【逻辑结构】维度。
+判断标准：论点是否清晰，论据是否充分，段落是否连贯，结构是否合理。
+
+待审文章：
+{article}
+
+只输出 JSON（score: 0-100 整数，issues 列表，每条含 severity）："""
+
+LANGUAGE_REVIEW_USER = """请审查以下文章的【语言文字】维度。
+判断标准：语言是否流畅，用词是否准确，有无错别字或病句。
+
+待审文章：
+{article}
+
+只输出 JSON（score: 0-100 整数，issues 列表，每条含 severity）："""

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/researcher_prompts.py

@@ -0,0 +1,49 @@
+EXTRACT_KEYWORDS_SYSTEM = """你是一名专业情报研究员（素材收集专家）。
+你的目标：为写作任务提取搜索关键词。
+你的工作背景：你在一家权威媒体机构任职，熟悉各类信息源的可信度评估。
+
+你只能做以下事情：
+1. 从写作意图中提取 3-6 个搜索关键词
+你不能写文章，不能评价内容好坏，只做关键词提取。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
+- 必须使用英文双引号 "，不要用中文引号 “”。
+- 关键词要具体、可搜索，避免过于宽泛的词。
+
+【完整示例】
+输入意图：写一篇分析今年新能源汽车出口增长的文章
+正确输出：
+{"keywords": ["新能源汽车出口", "汽车出口数据", "新能源汽车海外市场", "电动车出口增长", "汽车出口政策"]}"""
+
+EXTRACT_KEYWORDS_USER = """写作意图：{user_intent}
+文章类型：{article_type}
+目标读者：{target_audience}
+
+请提取 3-6 个搜索关键词，用于搜集相关素材。只输出 JSON：
+{{"keywords": ["关键词1", "关键词2"]}}"""
+
+RANK_MATERIALS_SYSTEM = """你是一名专业情报研究员（素材收集专家）。
+你的目标：对搜索结果按相关度排序和筛选。
+你的工作背景：你在一家权威媒体机构任职，熟悉各类信息源的可信度评估。
+
+你只能做以下事情：
+1. 从提供的素材列表中，选出与写作意图最相关的素材ID，按相关度从高到低排序
+你不能写文章，只做素材筛选排序。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
+- 必须使用英文双引号 "。
+- ranked_ids 中的 ID 必须来自素材列表，不要编造 ID。
+
+【完整示例】
+正确输出：
+{"ranked_ids": ["m003", "m001", "m005"], "summary": "共搜索到 8 条素材，精选 5 条最相关的"}"""
+
+RANK_MATERIALS_USER = """写作意图：{user_intent}
+
+素材列表（每条含ID和摘要）：
+{materials_list}
+
+请选出最相关的素材ID（最多8条），按相关度从高到低排序。只输出 JSON：
+{{"ranked_ids": ["id1", "id2"], "summary": "共搜索到X条素材，精选Y条"}}"""

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/prompts/writer_prompts.py

@@ -0,0 +1,57 @@
+OUTLINE_SYSTEM = """你是一名专业内容创作者（作家）。
+你的目标：根据素材和意图，规划清晰的文章大纲。
+你的工作背景：你有十年内容创作经验，擅长将零散素材组织成有逻辑的文章结构。
+
+你只能做以下事情：
+1. 分析素材，提取核心观点
+2. 规划文章章节结构和每节要点
+你不能写正文，只写大纲。
+
+【输出要求】
+- 只输出一个 JSON 对象，不要任何解释文字，不要 markdown 代码块。
+- 必须使用英文双引号 "，不要用中文引号 “”。
+- 规划 3-5 个章节；material_ids 中的 ID 必须来自给定素材，没有合适素材时填 []。
+
+【完整示例】
+正确输出：
+{
+  "title": "新能源汽车出口为何高速增长",
+  "sections": [
+    {"section_title": "引言：出口数据创新高", "key_points": ["今年出口量同比增长", "在全球市场的排名变化"], "material_ids": ["m001"]},
+    {"section_title": "增长背后的三大动因", "key_points": ["产业链成本优势", "海外政策利好", "产品竞争力提升"], "material_ids": ["m002", "m003"]},
+    {"section_title": "结论与展望", "key_points": ["未来增长面临的挑战", "行业趋势判断"], "material_ids": ["m004"]}
+  ]
+}"""
+
+OUTLINE_USER = """写作意图：{user_intent}
+文章类型：{article_type}
+目标读者：{target_audience}
+目标字数：{word_count_target}字{feedback_section}
+
+可用素材摘要：
+{materials_summary}
+
+请规划文章大纲（3-5个章节），只输出 JSON 格式。"""
+
+DRAFT_SECTION_SYSTEM = """你是一名专业内容创作者（作家）。
+你的目标：根据大纲和素材，写出一段高质量的文章内容。
+你的工作背景：你有十年内容创作经验，文字流畅，善于用事实说话。
+
+你只能做以下事情：
+1. 根据给定的章节要点和素材写这一段内容
+2. 观点必须来自素材，不编造数据
+你不能修改大纲，不能写其他章节，只写当前这一段。
+
+【输出要求】
+- 直接返回这一段的正文文字，不要 JSON，不要章节标题，不要解释。
+- 不要写"以下是正文""本段内容如下"之类的开场白。
+- 段落要连贯成文，不要罗列要点。"""
+
+DRAFT_SECTION_USER = """文章标题：{article_title}
+当前章节：{section_title}
+章节要点：{key_points}
+
+相关素材：
+{section_materials}
+{revision_section}
+请写出这一章节的正文内容（{target_words}字左右）。"""

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/__init__.py

@@ -0,0 +1,63 @@
+from .base import BaseSearchProvider, SearchResult
+from .internet_search import InternetSearchProvider
+from .intranet_stub import IntranetSearchProvider
+from .demo_search import DemoSearchProvider
+from .duckduckgo_search import DuckDuckGoSearchProvider
+
+
+def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
+    """
+    config.yaml 中配置：
+    ai_writing:
+      search_provider: internet   # internet(默认,DDG) / intranet / tavily / demo
+      intranet_search_url: http://...
+      intranet_search_api_key: ""
+      intranet_field_map:
+        content: "abstract"
+        source: "system_name"
+      ddg_region: "cn-zh"         # DuckDuckGo 区域，默认 cn-zh
+
+    internet 模式：DuckDuckGo 为主搜索（免 API Key），失败时 fallback 到 demo 演示素材。
+    """
+    cfg = ai_writing_config or {}
+    provider = cfg.get("search_provider", "internet")
+
+    if provider == "intranet":
+        return IntranetSearchProvider(
+            base_url=cfg.get("intranet_search_url", ""),
+            api_key=cfg.get("intranet_search_api_key", ""),
+            field_map=cfg.get("intranet_field_map", {}),
+        )
+    if provider == "tavily":
+        return _FallbackSearchProvider(InternetSearchProvider(), DemoSearchProvider())
+    if provider == "demo":
+        return DemoSearchProvider()
+
+    # 默认 internet 模式：DuckDuckGo + demo fallback
+    region = cfg.get("ddg_region", "cn-zh")
+    return _FallbackSearchProvider(DuckDuckGoSearchProvider(region=region), DemoSearchProvider())
+
+
+class _FallbackSearchProvider(BaseSearchProvider):
+    """当主 provider 返回空结果时自动切换到 fallback provider。"""
+
+    def __init__(self, primary: BaseSearchProvider, fallback: BaseSearchProvider):
+        self._primary = primary
+        self._fallback = fallback
+
+    async def search(self, query: str, max_results: int = 5, **kwargs):
+        results = await self._primary.search(query, max_results=max_results, **kwargs)
+        if not results:
+            results = await self._fallback.search(query, max_results=max_results, **kwargs)
+        return results
+
+    async def health_check(self) -> bool:
+        return await self._primary.health_check() or await self._fallback.health_check()
+
+
+__all__ = [
+    "BaseSearchProvider", "SearchResult",
+    "InternetSearchProvider", "IntranetSearchProvider",
+    "DemoSearchProvider", "DuckDuckGoSearchProvider",
+    "get_search_provider",
+]

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/base.py

@@ -0,0 +1,40 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import List
+
+
+@dataclass
+class SearchResult:
+    """统一搜索结果格式 —— 内外网共用"""
+    title: str
+    content: str
+    source: str
+    published_at: str
+    url: str = ""
+    relevance_score: float = 0.0
+    source_type: str = "internet"
+    recUuid: str = ""
+    extra_metadata: dict = field(default_factory=dict)
+
+
+class BaseSearchProvider(ABC):
+    """搜索提供者抽象基类。
+
+    对接内网步骤：
+    1. 继承此类，创建 IntranetSearchProvider
+    2. 实现 search() —— 调用内网 API，将响应映射到 SearchResult
+    3. 在 config.yaml 中设置 ai_writing.search_provider: intranet
+    4. 无需修改任何 Agent 代码
+    """
+
+    @abstractmethod
+    async def search(
+        self,
+        query: str,
+        max_results: int = 5,
+        date_from: str = "",
+        date_to: str = "",
+    ) -> List[SearchResult]: ...
+
+    @abstractmethod
+    async def health_check(self) -> bool: ...

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/demo_search.py

@@ -0,0 +1,183 @@
+"""离线演示搜索提供者 —— 当无法访问互联网/内网时返回模拟素材，供测试完整写作流程。"""
+from __future__ import annotations
+
+import hashlib
+from typing import List
+
+from .base import BaseSearchProvider, SearchResult
+
+# 演示素材库：按话题分类，用于关键词模糊匹配
+_DEMO_CORPUS = [
+    {
+        "keywords": ["新能源", "电动车", "汽车", "ev", "electric"],
+        "items": [
+            {
+                "title": "2025年新能源汽车市场规模突破万亿，渗透率达45%",
+                "content": "据中国汽车工业协会数据，2025年前三季度新能源汽车累计销量超过850万辆，同比增长32%，市场渗透率达到45.3%。比亚迪、吉利、长城等头部企业份额持续集中，纯电与插混比例趋近5:5。",
+                "source": "中国汽车工业协会",
+                "published_at": "2025-10-15",
+                "relevance_score": 0.95,
+            },
+            {
+                "title": "固态电池量产进入倒计时：多家车企宣布2026年装车计划",
+                "content": "宁德时代、比亚迪、丰田同步推进固态电池量产。固态电池能量密度可达400Wh/kg，较现有液态锂电池提升60%，续航里程有望突破1000公里。成本仍是主要瓶颈，预计量产初期每度电成本高出30%。",
+                "source": "电池中国网",
+                "published_at": "2025-09-20",
+                "relevance_score": 0.88,
+            },
+            {
+                "title": "充电基础设施缺口分析：农村地区桩车比仅1:12",
+                "content": "截至2025年8月，全国充电桩保有量达1120万根，城市地区桩车比约1:2.8，但农村及县域地区桩车比仅1:12，制约新能源汽车下沉市场。国家电网计划2026年前新增农村快充站3万座。",
+                "source": "中国充电联盟",
+                "published_at": "2025-08-30",
+                "relevance_score": 0.82,
+            },
+        ],
+    },
+    {
+        "keywords": ["人工智能", "ai", "大模型", "llm", "机器学习", "深度学习"],
+        "items": [
+            {
+                "title": "国内大模型进入「百亿参数普惠化」阶段，推理成本下降90%",
+                "content": "2025年第三季度，国内主要大模型服务商将百亿参数模型推理价格降至每百万token不足1元，较2024年同期下降90%。开源生态加速，Qwen3、DeepSeek-V3等模型在多项基准测试中超越GPT-4o。",
+                "source": "中国信通院",
+                "published_at": "2025-10-08",
+                "relevance_score": 0.93,
+            },
+            {
+                "title": "企业AI应用落地白皮书：制造业ROI最高达300%",
+                "content": "调研覆盖500家企业的白皮书显示，AI在质检、排产、预测性维护三个场景ROI最高，部分制造企业ROI超300%。阻碍因素：数据孤岛（67%）、人才不足（54%）、安全合规（48%）。",
+                "source": "麦肯锡中国",
+                "published_at": "2025-09-12",
+                "relevance_score": 0.87,
+            },
+            {
+                "title": "AI写作助手用户规模达2.3亿，内容质量争议持续",
+                "content": "国内AI写作工具月活用户突破2.3亿，覆盖学术论文、营销文案、新闻报道等场景。争议焦点：AI生成内容的版权归属、事实准确性（幻觉率约15%）、对写作从业者就业影响。",
+                "source": "艾瑞咨询",
+                "published_at": "2025-10-01",
+                "relevance_score": 0.79,
+            },
+        ],
+    },
+    {
+        "keywords": ["碳中和", "双碳", "绿色", "低碳", "碳排放", "可再生能源", "光伏", "风电"],
+        "items": [
+            {
+                "title": "光伏装机量突破900GW，度电成本降至0.18元",
+                "content": "国家能源局数据显示，截至2025年9月，全国光伏累计装机量达912GW，占总装机的28%。光伏度电成本已降至0.18元，首次低于煤电基准价，推动更多省份出现「光伏平价」区域。",
+                "source": "国家能源局",
+                "published_at": "2025-10-12",
+                "relevance_score": 0.92,
+            },
+            {
+                "title": "碳市场价格突破200元/吨，倒逼高耗能行业转型",
+                "content": "全国碳排放权交易市场碳价已突破200元/吨，较2024年底上涨65%。水泥、电解铝、造纸等高耗能行业面临显著成本压力，碳减排投入年均增速达40%。",
+                "source": "上海环境能源交易所",
+                "published_at": "2025-09-28",
+                "relevance_score": 0.85,
+            },
+        ],
+    },
+    {
+        "keywords": ["医疗", "健康", "药物", "生物", "医药", "病"],
+        "items": [
+            {
+                "title": "AI辅助诊断落地三甲医院，影像科漏诊率降低23%",
+                "content": "北京协和、上海瑞金等20家三甲医院数据显示，引入AI辅助诊断系统后，肺结节、眼底病变早期漏诊率降低23%，放射科医生平均阅片时间减少40%。监管层面，NMPA已批准38款AI医疗器械。",
+                "source": "医学界",
+                "published_at": "2025-09-18",
+                "relevance_score": 0.91,
+            },
+        ],
+    },
+    {
+        "keywords": ["经济", "GDP", "贸易", "出口", "产业", "制造"],
+        "items": [
+            {
+                "title": "2025年前三季度GDP同比增长5.1%，消费贡献率提升至58%",
+                "content": "国家统计局公布，2025年前三季度GDP同比增长5.1%，最终消费对经济增长贡献率达58%，较去年同期提升8个百分点。工业增加值增长5.8%，其中装备制造业增长9.2%，成重要支撑。",
+                "source": "国家统计局",
+                "published_at": "2025-10-18",
+                "relevance_score": 0.90,
+            },
+            {
+                "title": "跨境电商出口额突破2.4万亿元，东南亚成最大增量市场",
+                "content": "商务部数据显示，2025年前三季度跨境电商出口额达2.4万亿元，同比增长28%。东南亚市场增速最快（+45%），中东、南美紧随其后。Temu、速卖通、Shopee三平台合计占出口份额约62%。",
+                "source": "商务部",
+                "published_at": "2025-10-05",
+                "relevance_score": 0.84,
+            },
+        ],
+    },
+]
+
+# 通用兜底素材（与任意查询都能凑合搭配）
+_FALLBACK_ITEMS = [
+    {
+        "title": "行业研究方法论：如何构建有效的一手与二手数据体系",
+        "content": "高质量行业报告需整合政府统计、上市公司财报、行业协会数据与专家访谈。一手调研样本量建议≥200份，二手数据优先选用权威机构发布的年度报告与季度白皮书，并注明数据时效性与口径差异。",
+        "source": "产业研究院",
+        "published_at": "2025-08-15",
+        "relevance_score": 0.55,
+    },
+    {
+        "title": "政策解读：2025年产业发展专项支持资金管理办法",
+        "content": "工信部发布2025年产业发展专项资金管理办法，重点支持「专精特新」企业数字化改造、绿色制造和关键技术攻关。单项目最高补贴1000万元，要求申报企业研发投入不低于营收3%。",
+        "source": "工信部",
+        "published_at": "2025-07-20",
+        "relevance_score": 0.50,
+    },
+    {
+        "title": "市场竞争格局分析：头部集中与长尾分散并存",
+        "content": "多个新兴行业呈现「哑铃型」竞争格局：头部3-5家企业合计市占率超60%，长尾中小企业依靠细分场景差异化生存，中间层企业面临最大竞争压力。并购整合预期升温，2025年战略性并购交易同比增长35%。",
+        "source": "普华永道",
+        "published_at": "2025-09-01",
+        "relevance_score": 0.48,
+    },
+]
+
+
+def _score_topic(query: str, keywords: list[str]) -> int:
+    """简单关键词匹配打分"""
+    q = query.lower()
+    return sum(1 for kw in keywords if kw in q)
+
+
+class DemoSearchProvider(BaseSearchProvider):
+    """离线演示搜索提供者：根据查询关键词从内置语料库返回模拟素材。
+
+    仅用于开发测试；生产环境应配置 intranet 提供者。
+    """
+
+    async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
+        best_topic = max(_DEMO_CORPUS, key=lambda t: _score_topic(query, t["keywords"]), default=None)
+        best_score = _score_topic(query, best_topic["keywords"]) if best_topic else 0
+
+        items = []
+        if best_score > 0 and best_topic:
+            items.extend(best_topic["items"])
+
+        # 用通用素材补齐
+        for fb in _FALLBACK_ITEMS:
+            if len(items) >= max_results:
+                break
+            items.append(fb)
+
+        results = []
+        for i, it in enumerate(items[:max_results]):
+            url_hash = hashlib.md5(it["title"].encode()).hexdigest()[:8]
+            results.append(SearchResult(
+                title=it["title"],
+                content=it["content"],
+                source=it["source"],
+                published_at=it["published_at"],
+                url=f"https://demo.internal/{url_hash}",
+                relevance_score=it["relevance_score"],
+                source_type="demo",
+                recUuid=f"demo-{i:03d}",
+            ))
+        return results
+
+    async def health_check(self) -> bool:
+        return True

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/duckduckgo_search.py

@@ -0,0 +1,76 @@
+"""DuckDuckGo 搜索提供者 —— 免 API Key 互联网搜索。"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import List
+
+from .base import BaseSearchProvider, SearchResult
+
+logger = logging.getLogger(__name__)
+
+
+class DuckDuckGoSearchProvider(BaseSearchProvider):
+    """使用 DuckDuckGo 进行互联网搜索，无需 API Key。"""
+
+    def __init__(self, region: str = "cn-zh", max_timeout: int = 30):
+        self.region = region
+        self.max_timeout = max_timeout
+
+    async def search(
+        self,
+        query: str,
+        max_results: int = 5,
+        date_from: str = "",
+        date_to: str = "",
+    ) -> List[SearchResult]:
+        try:
+            results = await asyncio.to_thread(
+                self._sync_search, query, max_results
+            )
+            return results
+        except Exception as e:
+            logger.warning("DuckDuckGo search failed for query=%r: %s", query, e)
+            return []
+
+    def _sync_search(self, query: str, max_results: int) -> List[SearchResult]:
+        try:
+            from ddgs import DDGS
+        except ImportError:
+            logger.error("ddgs not installed. Run: uv add ddgs")
+            return []
+
+        ddgs = DDGS(timeout=self.max_timeout)
+        try:
+            raw = ddgs.text(
+                query,
+                region=self.region,
+                safesearch="moderate",
+                max_results=max_results,
+            )
+            items = list(raw) if raw else []
+        except Exception as e:
+            logger.warning("DDGS.text failed: %s", e)
+            return []
+
+        results = []
+        for item in items:
+            url = item.get("href", item.get("link", ""))
+            source = url.split("/")[2] if url and "/" in url else "web"
+            results.append(SearchResult(
+                title=item.get("title", ""),
+                content=item.get("body", item.get("snippet", ""))[:500],
+                source=source,
+                published_at=item.get("published", ""),
+                url=url,
+                relevance_score=0.75,
+                source_type="internet",
+            ))
+        return results
+
+    async def health_check(self) -> bool:
+        try:
+            results = await self.search("test", max_results=1)
+            return len(results) > 0
+        except Exception:
+            return False

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/internet_search.py

@@ -0,0 +1,66 @@
+import asyncio
+import logging
+from typing import List
+
+from .base import BaseSearchProvider, SearchResult
+
+logger = logging.getLogger(__name__)
+
+
+class InternetSearchProvider(BaseSearchProvider):
+    """互联网搜索，使用项目已有的 Tavily 搜索工具。"""
+
+    async def search(
+        self,
+        query: str,
+        max_results: int = 5,
+        date_from: str = "",
+        date_to: str = "",
+    ) -> List[SearchResult]:
+        try:
+            results = await asyncio.to_thread(self._sync_search, query, max_results)
+            return results
+        except Exception as e:
+            logger.warning("internet search failed for query=%r: %s", query, e)
+            return []
+
+    def _sync_search(self, query: str, max_results: int) -> List[SearchResult]:
+        from deerflow.config import get_app_config
+        try:
+            from tavily import TavilyClient
+        except ImportError:
+            logger.warning("tavily not installed, falling back to demo search")
+            return []
+
+        config = get_app_config().get_tool_config("web_search")
+        api_key = None
+        if config is not None and "api_key" in config.model_extra:
+            api_key = config.model_extra.get("api_key")
+
+        if not api_key:
+            logger.warning("Tavily API key not configured, falling back to demo search")
+            return []
+
+        client = TavilyClient(api_key=api_key)
+        res = client.search(query, max_results=max_results)
+
+        results = []
+        for item in res.get("results", []):
+            content = item.get("content", "")
+            results.append(SearchResult(
+                title=item.get("title", ""),
+                content=content[:500],
+                source=item.get("url", "").split("/")[2] if "/" in item.get("url", "") else "web",
+                published_at=item.get("published_date", ""),
+                url=item.get("url", ""),
+                relevance_score=item.get("score", 0.0),
+                source_type="internet",
+            ))
+        return results
+
+    async def health_check(self) -> bool:
+        try:
+            results = await self.search("test", max_results=1)
+            return True
+        except Exception:
+            return False

offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py

@@ -0,0 +1,57 @@
+from typing import List
+
+from .base import BaseSearchProvider, SearchResult
+
+
+class IntranetSearchProvider(BaseSearchProvider):
+    """内网搜索接口适配器。
+
+    内网系统需提供 HTTP 接口，推荐格式：
+    POST /api/search
+    请求体：{"query": "...", "page_size": 5}
+    响应体：{"total": 100, "results": [...]}
+
+    对接步骤：
+    1. 取消下方注释代码，调整字段映射
+    2. 在 config.yaml 的 intranet_field_map 中配置字段映射
+    3. 将 search_provider 改为 intranet
+    """
+
+    def __init__(self, base_url: str, api_key: str = "", timeout: int = 10, field_map: dict = None):
+        self.base_url = base_url
+        self.api_key = api_key
+        self.timeout = timeout
+        self.field_map = field_map or {}
+
+    def _map(self, item: dict, field: str) -> str:
+        key = self.field_map.get(field, field)
+        return str(item.get(key, item.get(field, "")))
+
+    async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
+        # ── 对接内网时，取消以下注释 ──────────────────────────────────────
+        # import httpx
+        # headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
+        # async with httpx.AsyncClient(timeout=self.timeout) as client:
+        #     resp = await client.post(
+        #         f"{self.base_url}/api/search",
+        #         json={"query": query, "page_size": max_results},
+        #         headers=headers,
+        #     )
+        #     resp.raise_for_status()
+        #     items = resp.json().get("results", [])
+        # return [
+        #     SearchResult(
+        #         title=self._map(it, "title"),
+        #         content=self._map(it, "content")[:500],
+        #         source=self._map(it, "source"),
+        #         published_at=self._map(it, "published_at"),
+        #         relevance_score=float(it.get(self.field_map.get("relevance_score", "relevance_score"), 0)),
+        #         recUuid=self._map(it, "recUuid"),
+        #         source_type="intranet",
+        #     )
+        #     for it in items
+        # ]
+        return []
+
+    async def health_check(self) -> bool:
+        return False

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

@@ -0,0 +1,108 @@
+from typing import Annotated, List, Literal, Optional, TypedDict
+import operator
+
+# ── 素材 ──────────────────────────────────────────────────────────────────────
+
+class Material(TypedDict):
+    id: str
+    title: str
+    content: str           # 截断到 500 字
+    source: str
+    published_at: str
+    relevance_score: float
+    source_type: str       # "internet" | "intranet"
+    url: str
+    recUuid: str
+
+class MaterialPackage(TypedDict):
+    keywords: List[str]
+    materials: List[Material]
+    summary: str
+
+# ── 大纲 ──────────────────────────────────────────────────────────────────────
+
+class OutlineSection(TypedDict):
+    section_title: str
+    key_points: List[str]
+    material_ids: List[str]
+
+class Outline(TypedDict):
+    title: str
+    sections: List[OutlineSection]
+
+# ── 草稿 ──────────────────────────────────────────────────────────────────────
+
+class DraftArticle(TypedDict):
+    title: str
+    outline: Outline
+    full_markdown: str
+    word_count: int
+    revision_count: int
+
+# ── 编辑审核 ──────────────────────────────────────────────────────────────────
+
+class ReviewIssue(TypedDict):
+    category: str    # "fact" | "logic" | "language"
+    severity: str    # "major" | "minor"
+    location: str
+    issue: str
+    suggestion: str
+
+class ReviewResult(TypedDict):
+    verdict: str     # "pass" | "reject"
+    fact_score: int
+    logic_score: int
+    language_score: int
+    overall_score: int
+    pass_threshold: int
+    revision_notes: List[ReviewIssue]
+
+# ── 用户干预 ──────────────────────────────────────────────────────────────────
+
+class UserInterventionPayload(TypedDict):
+    action: str
+    approved_material_ids: List[str]
+    extra_keywords: List[str]
+    edited_outline: Optional[Outline]
+    outline_feedback: str
+    user_revision_notes: str
+    override_verdict: bool
+
+# ── 主状态 ────────────────────────────────────────────────────────────────────
+
+WritingStatus = Literal[
+    "researching", "awaiting_material_confirm",
+    "writing_outline", "awaiting_outline_confirm",
+    "writing_draft", "awaiting_draft_confirm",
+    "reviewing", "awaiting_review_confirm",
+    "revising", "done", "error",
+]
+
+class AIWritingState(TypedDict):
+    # 写作配置
+    user_intent: str
+    article_type: str
+    target_audience: str
+    word_count_target: int
+
+    # Agent A 输出
+    material_package: Optional[MaterialPackage]
+
+    # Agent B 输出
+    current_outline: Optional[Outline]
+    drafts: Annotated[List[DraftArticle], operator.add]
+    current_draft: Optional[DraftArticle]
+
+    # Agent C 输出
+    review_result: Optional[ReviewResult]
+
+    # 用户干预数据（每个干预点会覆盖）
+    last_user_intervention: Optional[UserInterventionPayload]
+
+    # 流程控制
+    revision_count: int
+    max_revisions: int
+    status: WritingStatus
+
+    # SSE 进度推送队列（每个节点追加）
+    progress_events: Annotated[List[dict], operator.add]