wsk/deerflow-code
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更新

Browse Source
This commit is contained in:
董宏凯 committed 2026-05-21 08:31:33 +08:00
1 parent 628bbe14a8
commit 04664c8d42
2 files changed
+827
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

frontend-web/docs/ai-writing-优化.md
+501
View File
@@ -0,0 +1,501 @@
# AI 写作 — 迁移到 LangGraph Server 的分步实现方案
> **文档目的**：彻底解决 AI 写作在多 worker 部署下"SSE 时断时续 / resume 偶发 404"的根因，参照主聊天（`/workspace/chats/new`）的实现，把 AI 写作的 graph 注册成 LangGraph Server 上的一个 assistant，让 Server 接管线程、checkpoint、SSE、interrupt/resume 的全部协议。
>
> **配套文档**：[ai-writing-说明.md](./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()`）：
```python
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` 顶部新增：
```python
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`：
```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 通道改动最小、契合度最高。
**改造模板**（每个节点都按这个模板改）：
```python
# 旧：
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()`。新方案：
```python
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` 现有主聊天写法）：
```ts
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 个干预卡的提交逻辑**（旧 → 新）：
```ts
// 旧（自写 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: [...] } },
});
```
**启动写作**：
```ts
// 旧：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`
```python
# 旧：
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`。新逻辑：
```python
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](./ai-writing-说明.md) 上方案 C 应急止血，本方案作为长期改造按周排期。两者不冲突，方案 C 上线即受益，本方案完工后可下线方案 C 的双进程配置。
frontend-web/docs/ai-writing-说明.md
+326
View File
@@ -327,6 +327,332 @@ upstream gateway_pool {
---
## 六-A、方案 C 在 Docker 容器内的详细实施指南
> **目标**：在不引入新依赖、不动业务代码、不拆容器的前提下，让 AI 写作走专门的单 worker 进程，其它业务继续走多 worker。
>
> **基础设施**：本项目镜像基于 `python:3.12-slim-bookworm`，用 `uv` 管理依赖，Dockerfile 位于 `offline-backend-20260512/backend/Dockerfile`，默认入口是
>
> ```
> CMD ["sh", "-c", "cd backend && PYTHONPATH=. uv run --no-sync uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001"]
> ```
### 总览：单容器双 uvicorn 进程
```
┌─────────────── nginx (容器外或前置) ───────────────┐
│ │
/api/ai-writing/ ──┼──► container:8002 (uvicorn --workers 1) ───► AI 写作专用
其余 /api/* ──┼──► container:8001 (uvicorn --workers N) ───► 其他业务
└────────────────────────────────────────────────────┘
▲
│ (两个进程跑在同一个容器里，共享 .deer-flow 卷
│ ／同一个 MySQL ／同一个 sqlite checkpoints.db)
```
两进程都用 **同一个 `app.gateway.app:app`**，区别只在：
- `8002` 单 worker，专门接 `/api/ai-writing/*`
- `8001` 多 worker，接其他所有路由
- 容器内用 `supervisord` 同时托管这两个 uvicorn
### 步骤 1：在镜像里加 supervisord
编辑 `offline-backend-20260512/backend/Dockerfile`，给 **runtime stage**（第 67 行 `FROM python:3.12-slim-bookworm` 之后）补一段：
```dockerfile
# 安装 supervisord（约 4 MB，纯 Python，无需编译）
RUN apt-get update \
&& apt-get install -y --no-install-recommends supervisor \
&& rm -rf /var/lib/apt/lists/*
```
如果内网无法访问 debian 源，可用现有的 `APT_MIRROR` build-arg：
```bash
docker build \
--build-arg APT_MIRROR=mirrors.aliyun.com \
-f offline-backend-20260512/backend/Dockerfile \
-t deerflow-gateway:dual .
```
### 步骤 2：写 supervisord 配置
新建 `offline-backend-20260512/backend/deploy/supervisord.conf`：
```ini
[supervisord]
nodaemon=true ; 容器主进程，必须前台
user=root
logfile=/dev/null ; 让子进程日志直接走 stdout/stderr，由 docker logs 接管
logfile_maxbytes=0
pidfile=/tmp/supervisord.pid
; ─── 主 Gateway：多 worker，处理除 ai-writing 以外的所有业务 ───
[program:gateway-main]
command=uv run --no-sync uvicorn app.gateway.app:app
--host 0.0.0.0 --port 8001
--workers %(ENV_GATEWAY_MAIN_WORKERS)s
--proxy-headers --forwarded-allow-ips=*
directory=/app/backend
environment=PYTHONPATH=".",AI_WRITING_DEDICATED_PORT="8002"
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
stopsignal=TERM
stopwaitsecs=35 ; 给 lifespan 的 30s drain 留点余量
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
; ─── AI 写作专用：单 worker，处理 /api/ai-writing/* ───
[program:gateway-aiwriting]
command=uv run --no-sync uvicorn app.gateway.app:app
--host 0.0.0.0 --port 8002
--workers 1
--proxy-headers --forwarded-allow-ips=*
directory=/app/backend
environment=PYTHONPATH=".",DEER_FLOW_ROLE="ai-writing"
autostart=true
autorestart=true
stopasgroup=true
killasgroup=true
stopsignal=TERM
stopwaitsecs=35
stdout_logfile=/dev/stdout
stdout_logfile_maxbytes=0
stderr_logfile=/dev/stderr
stderr_logfile_maxbytes=0
```
**几个关键点**：
- `nodaemon=true` 让 supervisord 自己变成 PID 1，docker stop 发的 SIGTERM 才能正确传给它，它再传给两个 uvicorn 子进程
- `stopasgroup=true` + `killasgroup=true` 确保 uvicorn 多 worker 模式下所有子进程一起退
- `stopwaitsecs=35` 必须大于 lifespan 的 30s drain 超时，否则 in-flight 写作会被 SIGKILL
- `%(ENV_GATEWAY_MAIN_WORKERS)s` 让主进程 worker 数走环境变量，下面会传
把这个配置文件 COPY 进镜像（在 Dockerfile runtime stage `COPY --from=builder /app/backend ./backend` 之后加）：
```dockerfile
COPY backend/deploy/supervisord.conf /etc/supervisor/conf.d/deerflow.conf
```
### 步骤 3：把 Dockerfile 入口换成 supervisord
把 Dockerfile **runtime stage 最后一行** 改成：
```dockerfile
# 之前
CMD ["sh", "-c", "cd backend && PYTHONPATH=. uv run --no-sync uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001"]
# 之后
EXPOSE 8001 8002 2024
CMD ["supervisord", "-c", "/etc/supervisor/conf.d/deerflow.conf"]
```
> **`dev` stage 不用改**：开发环境本来就是单 worker、reload 模式，方案 C 是为生产准备的，dev stage 维持原 CMD。
### 步骤 4：暴露端口 / 挂卷 / 环境变量
启动容器时：
```bash
docker run -d --name deerflow-gateway \
-p 8001:8001 \
-p 8002:8002 \
-e GATEWAY_MAIN_WORKERS=4 \
-e DEER_FLOW_CONFIG_PATH=/app/config.yaml \
-v /your/host/path/.deer-flow:/app/.deer-flow \
-v /your/host/path/config.yaml:/app/config.yaml:ro \
-v /your/host/path/extensions_config.json:/app/extensions_config.json:ro \
deerflow-gateway:dual
```
**两个进程必须共享 `.deer-flow` 卷** —— 因为 `checkpoints.db` 在里面（`offline-backend-20260512/config.yaml` 里配的是相对路径 `.deer-flow/data/checkpoints.db`）。两个进程跑在同一个容器内，天然共享文件系统，**这一步不用做任何额外配置**，只要确认数据卷挂出去就行。
> ⚠️ 即便 8002 是单 worker，仍然推荐把 checkpointer 持久化到磁盘，因为：① 容器重启会重建进程；② 如果哪天临时改回多容器/多 worker，数据不丢；③ 跟主 gateway 共享 checkpoint，未来想做读副本也方便。
### 步骤 5：nginx 路由
如果你前面已经有 nginx（容器外或者另一个容器），把 AI 写作的路由独立出来：
```nginx
upstream gateway_main { server gateway-host:8001; }
upstream gateway_aiwriting { server gateway-host:8002; }
server {
listen 80;
server_name deerflow.example.com;
# SSE 长连接必须关 buffering
proxy_buffering off;
proxy_request_buffering off;
proxy_http_version 1.1;
proxy_set_header Connection "";
# AI 写作 → 单 worker 后端
location /api/ai-writing/ {
proxy_pass http://gateway_aiwriting;
proxy_read_timeout 3600s; # graph 跑得久，给足时间
proxy_send_timeout 3600s;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# 其它一切 API → 多 worker 后端
location /api/ {
proxy_pass http://gateway_main;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
}
# 前端静态资源不变
location / {
proxy_pass http://frontend:5174;
}
}
```
**`proxy_buffering off` 是必须的** —— 不关 buffering，nginx 会把 SSE 流缓冲到响应结束才一次性吐给前端，等于把流式写作变成「等几分钟啥都看不到 → 一下吐一整篇」。
如果前端是 Vite 直连后端（开发态没走 nginx），就只需要把 `frontend-web` 里 AI 写作相关的 `fetch`/`EventSource` 改成 `:8002`，但更推荐 **始终走 nginx**，对前端透明。
### 步骤 6：如果你用 docker-compose
完整 compose 示例：
```yaml
services:
gateway:
image: deerflow-gateway:dual
container_name: deerflow-gateway
restart: unless-stopped
environment:
GATEWAY_MAIN_WORKERS: "4"
DEER_FLOW_CONFIG_PATH: /app/config.yaml
ports:
- "8001:8001"
- "8002:8002" # 如果有外层 nginx，且 nginx 跟 gateway 同网段，可去掉这行只走内网
volumes:
- ./data/deer-flow:/app/.deer-flow
- ./config.yaml:/app/config.yaml:ro
- ./extensions_config.json:/app/extensions_config.json:ro
healthcheck:
test: ["CMD", "sh", "-c", "curl -fsS http://localhost:8001/health && curl -fsS http://localhost:8002/health"]
interval: 30s
timeout: 5s
retries: 3
start_period: 30s
stop_grace_period: 45s # 给 supervisord 的 35s + 余量，让两路 drain 跑完
nginx:
image: nginx:1.27-alpine
depends_on: [gateway]
ports: ["80:80"]
volumes:
- ./nginx.conf:/etc/nginx/conf.d/default.conf:ro
```
`stop_grace_period: 45s` 这一行很关键 —— docker-compose 默认只给 10s，到点直接 SIGKILL，in-flight 写作会全部丢。设到 45s 等于 supervisord 的 35s + 10s 余量。
### 步骤 7：验证（按这个清单逐项过）
```bash
# 1) 两个进程都起来了
docker exec deerflow-gateway sh -c 'ps -ef | grep -E "uvicorn|supervisord" | grep -v grep'
# 期望看到：
# 1 root supervisord -c /etc/supervisor/conf.d/deerflow.conf
# N root uvicorn app.gateway.app:app --port 8001 --workers 4
# M root uvicorn app.gateway.app:app --port 8002 --workers 1
# (8001 那一行会有 N 个 worker 子进程，8002 那一行只有 1 个)
# 2) 8002 只有 1 个 worker（这是方案 C 的命脉）
docker exec deerflow-gateway sh -c 'pgrep -f "port 8002" | wc -l'
# 期望 = 2（1 个 master + 1 个 worker）；如果 > 2，单 worker 没生效
# 3) 8001 不再警告多 worker
docker logs deerflow-gateway 2>&1 | grep -i 'WEB_CONCURRENCY'
# 期望：没有 ERROR 警告（因为 8001 走的是其他业务，8002 走的是单 worker）
# 4) 8002 进程能看到 ai_writing 持久化日志
docker logs deerflow-gateway 2>&1 | grep 'AsyncSqliteSaver'
# 期望至少 1 行：ai_writing: 使用 AsyncSqliteSaver (...)
# 5) /api/ai-writing/* 走 8002，其他走 8001
curl -sI http://deerflow.example.com/api/ai-writing/health # 看 nginx access log 落到 gateway_aiwriting
curl -sI http://deerflow.example.com/api/models # 看落到 gateway_main
# 6) 真跑一次 AI 写作，确认全程没有「resume 404 / 进度断流」
```
### 步骤 8：监控两个进程的健康
supervisord 自带 `unix_http_server` + `supervisorctl`，但对容器化场景更直接的是看 docker logs：
```bash
# 实时看两个进程的日志
docker logs -f deerflow-gateway 2>&1 | grep -E '(gateway-main|gateway-aiwriting)'
# 进程级状态
docker exec deerflow-gateway supervisorctl -c /etc/supervisor/conf.d/deerflow.conf status
# 期望：
# gateway-main RUNNING pid 7, uptime 0:05:12
# gateway-aiwriting RUNNING pid 18, uptime 0:05:12
# 单独重启 AI 写作进程（不影响主业务，前端只是 AI 写作页几秒不可用）
docker exec deerflow-gateway supervisorctl -c /etc/supervisor/conf.d/deerflow.conf restart gateway-aiwriting
```
### 资源开销（实测预估）
| 项 | 单 worker 镜像 | 双进程镜像 | 增量 |
|---|---|---|---|
| 镜像大小 | 基线 | +约 5 MB | supervisord 依赖 |
| 启动时 RSS | ~250 MB | ~450 MB | 多一份 Python runtime + 模型客户端预加载 |
| 稳态 RSS | 视业务而定 | +约 200 MB | 第二个 worker 的常驻内存 |
| CPU | 不变 | 不变 | 单 worker 进程空闲基本不占 CPU |
**结论**：内存额外占用约 200 MB，对 4 GB 以上的部署机基本无压力。
### 常见坑位
1. **`stop_grace_period` 没改大 → 重启时 in-flight 写作被秒杀**
表现：每次 `docker compose restart` 都有用户报告「写到一半没了」。修复见步骤 6。
2. **8002 没暴露 / nginx 没改 → 所有请求还是落 8001**
表现：症状跟没做方案 C 一样。验证步骤 5 的 curl 即可发现。
3. **两路用了不同的 `config.yaml`**
表现：8001 跟 8002 行为不一致。务必让两个进程**只读同一个文件**（在 compose 里只挂一次卷）。
4. **`.deer-flow` 卷没挂 → 8002 重启后会话全丢**
表现：用户刷新页面，AI 写作历史空白。修复：步骤 4 的 `-v` 必须有。
5. **AI 写作进程崩溃后 supervisord 反复重启 → 日志爆炸**
表现：`gateway-aiwriting` 状态在 RUNNING/EXITED/BACKOFF 间反复。
修复：先 `supervisorctl tail gateway-aiwriting stderr` 看真正原因（多半是 sqlite 权限或配置文件路径），再修，**别盲目调 `autorestart=false`**。
6. **想做 K8s 部署但还想用方案 C**
不要把两个进程塞一个 Pod，应该直接拆成两个 Deployment、共享 PVC（checkpoints.db）。这种情况下方案 C 已经变形成方案 F 的子集，建议直接走方案 F。
### 回滚方案
如果上线后发现问题，回滚只需要 1 步：
```bash
# 把 Dockerfile 最后一行换回原来的 CMD，重新 build + 重启
CMD ["sh", "-c", "cd backend && PYTHONPATH=. uv run --no-sync uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001"]
```
数据完全兼容（checkpoints.db、MySQL 表结构都跟单进程模式一致），**不会丢任何用户数据**，只是回到「resume 偶发 404、SSE 断断续续」的旧体验。
---
## 七、运维操作清单
### 部署前验证