wsk/deerflow-code

0 Fork 0

Files

T

董宏凯 cc8703577d ai写作

2026-05-17 13:08:45 +08:00

80 KiB

Raw Blame History

AI 智能写作 —— 杂志社多 Agent 协作系统（含用户干预 & 详细进度）

目标：通过角色扮演 + 流程编排，让性能较弱的本地大模型产出高质量文章；同时在每个关键节点暂停等待用户确认，确保最终结果符合预期。

整体架构
用户干预节点设计（核心）
开源参考项目
弱模型增强策略
三个 Agent 的角色设计
数据结构定义
后端实现（分步）
前端实现（分步）
内网接口预留设计
LangGraph 状态机流转图
提示词模板（弱模型专用）
SSE 协议 & 干预协议设计
前端 UI 组件详细设计
文件目录结构
分步开发检查清单

1. 整体架构

┌──────────────────────────────────────────────────────────────────────┐
│                         前端（现有页面）                               │
│  frontend-web/src/open-canvas/                                        │
│                                                                        │
│  ┌───────────────────────────┐  ┌──────────────────────────────────┐  │
│  │      AI写作左侧面板         │  │      CanvasArtifactPanel（右侧）  │  │
│  │  AIWritingPanel.tsx        │  │  （右侧保持不变，实时展示文章草稿）  │  │
│  │  ┌─────────────────────┐  │  │                                  │  │
│  │  │  1. 写作意图输入区    │  │  │  ┌────────────────────────────┐ │  │
│  │  ├─────────────────────┤  │  │  │  实时文章草稿（Markdown）     │ │  │
│  │  │  2. 进度时间线        │  │  │  │  每次修改都保留版本历史       │ │  │
│  │  │  ┌───────────────┐  │  │  │  └────────────────────────────┘ │  │
│  │  │  │ ✅ 研究院完成  │  │  │  │                                  │  │
│  │  │  │ 素材卡片列表▼ │  │  │  │                                  │  │
│  │  │  │ [可收起展开]  │  │  │  │                                  │  │
│  │  │  ├───────────────┤  │  │  │                                  │  │
│  │  │  │ ⏸ 等待确认    │  │  │  │                                  │  │
│  │  │  │ [用户干预卡片]│  │  │  │                                  │  │
│  │  │  ├───────────────┤  │  │  │                                  │  │
│  │  │  │ ✅ 大纲已生成 │  │  │  │                                  │  │
│  │  │  │ 大纲预览▼    │  │  │  │                                  │  │
│  │  │  │ [可收起展开]  │  │  │  │                                  │  │
│  │  │  ├───────────────┤  │  │  │                                  │  │
│  │  │  │ ⏸ 等待确认    │  │  │  │                                  │  │
│  │  │  └───────────────┘  │  │  │                                  │  │
│  └───────────────────────────┘  └──────────────────────────────────┘  │
└───────────────────────────────┬──────────────────────────────────────┘
                                │ SSE 流式推送（进度/干预/草稿）
                                │ REST 干预响应（用户决策提交）
                                ▼
┌──────────────────────────────────────────────────────────────────────┐
│              后端新路由（独立，URL 前缀 /api/ai-writing/）              │
│  app/gateway/routers/ai_writing.py                                    │
│  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     ← 查询当前流程状态                │
└───────────────────────────────┬──────────────────────────────────────┘
                                │
                                ▼
┌──────────────────────────────────────────────────────────────────────┐
│         AI Writing LangGraph Graph（独立于原有 Graph）                 │
│  packages/harness/deerflow/agents/ai_writing/graph.py                 │
│                                                                        │
│  intent_parser → researcher → [PAUSE:材料确认] → writer(outline)      │
│      → [PAUSE:大纲确认] → writer(draft) → editor                      │
│      → [PAUSE:终稿确认] → END                                          │
│                  ↑__________________________|（打回重写，最多3次）      │
└──────────────────────────────────────────────────────────────────────┘

2. 用户干预节点设计（核心）

整个流程共有 4 个用户干预点，每个干预点都会暂停流程，向前端推送 await_user 事件，等待用户操作后再恢复。

干预点总览

流程步骤                    干预点    用户可执行的操作
─────────────────────────   ──────    ───────────────────────────────────────
① 研究院搜集素材完成         PAUSE-1   确认素材 / 补充关键词重搜 / 手动删除素材
② 作家生成大纲完成           PAUSE-2   确认大纲 / 编辑大纲内容 / 要求重新规划
③ 作家完成初稿（每次修改）    PAUSE-3   确认定稿 / 给出修改意见 / 直接编辑文本
④ 编辑审核完成（打回时）      PAUSE-4   采纳编辑意见 / 忽略编辑意见强制定稿

PAUSE-1：素材确认

触发：研究院搜集完素材后暂停

前端展示：

素材卡片列表（默认收起，点击展开）
每张卡片：标题、来源、时间、相关度色条、正文摘要（可展开）
每张卡片有勾选框，支持删除不相关素材
"补充关键词"输入框，可追加搜索

用户操作选项：

[确认素材，开始写大纲]   ← 主操作，使用当前勾选素材继续
[重新搜索]              ← 清空素材重新来
[补充搜索: ___________] ← 追加关键词，合并到现有素材

PAUSE-2：大纲确认

触发：作家生成大纲后暂停

前端展示：

大纲树形视图（默认展开）
每个章节可直接点击编辑标题和要点
可拖拽调整章节顺序（可选）

用户操作选项：

[确认大纲，开始写作]     ← 使用当前大纲（含用户修改）继续
[重新规划大纲]           ← 清空大纲重新生成，可附加意见: ___

PAUSE-3：草稿确认

触发：每次作家完成草稿后暂停（包括修改稿）

前端展示：

右侧实时显示完整草稿
左侧显示"编辑打分 + 问题清单"（如已经过编辑）
版本切换导航（第1稿 / 第2稿 …）

用户操作选项：

[确认定稿]                      ← 以此为最终文章
[给出修改意见: ____________]    ← 用户自己写修改意见，绕过编辑流程直接指导重写
[让编辑继续审核]                ← 走编辑流程（默认）

PAUSE-4：编辑审核结果确认（仅在打回时）

触发：编辑判断"打回"且未达到最大循环次数

前端展示：

编辑评分 + 分项得分（事实/逻辑/语言）
问题清单（按严重程度排序）
每个问题条目：位置 + 问题描述 + 修改建议

用户操作选项：

[采纳编辑意见，让作家重写]      ← 走编辑反馈重写
[强制定稿（忽略编辑意见）]      ← 用当前草稿结束
[补充修改指令: ____________]    ← 用户自己的意见覆盖编辑意见

3. 开源参考项目

Stanford STORM（最核心参考）

项目：https://github.com/stanford-oval/storm
借鉴：Researcher + Writer + Editor 完整协作流程；素材 → 大纲 → 分段写作的两步写法；短 prompt + JSON 输出模式（弱模型友好）

GPT-Researcher

项目：https://github.com/assafelovic/gpt-researcher
借鉴：统一搜索接口抽象（context/retriever.py）；素材字段定义（标题/来源/时间/相关度）

CrewAI

项目：https://github.com/crewAIInc/crewAI
借鉴：role + goal + backstory 三段式角色锁定 Prompt；expected_output 强制 JSON 输出；任务链传参方式

LangGraph Human-in-the-loop（项目已有，重点参考）

文档：https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/
借鉴：interrupt() 实现流程暂停等待用户输入；Command(resume=...) 恢复流程；MemorySaver checkpointing 持久化状态

4. 弱模型增强策略

策略	实现方式	效果
任务最小化	每次 LLM 调用只做一件事，禁止混合任务	减少跑偏概率
角色锁定	System Prompt 开头：「你是X，你只做Y，不做其他」	防止角色混淆
强制 JSON	`response_format=json_object` + Prompt 末尾提供 JSON 模板	输出可解析，有 fallback
上下文压缩	传给下游的素材截断到 200 字摘要，不传原文	防止 context 溢出
两步写作	先写大纲（JSON），确认后再逐段写正文	防止弱模型一次生成长文失控
分项审查	编辑三项独立打分（事实/逻辑/语言），不混在一起	每次调用任务极简
Few-Shot 示例	每个 Agent 的 Prompt 内置 1 个期望输出的完整 JSON 示例	模型直接模仿，输出更稳定
迭代上限保护	编辑打回最多 3 次，超出强制定稿	防死循环
用户干预兜底	每个关键节点人工确认	彻底避免方向跑偏

5. 三个 Agent 的角色设计

Agent A —— 研究院（Researcher）

角色：专业情报研究员
单次调用任务：① 只拆关键词  ② 只排序素材（两次独立调用）
输入：用户写作意图
输出：MaterialPackage（结构化素材包）
工具：web_search（当前） / intranet_search（预留）

两步拆分（弱模型关键）：

Step 1：extract_keywords(intent) → {"keywords": [...]}（仅此一件事）
Step 2：对每个关键词独立搜索（并发），合并去重，截断到 500 字

输出字段：

{
  "keywords": ["关键词1", "关键词2"],
  "materials": [
    {
      "id": "m001",
      "title": "文章标题",
      "content": "正文摘要（最多500字）",
      "source": "来源域名或系统名称",
      "published_at": "2024-01-15",
      "relevance_score": 0.92,
      "source_type": "internet",
      "url": "https://...",
      "recUuid": "",
    }
  ],
  "summary": "共搜索到X条素材，精选Y条"
}

Agent B —— 作家（Writer）

角色：专业内容创作者
Phase 1：write_outline → 输出大纲 JSON（暂停等用户确认）
Phase 2：write_draft  → 按段落逐段写正文（逐段 SSE 推送）
输入：MaterialPackage + 用户意图 + 确认后的大纲
      （修改稿时额外附加：编辑意见 or 用户修改意见）
输出：DraftArticle（含 Markdown 全文）

Agent C —— 编辑（Editor）

角色：资深主编，标准极高
任务：三项独立审查 → 综合评分 → 通过/打回决定
      每项审查是独立的 LLM 调用（弱模型关键）
输入：DraftArticle + MaterialPackage（用于核实事实）
输出：ReviewResult（评分 + 分项问题清单）

三次独立 LLM 调用：

review_facts(draft, materials) → {"score": 85, "issues": [...]}
review_logic(draft) → {"score": 80, "issues": [...]}
review_language(draft) → {"score": 90, "issues": [...]}

6. 数据结构定义

6.1 后端 LangGraph State（Python）

文件：packages/harness/deerflow/agents/ai_writing/state.py

from typing import TypedDict, List, Optional, Annotated, Literal
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           # 知识库ID

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          # 见各干预点定义
    # PAUSE-1 素材确认
    approved_material_ids: List[str]   # 用户勾选保留的素材 ID
    extra_keywords: List[str]          # 补充搜索关键词
    # PAUSE-2 大纲确认
    edited_outline: Optional[Outline]  # 用户编辑后的大纲（None 表示未修改）
    outline_feedback: str              # 要求重新规划的意见
    # PAUSE-3 草稿确认
    user_revision_notes: str           # 用户自己的修改意见
    # PAUSE-4 编辑审核确认
    override_verdict: bool             # True = 强制定稿，False = 采纳编辑意见

# ── 主状态 ────────────────────────────────────────────────────────────
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       # 默认 3
    status: WritingStatus

    # SSE 进度推送队列（每个节点追加）
    progress_events: Annotated[List[dict], operator.add]

6.2 前端类型（TypeScript）

文件：frontend-web/src/open-canvas/types/ai-writing.ts

// ── 素材 ─────────────────────────────────────────────────────────────
export interface AIMaterial {
  id: string;
  title: string;
  content: string;
  source: string;
  publishedAt: string;
  relevanceScore: number;
  sourceType: 'internet' | 'intranet';
  url?: string;
  docId?: string;
  department?: string;
  classification?: string;
}

// ── 大纲 ─────────────────────────────────────────────────────────────
export interface OutlineSection {
  sectionTitle: string;
  keyPoints: string[];
  materialIds: string[];
}

export interface AIOutline {
  title: string;
  sections: OutlineSection[];
}

// ── 编辑审核 ──────────────────────────────────────────────────────────
export interface ReviewIssue {
  category: 'fact' | 'logic' | 'language';
  severity: 'major' | 'minor';
  location: string;
  issue: string;
  suggestion: string;
}

export interface ReviewResult {
  verdict: 'pass' | 'reject';
  factScore: number;
  logicScore: number;
  languageScore: number;
  overallScore: number;
  passThreshold: number;
  revisionNotes: ReviewIssue[];
}

// ── 写作状态 ──────────────────────────────────────────────────────────
export type WritingStatus =
  | 'idle'
  | 'researching'
  | 'awaiting_material_confirm'
  | 'writing_outline'
  | 'awaiting_outline_confirm'
  | 'writing_draft'
  | 'awaiting_draft_confirm'
  | 'reviewing'
  | 'awaiting_review_confirm'
  | 'revising'
  | 'done'
  | 'error';

// ── SSE 进度事件 ──────────────────────────────────────────────────────
export type ProgressEventType =
  | 'step_start'          // Agent 开始执行某步骤
  | 'step_done'           // Agent 完成某步骤
  | 'materials_ready'     // 素材搜集完毕
  | 'outline_ready'       // 大纲生成完毕
  | 'draft_chunk'         // 草稿文字流式片段（逐段）
  | 'draft_ready'         // 草稿完整版本
  | 'review_ready'        // 编辑审核完毕
  | 'await_user'          // 暂停，等待用户干预
  | 'resumed'             // 用户干预已提交，流程恢复
  | 'done'                // 全部完成
  | 'error';              // 出错

export interface ProgressEvent {
  type: ProgressEventType;
  sessionId: string;
  agentName?: string;    // '研究院' | '作家' | '编辑' | '系统'
  message: string;
  detail?: string;
  // type===materials_ready 时
  materials?: AIMaterial[];
  keywords?: string[];
  materialSummary?: string;
  // type===outline_ready 时
  outline?: AIOutline;
  // type===draft_chunk / draft_ready 时
  draftMarkdown?: string;
  draftTitle?: string;
  revisionCount?: number;
  // type===review_ready 时
  reviewResult?: ReviewResult;
  // type===await_user 时
  pausePoint?: 'material_confirm' | 'outline_confirm' | 'draft_confirm' | 'review_confirm';
  pauseContext?: Record<string, unknown>;  // 干预点所需的上下文数据
}

// ── 用户干预提交体 ────────────────────────────────────────────────────
export interface UserInterventionPayload {
  action: string;
  // PAUSE-1
  approvedMaterialIds?: string[];
  extraKeywords?: string[];
  // PAUSE-2
  editedOutline?: AIOutline;
  outlineFeedback?: string;
  // PAUSE-3
  userRevisionNotes?: string;
  // PAUSE-4
  overrideVerdict?: boolean;
}

// ── 写作请求 ──────────────────────────────────────────────────────────
export interface AIWritingRequest {
  userIntent: string;
  articleType: 'news' | 'analysis' | 'feature' | 'report';
  targetAudience: string;
  wordCountTarget: number;
  extraContext?: string;
}

// ── Hook 返回值 ───────────────────────────────────────────────────────
export interface AIWritingState {
  sessionId: string | null;
  status: WritingStatus;
  progressEvents: ProgressEvent[];
  materials: AIMaterial[];
  outline: AIOutline | null;
  currentDraftMarkdown: string;
  currentDraftTitle: string;
  reviewResult: ReviewResult | null;
  revisionCount: number;
  currentPause: ProgressEvent | null;  // 当前干预点事件（null 表示不在暂停状态）
  error: string | null;
}

7. 后端实现（分步）

步骤 7.1：目录结构

packages/harness/deerflow/agents/ai_writing/
├── __init__.py
├── state.py
├── graph.py
├── nodes/
│   ├── __init__.py
│   ├── intent_parser.py
│   ├── researcher.py
│   ├── writer_outline.py     # 只写大纲（独立节点）
│   ├── writer_draft.py       # 只写正文（独立节点）
│   └── editor.py
├── search/
│   ├── __init__.py
│   ├── base.py               # 抽象接口（内网预留）
│   ├── internet_search.py
│   └── intranet_stub.py
└── prompts/
    ├── researcher_prompts.py
    ├── writer_prompts.py
    └── editor_prompts.py

app/gateway/routers/
└── ai_writing.py             # 新建，不修改其他路由文件

步骤 7.2：搜索接口抽象（内网预留）

文件：packages/harness/deerflow/agents/ai_writing/search/base.py

from abc import ABC, abstractmethod
from typing import List
from dataclasses import dataclass, field

@dataclass
class SearchResult:
    """统一搜索结果格式 —— 内外网共用，对接内网只需映射到此结构"""
    title: str
    content: str
    source: str
    published_at: str
    url: str = ""
    relevance_score: float = 0.0
    source_type: str = "internet"   # "internet" | "intranet"
    # ── 内网扩展字段（按需填充）──────────────────────────────────────
    recUuid: str = ""                # 知识库 ID
    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: ...

内网接口桩：packages/harness/deerflow/agents/ai_writing/search/intranet_stub.py

from typing import List
from .base import BaseSearchProvider, SearchResult

class IntranetSearchProvider(BaseSearchProvider):
    """
    内网搜索接口适配器
    ═══════════════════════════════════════════════════════════
    内网系统需提供 HTTP 接口，推荐格式：

    POST /api/search
    Content-Type: application/json
    Authorization: Bearer {api_key}   （若需认证）

    请求体：
    {
        "query": "搜索关键词",
        "page_size": 5,
        "date_from": "2024-01-01",    （可选）
        "date_to": "2024-12-31"       （可选）
    }

    响应体：
    {
        "total": 100,
        "results": [
            {
                "title":           "文档标题",
                "content":         "摘要（建议500字以内）",
                "source":          "来源系统名称",
                "published_at":    "2024-01-15",
                "recUuid":          "DOC-2024-001",
                "url":             "intranet://...",   （可选）
                "relevance_score": 0.92                （可选，无则传0）
            }
        ]
    }

    字段名称不同时，在 config.yaml 的 intranet_field_map 中配置映射。
    ═══════════════════════════════════════════════════════════
    """

    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_field(self, item: dict, standard_field: str) -> str:
        """将内网字段名映射到标准字段名"""
        intranet_field = self.field_map.get(standard_field, standard_field)
        return item.get(intranet_field, item.get(standard_field, ""))

    async def search(self, query: str, max_results: int = 5, **kwargs) -> List[SearchResult]:
        # ── TODO：对接内网时，取消以下注释并调整 ──────────────────────
        # 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_field(it, "title"),
        #         content=self._map_field(it, "content")[:500],
        #         source=self._map_field(it, "source"),
        #         published_at=self._map_field(it, "published_at"),
        #         relevance_score=float(it.get(
        #             self.field_map.get("relevance_score", "relevance_score"), 0)),
        #         recUuid=self._map_field(it, "recUuid"),
        #         source_type="intranet",
        #     )
        #     for it in items
        # ]
        # ─────────────────────────────────────────────────────────────
        return []  # 未配置时返回空，不影响互联网模式

    async def health_check(self) -> bool:
        return False

搜索工厂：packages/harness/deerflow/agents/ai_writing/search/__init__.py

from .base import BaseSearchProvider, SearchResult
from .internet_search import InternetSearchProvider
from .intranet_stub import IntranetSearchProvider

def get_search_provider(ai_writing_config: dict) -> BaseSearchProvider:
    """
    config.yaml 中配置：
    ai_writing:
      search_provider: internet   # 或 intranet
      intranet_search_url: http://...
      intranet_search_api_key: ""
      intranet_field_map:
        content: "abstract"
        source: "system_name"
    """
    provider = ai_writing_config.get("search_provider", "internet")
    if provider == "intranet":
        return IntranetSearchProvider(
            base_url=ai_writing_config.get("intranet_search_url", ""),
            api_key=ai_writing_config.get("intranet_search_api_key", ""),
            field_map=ai_writing_config.get("intranet_field_map", {}),
        )
    return InternetSearchProvider()

步骤 7.3：LangGraph Graph 与 Interrupt 机制

文件：packages/harness/deerflow/agents/ai_writing/graph.py

import asyncio
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command
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：素材确认干预点"""
    user_input = interrupt({
        "pause_point": "material_confirm",
        "message": "研究院已完成素材搜集，请确认素材后继续。",
        "materials": state["material_package"]["materials"],
        "keywords": state["material_package"]["keywords"],
        "summary": state["material_package"]["summary"],
    })
    # 用户提交后，user_input 是 UserInterventionPayload
    # 根据用户操作过滤素材
    action = user_input.get("action", "confirm")
    if action == "confirm":
        approved_ids = set(user_input.get("approvedMaterialIds", [
            m["id"] for m in state["material_package"]["materials"]
        ]))
        filtered_materials = [
            m for m in state["material_package"]["materials"]
            if m["id"] in approved_ids
        ]
        updated_pkg = {**state["material_package"], "materials": filtered_materials}
        return {
            "material_package": updated_pkg,
            "last_user_intervention": user_input,
            "status": "writing_outline",
            "progress_events": [{"type": "resumed", "message": "用户已确认素材，开始规划大纲"}]
        }
    # action == "re_search"：重新搜索（返回 researcher 节点，通过路由控制）
    extra_keywords = user_input.get("extraKeywords", [])
    return {
        "last_user_intervention": {**user_input, "_extra_keywords": extra_keywords},
        "status": "researching",
        "progress_events": [{"type": "resumed", "message": "重新搜索中..."}]
    }

async def pause_outline_confirm(state: AIWritingState) -> dict:
    """PAUSE-2：大纲确认干预点"""
    user_input = interrupt({
        "pause_point": "outline_confirm",
        "message": "作家已完成大纲规划，请确认或编辑大纲后继续。",
        "outline": state["current_outline"],
    })
    action = user_input.get("action", "confirm")
    if action == "confirm":
        edited_outline = user_input.get("editedOutline") or state["current_outline"]
        return {
            "current_outline": edited_outline,
            "last_user_intervention": user_input,
            "status": "writing_draft",
            "progress_events": [{"type": "resumed", "message": "用户已确认大纲，开始写作"}]
        }
    # action == "re_outline"
    feedback = user_input.get("outlineFeedback", "")
    return {
        "last_user_intervention": {**user_input, "_feedback": feedback},
        "status": "writing_outline",
        "progress_events": [{"type": "resumed", "message": f"重新规划大纲（意见：{feedback[:50]}）"}]
    }

async def pause_draft_confirm(state: AIWritingState) -> dict:
    """PAUSE-3：草稿确认干预点"""
    user_input = interrupt({
        "pause_point": "draft_confirm",
        "message": "草稿已完成，请确认或给出修改意见。",
        "draft_title": state["current_draft"]["title"],
        "revision_count": state["current_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", "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", "message": f"按用户意见重写（{notes[:50]}）"}]
        }
    # action == "to_editor"：走编辑流程
    return {
        "last_user_intervention": user_input,
        "status": "reviewing",
        "progress_events": [{"type": "resumed", "message": "提交编辑审核"}]
    }

async def pause_review_confirm(state: AIWritingState) -> dict:
    """PAUSE-4：编辑打回确认（仅 verdict==reject 时触发）"""
    user_input = 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"):
        return {
            "last_user_intervention": user_input,
            "status": "done",
            "progress_events": [{"type": "resumed", "message": "用户强制定稿"}]
        }
    # 采纳编辑意见 or 附加用户意见
    extra_notes = user_input.get("userRevisionNotes", "")
    return {
        "last_user_intervention": {**user_input, "_extra_notes": extra_notes},
        "revision_count": state.get("revision_count", 0) + 1,
        "status": "revising",
        "progress_events": [{"type": "resumed", "message": "按编辑意见重写"}]
    }

# ── 条件路由 ──────────────────────────────────────────────────────────

def after_material_pause(state: AIWritingState) -> str:
    if state["status"] == "researching":
        return "researcher"  # 重新搜索
    return "writer_outline"

def after_outline_pause(state: AIWritingState) -> str:
    if state["status"] == "writing_outline":
        return "writer_outline"  # 重新规划
    return "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", {})
    if review.get("verdict") == "pass":
        return "pause_draft_confirm"  # 编辑通过 → 草稿确认
    rev_count = state.get("revision_count", 0)
    max_rev = state.get("max_revisions", 3)
    if rev_count >= max_rev:
        return "pause_draft_confirm"  # 超出次数强制到草稿确认
    return "pause_review_confirm"     # 编辑打回 → 审核确认

def after_review_pause(state: AIWritingState) -> str:
    if state["status"] == "done":
        return "end"
    return "writer_draft"  # 重写

# ── 组装 Graph ────────────────────────────────────────────────────────

def build_ai_writing_graph():
    graph = StateGraph(AIWritingState)

    # 节点
    graph.add_node("intent_parser",        intent_parser_node)
    graph.add_node("researcher",           researcher_node)
    graph.add_node("pause_material",       pause_material_confirm)   # PAUSE-1
    graph.add_node("writer_outline",       writer_outline_node)
    graph.add_node("pause_outline",        pause_outline_confirm)    # PAUSE-2
    graph.add_node("writer_draft",         writer_draft_node)
    graph.add_node("pause_draft",          pause_draft_confirm)       # PAUSE-3
    graph.add_node("editor",               editor_node)
    graph.add_node("pause_review",         pause_review_confirm)     # PAUSE-4

    # 边
    graph.set_entry_point("intent_parser")
    graph.add_edge("intent_parser", "researcher")
    graph.add_edge("researcher",    "pause_material")

    graph.add_conditional_edges("pause_material", after_material_pause,
                                 {"researcher": "researcher", "writer_outline": "writer_outline"})

    graph.add_edge("writer_outline", "pause_outline")
    graph.add_conditional_edges("pause_outline", after_outline_pause,
                                 {"writer_outline": "writer_outline", "writer_draft": "writer_draft"})

    graph.add_edge("writer_draft", "pause_draft")
    graph.add_conditional_edges("pause_draft", after_draft_pause,
                                 {"end": END, "writer_draft": "writer_draft", "editor": "editor"})

    graph.add_conditional_edges("editor", after_review,
                                 {"pause_draft_confirm": "pause_draft",
                                  "pause_review_confirm": "pause_review"})

    graph.add_conditional_edges("pause_review", after_review_pause,
                                 {"end": END, "writer_draft": "writer_draft"})

    # 使用 MemorySaver 支持 interrupt/resume（生产环境换成 PostgresSaver）
    checkpointer = MemorySaver()
    return graph.compile(checkpointer=checkpointer, interrupt_before=None)

AI_WRITING_GRAPH = build_ai_writing_graph()

步骤 7.4：后端 HTTP 路由

文件：app/gateway/routers/ai_writing.py

import asyncio
import json
import uuid
from typing import AsyncGenerator, Optional
from fastapi import APIRouter, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from langgraph.types import Command
from deerflow.agents.ai_writing.graph import AI_WRITING_GRAPH

router = APIRouter()

# ── 请求/响应模型 ─────────────────────────────────────────────────────

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

# ── 工具函数 ──────────────────────────────────────────────────────────

def sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"

def _make_thread_config(session_id: str, model_name: str = ""):
    from deerflow.models.factory import get_default_llm
    llm = get_default_llm(model_name=model_name or None)
    return {
        "configurable": {
            "thread_id": session_id,
            "llm": llm,
            "ai_writing_config": {},  # 从全局 config 读取 search_provider 等
        }
    }

# ── 路由 ──────────────────────────────────────────────────────────────

@router.post("/start", response_model=StartWritingResponse)
async def start_writing(body: StartWritingRequest):
    """启动写作流程，返回 session_id（后续用此 ID 监听 SSE 和提交干预）"""
    session_id = str(uuid.uuid4())
    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": [],
    }
    config = _make_thread_config(session_id, body.model_name)

    # 异步启动 Graph（不等待完成，通过 SSE 推送进度）
    asyncio.create_task(_run_graph(session_id, initial_state, config))
    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):
    """用户提交干预决策，恢复暂停的流程"""
    config = _make_thread_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,
    }
    # LangGraph Command(resume=...) 恢复被 interrupt() 暂停的流程
    asyncio.create_task(
        AI_WRITING_GRAPH.ainvoke(Command(resume=intervention), config=config)
    )
    return {"message": "已提交，流程恢复中"}


@router.get("/{session_id}/state")
async def get_state(session_id: str):
    """查询当前流程状态快照"""
    config = _make_thread_config(session_id)
    state = await AI_WRITING_GRAPH.aget_state(config)
    if not state:
        raise HTTPException(status_code=404, detail="Session not found")
    return state.values


# ── 内部辅助 ──────────────────────────────────────────────────────────

# 每个 session 的事件队列
_event_queues: dict[str, asyncio.Queue] = {}

async def _run_graph(session_id: str, initial_state: dict, config: dict):
    """后台运行 Graph，将节点输出推入事件队列"""
    queue = asyncio.Queue()
    _event_queues[session_id] = queue

    try:
        async for chunk in AI_WRITING_GRAPH.astream(
            initial_state, config=config, stream_mode="updates"
        ):
            node_name = list(chunk.keys())[0]
            node_output = chunk[node_name]

            # 推送 progress_events
            for evt in node_output.get("progress_events", []):
                await queue.put(("progress", evt))

            # 素材就绪
            if pkg := node_output.get("material_package"):
                await queue.put(("materials_ready", {
                    "keywords": pkg["keywords"],
                    "materials": pkg["materials"],
                    "summary": pkg["summary"],
                }))

            # 大纲就绪
            if outline := node_output.get("current_outline"):
                await queue.put(("outline_ready", {"outline": outline}))

            # 草稿就绪
            if draft := node_output.get("current_draft"):
                await queue.put(("draft_ready", {
                    "title": draft["title"],
                    "markdown": draft["full_markdown"],
                    "revision_count": draft["revision_count"],
                }))

            # 编辑审核就绪
            if review := node_output.get("review_result"):
                await queue.put(("review_ready", {"review_result": review}))

            # interrupt() 触发 → 推送 await_user 事件
            # LangGraph 在 interrupt 时会抛出特殊异常，
            # 通过 stream_mode="updates" 时以 "__interrupt__" 节点出现
            if node_name == "__interrupt__":
                interrupt_data = node_output.get("value", {})
                await queue.put(("await_user", interrupt_data))

        await queue.put(("done", {"message": "写作完成"}))

    except Exception as e:
        await queue.put(("error", {"message": str(e)}))
    finally:
        await queue.put(None)  # 哨兵，关闭流


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, {"session_id": session_id, **data})
        await asyncio.sleep(0)

    _event_queues.pop(session_id, None)

注册路由（在 app/gateway/app.py 末尾追加，不修改已有代码）：

from app.gateway.routers import ai_writing
app.include_router(ai_writing.router, prefix="/api/ai-writing", tags=["ai-writing"])

8. 前端实现（分步）

步骤 8.1：API 客户端

文件：frontend-web/src/open-canvas/api/ai-writing.ts

import type { AIWritingRequest, ProgressEvent, UserInterventionPayload } from '../types/ai-writing';

const BASE = '/api/ai-writing';

/** 启动写作流程，返回 session_id */
export async function startWritingSession(req: AIWritingRequest): Promise<string> {
  const resp = await fetch(`${BASE}/start`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      user_intent: req.userIntent,
      article_type: req.articleType,
      target_audience: req.targetAudience,
      word_count_target: req.wordCountTarget,
    }),
  });
  if (!resp.ok) throw new Error(`启动失败: ${resp.status}`);
  const data = await resp.json();
  return data.session_id;
}

/** 监听 SSE 进度流 */
export function subscribeToProgress(
  sessionId: string,
  onEvent: (event: ProgressEvent) => void,
  signal?: AbortSignal,
): () => void {
  const es = new EventSource(`${BASE}/${sessionId}/stream`);

  const eventTypes = [
    'progress', 'materials_ready', 'outline_ready',
    'draft_ready', 'review_ready', 'await_user',
    'resumed', 'done', 'error',
  ] as const;

  eventTypes.forEach((type) => {
    es.addEventListener(type, (e: MessageEvent) => {
      try {
        const data = JSON.parse(e.data);
        onEvent({ type, sessionId, ...data } as ProgressEvent);
      } catch {
        // 忽略
      }
    });
  });

  signal?.addEventListener('abort', () => es.close());
  return () => es.close();
}

/** 提交用户干预决策 */
export async function resumeWriting(
  sessionId: string,
  payload: UserInterventionPayload,
): Promise<void> {
  const resp = await fetch(`${BASE}/${sessionId}/resume`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json' },
    body: JSON.stringify({
      action: payload.action,
      approved_material_ids: payload.approvedMaterialIds ?? [],
      extra_keywords: payload.extraKeywords ?? [],
      edited_outline: payload.editedOutline ?? null,
      outline_feedback: payload.outlineFeedback ?? '',
      user_revision_notes: payload.userRevisionNotes ?? '',
      override_verdict: payload.overrideVerdict ?? false,
    }),
  });
  if (!resp.ok) throw new Error(`提交干预失败: ${resp.status}`);
}

步骤 8.2：核心 Hook

文件：frontend-web/src/open-canvas/hooks/useAIWriting.ts

import { useState, useCallback, useRef, useEffect } from 'react';
import { startWritingSession, subscribeToProgress, resumeWriting } from '../api/ai-writing';
import type {
  AIWritingRequest, AIWritingState, ProgressEvent,
  UserInterventionPayload, AIMaterial, AIOutline, ReviewResult,
} from '../types/ai-writing';
import { useCanvasGraphContext } from '../contexts/CanvasGraphContext';

const INITIAL_STATE: AIWritingState = {
  sessionId: null,
  status: 'idle',
  progressEvents: [],
  materials: [],
  outline: null,
  currentDraftMarkdown: '',
  currentDraftTitle: '',
  reviewResult: null,
  revisionCount: 0,
  currentPause: null,
  error: null,
};

export function useAIWriting() {
  const { setArtifact } = useCanvasGraphContext();
  const [state, setState] = useState<AIWritingState>(INITIAL_STATE);
  const abortRef = useRef<AbortController | null>(null);
  const unsubscribeRef = useRef<(() => void) | null>(null);

  const handleEvent = useCallback((event: ProgressEvent) => {
    setState((prev) => {
      const base = { ...prev, progressEvents: [...prev.progressEvents, event] };

      switch (event.type) {
        case 'materials_ready':
          return { ...base, materials: event.materials ?? [], status: 'awaiting_material_confirm', currentPause: event };

        case 'outline_ready':
          return { ...base, outline: event.outline ?? null, status: 'awaiting_outline_confirm', currentPause: event };

        case 'draft_ready':
          return {
            ...base,
            currentDraftMarkdown: event.draftMarkdown ?? '',
            currentDraftTitle: event.draftTitle ?? '',
            revisionCount: event.revisionCount ?? 0,
            status: 'awaiting_draft_confirm',
            currentPause: event,
          };

        case 'review_ready':
          return { ...base, reviewResult: event.reviewResult ?? null };

        case 'await_user':
          return { ...base, status: `awaiting_${event.pausePoint ?? 'unknown'}` as AIWritingState['status'], currentPause: event };

        case 'resumed':
          return { ...base, currentPause: null };

        case 'done':
          return { ...base, status: 'done', currentPause: null };

        case 'error':
          return { ...base, status: 'error', error: event.message ?? '未知错误' };

        default:
          return base;
      }
    });

    // 草稿实时同步到右侧 Artifact 面板
    if (event.type === 'draft_ready' && event.draftMarkdown) {
      setArtifact({
        currentIndex: 0,
        contents: [{
          index: 0,
          type: 'text',
          title: event.draftTitle ?? '草稿',
          fullMarkdown: event.draftMarkdown,
        }],
      });
    }
  }, [setArtifact]);

  const startWriting = useCallback(async (request: AIWritingRequest) => {
    if (state.status !== 'idle' && state.status !== 'done' && state.status !== 'error') return;

    abortRef.current?.abort();
    abortRef.current = new AbortController();
    setState({ ...INITIAL_STATE, status: 'researching' });

    try {
      const sessionId = await startWritingSession(request);
      setState((prev) => ({ ...prev, sessionId }));
      const unsubscribe = subscribeToProgress(sessionId, handleEvent, abortRef.current.signal);
      unsubscribeRef.current = unsubscribe;
    } catch (err) {
      setState((prev) => ({ ...prev, status: 'error', error: String(err) }));
    }
  }, [state.status, handleEvent]);

  const submitIntervention = useCallback(async (payload: UserInterventionPayload) => {
    if (!state.sessionId) return;
    try {
      await resumeWriting(state.sessionId, payload);
      setState((prev) => ({ ...prev, currentPause: null }));
    } catch (err) {
      setState((prev) => ({ ...prev, error: String(err) }));
    }
  }, [state.sessionId]);

  const stopWriting = useCallback(() => {
    abortRef.current?.abort();
    unsubscribeRef.current?.();
    setState((prev) => ({ ...prev, status: 'idle' }));
  }, []);

  const reset = useCallback(() => {
    abortRef.current?.abort();
    unsubscribeRef.current?.();
    setState(INITIAL_STATE);
  }, []);

  useEffect(() => () => { abortRef.current?.abort(); }, []);

  return { state, startWriting, submitIntervention, stopWriting, reset };
}

9. 内网接口预留设计

9.1 配置切换（config.yaml）

ai_writing:
  search_provider: internet         # ← 切换为 intranet 即启用内网模式
  intranet_search_url: ""           # 内网搜索服务地址
  intranet_search_api_key: ""       # 认证 Token（若需要）
  intranet_search_timeout: 10
  intranet_field_map:               # 内网字段名映射（与标准字段不同时填写）
    title: "title"
    content: "abstract"             # 示例：内网叫 abstract
    source: "system_name"           # 示例：内网叫 system_name
    published_at: "create_time"
    relevance_score: "score"
    recUuid: "id"

9.2 内网素材标准字段

标准字段	类型	必须	说明
`title`	string	✅	文章/文档标题
`content`	string	✅	正文摘要，500 字以内
`source`	string	✅	来源系统名（如"研发知识库"）
`published_at`	string	✅	发布日期 YYYY-MM-DD
`relevance_score`	float 0-1	⬜	无则传 0
`recUuid`	string	⬜	知识库文档 ID

9.3 对接内网步骤清单

确认内网搜索 API 地址和认证方式
在 config.yaml 的 intranet_field_map 中配置字段映射
取消 intranet_stub.py 中的注释代码，调整字段映射
将 config.yaml 中 search_provider 改为 intranet
调用 GET /api/ai-writing/health 验证连通性
验证内网素材卡片在前端正常展示（包括 department、classification 字段）

10. LangGraph 状态机流转图

START
  │
  ▼
intent_parser ──── 意图解析（结构化写作配置）
  │
  ▼
researcher ──────── Agent A：关键词拆解 → 并发搜索 → 去重排序
  │
  ▼
pause_material ──── ⏸ PAUSE-1：推送 await_user 事件，等待用户确认素材
  │ (interrupt)
  ├── action=confirm    ──────────────────────────▶ writer_outline
  └── action=re_search  ──────────────────────────▶ researcher（重搜）

writer_outline ─── Agent B Phase1：写大纲（JSON）
  │
  ▼
pause_outline ───── ⏸ PAUSE-2：推送 await_user 事件，等待用户确认大纲
  │ (interrupt)
  ├── action=confirm（含用户可能的编辑）─────────▶ writer_draft
  └── action=re_outline ─────────────────────────▶ writer_outline（重规划）

writer_draft ────── Agent B Phase2：按大纲逐段写正文
  │
  ▼
pause_draft ─────── ⏸ PAUSE-3：推送 await_user 事件，等待用户确认草稿
  │ (interrupt)
  ├── action=finalize ────────────────────────────▶ END（用户直接定稿）
  ├── action=user_revise ─────────────────────────▶ writer_draft（按用户意见重写）
  └── action=to_editor ────────────────────────────▶ editor

editor ─────────── Agent C：三项独立审查，综合评分
  │
  ├── verdict=pass ────────────────────────────────▶ pause_draft（草稿确认）
  ├── verdict=reject & count < max ───────────────▶ pause_review（PAUSE-4）
  └── verdict=reject & count >= max ──────────────▶ pause_draft（强制到草稿确认）

pause_review ────── ⏸ PAUSE-4：推送编辑意见，等待用户决策
  │ (interrupt)
  ├── override_verdict=true ──────────────────────▶ END（强制定稿）
  └── override_verdict=false ─────────────────────▶ writer_draft（按编辑意见重写）

END

11. 提示词模板（弱模型专用）

以下是针对弱模型优化的三段式提示词框架（借鉴 CrewAI 的 role+goal+backstory 结构）：

研究院 System Prompt

你是一名专业情报研究员（研究院）。
你的目标：为写作任务搜集高质量素材。
你的工作背景：你在一家权威媒体机构任职，熟悉各类信息源的可信度评估。

你只能做以下事情：
1. 从写作意图中提取搜索关键词
2. 对搜索结果按相关度排序
你不能写文章，不能评价内容好坏，只做信息收集。

所有输出必须是合法的 JSON 格式，参考示例：
{"keywords": ["关键词1", "关键词2", "关键词3"]}

作家 System Prompt（大纲阶段）

你是一名专业内容创作者（作家）。
你的目标：根据素材和意图，规划清晰的文章大纲。
你的工作背景：你有十年内容创作经验，擅长将零散素材组织成有逻辑的文章结构。

你只能做以下事情：
1. 分析素材，提取核心观点
2. 规划文章章节结构和每节要点
你不能写正文，只写大纲。

所有输出必须是合法的 JSON，严格参考以下示例格式：
{
  "title": "文章标题",
  "sections": [
    {
      "section_title": "第一章标题",
      "key_points": ["要点一", "要点二"],
      "material_ids": ["m001"]
    }
  ]
}

作家 System Prompt（写作阶段）

你是一名专业内容创作者（作家）。
你的目标：根据大纲和素材，写出一段高质量的文章内容。
你的工作背景：你有十年内容创作经验，文字流畅，善于用事实说话。

你只能做以下事情：
1. 根据给定的章节要点和素材写这一段内容
2. 观点必须来自素材，不编造数据
你不能修改大纲，不能写其他章节，只写当前这一段。

直接返回段落文字，不要 JSON，不要标题，不要解释。

编辑 System Prompt

你是一名资深主编（编辑），标准极高，要求严格。
你的目标：审查文章质量，给出客观评分和具体修改意见。
你的工作背景：你在一家权威媒体担任主编十五年，见过各种质量的文章。

你只能做以下事情：
1. 从 [fact/logic/language] 某一个维度审查文章
2. 给出该维度的评分（0-100）和具体问题列表
你不能改写文章，不能评价其他维度，只做当前指定维度的审查。

所有输出必须是合法的 JSON，严格参考以下示例：
{
  "score": 82,
  "issues": [
    {
      "location": "第二段",
      "issue": "数据来源不明确",
      "suggestion": "引用素材 m001 中的具体数据，并注明来源"
    }
  ]
}

12. SSE 协议 & 干预协议设计

SSE 事件类型

event	触发时机	data 关键字段
`connected`	连接建立	`session_id`
`step_start`	Agent 开始某步骤	`agent_name`, `message`
`step_done`	Agent 完成某步骤	`agent_name`, `message`
`materials_ready`	研究院搜索完成	`keywords`, `materials[]`, `summary`
`outline_ready`	大纲生成完成	`outline{title, sections[]}`
`draft_ready`	草稿完成	`title`, `markdown`, `revision_count`
`review_ready`	编辑审核完成	`review_result{verdict, scores, notes[]}`
`await_user`	流程暂停，等待干预	`pause_point`, 干预所需上下文数据
`resumed`	用户干预已提交，流程恢复	`message`
`done`	全部完成	`message`
`error`	出错	`message`

干预 API

POST /api/ai-writing/{session_id}/resume
Content-Type: application/json

{
  "action": "confirm",               // 操作类型（见各干预点定义）
  "approved_material_ids": [...],    // PAUSE-1
  "extra_keywords": [...],           // PAUSE-1
  "edited_outline": {...},           // PAUSE-2（null=不修改）
  "outline_feedback": "",            // PAUSE-2（重规划时的意见）
  "user_revision_notes": "",         // PAUSE-3 或 PAUSE-4
  "override_verdict": false          // PAUSE-4
}

各干预点的 action 值：

干预点	action 值	说明
PAUSE-1	`confirm`	确认当前素材继续
PAUSE-1	`re_search`	重新搜索（可附 extra_keywords）
PAUSE-2	`confirm`	确认大纲（可附 edited_outline）
PAUSE-2	`re_outline`	重新规划（可附 outline_feedback）
PAUSE-3	`finalize`	直接定稿
PAUSE-3	`user_revise`	按用户意见重写（附 user_revision_notes）
PAUSE-3	`to_editor`	提交编辑审核
PAUSE-4	`accept_review`	采纳编辑意见重写
PAUSE-4	`force_finalize`	强制定稿（override_verdict=true）
PAUSE-4	`custom_revise`	附加用户意见重写（user_revision_notes）

13. 前端 UI 组件详细设计

13.1 组件树

AIWritingPanel.tsx                    （左侧面板根组件）
├── AIWritingForm.tsx                  （写作意图输入表单）
├── AIWritingTimeline.tsx             （进度时间线，纵向排列）
│   ├── TimelineStep.tsx              （单个步骤卡片）
│   │   ├── StepHeader.tsx            （Agent名 + 状态图标 + 时间）
│   │   └── StepContent.tsx          （步骤详情，可收起展开）
│   │
│   ├── MaterialsCard.tsx             （素材卡片区，PAUSE-1 触发展示）
│   │   ├── MaterialItem.tsx          （单条素材卡片，默认收起）
│   │   │   ├── 标题 + 来源 + 时间
│   │   │   ├── 相关度色条
│   │   │   └── 正文摘要（可展开）
│   │   └── MaterialsToolbar.tsx      （全选/全取消/补充搜索输入）
│   │
│   ├── OutlineCard.tsx               （大纲展示卡，PAUSE-2 触发展示）
│   │   ├── OutlineSection.tsx        （可编辑的章节行）
│   │   └── OutlineToolbar.tsx        （确认/重规划按钮）
│   │
│   ├── ReviewCard.tsx                （编辑审核结果，PAUSE-4 触发展示）
│   │   ├── ScoreBar.tsx              （三项评分进度条）
│   │   └── IssueList.tsx             （问题清单，按严重程度分组）
│   │
│   └── InterventionCard.tsx          （用户干预操作区，每个 PAUSE 点渲染）
│       ├── 操作按钮组
│       └── 补充意见输入框
│
└── AIWritingStatus.tsx               （底部状态栏：当前 Agent + 进度）

13.2 MaterialItem 组件规格

┌─────────────────────────────────────────────────────────┐
│ ☑  [标题（最多两行截断）]                              ▶ │  ← 点击展开
│    来源网站  ·  2024-01-15  ·  ████████░░ 82%           │
└─────────────────────────────────────────────────────────┘

展开后：
┌─────────────────────────────────────────────────────────┐
│ ☑  [标题]                                              ▼ │
│    来源网站  ·  2024-01-15  ·  ████████░░ 82%           │
│    ─────────────────────────────────────────────────    │
│    [正文摘要，最多 200 字，超出显示"...展开"]             │
│               │
│    [查看原文 →]（若有 url）                              │
└─────────────────────────────────────────────────────────┘

相关度色条颜色规则：

>= 0.8：绿色（高相关）
0.5 ~ 0.8：橙色（中相关）
< 0.5：红色（低相关，建议取消勾选）

13.3 OutlineCard 组件规格

┌──────────────────────────────────────────────────────────┐
│ 大纲（可编辑）                              [全部展开▼]   │
├──────────────────────────────────────────────────────────┤
│ 文章标题：[可内联编辑]                                    │
│                                                          │
│ ▶ 第一部分：引言与背景（点击展开要点）                    │
│ ▶ 第二部分：核心分析                                      │
│ ▶ 第三部分：结论与建议                                    │
│                                                          │
│ （展开后）                                               │
│ ▼ 第二部分：核心分析                                      │
│   • 要点一（可编辑）                                      │
│   • 要点二（可编辑）                                      │
│   + 添加要点                                             │
│   引用素材：m001 m002（Tag 样式）                        │
├──────────────────────────────────────────────────────────┤
│ [确认大纲，开始写作]    [重新规划大纲...]                 │
└──────────────────────────────────────────────────────────┘

13.4 InterventionCard 组件规格

┌──────────────────────────────────────────────────────────┐
│ ⏸  需要您的确认才能继续                                   │
│ ─────────────────────────────────────────────────────── │
│ [PAUSE-1 素材确认时]                                     │
│   已搜集到 8 条素材（已为您默认全选）                     │
│   您可以取消勾选不相关的素材，或补充关键词                │
│                                                          │
│   补充关键词：[___________________] [+追加搜索]          │
│                                                          │
│   [✓ 确认素材，开始规划大纲]   [↺ 重新搜索]              │
└──────────────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────────────┐
│ ⏸  需要您的确认才能继续                                   │
│ ─────────────────────────────────────────────────────── │
│ [PAUSE-3 草稿确认时]                                     │
│   初稿已完成（共 856 字）                                 │
│   编辑评分：81分（通过线 80 分）                          │
│                                                          │
│   您的修改意见（可选）：                                   │
│   [___________________________________]                  │
│                                                          │
│   [✓ 确认定稿]  [✎ 按我的意见修改]  [→ 让编辑审核]       │
└──────────────────────────────────────────────────────────┘

13.5 ReviewCard 组件规格

┌──────────────────────────────────────────────────────────┐
│ 编辑审核结果（第 1 次修改）             综合评分：78 / 100 │
│                                                          │
│ 事实准确性  ████████░░  82  （通过）                     │
│ 逻辑结构    ██████░░░░  72  （需改进）                   │
│ 语言文字    █████████░  88  （优秀）                     │
│                                                          │
│ ── 修改意见 ────────────────────────────────────────── │
│                                                          │
│ [major] 逻辑 · 第三段：论点跳跃，缺少过渡               │
│   → 建议：在第三段开头增加连接上下文的过渡句              │
│                                                          │
│ [minor] 事实 · 第二段：引用数据未标注来源               │
│   → 建议：补充"据素材 m002 显示"                         │
│                                                          │
│ 您的补充意见（可选）：                                    │
│ [___________________________________]                   │
│                                                          │
│ [↺ 采纳编辑意见，让作家重写]   [✓ 忽略编辑，强制定稿]   │
└──────────────────────────────────────────────────────────┘

13.6 进度时间线整体效果

左侧面板时间线（纵向）

○ ── 研究院                          ← 完成（绿色圆点）
│    搜集到 8 条素材
│    ┌───────────────────────────┐
│    │ 素材列表（默认收起）    ▶ │   ← 点击展开素材卡片
│    └───────────────────────────┘
│    ┌──────────────────────────────────────────────┐
│    │ ⏸ 需要您确认素材，才能继续                    │
│    │ [✓ 确认] [↺ 重新搜索]                        │
│    └──────────────────────────────────────────────┘

○ ── 作家（大纲）                     ← 完成（绿色圆点）
│    已规划 3 个章节
│    ┌───────────────────────────┐
│    │ 大纲预览（默认收起）    ▶ │
│    └───────────────────────────┘
│    ┌──────────────────────────────────────────────┐
│    │ ⏸ 请确认大纲后继续写作                       │
│    │ [✓ 确认] [✎ 编辑大纲] [↺ 重规划]            │
│    └──────────────────────────────────────────────┘

◌ ── 作家（写作）                     ← 进行中（动画圆点）
│    正在写作第二章...

◌ ── 编辑                             ← 等待中（灰色圆点）

◌ ── 定稿                             ← 等待中

14. 文件目录结构

# 后端新增（不修改任何已有文件）
offline-backend-20260512/backend/
├── app/gateway/routers/
│   └── ai_writing.py                       ← 新建
└── packages/harness/deerflow/agents/
    └── ai_writing/                          ← 新建目录
        ├── __init__.py
        ├── state.py
        ├── graph.py
        ├── nodes/
        │   ├── __init__.py
        │   ├── intent_parser.py
        │   ├── researcher.py
        │   ├── writer_outline.py
        │   ├── writer_draft.py
        │   └── editor.py
        ├── search/
        │   ├── __init__.py
        │   ├── base.py                      ← 抽象接口（内网预留）
        │   ├── internet_search.py
        │   └── intranet_stub.py             ← 内网桩
        └── prompts/
            ├── researcher_prompts.py
            ├── writer_prompts.py
            └── editor_prompts.py

# 前端新增（不修改已有组件）
frontend-web/src/open-canvas/
├── api/
│   └── ai-writing.ts                        ← 新建
├── types/
│   └── ai-writing.ts                        ← 新建
├── hooks/
│   └── useAIWriting.ts                      ← 新建
└── components/
    └── ai-writing/                           ← 新建目录
        ├── AIWritingPanel.tsx               ← 左侧面板根组件
        ├── AIWritingForm.tsx                ← 写作意图输入表单
        ├── AIWritingTimeline.tsx            ← 进度时间线
        ├── TimelineStep.tsx                 ← 单步骤卡片
        ├── MaterialsCard.tsx                ← 素材卡片区
        ├── MaterialItem.tsx                 ← 单条素材
        ├── MaterialsToolbar.tsx             ← 素材操作工具栏
        ├── OutlineCard.tsx                  ← 大纲展示与编辑
        ├── OutlineSection.tsx               ← 单章节行
        ├── ReviewCard.tsx                   ← 编辑审核结果
        ├── ScoreBar.tsx                     ← 评分进度条
        ├── IssueList.tsx                    ← 问题清单
        └── InterventionCard.tsx             ← 用户干预操作区

15. 分步开发检查清单

阶段一：后端骨架（2 天）

创建 ai_writing/ 目录结构（所有 __init__.py）
实现 state.py（完整 AIWritingState 定义）
实现 search/base.py（抽象接口）
实现 search/internet_search.py（接入项目已有搜索工具）
实现 search/intranet_stub.py（桩，返回空列表）
实现 nodes/intent_parser.py（意图解析 → 结构化配置）
实现 nodes/researcher.py（Agent A，两步：关键词+搜索）
实现 nodes/writer_outline.py（Agent B Phase1，只写大纲）
实现 nodes/writer_draft.py（Agent B Phase2，逐段写正文）
实现 nodes/editor.py（Agent C，三次独立调用）
实现 graph.py（含 4 个 interrupt 干预节点）
实现 app/gateway/routers/ai_writing.py（start + stream + resume + state）
在 app.py 注册路由（追加一行）
curl 测试 /api/ai-writing/start 和 /api/ai-writing/{id}/stream

阶段二：后端干预验证（1 天）

测试 PAUSE-1（素材确认）：验证 interrupt() 暂停 + resume 恢复
测试 PAUSE-2（大纲确认）：验证编辑大纲后继续写作
测试 PAUSE-3（草稿确认）：验证三种 action（定稿/重写/编辑）
测试 PAUSE-4（审核确认）：验证打回循环和强制定稿
测试最大循环上限（revision_count >= max_revisions）

阶段三：前端进度展示（2 天）

创建 types/ai-writing.ts（完整类型定义）
创建 api/ai-writing.ts（start + SSE subscribe + resume）
创建 hooks/useAIWriting.ts（状态管理 + 事件处理）
创建 AIWritingTimeline.tsx + TimelineStep.tsx（时间线骨架）
创建 ScoreBar.tsx（评分进度条）
在 CanvasResizableLayout.tsx 引入 AIWritingPanel（新增 import，不修改现有逻辑）

阶段四：素材卡片（1 天）

创建 MaterialItem.tsx（收起/展开，相关度色条，勾选框）
创建 MaterialsCard.tsx（素材列表，默认收起，显示素材数量）
创建 MaterialsToolbar.tsx（全选/取消，补充关键词输入）
创建 InterventionCard.tsx for PAUSE-1（确认/重搜按钮）
联调测试：素材 SSE 事件 → 素材列表渲染 → 用户确认 → resume 请求

阶段五：大纲交互（1 天）

创建 OutlineSection.tsx（可内联编辑章节标题和要点）
创建 OutlineCard.tsx（大纲树形展示，默认收起）
创建 InterventionCard.tsx for PAUSE-2（确认/重规划按钮 + 反馈输入）
测试：编辑大纲后提交 → 后端收到 edited_outline → 按修改后大纲写作

阶段六：草稿确认与编辑审核（1 天）

创建 IssueList.tsx（问题清单，按 major/minor 分组）
创建 ReviewCard.tsx（评分 + 问题清单）
创建 InterventionCard.tsx for PAUSE-3 和 PAUSE-4
测试完整流程（搜索 → 大纲确认 → 写作 → 编辑 → 打回 → 重写 → 定稿）

阶段七：弱模型调优（2 天）

为每个 Agent 添加 Few-Shot JSON 示例
添加完善的 JSON 解析容错（try/except + fallback 策略）
调整各 Agent 的 prompt，确保弱模型返回合法 JSON
调整 pass_threshold（从 80 开始，根据实际测试调整）
测试素材截断逻辑（确保不超过 token 限制）

阶段八：内网对接（未来）

确认内网搜索 API 文档和认证方式
在 config.yaml 配置 intranet_field_map 字段映射
取消 intranet_stub.py 中的注释代码并调整
切换 search_provider: intranet 测试连通性
验证内网素材卡片（包含 department、classification 字段）正常展示

架构原则：后端新路由用 /api/ai-writing/ 前缀完全隔离；前端新建独立组件目录，通过在 CanvasResizableLayout.tsx 中新增导入接入，不修改任何已有组件。参考：Stanford STORM 多 Agent 写作流程、LangGraph HITL interrupt 机制、 CrewAI 角色锁定 Prompt 设计、GPT-Researcher 搜索接口抽象。

80 KiB Raw Blame History Unescape Escape

AI 智能写作 —— 杂志社多 Agent 协作系统（含用户干预 & 详细进度）

目录

1. 整体架构

2. 用户干预节点设计（核心）

干预点总览

PAUSE-1：素材确认

PAUSE-2：大纲确认

PAUSE-3：草稿确认

PAUSE-4：编辑审核结果确认（仅在打回时）

3. 开源参考项目

Stanford STORM（最核心参考）

GPT-Researcher

CrewAI

LangGraph Human-in-the-loop（项目已有，重点参考）

4. 弱模型增强策略

5. 三个 Agent 的角色设计

Agent A —— 研究院（Researcher）

Agent B —— 作家（Writer）

Agent C —— 编辑（Editor）

6. 数据结构定义

6.1 后端 LangGraph State（Python）

6.2 前端类型（TypeScript）

7. 后端实现（分步）

步骤 7.1：目录结构

步骤 7.2：搜索接口抽象（内网预留）

步骤 7.3：LangGraph Graph 与 Interrupt 机制

步骤 7.4：后端 HTTP 路由

8. 前端实现（分步）

步骤 8.1：API 客户端

步骤 8.2：核心 Hook

9. 内网接口预留设计

9.1 配置切换（config.yaml）

9.2 内网素材标准字段

9.3 对接内网步骤清单

10. LangGraph 状态机流转图

11. 提示词模板（弱模型专用）

研究院 System Prompt

作家 System Prompt（大纲阶段）

作家 System Prompt（写作阶段）

编辑 System Prompt

12. SSE 协议 & 干预协议设计

SSE 事件类型

干预 API

13. 前端 UI 组件详细设计

13.1 组件树

13.2 MaterialItem 组件规格

13.3 OutlineCard 组件规格

13.4 InterventionCard 组件规格

13.5 ReviewCard 组件规格

13.6 进度时间线整体效果

14. 文件目录结构

15. 分步开发检查清单

阶段一：后端骨架（2 天）

阶段二：后端干预验证（1 天）

阶段三：前端进度展示（2 天）

阶段四：素材卡片（1 天）

阶段五：大纲交互（1 天）

阶段六：草稿确认与编辑审核（1 天）

阶段七：弱模型调优（2 天）

阶段八：内网对接（未来）

80 KiB

Raw Blame History