Files
deerflow-code/frontend-web/docs/ai-writing-优化.md
T
2026-05-21 08:31:33 +08:00

29 KiB
Raw Blame History

AI 写作 — 迁移到 LangGraph Server 的分步实现方案

文档目的:彻底解决 AI 写作在多 worker 部署下"SSE 时断时续 / resume 偶发 404"的根因,参照主聊天(/workspace/chats/new)的实现,把 AI 写作的 graph 注册成 LangGraph Server 上的一个 assistant,让 Server 接管线程、checkpoint、SSE、interrupt/resume 的全部协议。

配套文档:ai-writing-说明.md 里描述了为什么会出问题、当前已落地的兜底方案,以及方案 B/C/F 三条"绕过"路径。本文档是其中"长期方案"的具体实现路线 —— 不是绕过多 worker,而是从架构层面让它天然不受 worker 数影响。

更新日期:2026-05-20


一、为什么走这条路

主聊天和 AI 写作功能形态几乎一致(多步流式、可中断/恢复、有持久化),但主聊天完全不受多 worker 影响。差别在于:

维度 主聊天(现状) AI 写作(现状) AI 写作(迁移后)
前端流接入 useStream (SDK) fetch + 自写 EventSource useStream (SDK)
后端运行时 LangGraph Server(独立进程,端口 2024 / 经 nginx 走 /api/langgraph) 嵌在 FastAPI Gateway 路由里直接跑 LangGraph Server(同主聊天)
线程状态 Server 自带持久化 checkpointer 自管 sqlite + 一堆模块级 dict Server 自带持久化 checkpointer
SSE 通道 Server ↔ SDK 协议层 自写 asyncio.Queue per session_id Server ↔ SDK 协议层
多 worker 任何 worker 都能查 Server per-worker 内存 dict 隔离 任何 worker 都能查 Server
Interrupt/Resume Server 协议(Command(resume=...)) 自写 _session_pending_pause + /resume 路由 Server 协议

核心收益:迁完后 _event_queues、_session_pending_pause、_session_repos、_session_user_ids、_session_interventions、_session_article_types、_session_error_flags、_session_pending_pause、_inflight_graph_tasks、_recover_session()、自管 sqlite checkpointer 这一整套"per-process 内存状态"全部可以删掉 —— 不是"绕过"问题,是让问题消失。

代价:节点要改造(流式 chunk 改走 LangGraph 原生 StreamWriter/state update)、前端干预卡的交互要重写、业务表写入路径要从"路由里写"挪到"节点里写"。约 6 人日核心改造 + 50% buffer,建议按 9 人日排期。


二、现状盘点:要删 / 要留 / 要改

2.1 要删的(迁完即可移除)

位置 内容 备注
app/gateway/routers/ai_writing.py POST /start、GET /{id}/stream、POST /{id}/resume、GET /{id}/state 由 LangGraph Server 的 runs.stream / threads.create / runs.wait 替代
同上 _event_queues、_session_pending_pause、_session_model_names、_session_repos、_session_user_ids、_session_interventions、_session_error_flags、_session_article_types 这 8 个模块级 dict 状态全部交给 LangGraph Server
同上 _recover_session()、_inflight_graph_tasks、_spawn_graph_task()、wait_for_inflight_graph_tasks() 跨 worker 恢复 / drain 都由 Server 内建机制处理
同上 _make_config() 里注入 sse_queue / sse_session_id 的逻辑 流式输出改走 LangGraph 原生通道
packages/harness/deerflow/agents/ai_writing/graph.py _get_async_checkpointer()、get_ai_writing_checkpointer()、shutdown_ai_writing_graph()、AI_WRITING_GRAPH 单例、_async_checkpointer_singleton checkpoint 由 langgraph.json 的全局 checkpointer 接管
app/gateway/ai_writing_cleanup.py "联动清理 sqlite checkpointer thread" 那一段 改成调 langgraph_sdk 的 threads.delete()
offline-backend-20260512/config.yaml checkpointer: 段里 AI 写作专用的 sqlite 路径(如果跟主 checkpointer 不同) 共用主配置

2.2 要保留的

位置 内容 理由
packages/harness/deerflow/persistence/ai_writing_sessions/ MySQL 业务表(标题、最终 draft、completed_interventions、transcript、article_type、status…) 列表页 / 历史回显 / 全文搜索都依赖业务表,不能让 LangGraph checkpoint 取代它 —— checkpoint 是流程状态,业务表是产品形态
GET/PATCH/DELETE /api/ai-writing/sessions[/{id}] 列表 / 重命名 / 删除 只查业务表,不碰 graph
PUT /api/ai-writing/sessions/{id}/transcript 时间线持久化 维持锦上添花的回显方案
GET/POST/PATCH/DELETE /api/ai-writing/article-types[/{id}] 文章类型 CRUD 纯业务路由,跟 graph 无关
POST /api/ai-writing/admin/cleanup admin 清理入口 路由保留,内部实现改
4 个 interrupt 暂停点 + 5 个节点 + 全部 state 字段 业务逻辑零改动 pause_material_confirm / pause_outline_confirm / pause_section_help / pause_draft_confirm / pause_review_confirm 用的就是 LangGraph 原生 interrupt(),直接吃迁移红利

2.3 要改的

位置 改造点
graph.py 删自管 checkpointer,导出 make_ai_writing_graph(config) → CompiledStateGraph,由 Server 注入 checkpointer
langgraph.json graphs 里增加 "ai_writing": "deerflow.agents.ai_writing.graph:make_ai_writing_graph"
各 node(researcher.py / writer_outline.py / writer_draft.py / editor.py) 流式输出从 sse_queue.put_nowait(...) 改成 LangGraph 原生 get_stream_writer() 推 custom 事件
各 node DB 写入(草稿、审核结果、status)从「路由里调 repo」改成「节点里通过 RunnableConfig 拿 thread_id 调 repo」
各 node "创建 session 行"从 /start 路由里挪到 intent_parser_node(或前端 onCreated 回调里调一个轻量级 upsert API)
langgraph_auth.py 给 ai_writing assistant 加授权规则(保证只能访问自己的 thread)
前端 frontend-web/src/open-canvas/ 完全弃用现有 api/ai-writing.ts 的 fetch + EventSource,新写 useAIWritingStream hook(基于 @langchain/langgraph-sdk/react 的 useStream)
前端 4 个干预卡 提交回调从 POST /resume 改为 thread.submit(new Command({ resume: payload }))
前端"启动写作"按钮 从 POST /start + 拿 session_id + 建 EventSource 改成 thread.submit({...args}) 触发 onCreated 回调
tests/test_ai_writing_session_recovery.py 整体删除 —— 跨 worker 恢复由 LangGraph Server 接管,不再是 AI 写作自己的事
新增 tests/test_ai_writing_assistant.py 通过 LangGraph SDK in-process 跑一遍 graph,覆盖 interrupt / resume / custom stream / state 持久化

三、分步实施

阶段 0:可行性预演(0.5 d)

目标:在不动现有代码的前提下,验证"把 ai_writing graph 注册到 LangGraph Server 并通过 SDK 跑通"这条路完全可行。

  1. 临时在 langgraph.json 的 graphs 里加一条 "ai_writing_probe": "deerflow.agents.ai_writing.graph:build_ai_writing_graph"(直接复用现有 build_ai_writing_graph,不传 checkpointer 让 Server 注入)。
  2. 起 make dev,确认 LangGraph Server 把 ai_writing_probe 列入 assistant 列表。
  3. 用 Python SDK 客户端(langgraph_sdk.get_client()):
    from langgraph_sdk import get_client
    client = get_client(url="http://localhost:8001/api/langgraph")
    thread = await client.threads.create()
    async for chunk in client.runs.stream(
        thread["thread_id"], "ai_writing_probe",
        input={"user_intent": "测试", "article_type": "analysis", ...},
        stream_mode=["values", "messages-tuple", "custom"],
    ): print(chunk)
    # 跑到 pause_material_confirm 应该收到 interrupt 事件
    await client.runs.stream(thread["thread_id"], "ai_writing_probe",
        command={"resume": {"action": "confirm", "approvedMaterialIds": [...]}})
    
  4. 验证:interrupt 能正确暂停、Command(resume=...) 能续跑、state 能持久化到 Server checkpoint。
  5. 跑通后回滚 langgraph.json —— 这一步只是"摸路"。

通过条件:4 个 pause 点全部能用 SDK 跑通 interrupt + resume。如果某个 pause 点的 payload 结构跟 SDK 协议不兼容(例如用了非 JSON 可序列化的对象),阶段 0 就要解决,否则会在阶段 2 翻车。


阶段 1:graph 注册为 assistant(0.5 d)

改动文件:packages/harness/deerflow/agents/ai_writing/graph.py、offline-backend-20260512/backend/langgraph.json

  1. graph.py 顶部新增:
    from langchain_core.runnables import RunnableConfig
    
    def make_ai_writing_graph(config: RunnableConfig | None = None):
        """LangGraph Server 入口:不接收 checkpointer,由 Server 注入。"""
        g = StateGraph(AIWritingState)
        # ... 跟现在 build_ai_writing_graph 完全一样的节点 / 边定义 ...
        return g.compile()   # 注意:不传 checkpointer
    
  2. 旧的 build_ai_writing_graph(checkpointer=None)、get_ai_writing_graph()、_get_async_checkpointer()、get_ai_writing_checkpointer()、shutdown_ai_writing_graph() 全部先标 @deprecated 暂留 —— 阶段 7 再删,避免阶段 2/3 出错时无法快速回退。
  3. langgraph.json:
    {
      "graphs": {
        "lead_agent": "deerflow.agents:make_lead_agent",
        "canvas_agent": "deerflow.agents.canvas_agent:make_canvas_agent",
        "ai_writing": "deerflow.agents.ai_writing.graph:make_ai_writing_graph"
      }
    }
    
  4. checkpointer.path 已经指向 make_checkpointer(看 langgraph.json:15-17),AI 写作直接共享主 checkpointer 即可,不要重新搞一套。

通过条件:启动 LangGraph Server 后 GET /api/langgraph/assistants 能列出 ai_writing,state schema 字段齐全。


阶段 2:节点改造 —— 流式输出走原生通道(1.5 d)

核心问题:现在的 writer_draft_node / writer_outline_node / researcher_node 通过 config["configurable"]["sse_queue"] 把 LLM token chunk 推给前端。LangGraph Server 不会再注入 sse_queue,必须改用原生流通道。

选型:用 langgraph.config.get_stream_writer() 推 custom 事件(保留现有"非聊天消息、自定义协议"的形态)。

不选 messages-tuple 模式的原因:草稿生成不是聊天消息流,而是大段 markdown;硬塞成 AIMessage 会让前端渲染逻辑变得很别扭。custom 通道改动最小、契合度最高。

改造模板(每个节点都按这个模板改):

# 旧:
from .nodes._utils import push_sse_event
async def writer_draft_node(state, config):
    sse_queue = config["configurable"].get("sse_queue")
    sse_queue.put_nowait({"event": "progress", "data": {...}})
    async for chunk in llm.astream(...):
        sse_queue.put_nowait({"event": "draft_chunk", "data": {"text": chunk}})

# 新:
from langgraph.config import get_stream_writer
async def writer_draft_node(state, config):
    writer = get_stream_writer()   # 注意:必须在 async 函数体内调,不能模块级缓存
    writer({"type": "progress", "agent": "writer", "message": "开始写作"})
    async for chunk in llm.astream(...):
        writer({"type": "draft_chunk", "text": chunk})

业务表写入改造:现在 writer_draft_node 收尾时通过 config["configurable"]["sse_session_id"] + _session_repos[sid] 调 repo.save_draft()。新方案:

async def writer_draft_node(state, config):
    thread_id = config["configurable"]["thread_id"]   # LangGraph Server 注入
    # 从 thread metadata / state 里拿到 user_id(在阶段 4 写入)
    user_id = state.get("_user_id") or config["configurable"].get("user_id")
    repo = get_ai_writing_session_store()  # 进程内全局,无 per-session 字典
    try:
        await repo.save_draft(thread_id, user_id, state["current_draft"])
    except Exception:
        logger.exception("save_draft failed (non-fatal)")

注意事项:

  • get_stream_writer() 在子图 / 工具函数里会拿到当前 graph 的 writer,不要在模块 import 时缓存。
  • 节点要兼容"测试场景下没有 LangGraph runtime":包一层 try: writer = get_stream_writer() except ...,单测里 writer 是 no-op。
  • progress_events 字段保留写入 state(现在节点都已经 return {"progress_events": [...]}),这样 history / 回放仍然能在 state 里看到完整轨迹;custom stream 只是为了实时推送。

通过条件:手动跑一次 make dev,用 SDK 客户端从前一阶段验证脚本流式拿到 draft_chunk 事件。


阶段 3:前端接入 LangGraph SDK(1 d)

核心改动:frontend-web/src/open-canvas/api/ai-writing.ts 整文件作废,改写 frontend-web/src/open-canvas/hooks/useAIWritingStream.ts。

新 hook 骨架(参考 frontend-web/src/core/threads/hooks.ts:207 现有主聊天写法):

import { useStream } from "@langchain/langgraph-sdk/react";
import { Command } from "@langchain/langgraph-sdk";
import { getAPIClient } from "@/core/api/api-client";

interface AIWritingValues {
  user_intent: string;
  article_type: string;
  status: string;
  material_package?: MaterialPackage;
  current_outline?: Outline;
  current_draft?: DraftArticle;
  review_result?: ReviewResult;
  progress_events: any[];
}

export function useAIWritingStream(sessionId: string | "new") {
  const stream = useStream<AIWritingValues, { CustomEventType: AIWritingCustomEvent }>({
    client: getAPIClient(),
    assistantId: "ai_writing",
    threadId: sessionId === "new" ? undefined : sessionId,
    reconnectOnMount: true,
    onCreated(meta) {
      // 后端会用同一个 thread_id 做 ai_writing_sessions.id;通知 UI 写入 URL
      navigateTo(`/workspace/ai-writing/${meta.thread_id}`);
    },
    onCustomEvent(event) {
      // 节点 push 来的 draft_chunk / progress / notice
      pushUIEvent(event);
    },
  });
  return stream;
}

4 个干预卡的提交逻辑(旧 → 新):

// 旧(自写 fetch):
await fetch(`/api/ai-writing/${sessionId}/resume`, {
  method: "POST",
  body: JSON.stringify({ action: "confirm", approvedMaterialIds: [...] }),
});

// 新(SDK 原生 Command):
stream.submit(undefined, {
  command: { resume: { action: "confirm", approvedMaterialIds: [...] } },
});

启动写作:

// 旧:POST /start → 返回 session_id → 建 EventSource
// 新:
stream.submit({
  user_intent, article_type, target_audience, word_count_target, keyword_count, writing_mode,
}, {
  config: { configurable: { model_name } },
  metadata: { feature: "ai_writing", article_type },   // 走业务表筛选
});

列表页 / 历史回显:仍然走 GET /api/ai-writing/sessions 业务 API,无改动。点进某条历史记录时:

  • 拿到 sessionId (== LangGraph thread_id)
  • useAIWritingStream(sessionId) 自动调 client.threads.getState() 拉最新 state,渲染最后一次草稿 / 干预点 / interrupt 等待
  • 已结束的会话 (status == "done") 直接读业务表的 draft_markdown 字段,不挂 SDK stream(省 Server 资源)

通过条件:手动跑完一个完整流程(启动 → 4 次干预 → 拿到草稿),过程中刷新页面 / 切 worker,进度都不丢。


阶段 4:业务表写入解耦(0.5 d)

问题:现在 /start 路由先 INSERT 一行 ai_writing_sessions(id=session_id, ...),再启动 graph。迁移后前端用 stream.submit(...) 直接触发 LangGraph 建 thread + 跑 graph,业务表行没人建 → 列表页查不到。

两个方案二选一:

方案 4A(推荐):节点里建业务表行

  • intent_parser_node 开头:await repo.upsert_session(thread_id, user_id, user_intent, article_type, status="researching")
  • 优点:单一数据源(节点),不依赖前端做"双写"。
  • 缺点:用户从前端点"启动" → 看到列表里出现这条 → 之间有 200~500ms 延迟(intent_parser 跑完才入库),需要前端 onCreated 后乐观插入一条临时行。

方案 4B:前端 onCreated 时调一个轻量 upsert API

  • 新增 POST /api/ai-writing/sessions/{thread_id}/init:只插业务表行,不碰 graph。
  • 前端 onCreated(meta) 回调里立刻 fire-and-forget 调它。
  • 优点:列表瞬间出现新行。
  • 缺点:前端要多调一次 API,且失败要兜底;多了一条"双方都能建行"的路径。

推荐 4A —— 单一数据源比 UI 响应更重要,200ms 延迟用 loading skeleton 即可。

用户身份传递:节点要拿到 user_id 写入业务表。两条路:

  • LangGraph Server 已有 langgraph_auth.py 注入 auth context,在节点里 config["configurable"].get("langgraph_auth_user") 取(具体字段名以现有实现为准)。
  • 或者前端 submit() 时 metadata 里带上 user_id,节点从 config["configurable"]["metadata"] 拿。

通过条件:从启动到列表显示新行,延迟 < 1s;多用户场景下不会出现"A 用户的 thread 落到 B 的业务表行"。


阶段 5:transcript 持久化(0.2 d)

当前:前端 PUT /api/ai-writing/sessions/{id}/transcript 全量覆盖 ai_writing_sessions.transcript 列。

改造选择:

  • 维持现状(推荐):transcript 是前端组装的时间线 blob,让前端继续 fire-and-forget PUT 即可,跟迁移正交。
  • 或者完全弃用 transcript 列,直接从 client.threads.getHistory() + getState() 还原 —— 工作量大且需要重写回显逻辑,不推荐第一版做。

通过条件:原 PUT 接口照常工作;新建会话也能落库。


阶段 6:admin 清理改造(0.3 d)

改动文件:app/gateway/ai_writing_cleanup.py

# 旧:
saver = get_ai_writing_checkpointer()
await saver.adelete_thread(sid)   # 自管 sqlite

# 新:
from langgraph_sdk import get_client
lg_client = get_client(url=internal_langgraph_url())
await lg_client.threads.delete(sid)

活跃保护:旧逻辑是看 sid in _event_queues。新逻辑:

state = await lg_client.threads.get_state(sid)
is_active = (
    state.get("next")                   # graph 还有未执行的节点
    or state.get("tasks")               # 当前有 pending task
    or state.get("metadata", {}).get("interrupted")  # 卡在 interrupt 上
)
if is_active:
    skipped_active += 1
    continue

通过条件:admin cleanup 接口照常返回 deleted_db / deleted_checkpoints / skipped_active,且能正确跳过仍在 interrupt 等待用户的会话。


阶段 7:删冗余代码(0.3 d)

按 §2.1 列表逐项删除。建议单独开一个 PR,跟阶段 1~6 解耦:

  • 前 6 个 PR 全部上线、灰度验证至少 1 周
  • 确认旧代码路径再无流量后,再走清理 PR
  • 这样万一灰度阶段需要回滚,旧路径还在

通过条件:make test 全绿,grep -r "_event_queues\|_session_pending_pause\|_recover_session" app/ packages/ 零命中。


阶段 8:测试 + 灰度(1 d)

新增测试:

  • tests/test_ai_writing_assistant.py —— 用 langgraph_sdk 客户端打 LangGraph Server,覆盖:
    • 完整流程:start → 4 个 interrupt → resume → done
    • state 持久化:跑到 pause_outline 后重建客户端,能 getState 看到 outline
    • custom stream:能收到 draft_chunk 等事件
    • 错误兜底:节点抛错时 status 落 error、业务表 status 同步
  • 保留 tests/test_portable_long_text.py / tests/test_ai_writing_transcript_endpoint.py / tests/test_ai_writing_session_repo.py —— 这几个跟业务表强相关,不受迁移影响。

删除测试:

  • tests/test_ai_writing_session_recovery.py —— 跨 worker 恢复改由 LangGraph Server 提供,不再是 AI 写作自己的能力。
  • tests/test_ai_writing_cleanup.py 里关于"drain + checkpointer 降级 + SSE 序列化"的几个用例 —— 这些机制已删,对应用例失效。

灰度策略:

  • 前端开关:VITE_AI_WRITING_BACKEND = legacy | langgraph,默认 legacy
    • legacy → 走老的 fetch + EventSource + /resume
    • langgraph → 走新 useAIWritingStream
  • 后端:旧路由(/start /stream /resume /state)保留两个发版周期,逐步下线
  • 数据兼容:业务表 schema 完全不变,两条路径都用同一张表
  • 回滚成本:改回 VITE_AI_WRITING_BACKEND=legacy 重新部署即可,无 DB migration

通过条件:

  • 灰度环境跑 100+ 写作流程,对比新旧两条路径的成功率、SSE 中断率、resume 成功率
  • 新路径在 worker=4 部署下 SSE 中断率 = 0(旧路径基线约 5~15%)

四、关键设计决策与权衡

4.1 为什么不让 LangGraph checkpoint 替代业务表?

LangGraph checkpoint 存的是"图执行状态",业务表存的是"产品形态"。两者职责不同:

  • 会话列表筛选/排序:业务表 WHERE user_id = ? AND status = ? ORDER BY updated_at —— LangGraph checkpoint 没法这样查(它是 (thread_id, checkpoint_ns) 主键)
  • 全文搜索 draft:业务表 LIKE '%关键词%' —— checkpoint 是序列化 blob
  • transcript 自定义 UI 协议:前端组装的时间线 JSON,跟 graph state 不一一对应

结论:两者并存,thread_id 作为外键。这是主聊天的做法(threads_meta + LangGraph checkpoint),AI 写作迁完后也是同样的形态。

4.2 为什么 graph 节点里直接调 repo,不发到外面消费?

考虑过"节点只 emit custom event,路由层订阅事件后写业务表" —— 但这又把状态分散回路由层了。

直接在节点里调 await repo.save_draft(...):

  • 节点拿得到 thread_id(== session_id == business id),主键一致
  • repo 是单例(DB 连接池),不依赖 per-session 字典
  • 写入失败包 try/except 即可(草稿落 state 比落 DB 更关键,DB 失败不阻断流程)

4.3 为什么用 custom stream 而不是 messages-tuple?

writer_draft 输出是大段 markdown,强行包成 AIMessage 后:

  • 前端要把它从消息列表里拎出来当草稿渲染(混杂)
  • 大纲 / 审核结果是结构化 JSON,更不像聊天消息
  • progress_events / notice / draft_chunk / outline_chunk 几种事件类型,全塞 message 通道得发明 metadata convention

custom stream 是为这种"非聊天但需要流式 push"场景设计的,匹配度更高。

4.4 在 Server 进程里跑 graph,资源够用吗?

LangGraph Server 是独立 Python 进程,跟 Gateway 共享同一台机器但不共享内存。新增一个 assistant 的常驻开销 ~10 MB(graph 编译后的 schema 缓存),每个并发会话峰值 ~50 MB(LLM client + state)。

实测建议:在阶段 0 之后跑一次 10 个并发的压测,观察 Server RSS 增量。若一台机器 4 GB 内存,理论支撑 60~80 并发 AI 写作,足够 50 人规模团队使用。


五、风险清单

# 风险 缓解
R1 LangGraph Server 升级版本时 interrupt() / Command(resume=...) 协议有 breaking change 在 pyproject.toml 里 pin langgraph minor 版本;升级前先在阶段 0 那个 probe 跑一遍冒烟
R2 前端 useStream 的 reconnectOnMount 行为跟自写 EventSource 不完全一致,刷新页面后某些情况下不会自动续接 SSE 阶段 3 手动测刷新场景;如不达预期,配合 client.threads.getState() 拉一次 state 兜底
R3 节点里 get_stream_writer() 在单测里没有 LangGraph runtime → 抛错 节点 wrap try: writer = get_stream_writer() except: writer = lambda x: None
R4 多用户场景下 langgraph_auth.py 没对 ai_writing assistant 做隔离,可能跨用户拿到别人的 thread state 阶段 4 一并配置 auth;新增专项测试
R5 client.threads.delete() 是异步删,admin 清理脚本调完后立即查 state 还能短暂查到 → skipped_active 误判 删之前先 lock thread 或调用 client.runs.cancel() 强制收敛
R6 LangGraph Server 自带 checkpoint 仍然是 sqlite(langgraph.json 里配的),在容器重启时跟 AI 写作原有的 .deer-flow/data/checkpoints.db 是同一个文件 → schema 共用 阶段 1 之前先确认主 checkpointer 配置路径,避免双开 sqlite 文件锁竞争
R7 现有用户在迁移窗口期发起的会话,灰度切到新路径后无法续跑(state 在旧 sqlite,新路径查不到) 迁移前 1 小时禁止新会话;或写一个一次性脚本把旧 checkpointer 的 thread 复制到主 checkpointer

六、工时与里程碑

阶段 工时 可独立交付
阶段 0:可行性预演 0.5 d ✅
阶段 1:graph 注册 0.5 d ❌(后端只生效,前端没接)
阶段 2:节点改造 1.5 d ❌
阶段 3:前端接入 1 d ✅(带开关上灰度)
阶段 4:业务表写入解耦 0.5 d ❌(跟阶段 3 合并发)
阶段 5:transcript 0.2 d ✅
阶段 6:admin 清理改造 0.3 d ✅
阶段 7:删冗余 0.3 d ✅(独立 PR,灰度后再发)
阶段 8:测试 + 灰度 1 d ✅
核心工时 5.8 d
+ 50% buffer 8.7 d ≈ 9 d

建议排期:

  • 第 1 周:阶段 0 + 1 + 2(后端完整、本地能跑通)
  • 第 2 周:阶段 3 + 4 + 5 + 6(前端 + 业务集成)
  • 第 3 周:阶段 8(测试 + 灰度)
  • 灰度 1~2 周后:阶段 7(删冗余)

七、回滚预案

每个阶段都设计为可独立回滚:

场景 回滚动作
阶段 3 上线后发现 SSE 不稳 前端开关 VITE_AI_WRITING_BACKEND=legacy,重新部署,0 dataloss
阶段 7 删冗余代码后发现新路径有问题 从 git revert 删除 commit,恢复旧路由代码;业务表/checkpoint 不受影响
阶段 1 注册 assistant 后 LangGraph Server 起不来 从 langgraph.json 移除 ai_writing 那一行,重启 Server,回到迁移前状态

数据兼容承诺:

  • 业务表 schema 完全不变
  • thread_id (== ai_writing_sessions.id) 含义保持一致
  • transcript 列含义不变

最坏回滚:把 VITE_AI_WRITING_BACKEND 改回 legacy、langgraph.json 删 ai_writing 那行,回到迁移前的"方案 C/F/B 路径"。已经走完的会话留在新 checkpoint 里成"孤儿",但业务表里的草稿和审核结果完整无损。


八、迁移成功后能享受的"免费红利"

  1. 完整 history API:client.threads.getHistory(thread_id) 拿到每一步 checkpoint,做"回到某一步重试"非常便宜
  2. time travel:用户能从任意 checkpoint 分叉重写
  3. 跟主聊天共享基础设施:升级 LangGraph 版本时一并升级,不用维护两套 SSE / checkpoint 协议
  4. 天然支持水平扩容:哪天上 K8s 多副本部署,AI 写作跟主聊天一样开箱即用
  5. 天然支持 IM 渠道接入:app/channels/ 那套 Feishu / Slack 实现是 langgraph-sdk 的,迁完后 AI 写作也能在 IM 里用,零额外代码

九、相关文件索引(迁移前后参照)

现状(迁前)

  • app/gateway/routers/ai_writing.py:148-176 —— 8 个模块级 dict + drain task
  • packages/harness/deerflow/agents/ai_writing/graph.py:370-538 —— 自管 checkpointer
  • frontend-web/src/open-canvas/api/ai-writing.ts —— 自写 fetch + EventSource

参照(主聊天迁后形态)

  • frontend-web/src/core/threads/hooks.ts:207 —— useStream 用法
  • frontend-web/src/core/api/api-client.ts —— LangGraph SDK Client 初始化
  • offline-backend-20260512/backend/langgraph.json —— assistant 注册位置
  • offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/ —— graph 工厂参考写法
  • offline-backend-20260512/backend/packages/harness/deerflow/runtime/checkpointer/async_provider.py —— 共享 checkpointer 入口

目标(迁后)

  • 新增:packages/harness/deerflow/agents/ai_writing/graph.py:make_ai_writing_graph
  • 新增:frontend-web/src/open-canvas/hooks/useAIWritingStream.ts
  • 删除:app/gateway/routers/ai_writing.py 中的 /start /stream /resume /state 路由 + 8 个 dict
  • 删除:frontend-web/src/open-canvas/api/ai-writing.ts 中的 startWriting / streamSession / resume 全部 fetch 调用
  • 简化:app/gateway/ai_writing_cleanup.py(改用 SDK client 删 thread)

十、决策摘要(给 PM / 维护者)

TL;DR

  1. 主聊天不受多 worker 影响,是因为它把状态托管给了独立的 LangGraph Server 进程,前端直接通过 SDK 跟 Server 通信。AI 写作要做同样的事。
  2. 改造范围:后端 graph 注册 + 节点流通道改造 + 前端 hook 换 SDK + 业务表写入解耦。核心约 6 人日,含 buffer 9 人日。
  3. 业务表、节点逻辑、4 个干预点、UI 形态全部保留,用户感知层面零变化。
  4. 改造完一次性获得"刷新不丢进度、多 worker 不裂、跟主聊天共享基础设施、IM 渠道免费支持"等长期红利。
  5. 建议跟"方案 C"并行:先按 ai-writing-说明.md 上方案 C 应急止血,本方案作为长期改造按周排期。两者不冲突,方案 C 上线即受益,本方案完工后可下线方案 C 的双进程配置。