diff --git a/frontend-web/docs/AI_WRITING_BACKEND_DEV_ZH.md b/frontend-web/docs/AI_WRITING_BACKEND_DEV_ZH.md new file mode 100644 index 00000000..ab2f9732 --- /dev/null +++ b/frontend-web/docs/AI_WRITING_BACKEND_DEV_ZH.md @@ -0,0 +1,845 @@ +# AI 写作 — 后端开发文档 + +> **本文档目的**:帮助后端开发者快速理解 AI 写作子系统的代码结构、Graph 拓扑、数据流、持久化、容错策略和扩展点。 +> +> **配套文档**: +> - 前端:`frontend-web/docs/ai-writing-frontend-dev.md` +> - 故障排查:`frontend-web/docs/ai-writing-说明.md` +> - 前端 API 字段索引:`frontend-web/docs/ai-writing-assistant-backend-api.md` + +--- + +## 一、产品定位与技术架构 + +### 产品定位 +**杂志社协作写作子系统**:由 4 个 AI Agent(**素材收集专家 / 作家 / 编辑 / 系统**)协作完成一篇文章,通过 4 个用户干预点参与决策。 + +### 技术栈 +- **编排框架**:LangGraph(独立 StateGraph,不复用主 lead agent 的 middleware 链) +- **节点驱动**:异步 LLM 调用(`langchain_openai` 等),支持流式 token 推送到 SSE +- **状态持久化**:LangGraph checkpointer(SQLite 文件,跨重启可恢复) +- **业务数据**:MySQL `ai_writing_sessions` 表 +- **协议**:SSE(Server-Sent Events)推进度 + REST 控制流 + +### 与主 Agent 的关系 +**完全独立**。AI 写作 graph 不通过 `make_lead_agent` 也不走主 agent 的 18 层 middleware。它有自己的 StateGraph、自己的 checkpointer、自己的路由前缀(`/api/ai-writing/*`),跟主 agent 解耦。 + +--- + +## 二、目录结构 + +``` +offline-backend-20260512/backend/ +├── app/gateway/ +│ ├── routers/ +│ │ └── ai_writing.py # FastAPI 路由 + 模块级状态字典 + SSE 生成器(1000+ 行,核心) +│ ├── ai_writing_cleanup.py # 定期清理 + cron 调度(独立模块) +│ └── app.py # lifespan 启动/停止清理任务、检测多 worker、关闭 checkpointer +│ +└── packages/harness/deerflow/ + ├── agents/ai_writing/ + │ ├── __init__.py # 暴露 AI_WRITING_GRAPH + 工厂函数 + │ ├── graph.py # ⭐ StateGraph 定义 + 4 个 pause 节点 + checkpointer 工厂(538 行) + │ ├── state.py # AIWritingState TypedDict(121 行) + │ │ + │ ├── nodes/ # ── Graph 节点 ── + │ │ ├── _utils.py # LLM 工具:parse_json, ThinkTagParser, llm_json + │ │ ├── intent_parser.py # 节点 1:意图解析 + │ │ ├── researcher.py # 节点 2:素材检索(支持流式 progress) + │ │ ├── writer_outline.py # 节点 3:大纲规划 + │ │ ├── writer_draft.py # 节点 4:草稿生成(按章节流式) + │ │ └── editor.py # 节点 5:编辑审核(3 维度并行流式) + │ │ + │ ├── prompts/ # ── Prompt 模板 ── + │ │ ├── __init__.py + │ │ ├── researcher_prompts.py + │ │ ├── writer_prompts.py + │ │ └── editor_prompts.py + │ │ + │ └── search/ # ── 素材检索 provider ── + │ ├── base.py # SearchResult / SearchProvider 接口 + │ ├── duckduckgo_search.py + │ ├── internet_search.py # 互联网检索(默认) + │ ├── intranet_stub.py # 内网检索占位(按部署接入) + │ └── demo_search.py # 离线 demo 数据 + │ + └── persistence/ai_writing_sessions/ + ├── __init__.py # AIWritingSessionRepository 工厂 + ├── model.py # ORM:AIWritingSessionRow + └── sql.py # CRUD 实现(含清理用的 list_older_than / delete_by_ids) +``` + +--- + +## 三、Graph 拓扑(最重要) + +### StateGraph 节点 + 边 + +``` + START + │ + ↓ + ┌─────────────────┐ + │ intent_parser │ 解析用户意图(确定 article_type / target_audience) + └────────┬────────┘ + ↓ + ┌─────────────────┐ + │ researcher │ ← (re_search 回到这里) + └────────┬────────┘ + ↓ (after_researcher: ⼟材 ready) + ┌──────────────────────┐ + │ pause_material │ ⚙️ PAUSE-1:素材确认 + └─┬──────────┬─────────┘ + │ confirm │ re_search + ↓ └→ researcher + ┌─────────────────┐ + │ writer_outline │ ← (re_outline 回到这里) + └────────┬────────┘ + ↓ + ┌──────────────────────┐ + │ pause_outline │ ⚙️ PAUSE-2:大纲确认 + └─┬──────────┬─────────┘ + │ confirm │ re_outline + ↓ └→ writer_outline + ┌─────────────────┐ + │ writer_draft │ ← (素材补充后回到这里) + └─┬───────────────┘ + ↓ (after_writer_draft) + ┌──────┴────────────────┐ + ↓ ↓ + ┌──────────────────┐ ┌────────────────────┐ + │ pause_section_ │ │ pause_draft │ ⚙️ PAUSE-3:草稿确认 + │ help(素材不足)│ │ │ + └─┬─────────────┬──┘ └─┬────────┬───┬─────┘ + │ supplement │ del/ │ │ │ to_editor + │↓ │ loose │ │ user_revise + research writer_ │ final ↓ + draft │ ize writer_draft + │↓ + done + │ + (to_editor)│ + ↓ + ┌─────────────────┐ + │ editor │ 3 维度并行流式审核 + └────────┬────────┘ + ↓ (after_review) + ┌──────────────────────┐ + │ pause_review │ ⚙️ PAUSE-4:审核确认 + └─┬──────────┬─────────┘ + │ confirm │ revise + │↓ └→ writer_draft + done +``` + +### 节点定义位置一览 + +| 节点名 | 实现文件 | 类型 | 流式? | +|---|---|---|---| +| `intent_parser` | `nodes/intent_parser.py` | async function | 否 | +| `researcher` | `nodes/researcher.py` | async function | 是(research_progress) | +| `pause_material` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `writer_outline` | `nodes/writer_outline.py` | async function | 是(outline_chunk) | +| `pause_outline` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `writer_draft` | `nodes/writer_draft.py` | async function | 是(draft_chunk,按章节) | +| `pause_section_help` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `pause_draft` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `editor` | `nodes/editor.py` | async function | 是(review_chunk,3 路并行) | +| `pause_review` | `graph.py` 顶部 | async, 调 `interrupt()` | — | + +### 条件路由函数(graph.py) + +| 函数 | 在哪个节点之后 | 路由依据 | +|---|---|---| +| `after_researcher` | researcher | 是否找到素材 | +| `after_material_pause` | pause_material | `state["status"]` 是 researching 还是 writing_outline | +| `after_outline_pause` | pause_outline | 同上,writing_outline 还是 writing_draft | +| `after_writer_draft` | writer_draft | 是否有 blocked_sections | +| `after_section_help_pause` | pause_section_help | status | +| `after_draft_pause` | pause_draft | done / revising / reviewing | +| `after_review` | editor | review_result.verdict 是 pass 还是 reject + revision_count 上限 | +| `after_review_pause` | pause_review | done / writing_draft | + +--- + +## 四、State 定义(state.py) + +```python +class AIWritingState(TypedDict): + # —— 写作配置(启动时填充)—— + user_intent: str + article_type: str + target_audience: str + word_count_target: int + keyword_count: int # 1-10,默认 6 + writing_mode: str # "strict" | "loose" + + # —— Agent 产出 —— + material_package: Optional[MaterialPackage] # 素材收集专家 + current_outline: Optional[Outline] # 作家 + drafts: Annotated[List[DraftArticle], operator.add] # 累计所有版本 + current_draft: Optional[DraftArticle] # 最新草稿 + review_result: Optional[ReviewResult] # 编辑 + + # —— 用户干预 —— + last_user_intervention: Optional[UserInterventionPayload] + blocked_sections: Optional[List[BlockedSection]] # 素材不足的章节 + + # —— 流程控制 —— + revision_count: int + max_revisions: int # 默认 3 + status: WritingStatus + + # —— SSE 推送队列 —— + progress_events: Annotated[List[dict], operator.add] # 累加,不覆盖 +``` + +### State 字段的特殊语义 + +- **`Annotated[..., operator.add]`** — LangGraph 的累加 reducer,节点 return 时 `progress_events: [evt]` 会**追加**到现有列表,而不是替换 +- **TypedDict** — 运行时是普通 dict,类型注解仅供 IDE 提示 +- **节点输出** — 节点 return 的 dict 只需要包含**要更新的字段**,未指定的字段保持不变 + +--- + +## 五、路由层(app/gateway/routers/ai_writing.py) + +### 14 个接口一览 + +| HTTP | 路径 | 用途 | 鉴权 | +|---|---|---|---| +| 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` | 查 graph 状态快照(调试用) | 用户 | +| GET | `/api/ai-writing/sessions` | 列出当前用户的历史会话 | 用户 | +| GET | `/api/ai-writing/sessions/{id}` | 获取单个历史会话 | 用户 | +| PATCH | `/api/ai-writing/sessions/{id}` | 重命名会话 | 用户 | +| DELETE | `/api/ai-writing/sessions/{id}` | 删除会话 | 用户 | +| PUT | `/api/ai-writing/sessions/{id}/transcript` | 保存对话时间线(**fire-and-forget**) | 用户 | +| GET | `/api/ai-writing/article-types` | 列出文章类型 | 用户 | +| POST | `/api/ai-writing/article-types` | 新增文章类型 | 用户(建议改成 admin) | +| PATCH | `/api/ai-writing/article-types/{id}` | 更新文章类型 | 用户 | +| DELETE | `/api/ai-writing/article-types/{id}` | 删除文章类型 | 用户 | +| POST | `/api/ai-writing/admin/cleanup` | **手动触发清理** | admin only | + +### 模块级状态字典(关键设计 — 也是 worker 限制的根源) + +```python +# 每个 session_id 对应一份内存状态。多 worker 部署时每个 worker 一份,互不共享。 +_event_queues: dict[str, asyncio.Queue] # SSE 事件队列 +_session_pending_pause: dict[str, tuple] # 挂起的 await_user 事件(供重连重投) +_session_model_names: dict[str, str] # 用户选的模型 +_session_interventions: dict[str, list] # 累计干预历史 +_session_repos: dict[str, repo] # DB 仓库引用 +_session_user_ids: dict[str, str] # 用户 id +_session_article_types: dict[str, list] # 动态文章类型列表 +_session_error_flags: dict[str, bool] +_inflight_graph_tasks: set[asyncio.Task] # in-flight task 注册表(lifespan drain 用) +_recovery_lock: asyncio.Lock # 跨 worker 会话恢复的并发锁 +``` + +⚠️ **关键限制**:这些都是 per-process 内存。多 worker 部署下,worker B 看不到 worker A 的状态 → SSE 事件分发失败。详见 `frontend-web/docs/ai-writing-说明.md` 第五、六章。 + +### `_stream_graph_segment`(graph 推进函数) + +`/start` 和 `/resume` 都调用这个函数(用 `_spawn_graph_task` 包装),负责: +1. 调 `graph.astream(input_or_command, config=config, stream_mode="updates")` +2. 把每个 chunk 解析成 SSE 事件 put 进 `_event_queues[session_id]` +3. 遇到 `__interrupt__` → 推 `await_user` 事件 → 返回(不关闭 queue) +4. 遇到 END → 推 `done` 事件 → 关闭 queue(put None 哨兵) +5. 任何异常 → 推 `error` 事件 → 关闭 queue +6. finally 段:更新 DB 最终状态、清理所有模块字典 + +### SSE 生成器 `_sse_generator` + +`GET /stream` 的实现,从 `_event_queues[session_id]` 拿事件 → `_sse(event, data)` 序列化 → yield 给 fastapi `StreamingResponse`。 + +**关键逻辑**: +- 队列不存在时**等待 5 秒**,期间触发 `_recover_session`(跨 worker / 进程重启恢复) +- `_sse()` 三层兜底:`default=str` → 手写 notice 帧 → 永不抛 TypeError +- 收到 None 哨兵 → 关闭流 + +### `_recover_session`(跨 worker 恢复) + +当 worker 内存里没有 session_id 但 sqlite checkpointer 有快照时: +1. 从 checkpointer 读出 state snapshot +2. 重建 `_event_queues[sid]` 队列 +3. 从 snapshot 提取当前 interrupt 负载,put 进队列 +4. 同时尽量恢复 `_session_repos`、`_session_user_ids` 等 + +返回三种状态: +- `"interrupted"` — 图停在干预点,已重建队列 +- `"finished"` — 图已完成或中断在节点中间,推 done/error 事件后关闭 +- `"unrecoverable"` — sqlite 也没有这个 thread_id → 真的丢了 + +### `_spawn_graph_task` / `wait_for_inflight_graph_tasks`(优雅关闭) + +- `_spawn_graph_task(coro)` — 启 task 并自动登记到 `_inflight_graph_tasks`,完成时反登记 +- `wait_for_inflight_graph_tasks(timeout)` — lifespan shutdown 等所有 task 跑到下一个 interrupt / END(默认 30s),超时则 cancel 强制结束 + +--- + +## 六、Graph 与持久化(graph.py) + +### Lazy 单例 + 自动降级 + +```python +AI_WRITING_GRAPH = None # 模块级槽位(测试可 monkeypatch) +_async_checkpointer_singleton = None # checkpointer 单例 + +async def get_ai_writing_graph(): + """懒加载 graph。第一次调用时按 config.yaml 决定 checkpointer。""" + global AI_WRITING_GRAPH + if AI_WRITING_GRAPH is not None: + return AI_WRITING_GRAPH + checkpointer = await _get_async_checkpointer() + AI_WRITING_GRAPH = build_ai_writing_graph(checkpointer=checkpointer) + return AI_WRITING_GRAPH +``` + +### Checkpointer 优先级(_get_async_checkpointer) + +``` +1. config.yaml 的 checkpointer 段 + ├── type: sqlite → AsyncSqliteSaver(aiosqlite.connect(...)) + ├── type: postgres → AsyncPostgresSaver(AsyncConnectionPool(...)) + └── type: memory → InMemorySaver + +2. 任何初始化失败 → InMemorySaver(降级,打 warning) +3. 未配置 checkpointer 段 → InMemorySaver(降级,打 warning) +4. 未知 type(如 "mysql") → InMemorySaver(降级,打 warning) +``` + +**核心契约**:`_get_async_checkpointer()` **绝不抛异常**,永远返回一个可用的 saver。这是「写作流程能完整继续」契约的关键。 + +### Shutdown + +`shutdown_ai_writing_graph()` 在 lifespan shutdown 时调用: +- 关闭 `aiosqlite.Connection` / `AsyncConnectionPool` +- 清空 `AI_WRITING_GRAPH` / `_async_checkpointer_singleton` + +调用顺序(详见 `app/gateway/app.py` lifespan): +1. 停清理任务 +2. `wait_for_inflight_graph_tasks(30s)` +3. `shutdown_ai_writing_graph()` + +--- + +## 七、节点详解(nodes/) + +### `nodes/_utils.py`(公共工具) + +| 函数 / 类 | 用途 | +|---|---| +| `get_model(config, model_name)` | 从 config.configurable 拿 model_name 或回退默认 | +| `parse_json(text, fallback)` | 鲁棒 JSON 解析(支持代码块、不完整 JSON、转义错误) | +| `_extract_balanced(text)` | 找出第一对平衡的 `{...}` | +| `_repair_json(text)` | 修复常见的 JSON 错误(缺尾引号等) | +| `ThinkTagParser` | 流式解析 `...` 标签(思考 / 正文分流) | +| `llm_json(system, user, fallback, config)` | 调 LLM 并解析 JSON 输出 | +| `count_words(text)` | 中文 + 英文混合计字数 | + +### `nodes/intent_parser.py` — 意图解析 + +- 调 LLM 把用户意图(`user_intent` 字符串)解析成结构化字段:`article_type`、`target_audience`、`word_count_target` +- 用户已经在 `AIWritingRequest` 里指定的字段会**覆盖** LLM 推断 +- 输出:更新 state 的对应字段,推 `step_done` 事件 + +### `nodes/researcher.py` — 素材检索 + +- 根据 `user_intent` 用 LLM 生成 N 个关键词(N = `keyword_count`,默认 6) +- 用 `search/*` 的 provider 并行检索 +- 流式推 `research_progress` 事件("正在搜索关键词 X 中...") +- 输出:`material_package = {keywords, materials, summary}`,推 `materials_ready` + +### `nodes/writer_outline.py` — 大纲规划 + +- 输入:`user_intent` + `material_package` +- 调 LLM 流式生成大纲 JSON,期间推 `outline_chunk`(正文 + thinking 分流) +- 解析最终 JSON → `current_outline`,推 `outline_ready` + +### `nodes/writer_draft.py` — 草稿生成(最复杂) + +- 按大纲**逐章节**调 LLM 生成正文,每章节独立流式推 `draft_chunk` +- 检测 LLM 是否「拒答」(`_looks_like_refusal`),如果是则换提示重试 +- **严格模式下**:如果章节关联的素材不足,把该章节加入 `blocked_sections`,触发 `pause_section_help` +- 完成所有章节后拼成 `current_draft.full_markdown`,推 `draft_ready` + +### `nodes/editor.py` — 编辑审核(3 维度并行) + +- 3 个维度(**fact / logic / language**)并行调 LLM +- 每个维度独立流式推 `review_chunk`,完成后推 `review_item_done` +- 三个都完成后聚合成 `review_result`,推 `review_ready` +- 评分加权(fact 40% + logic 30% + language 30%),超过 `pass_threshold`(默认 80)verdict = `pass` + +--- + +## 八、持久化(persistence/ai_writing_sessions/) + +### `model.py` — ORM + +```python +class AIWritingSessionRow(Base): + __tablename__ = "ai_writing_sessions" + id: Mapped[str] # = session_id(uuid) + user_id: Mapped[str | None] # NULL = no-auth 模式 + title: Mapped[str] # 从 user_intent 截取前 50 字 + user_intent: Mapped[str | None] + status: Mapped[str] # in_progress / draft_ready / review_ready / done / error + draft_title: Mapped[str | None] + draft_markdown: Mapped[str | None] + completed_interventions: Mapped[str | None] # JSON 字符串 + review_result: Mapped[str | None] # JSON 字符串 + transcript: Mapped[str | None] # PortableLongText(MySQL→LONGTEXT,4GB) + created_at / updated_at: BeijingDateTime +``` + +### `sql.py` — Repository + +| 方法 | 用途 | +|---|---| +| `create(id, title, user_intent, user_id)` | 启动写作时建一行 | +| `get(session_id, user_id)` | 取单个(带 user_id 权限校验) | +| `list(user_id, limit)` | 列出当前用户的会话 | +| `update(session_id, user_id, ...)` | 更新指定字段(其他保持不变) | +| `delete(session_id, user_id)` | 删除(带权限校验) | +| `list_older_than(cutoff, only_finished, limit)` | 清理用:列出过期 session_id | +| `delete_by_ids(session_ids)` | 清理用:批量删除 | + +### Alembic 迁移 + +- `20260517_03_ai_writing_transcript.py` — 加 transcript 列(TEXT) +- `20260519_02_ai_writing_transcript_longtext.py` — MySQL 上把 transcript 改成 LONGTEXT(修 64KB 限制导致的 500) + +--- + +## 九、定期清理(ai_writing_cleanup.py) + +### 设计目标 +- 删超过 `retention_days`(默认 7)的旧会话 +- 联动清理:MySQL 业务行 + SQLite checkpointer 里的 thread 状态 +- **绝对不删正在写的会话**(活跃保护) + +### 关键函数 + +```python +async def run_cleanup_once(*, retention_days, only_finished=False, batch_size=500) -> dict +``` + +返回 `{deleted_db, deleted_checkpoints, skipped_active}`。 + +执行流程: +1. 调 `repo.list_older_than(cutoff, only_finished, batch_size)` 拿过期 id +2. 从 `_event_queues.keys()` 拿活跃 session_id 集合 → **过滤掉所有仍在跑的** +3. 对每个待删 id:调 `checkpointer.adelete_thread(id)` 删 checkpoint +4. 调 `repo.delete_by_ids(ids)` 批量删 MySQL +5. 循环直到本批 < batch_size + +### Cron 调度 + +`cleanup_scheduler_loop()` 常驻后台任务: +- 用 `croniter` 解析 `ai_writing.cleanup.cron`(默认 `"0 3 * * *"`,每天凌晨 3 点) +- `asyncio.sleep` 到下次触发时间 +- 多 worker 用文件锁(`runtime_home() / ".ai_writing_cleanup.lock"`)做 leader election + +### Admin 手动触发 + +`POST /api/ai-writing/admin/cleanup`: +- 鉴权:`system_role == "admin"` +- 可选覆盖 `retention_days` / `only_finished` / `batch_size` +- 即使传 `retention_days: 0` 也不会秒杀活跃会话(活跃保护硬保证) + +--- + +## 十、配置(config.yaml) + +```yaml +# Checkpointer:graph 状态持久化 +checkpointer: + type: sqlite # sqlite | postgres | memory + connection_string: .deer-flow/data/checkpoints.db + +# 清理调度 +ai_writing: + cleanup: + enabled: true + retention_days: 7 + cron: "0 3 * * *" # 每天凌晨 3 点 + delete_only_finished: false + batch_size: 500 + + # (可选)素材检索 provider 配置 + search: + provider: deerflow.agents.ai_writing.search.internet_search:WebSearchProvider + ... +``` + +`_ai_writing_config()` 函数从 `get_app_config().ai_writing` 里读 dict,节点通过 `config.configurable.ai_writing_config` 拿到。 + +--- + +## 十一、SSE 事件协议(后端 → 前端契约) + +后端通过 `_sse(event_type, data)` 推 `event: \ndata: \n\n` 格式。 + +### 推出事件的位置 + +| 事件 | 推出位置 | +|---|---| +| `connected` | `_sse_generator` 入口 | +| `step_start` / `step_done` | 节点内部主动 emit | +| `research_progress` | `researcher_node` 内的 `_emit` | +| `materials_ready` | `_stream_graph_segment` 检测到 `material_package` 字段 | +| `outline_chunk` | `writer_outline_node` 流式推(通过 `sse_queue`) | +| `outline_ready` | `_stream_graph_segment` 检测到 `current_outline` | +| `draft_chunk` | `writer_draft_node` 按章节流式推 | +| `draft_ready` | `_stream_graph_segment` 检测到 `current_draft` | +| `review_chunk` / `review_item_done` | `editor_node` 3 维度并行流式 | +| `review_ready` | `_stream_graph_segment` 检测到 `review_result` | +| `await_user` | `_stream_graph_segment` 捕获 `__interrupt__` | +| `resumed` | pause 节点 return 时塞进 `progress_events` | +| `notice` | 节点内手动推(非致命告警) | +| `done` | `_stream_graph_segment` 正常结束 | +| `error` | `_stream_graph_segment` except 分支 | + +### 节点如何流式推 chunk + +通过 `config.configurable.sse_queue`: + +```python +async def writer_outline_node(state, config): + sse_queue = config.get("configurable", {}).get("sse_queue") + sse_session_id = config.get("configurable", {}).get("sse_session_id") + + async for chunk in llm.astream(prompt): + if sse_queue: + await sse_queue.put(("outline_chunk", { + "session_id": sse_session_id, + "text": chunk.content, + "is_thinking": False, + })) + + return {"current_outline": parsed, ...} +``` + +`sse_queue` 是路由层在 `_make_config` 里注入的 `_event_queues[session_id]`。 + +--- + +## 十二、扩展指南:常见开发任务 + +### 任务 1:新增一个 Graph 节点 + +**例**:在 `editor` 之后加一个「术语统一」节点。 + +1. **写节点**:在 `nodes/` 新建 `terminology.py`: + ```python + async def terminology_node(state, config): + # ... 读 current_draft 改术语 + return { + "current_draft": new_draft, + "progress_events": [{"type": "step_done", "agent_name": "术语统一", "message": "..."}], + } + ``` + +2. **注册到 graph**:在 `graph.py` 的 `build_ai_writing_graph` 里: + ```python + from .nodes.terminology import terminology_node + g.add_node("terminology", terminology_node) + # 改 editor 后的边 + g.add_edge("editor", "terminology") + g.add_edge("terminology", "pause_review") + ``` + +3. **State 加字段**(如果需要):`state.py` 加新字段到 `AIWritingState` + +4. **前端事件支持**(如果要新事件类型):见前端文档「扩展指南」 + +5. **测试**:在 `tests/` 加单元测试,至少测节点能正常 return 和 graph 拓扑能编译 + +### 任务 2:新增一个用户干预点 + +1. **写 pause 节点**:在 `graph.py` 顶部加: + ```python + async def pause_xxx(state): + user_input = interrupt({ + "pause_point": "xxx_confirm", + "message": "...", + "context_field": state["..."], + }) + action = user_input.get("action", "confirm") + if action == "confirm": + return {"status": "next_status", "progress_events": [{"type": "resumed", ...}]} + # ... 其他分支 + ``` + +2. **路由层 ResumeRequest 加字段**(如果干预需要新字段): + ```python + class ResumeRequest(BaseModel): + # ... + new_field: Optional[str] = None + ``` + +3. **`resume_writing` 路由把字段塞进 Command(resume=...)**: + ```python + intervention = { + # ... + "newField": body.new_field, + } + ``` + +4. **graph 拓扑加 pause 节点的边**:在 build_ai_writing_graph 里 + +5. **前端配合**:见前端文档 + +### 任务 3:新增一种素材检索 provider + +1. 在 `agents/ai_writing/search/` 加新文件,实现 `SearchProvider` 接口(参考 `internet_search.py`): + ```python + class MyProvider(SearchProvider): + async def search(self, keyword: str, limit: int) -> list[SearchResult]: + ... + ``` + +2. 配置 `config.yaml`: + ```yaml + ai_writing: + search: + provider: my_pkg.search:MyProvider + # provider 特定的配置... + ``` + +3. `researcher_node` 通过 `config.configurable.ai_writing_config.search.provider` 用 `resolve_class` 动态加载 + +### 任务 4:修改 LLM Prompt + +所有 prompt 集中在 `agents/ai_writing/prompts/`: +- `researcher_prompts.py` — 检索专家 +- `writer_prompts.py` — 大纲 + 草稿 +- `editor_prompts.py` — 审核 + +改 prompt 后**强烈建议**: +1. 在 `tests/` 加针对该 prompt 输出的回归测试(用 LLM mock 或固定输入) +2. 跑现有测试确认拓扑兼容 + +### 任务 5:扩展持久化字段 + +例:给 `ai_writing_sessions` 加一个 `total_tokens` 列。 + +1. `model.py` 加列: + ```python + total_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True) + ``` + +2. **必须**新建 alembic 迁移: + ```bash + PYTHONPATH=. uv run alembic revision -m "add total_tokens column" + ``` + 然后填 upgrade/downgrade 逻辑(参考 `20260519_02_ai_writing_transcript_longtext.py`) + +3. `sql.py` 的 `update()` / `to_dict()` 加字段处理 + +4. `routers/ai_writing.py` 的 `SessionResponse` 加字段 + +5. 测试:在 `tests/test_ai_writing_session_repo.py` 加新字段的 round-trip 测试 + +--- + +## 十三、容错策略(关键设计原则) + +### 「写作流程能完整继续」契约 + +后端承诺:**除非 Pydantic 校验失败(4xx),AI 写作的 4 个核心接口任何后端组件失败都不阻断当次写作的完整执行**。 + +### 落实位置 + +| 失败点 | 兜底位置 | 行为 | +|---|---|---| +| sqlite checkpointer 初始化失败 | `_get_async_checkpointer` | 自动降级 InMemorySaver | +| SSE JSON 序列化失败 | `_sse()` | 三层兜底(default=str → notice 帧) | +| DB `create()` session 行失败 | `start_writing` | try/except 吞掉,session_id 仍返回 | +| DB `update(draft)` 失败 | `_stream_graph_segment` | try/except 吞掉,graph 继续推 | +| DB `update(review)` 失败 | 同上 | 同上 | +| DB `update(transcript)` 失败 | `save_transcript` 路由 | 返回 `success=false`,不抛 5xx | +| 文章类型加载失败 | `start_writing` | 退回 `[]` | +| Graph 编译失败 | `start_writing` | 抛 503(极端情况,节点 import 错) | +| Graph 运行时节点抛错 | `_stream_graph_segment` | except 推 error 事件,session 结束 | +| Cleanup 误删活跃会话 | `run_cleanup_once` | 活跃保护硬过滤 | + +### 容器优雅关闭 + +lifespan shutdown 顺序: +1. 停 cleanup scheduler +2. `wait_for_inflight_graph_tasks(30s)` — 等所有 in-flight 写作跑到下一个 checkpoint +3. 超时未结束的 task `cancel()` + 2s ack 等待 +4. `shutdown_ai_writing_graph()` — 关闭 aiosqlite Connection / pg pool + +--- + +## 十四、测试覆盖 + +| 测试文件 | 数量 | 覆盖范围 | +|---|---|---| +| `tests/test_ai_writing_session_repo.py` | 7 | Repo CRUD + 清理用的 list_older_than / delete_by_ids | +| `tests/test_ai_writing_session_recovery.py` | 8 | `_recover_session` / `_extract_interrupt` / `resume_writing` 跨 worker 恢复 | +| `tests/test_ai_writing_transcript_endpoint.py` | 5 | transcript 保存兜底 | +| `tests/test_ai_writing_cleanup.py` | 26 | 清理 + drain + checkpointer 降级 + SSE 序列化 + admin 接口 + 多 worker 检测 | +| `tests/test_portable_long_text.py` | 5 | LONGTEXT 列类型 + 迁移链 | + +跑全部: +```bash +PYTHONPATH=. uv run pytest tests/test_ai_writing*.py tests/test_portable_long_text.py -v +``` + +期望 **51 个测试全过**。 + +--- + +## 十五、关键调试 / 排查命令 + +### 看 graph 当前状态 + +```python +# 进 python shell +from deerflow.agents.ai_writing.graph import get_ai_writing_graph +graph = await get_ai_writing_graph() +snap = await graph.aget_state({"configurable": {"thread_id": ""}}) +print(snap.values, snap.next, snap.tasks) +``` + +### 直接读 sqlite checkpointer + +```bash +sqlite3 /app/.deer-flow/data/checkpoints.db +.tables # 看表(checkpoints / checkpoint_writes 等) +SELECT thread_id, checkpoint_ns, checkpoint_id FROM checkpoints + WHERE thread_id = '' LIMIT 5; +``` + +### 查 MySQL 业务行 + +```sql +SELECT id, status, draft_title, updated_at, + JSON_LENGTH(transcript) AS transcript_size + FROM ai_writing_sessions + WHERE id = ''; +``` + +### 查内存中的活跃会话(调试时手动加) + +```python +# 临时在路由加一个 admin 接口 +@router.get("/admin/inflight") +async def admin_inflight(): + return { + "event_queues": list(_event_queues.keys()), + "inflight_tasks": len(_inflight_graph_tasks), + "pending_pause": list(_session_pending_pause.keys()), + } +``` + +### 日志关键字 + +```bash +docker logs 2>&1 | grep -E 'ai_writing|AsyncSqliteSaver|checkpointer|recovered' +``` + +期望看到(启动时): +- `ai_writing: 使用 AsyncSqliteSaver (xxx.db)` — 持久化生效 +- `ai_writing_cleanup: 已启动 cron=...` — 清理调度起来 + +警告: +- `ai_writing: 持久化 checkpointer 不可用,退回 InMemorySaver` — sqlite 故障 +- `⚠ 检测到 WEB_CONCURRENCY=N` — 多 worker 警告(详见 `ai-writing-说明.md`) + +--- + +## 十六、性能与限制 + +### 单 session 资源占用 +- 内存:~50KB(state + progress_events 累计) +- sqlite checkpoint:每次 state update 写一行(一篇完整文章约 20-50 行) +- LLM tokens:约 30K-100K(取决于素材量 + 字数 + 修订次数) + +### 并发上限 +- 单 worker 内:理论上百级 session 并发(IO bound),实际 LLM API rate limit 才是瓶颈 +- 多 worker:**当前 in-memory dict 隔离,详见 ai-writing-说明.md 第六章** + +### 长会话风险 +- progress_events 累加无界 → 长会话 state 可能膨胀到几 MB +- transcript JSON 同理 → MySQL 行可能上 MB(但 LONGTEXT 列上限 4GB,影响读取性能) + +### 优化方向 +- progress_events 节流(聚合 chunk 事件) +- transcript 增量保存(diff 而非全量覆盖) +- 节点级 LLM 调用加 retry(tenacity) + +--- + +## 十七、与主 Agent 的关系 + +| 维度 | 主 Agent | AI 写作 | +|---|---|---| +| Graph 入口 | `make_lead_agent` | `AI_WRITING_GRAPH` | +| Middleware | 18 层中间件 | 无(直接节点链) | +| State | `ThreadState` | `AIWritingState` | +| Checkpointer | `make_checkpointer()`(共享) | `_get_async_checkpointer()`(同一文件,不同 thread_id 命名空间) | +| 路由前缀 | `/api/langgraph/*` `/api/threads/*` | `/api/ai-writing/*` | +| 流式协议 | LangGraph stream_mode(messages-tuple/values) | 自定义 SSE 事件 | +| 用户干预 | 通过 `ask_clarification` 工具 | 通过 LangGraph `interrupt()` | +| 持久化业务表 | `threads_meta`、`runs` | `ai_writing_sessions` | + +二者**完全独立**,仅在「共享同一个 sqlite checkpointer 文件」这一点上有间接关联(也可以拆开,性能层面没区别)。 + +--- + +## 十八、未来 RFC 方向 + +| 主题 | 描述 | +|---|---| +| 多 worker 部署支持 | 把 `_event_queues` 等搬 Redis Pub/Sub,参考 `ai-writing-说明.md` 方案 B | +| Graph 拓扑可配置 | 现在节点链路写死,未来按 article_type 走不同拓扑(如「研报」vs「散文」) | +| 节点级重试 | LLM 抽风时自动重试 N 次,参考 `tenacity` | +| 素材跨 session 复用 | 同一用户多次写作可能用同一批素材 → 缓存层 | +| 多模态素材 | 支持图片素材输入和图文混合输出 | +| 中文标点 / 排版规则注入 | 在节点 prompt 里强化中文出版规范 | + +--- + +## 附录 A:相关文件速查 + +| 文件 | 行数 | 用途 | +|---|---|---| +| `routers/ai_writing.py` | 1006 | 路由 + 状态字典 + SSE + drain | +| `agents/ai_writing/graph.py` | 538 | StateGraph + checkpointer 工厂 | +| `agents/ai_writing/state.py` | 121 | AIWritingState 定义 | +| `agents/ai_writing/nodes/_utils.py` | 242 | LLM 工具 | +| `agents/ai_writing/nodes/intent_parser.py` | 131 | 节点:意图解析 | +| `agents/ai_writing/nodes/researcher.py` | 271 | 节点:素材检索 | +| `agents/ai_writing/nodes/writer_outline.py` | 147 | 节点:大纲规划 | +| `agents/ai_writing/nodes/writer_draft.py` | 318 | 节点:草稿生成 | +| `agents/ai_writing/nodes/editor.py` | 180 | 节点:编辑审核 | +| `app/gateway/ai_writing_cleanup.py` | ~280 | 清理 + 调度 | +| `persistence/ai_writing_sessions/sql.py` | ~200 | Repository | +| `persistence/ai_writing_sessions/model.py` | ~35 | ORM | + +--- + +## 附录 B:常见踩坑 + +| 现象 | 根因 | 解决 | +|---|---|---| +| `/start` 报 500 | graph 编译失败(多半是 import 错) | 看启动日志 stack trace | +| `/resume` 报 404 | session_id 不存在或 checkpointer 没快照 | 看 `ai_writing recover` 日志 | +| transcript 500 | DB 列容量 / 网络抖 | ✅ 已修,前端 fire-and-forget | +| SSE 卡住不动 | 多 worker SSE 隔离 / 反代超时 | 见 `ai-writing-说明.md` | +| `draft_chunk` 丢失内容 | `ThinkTagParser` 在 `` 标签边界处理 | 看 `_utils.py:ThinkTagParser.feed` 单测 | +| graph 卡在某节点不退出 | LLM API 卡住 / 无限重试 | 看节点日志,加 timeout | +| cleanup 没跑 | cron 表达式配错 / 文件锁被其他 worker 持有 | 看 `ai_writing_cleanup` 日志 | +| 草稿在前端看不到但 MySQL 有 | 多 worker 导致 SSE 事件丢失 | 同上 | +| 配 mysql 做 checkpointer 报错 | LangGraph 不支持 MySQL checkpointer | 用 sqlite 或 postgres | + +--- + +如需补充内容,请直接编辑本文件,并在 git commit message 里注明改动原因。 diff --git a/frontend-web/docs/ai-writing-frontend-dev.md b/frontend-web/docs/ai-writing-frontend-dev.md new file mode 100644 index 00000000..42c32076 --- /dev/null +++ b/frontend-web/docs/ai-writing-frontend-dev.md @@ -0,0 +1,538 @@ +# AI 写作 — 前端开发文档 + +> **本文档目的**:帮助前端开发者快速理解 AI 写作子系统的代码结构、数据流、状态机和扩展点。 +> +> **配套文档**: +> - `ai-writing-说明.md` — 故障排查与运维向解决方案 +> - `ai-writing-assistant-backend-api.md` — 后端 API 详细字段定义 +> - `ai-writing-assistant-source-map.md` — 旧版前端结构索引(已合并到本文档) + +--- + +## 一、产品定位 + +AI 写作是 DeerFlow 的「**杂志社协作写作**」子产品,由 LangGraph 后端驱动 4 个 AI Agent(**素材收集专家 / 作家 / 编辑 / 系统**)协作完成一篇文章。用户通过 **4 个干预点**(素材确认、大纲确认、草稿确认、审核确认)参与决策,UI 实时呈现进度。 + +### UI 入口 +- 路由:`/canvas/ai-writing`(在 `WorkspaceRoutes.tsx` 注册) +- 入口页:`src/open-canvas/pages/AIWritingPage.tsx` +- 双栏布局:左 AI 写作面板(聊天 + 干预),右草稿预览 + +--- + +## 二、目录结构 + +``` +frontend-web/src/open-canvas/ +├── pages/ +│ └── AIWritingPage.tsx # 入口页(双栏 Resizable 布局) +│ +├── api/ # ── 后端 API 适配层 ── +│ ├── ai-writing.ts # /start /resume + SSE 订阅(核心) +│ ├── ai-writing-sessions.ts # 历史会话 CRUD + transcript 保存 +│ └── ai-writing-article-types.ts # 文章类型 CRUD +│ +├── hooks/ # ── 状态管理 ── +│ ├── useAIWriting.ts # 写作主状态机(最重要) +│ ├── useAIWritingSessions.ts # 历史会话列表 hook +│ └── useArticleTypes.ts # 文章类型选项 hook +│ +├── components/ai-writing/ # ── UI 组件 ── +│ ├── AIWritingPanel.tsx # 左侧主面板(聊天 + 干预编排) +│ ├── AIWritingDraftPanel.tsx # 右侧草稿预览面板 +│ ├── AIWritingPageHeader.tsx # 顶部标题/历史下拉 +│ ├── AIWritingForm.tsx # 写作请求表单(首次启动用) +│ │ +│ ├── AIWritingTimeline.tsx # 时间线主容器 +│ ├── TimelineStep.tsx # 单条进度事件 +│ ├── CompletedInterventionCard.tsx # 已完成的干预记录 +│ │ +│ ├── InterventionCard.tsx # 4 个干预点 + 求助点的卡片(核心) +│ ├── MaterialsCard.tsx # 素材确认 +│ ├── MaterialItem.tsx # 单个素材项 +│ ├── MaterialsToolbar.tsx # 素材筛选/搜索工具栏 +│ ├── OutlineCard.tsx # 大纲确认(含编辑) +│ ├── OutlineSection.tsx # 大纲单章节 +│ ├── ReviewCard.tsx # 审核确认 +│ ├── IssueList.tsx # 审核意见列表 +│ ├── ScoreBar.tsx # 评分进度条 +│ │ +│ ├── OutlineStreamingView.tsx # 大纲流式打字机 +│ ├── DraftStreamingView.tsx # 草稿流式打字机 +│ ├── ReviewStreamingView.tsx # 审核流式打字机 +│ ├── WritingSkeletonCard.tsx # 加载骨架屏 +│ ├── RevisionSummaryCard.tsx # 修改前后对比摘要 +│ │ +│ ├── AIWritingHistoryDropdown.tsx # 历史会话下拉 +│ └── AIWritingHistoryView.tsx # 历史会话只读回显 +│ +└── types/ + └── ai-writing.ts # 全部类型定义(重要参考) +``` + +--- + +## 三、数据流总览 + +``` +┌─────────────────────────────────────────────────────────────────┐ +│ AIWritingPanel │ +│ ┌──────────────────────────────────────────────────────────┐ │ +│ │ useAIWriting() 主状态 hook │ │ +│ │ ┌────────────────────────────────────────────────────┐ │ │ +│ │ │ startWriting() ─── POST /api/ai-writing/start │ │ │ +│ │ │ ↓ 返回 session_id │ │ │ +│ │ │ │ │ │ +│ │ │ subscribeToProgress(sessionId) │ │ │ +│ │ │ ─── GET /api/ai-writing/{id}/stream (SSE) │ │ │ +│ │ │ ─── 持续推 ProgressEvent 给 handleEvent │ │ │ +│ │ │ ─── handleEvent reducer → setState │ │ │ +│ │ │ │ │ │ +│ │ │ submitIntervention() │ │ │ +│ │ │ ─── POST /api/ai-writing/{id}/resume │ │ │ +│ │ │ ─── graph 续推 → 新事件继续走 SSE │ │ │ +│ │ │ │ │ │ +│ │ │ saveAIWritingTranscript()(fire-and-forget) │ │ │ +│ │ │ ─── PUT /api/ai-writing/sessions/{id}/transcript │ │ +│ │ └────────────────────────────────────────────────────┘ │ │ +│ └──────────────────────────────────────────────────────────┘ │ +│ │ +│ 渲染: │ +│ ├ AIWritingForm(idle 状态) │ +│ ├ AIWritingTimeline(运行中) │ +│ │ ├ TimelineStep × N(progressEvents) │ +│ │ └ CompletedInterventionCard × N(已完成的干预) │ +│ ├ {Outline,Draft,Review}StreamingView(流式打字机) │ +│ ├ InterventionCard(当前 currentPause 存在时弹出) │ +│ └ WritingSkeletonCard(运行中无 pause 时的骨架屏) │ +└─────────────────────────────────────────────────────────────────┘ +``` + +--- + +## 四、状态机(useAIWriting.ts) + +### 状态字段(完整定义见 `types/ai-writing.ts`) + +```typescript +interface AIWritingState { + // —— 核心标识 —— + sessionId: string | null; // 后端返回的会话 id + status: WritingStatus; // 当前阶段(13 种状态之一) + isPaused: boolean; // 是否在干预点等待用户 + error: string | null; + + // —— 配置 —— + request: AIWritingRequest | null; // 用户提交的写作请求 + + // —— 累计数据 —— + progressEvents: ProgressEvent[]; // SSE 事件全量历史(供时间线渲染) + materials: AIMaterial[]; // 素材列表(素材就绪后填充) + outline: AIOutline | null; // 大纲(大纲就绪后填充) + currentDraftMarkdown: string; // 草稿正文(草稿就绪后填充) + currentDraftTitle: string; + reviewResult: ReviewResult | null; + revisionCount: number; + completedInterventions: CompletedIntervention[]; // 已完成的干预历史 + + // —— 干预 —— + currentPause: ProgressEvent | null; // 当前挂起的 await_user 事件 + + // —— 流式视图状态 —— + streamingDraftTitle: string; + streamingDraftSections: StreamingDraftSection[]; + outlineStreamingText: string; + outlineThinkingText: string; + researchProgress: string; + streamingReviewItems: StreamingReviewItem[]; + revisionSummary: RevisionSummary | null; + pendingRevisionContext: { /* ... */ } | null; +} +``` + +### 状态转移图(写作状态机) + +``` + ┌──────┐ + │ idle │ ←─────────────────────────────────────┐ + └──┬───┘ │ + │ startWriting() │ + ↓ │ + ┌─────────────────┐ │ + │ researching │ │ + └──┬──────────────┘ │ + ↓ materials_ready │ + ┌─────────────────────────────┐ │ + │ awaiting_material_confirm │ ← await_user │ + └──┬──────────────────────────┘ │ + ↓ submit(confirm) │ + ┌─────────────────┐ │ + │ writing_outline │ ← (re_search 路径) │ + └──┬──────────────┘ │ + ↓ outline_ready │ + ┌─────────────────────────────┐ │ + │ awaiting_outline_confirm │ │ + └──┬──────────────────────────┘ │ + ↓ submit(confirm) │ + ┌─────────────────┐ │ + │ writing_draft │ ← (re_outline 路径) │ + └──┬──────────────┘ │ + │ │ + ┌───────┴────────┐ │ + ↓ ↓ │ + awaiting_section_ awaiting_draft_confirm │ + help (素材不足) └─┬───┬───┬──────────────┐ │ + │ (supplement) │ │ │ to_editor │ │ + │ ↑回到 research │ │ ↓ │ │ + │ │ │ reviewing │ │ + │ │ │ └─┬──────┐ │ │ + │ │ │ ↓ ↓ │ │ + │ │ │ awaiting_review_confirm │ + │ │ │ └─┬────────┐ │ + │ │ │ ↓ submit ↓ │ + │ │ │ revising done ───────────────┘ + │ │ │ │ + │ │ │ └─→ writing_draft (循环修订) + │ │ │ + │ │ │ user_revise + │ │ └─→ revising + │ │ + │ │ finalize + │ └─→ done + │ + └──→ (delete/loose) writing_draft 继续 +``` + +### handleEvent reducer(核心) + +`useAIWriting.ts` 里的 `handleEvent` 是个**纯 reducer**,根据 SSE 事件类型更新状态: + +| 事件类型 | 主要影响的状态字段 | +|---|---| +| `connected` | 仅记录到 progressEvents | +| `step_start` | 切到对应 `status`(如「编辑」→ reviewing) | +| `step_done` | 仅记录 | +| `research_progress` | `researchProgress`(实时检索进度文案) | +| `materials_ready` | `materials`、清空 researchProgress | +| `outline_chunk` | `outlineStreamingText` / `outlineThinkingText`(打字机) | +| `outline_ready` | `outline`、清空流式文本 | +| `draft_chunk` | `streamingDraftSections`(按章节拼接打字机) | +| `draft_ready` | `currentDraftMarkdown` / `currentDraftTitle` | +| `review_chunk` | `streamingReviewItems[i].streamingText`(逐条打字机) | +| `review_item_done` | `streamingReviewItems[i]` 标记 isDone + 结构化结果 | +| `review_ready` | `reviewResult` | +| `await_user` | `currentPause = event`,`isPaused = true`,**弹出 InterventionCard** | +| `resumed` | `currentPause = null` | +| `notice` | 仅记录(非致命告警) | +| `done` | `status = 'done'` | +| `error` | `status = 'error'`,`error = event.message` | + +--- + +## 五、API 层(api/) + +### 1. `api/ai-writing.ts` — 核心写作 API + +```typescript +// 启动写作 → 返回 session_id +export async function startWritingSession(req: AIWritingRequest): Promise + +// 订阅 SSE 进度流 → 返回取消订阅函数 +export function subscribeToProgress( + sessionId: string, + onEvent: (event: ProgressEvent) => void, +): () => void + +// 提交用户干预 → 让 graph 续推 +export async function resumeWriting( + sessionId: string, + payload: UserInterventionPayload, +): Promise +``` + +**关键实现细节**: +- **snake_case ↔ camelCase 转换**:后端返回 snake_case,前端类型是 camelCase。所有 normalize 函数(`normalizeMaterial`、`normalizeOutlineSection`...)集中在这一层做转换 +- **SSE 事件白名单**:`SSE_EVENT_TYPES` 数组列出所有合法事件名,未知事件会被静默忽略 +- **取消订阅**:`subscribeToProgress` 返回的函数会调用 `AbortController.abort()` 关闭 fetch stream +- **重连机制**:当前**没有内置重连** — 如果 SSE 断开(网络抖动、反代超时),需要用户主动重启写作或刷新页面 + - 后端 `_recover_session` 能让 SSE 重连时从 checkpointer 恢复事件,但前端没主动重连逻辑 + +### 2. `api/ai-writing-sessions.ts` — 历史会话 + +```typescript +// CRUD +listAIWritingSessions() : Promise +getAIWritingSession(id) : Promise +renameAIWritingSession(id, title) : Promise +deleteAIWritingSession(id) : Promise + +// 时间线持久化(fire-and-forget,永不抛错) +saveAIWritingTranscript(id, transcript) : Promise +``` + +**关键设计**:`saveAIWritingTranscript` 内部用 try/catch 吞掉所有异常(网络错误、HTTP 500 等),只 `console.warn` 不影响主流程。这是**「锦上添花」契约**的前端侧。 + +### 3. `api/ai-writing-article-types.ts` — 文章类型管理 + +| 函数 | 用途 | +|---|---| +| `listArticleTypes()` | 表单下拉框选项 | +| `createArticleType()` | 管理员新增 | +| `updateArticleType()` | 管理员编辑 | +| `deleteArticleType()` | 管理员删除 | + +--- + +## 六、组件树详解 + +### 顶层:`AIWritingPanel`(457 行,最大) + +负责: +1. 调用 `useAIWriting()` 拿主状态 +2. 调用 `useAIWritingSessions()` 拿历史列表 +3. 协调表单 / 时间线 / 干预卡 / 流式视图 / 草稿同步 +4. **自动滚底逻辑**:检测用户是否手动上划,决定是否跟随最新内容 +5. **草稿同步到右侧画布**:监听 `currentDraftMarkdown` 变化,调 `setArtifact` 更新右侧 +6. **历史会话回显**:当 `historySession` 非空时,渲染 `AIWritingHistoryView` 而不是实时面板 + +### 干预卡:`InterventionCard`(663 行,第二大) + +按 `pausePoint` 渲染**不同的卡片**: + +| pausePoint | 卡片 | 用户能做的 | +|---|---|---| +| `material_confirm` | `MaterialsCard` | 勾选保留素材 / 加关键词重新搜 | +| `outline_confirm` | `OutlineCard` | 直接编辑大纲 / 给反馈让重写 | +| `draft_confirm` | 草稿确认卡 | 定稿 / 给修订意见 / 提交编辑审核 | +| `review_confirm` | `ReviewCard` | 勾选采纳的意见 / 强制通过 | +| `section_help` | 求助卡 | 逐章选择:补充检索 / 删除 / 按通用知识续写 | + +### 时间线:`AIWritingTimeline` + +迭代 `progressEvents`,每个 `step_start` / `step_done` / `notice` 渲染成一条 `TimelineStep`。`CompletedInterventionCard` 穿插显示已完成的干预记录。 + +### 流式视图:`{Outline,Draft,Review}StreamingView` + +- **OutlineStreamingView**:打字机渲染 `outlineStreamingText`,支持 `...` 折叠显示 +- **DraftStreamingView**:按章节流式拼接 `streamingDraftSections`,每章节独立打字机 +- **ReviewStreamingView**:3 条审核(事实 / 逻辑 / 语言)并行打字机,完成后切到结构化评分 + +### 历史回显:`AIWritingHistoryView` + +只读模式,从 `AIWritingSessionSummary` 重建 UI。**当 `transcript` 字段有值时**,从里面拿到完整 `progressEvents`,UI 表现跟实时一模一样。`transcript` 没有时退回到只显示最终草稿。 + +--- + +## 七、SSE 事件协议(前后端契约) + +后端通过 SSE 推 `event: \ndata: \n\n` 格式,前端用浏览器原生 `fetch + ReadableStream` 解析(不用 `EventSource`,因为 EventSource 不能自定义 header)。 + +### 解析流程 + +``` +ReadableStream → TextDecoder + → 按 "\n\n" 分帧 + → 每帧匹配 /^event: (.+)\ndata: (.+)$/ + → JSON.parse(data) + → normalize(parsed)(snake → camel) + → onEvent(normalized) +``` + +### 事件载荷字段(不完全列表) + +| 字段 | 出现在哪些事件 | 用途 | +|---|---|---| +| `type` | 全部 | 事件类别 | +| `sessionId` | 全部 | 标识 | +| `message` | 大多数 | 显示给用户的文案 | +| `agentName` | step_* | 「素材收集专家」/「作家」/「编辑」/「系统」 | +| `materials` | materials_ready | 素材数组 | +| `outline` | outline_ready | 大纲对象 | +| `draftMarkdown` / `draftTitle` | draft_ready | 草稿内容 | +| `reviewResult` | review_ready | 完整审核结果 | +| `pausePoint` | await_user | 哪个干预点 | +| `pauseContext` | await_user | 干预所需的上下文(如待确认的素材列表) | +| `blockedSections` | await_user (section_help) | 素材不足的章节列表 | +| `title` / `detail` / `raw` | notice / error | 非致命告警的标题、详情、原始错误 | + +--- + +## 八、扩展指南:常见开发任务 + +### 新增一个干预点 + +1. **前端类型扩展**: + - 在 `types/ai-writing.ts` 的 `ProgressEvent.pausePoint` 联合类型里加新 case + - 在 `WritingStatus` 加对应的 `awaiting_xxx` + - 在 `CompletedIntervention.pausePoint` 加 case + +2. **写 InterventionCard 分支**:在 `InterventionCard.tsx` 里加新 `pausePoint` 分支,做 UI + 提交逻辑 + +3. **后端配合**(详见后端开发文档): + - `graph.py` 加 `pause_xxx_confirm` 节点 + - graph 拓扑加这个节点的边 + +4. **完成干预后的清理**:在 `useAIWriting.ts` 的 `submitIntervention` 函数里,**确保把这次干预 push 到 `completedInterventions`**,让历史能看到 + +### 新增一个 SSE 事件类型 + +1. `types/ai-writing.ts`: + - `ProgressEventType` 联合类型加新值 + - `ProgressEvent` 接口加新字段(如果有载荷) + +2. `api/ai-writing.ts`: + - `SSE_EVENT_TYPES` 数组加新值(不加会被忽略) + - `normalizeEvent` 函数里加字段转换(如果是 snake_case 需要 normalize) + +3. `useAIWriting.ts` 的 `handleEvent`:加新 case 处理状态更新 + +4. 组件层:根据需要在时间线 / 干预卡 / 流式视图里渲染 + +### 修改一个流式打字机的行为 + +每个 streaming view 内部都是独立的: +- 状态在 `useAIWriting.ts` 的 `streaming*` 系列字段 +- 渲染在 `*StreamingView.tsx` +- 流式 chunk 处理在 `handleEvent` 的 `outline_chunk` / `draft_chunk` / `review_chunk` 分支 + +### 新增草稿到右侧画布的逻辑 + +`AIWritingPanel.tsx` 里 `useEffect` 监听 `state.currentDraftMarkdown` → 调 `setArtifact` 更新右侧 `CanvasGraphContext`。如果需要新的 artifact 类型,看 `contexts/CanvasGraphContext.tsx` 的 artifact 接口。 + +--- + +## 九、容错与边界处理 + +### 已落实的容错(不会阻断写作) +- ✅ `saveAIWritingTranscript` 失败仅 `console.warn` +- ✅ 历史会话列表加载失败仅 `console.warn` +- ✅ 重命名/删除会话失败静默吞掉 +- ✅ 后端 SSE 推 `notice` / `error` 事件时,前端只渲染不中断 +- ✅ JSON parse 失败时 SSE 帧被静默丢弃(不影响后续帧) + +### 已知边界(可能影响 UI 体验) +- ⚠️ **SSE 断线无自动重连** — 网络抖动 / 反代超时后用户需要手动刷新 +- ⚠️ **多 worker 部署下 SSE 事件丢失**(详见 `ai-writing-说明.md`)—— 前端只能感知到「卡住」,没有重连恢复机制 +- ⚠️ `outline_chunk` / `draft_chunk` / `review_chunk` 的 `text` / `is_thinking` 字段是 snake_case,前端类型是 `text` / `isThinking` 但 normalize 时是**保留 snake_case 直接读**(参考 `useAIWriting.ts` 的 `(event as unknown as { text: string }).text`),这是历史包袱,未来 normalize 函数应统一处理 + +--- + +## 十、调试技巧 + +### 看 SSE 原始事件 +浏览器 DevTools → Network → 找 `/api/ai-writing//stream` 请求 → Response(注意是 EventStream 格式,不能直接展开 JSON)。 + +### 查当前 hook 状态 +在浏览器 console 里: +```js +// 需要安装 React DevTools,在 Components tab 找到 AIWritingPanel +// 或在 useAIWriting 里临时加: +useEffect(() => { (window as any).__aiwriting = state; }, [state]); +// 然后在 console 里 __aiwriting 看 +``` + +### 模拟事件 +开发期可以注释掉 `subscribeToProgress`,手动调 `handleEvent({...})` 测试 reducer 行为。 + +### 看后端 transcript +``` +GET /api/ai-writing/sessions/ +``` +返回里 `transcript` 字段就是前端组装过的完整时间线 JSON。 + +--- + +## 十一、性能注意点 + +### 大数组渲染 +- `progressEvents` 在长写作里可能上千条 → `AIWritingTimeline` 应该考虑虚拟滚动(当前未实现,会议室会议讨论过) +- `streamingDraftSections` 章节多时也是潜在瓶颈 + +### transcript PUT 频率 +- 当前每次 `progressEvents` 变化都会触发保存(实际上是 useEffect 监听) +- 大量 chunk 事件会导致**每秒 N 次 PUT**,后端 MySQL 写压力大 +- **优化建议**:在 `useAIWriting.ts` 给 saveAIWritingTranscript 加 debounce(1-2 秒),或者只在 `await_user` / `done` / `error` 关键节点保存 + +### 浏览器内存 +- progressEvents 不主动清理 → 长会话可能占用几十 MB +- 完成一篇文章后建议 `reset()` 清空状态 + +--- + +## 十二、相关后端接口(速查) + +| 接口 | 用途 | 前端调用方 | +|---|---|---| +| `POST /api/ai-writing/start` | 启动写作 | `startWritingSession` | +| `GET /api/ai-writing/{id}/stream` | SSE 进度流 | `subscribeToProgress` | +| `POST /api/ai-writing/{id}/resume` | 提交干预 | `resumeWriting` | +| `GET /api/ai-writing/{id}/state` | 查 graph 状态快照(调试用) | 暂未调用 | +| `GET /api/ai-writing/sessions` | 历史列表 | `listAIWritingSessions` | +| `GET /api/ai-writing/sessions/{id}` | 历史详情 | `getAIWritingSession` | +| `PATCH /api/ai-writing/sessions/{id}` | 重命名 | `renameAIWritingSession` | +| `DELETE /api/ai-writing/sessions/{id}` | 删除会话 | `deleteAIWritingSession` | +| `PUT /api/ai-writing/sessions/{id}/transcript` | 保存时间线(fire-and-forget) | `saveAIWritingTranscript` | +| `GET /api/ai-writing/article-types` | 文章类型下拉 | `listArticleTypes` | +| `POST /api/ai-writing/admin/cleanup` | admin 手动清理 | 暂未在 UI 暴露 | + +--- + +## 十三、未来优化方向 + +| 项目 | 说明 | 工作量 | +|---|---|---| +| transcript PUT 防抖 | 减少 MySQL 写压力 | 1 小时 | +| SSE 自动重连 | 网络抖动 / 反代超时后能恢复 | 0.5 天 | +| 时间线虚拟滚动 | 处理超长 progressEvents | 0.5 天 | +| 干预卡组件解耦 | 663 行的 InterventionCard 拆成 5 个独立卡片 | 1 天 | +| AbortController 串行化 | 避免重启写作时旧 SSE 还没断 | 0.5 天 | +| 后端字段统一 camelCase | 消除 snake_case ↔ camelCase 转换层 | 需后端配合 | + +--- + +## 附录 A:所有类型一览 + +完整定义在 `types/ai-writing.ts`,主要类型: + +```typescript +// 业务实体 +AIMaterial // 单个素材 +AIOutline // 大纲 +OutlineSection // 大纲单章节 +ReviewIssue // 审核问题 +ReviewResult // 完整审核结果 +BlockedSection // 素材不足的章节 +SectionDecision // 章节求助决策 + +// 流程 +WritingStatus // 13 种状态 +ProgressEventType // 17 种事件类型 +ProgressEvent // SSE 事件载荷 +UserInterventionPayload // 干预提交体 +AIWritingRequest // 启动写作的请求 +CompletedIntervention // 已完成的干预记录 + +// 流式视图 +StreamingDraftSection +StreamingReviewItem +RevisionSummary + +// Hook 状态 +AIWritingState // useAIWriting 的完整状态 +``` + +--- + +## 附录 B:常见踩坑 + +| 现象 | 原因 | 解决 | +|---|---|---| +| 前端类型 camelCase 但拿到 undefined | 后端 snake_case 没 normalize | 在 `api/ai-writing.ts` 的 normalize 函数里加字段 | +| 进度卡住不动 | SSE 断了或落到错的 worker | 刷新页面,看 `ai-writing-说明.md` 多 worker 部分 | +| 历史会话回显时间线缺失 | transcript 保存失败 | DB 列容量 / 后端日志查 `save_transcript` | +| 修改大纲后没生效 | 用户 PATCH editedOutline 但没勾选 confirm | InterventionCard 大纲分支检查 action 字段 | +| 草稿渲染换行错乱 | LLM 输出包含奇怪 markdown | `DraftStreamingView` 里 markdown parser 设置 | + +--- + +如需补充内容,请直接编辑本文件,并在 git commit message 里注明改动原因。 diff --git a/frontend-web/docs/ai-writing-说明.md b/frontend-web/docs/ai-writing-说明.md new file mode 100644 index 00000000..cf343693 --- /dev/null +++ b/frontend-web/docs/ai-writing-说明.md @@ -0,0 +1,461 @@ +# AI 写作 — 故障排查与解决方案说明 + +> **本文档目的**:完整记录 AI 写作子系统在内网 Docker 部署后出现的问题、根因、已落地的修复、剩余风险、以及未来的解决方案路径。 +> +> **适用读者**:维护 AI 写作功能的前端/后端开发者、内网部署运维。 +> +> **更新日期**:2026-05-20 + +--- + +## 一、问题背景 + +### 起因 +内网 Docker 部署后,AI 写作出现两类报错: + +| 接口 | 报错 | 频率 | +|---|---|---| +| `PUT /api/ai-writing/sessions/{id}/transcript` | HTTP 500 | 经常 | +| `POST /api/ai-writing/{id}/resume` | HTTP 4xx(用户记忆为 400,实际可能是 404) | 经常,且没有 Docker 重启时也发生 | + +### 这两个接口的作用 + +#### 1. `PUT /sessions/{id}/transcript` — 对话时间线持久化 +- **职责**:保存前端组装的完整对话时间线(progressEvents + completedInterventions)到 MySQL +- **调用时机**:前端在写作过程中**反复 PUT**(事件累计、干预完成、草稿生成等节点全量覆盖) +- **存到哪**:MySQL 的 `ai_writing_sessions.transcript` 列(LONGTEXT) +- **用途**:用户**关闭页面后再回来 / 刷新 / 看历史会话**时,从 DB 拉回这段时间线,UI 上回显出「跟实时一样的对话历史」 +- **定位**:**纯锦上添花** — 失败了用户当下写作完全不受影响,只是历史回显会缺失/不完整 + +#### 2. `POST /{id}/resume` — 写作流程的核心控制接口 +- **职责**:提交用户在干预点的决策,让 LangGraph 从挂起的 interrupt 点继续推进 +- **调用时机**:用户在 4 个 interrupt 点(素材确认 / 大纲确认 / 草稿确认 / 审核确认)做出决定后 +- **传什么**:用户的决策内容(确认的素材 id、修改后的大纲、用户的修订意见、章节求助决策…) +- **定位**:**流程主干** — 它失败,用户就**真的卡死**在干预点 + +--- + +## 二、根因分析 + +### 根因 1:`transcript` 列容量不足 + 路由层无兜底 +- MySQL `TEXT` 列上限 64 KB +- 用户写作过程中累计 `progressEvents`(含每个流式 chunk)+ `completedInterventions` 后,JSON blob 经常超过 64 KB +- `session.commit()` 抛 `DataError: Data too long for column 'transcript'` +- 原路由没有 try/except,异常直接冒泡成 HTTP 500 + +### 根因 2:内存中的会话状态在多 worker 部署下隔离 +后端用了大量模块级 dict 维护会话状态: + +```python +_event_queues: dict[str, asyncio.Queue] # SSE 事件队列 +_session_pending_pause: dict[str, tuple] # 挂起的 await_user 事件 +_session_model_names: dict[str, str] # 用户选的模型 +_session_interventions: dict[str, list] # 累计干预历史 +_session_repos: dict[str, repo] # DB 仓库引用 +_session_user_ids: dict[str, str] # 用户 id +_session_article_types: dict[str, list] # 文章类型列表 +``` + +**uvicorn / gunicorn 多 worker 是多进程共享 socket**,OS 内核决定哪个 worker 接到请求,应用层和 nginx 都管不着。所以: + +``` +T0: POST /start → worker A 创建 _event_queues[sid] = QueueA + graph 推进任务跑在 worker A +T1: GET /stream → 落 worker B,没有 QueueA 的引用 + → 触发 _recover_session,从 sqlite 重建 QueueB + → SSE 推 QueueB 但事件都在 QueueA 里 +T2: POST /resume → 落 worker C,同样无法看到 A/B 的状态 +``` + +**结果**:SSE 时断时续、resume 偶发 404「会话状态已丢失」。 + +### 根因 3:Graph 用了内存 checkpointer +原代码: +```python +# ai_writing/graph.py +checkpointer = MemorySaver() +``` +进程内的 `MemorySaver` 重启即丢,**容器重启或单 worker 进程被回收**后所有进行中会话都拿不到。 + +### 根因 4:路径配错 +`config.yaml` 里 sqlite 路径写成了 Windows 绝对路径 `D:\cursor-work\...`,Linux 容器里根本无法解析。 + +--- + +## 三、已落地的修复(已合入 dhk-1 分支) + +### 修复 1:`transcript` 列改 LONGTEXT + 路由兜底 + +| 文件 | 改动 | +|---|---| +| `packages/harness/deerflow/persistence/types.py` | 新增 `PortableLongText` 类型(MySQL→LONGTEXT,SQLite/PG→TEXT) | +| `packages/harness/deerflow/persistence/ai_writing_sessions/model.py` | `transcript` 列改成 `PortableLongText` | +| `packages/harness/deerflow/persistence/migrations/versions/20260519_02_ai_writing_transcript_longtext.py` | Alembic 迁移:MySQL 上 `ALTER ... type_=LONGTEXT()` | +| `app/gateway/routers/ai_writing.py` | `save_transcript` 整段 try/except 包裹,**任何异常都返回 `{success: false, reason}`,永远不抛 5xx** | + +**结果**:transcript 500 永远不会再发生。前端已经做了 fire-and-forget 容错,失败只 `console.warn`,写作流程不阻断。 + +### 修复 2:Checkpointer 改为持久化 SQLite + +| 文件 | 改动 | +|---|---| +| `packages/harness/deerflow/agents/ai_writing/graph.py` | `MemorySaver()` → 懒加载 `AsyncSqliteSaver` | +| 同上 | 新增 `get_ai_writing_graph()` / `get_ai_writing_checkpointer()` / `shutdown_ai_writing_graph()` | +| `offline-backend-20260512/config.yaml` | sqlite 路径改成相对路径 `.deer-flow/data/checkpoints.db` | + +**关键容错**:sqlite 初始化失败(磁盘满 / 权限 / aiosqlite 未装 / 损坏)会**自动降级 InMemorySaver**,写作流程能跑完,但跨重启不可恢复。 + +### 修复 3:跨 worker / 进程重启会话恢复 + +| 文件 | 改动 | +|---|---| +| `app/gateway/routers/ai_writing.py` | 新增 `_recover_session(session_id, request)` | + +当 worker 内存里没有该 session 时,从 sqlite checkpointer 读快照,重建 `_event_queues`、重新注入 `await_user` 事件。 + +**效果**:worker B 收到的 resume 请求,能从 sqlite 找到 worker A 写下的 checkpoint,**resume 接口本身能跑通**(但 SSE 事件流仍隔离 — 见下文剩余风险)。 + +### 修复 4:多层失败兜底,保证写作流程能继续 + +| 改动 | 作用 | +|---|---| +| `_sse(event, data)` 加 `default=str` + notice 帧降级 | JSON 序列化失败不会让 SSE 断开 | +| 所有 DB 写入(草稿/审核/干预/status)已包 try/except | DB 异常吞掉,graph 继续推 | +| 文章类型加载失败 → 退回 `[]` | intent_parser 仍能跑 | +| `/start` 加 503 兜底 | graph 构建失败时给前端可读错误而不是 500 | + +### 修复 5:定期清理 + admin 触发接口 + +| 文件 | 改动 | +|---|---| +| `app/gateway/ai_writing_cleanup.py` | 新文件:cron 调度 + 联动清理(MySQL 行 + sqlite checkpointer thread) | +| `packages/harness/deerflow/persistence/ai_writing_sessions/sql.py` | 新增 `list_older_than()` / `delete_by_ids()` | +| `app/gateway/app.py` | lifespan 启动/停止清理任务 | +| `app/gateway/routers/ai_writing.py` | 新增 `POST /api/ai-writing/admin/cleanup` 手动触发接口(admin only) | +| `config.yaml` | 新增 `ai_writing.cleanup` 段(默认每天凌晨 3 点跑,保留 7 天) | + +**关键安全机制**:清理永远跳过 `_event_queues` 里仍在跑的会话(**活跃保护**),即使 admin 手动传 `retention_days: 0` 也不会秒杀正在写作的用户。 + +### 修复 6:优雅关闭 — In-flight Drain + +| 文件 | 改动 | +|---|---| +| `app/gateway/routers/ai_writing.py` | 新增 `_inflight_graph_tasks` + `_spawn_graph_task()` + `wait_for_inflight_graph_tasks(timeout)` | +| `app/gateway/app.py` | lifespan shutdown 顺序:取消清理 → **等 in-flight 写作 task 收尾(30s 超时)** → 关 checkpointer connection | + +容器收到 SIGTERM 时,给正在跑的写作 30 秒窗口跑到下一个 interrupt / END,已生成的 checkpoint 都安全落盘。 + +### 修复 7:多 worker 配置陷阱检测 + +| 文件 | 改动 | +|---|---| +| `app/gateway/app.py` | lifespan startup 检测 `WEB_CONCURRENCY > 1` 或 `/proc` 里同名 uvicorn 进程数 > 1 → 打 ERROR 日志 | + +下次启动容器看 docker logs 第一屏就能看到强提醒。 + +--- + +## 四、当前状态评估 + +### transcript 500 +✅ **彻底修复**,跟 worker 数无关。LONGTEXT 列 + 路由兜底双保险。 + +### resume 报错 +⚠️ **大幅缓解**: +- 接口本身(在 worker > 1 下)能跑通 +- sqlite 共享让 worker B 也能恢复 worker A 创建的 session + +❌ **未彻底解决**: +- SSE 事件流仍 per-worker 隔离 +- 用户可能看到「进度时有时无、需要刷新页面才能看到草稿」 +- 但**草稿、审核结果**都落到 MySQL,刷新 `/sessions/{id}` 总能看到最终成品 + +### 「写作流程能不能跑完」 +✅ **能**。最坏情况下: +- 用户感受:进度断断续续,需要刷新几次 +- 数据保证:草稿和审核都落 MySQL,**写出来的内容不会丢** + +--- + +## 五、剩余风险 + +### 风险 A:多 worker 下 SSE 事件流隔离 ⚠️ 最重要 + +**症状**:用户 SSE 长连接绑在 worker B,但 graph 推进任务跑在 worker A,事件 put 进 A 的 queue → SSE 收不到。 + +**为什么不能用 `--workers 1` 解决**:业务团队反馈其他业务功能需要多 worker。 + +**解决方案见第六章**。 + +### 风险 B:容器优雅关闭时 in-flight 写作进度 + +**当前缓解**:30 秒 drain timeout 让 graph 跑到下一个 checkpoint。 + +**完美方案**(如有需求):无法 100% 防护,只能尽量缩短「未落 checkpoint」窗口。 + +### 风险 C:节点级别 LLM 抽风 + +某个写作节点调用 LLM 时网络抖动 / 限流 → astream 抛错 → session 结束。 + +**当前缓解**:astream level 已 try/except,推 error 事件。用户从 checkpoint 重试。 + +**改进方向**(不在本次修复范围): +- 给节点加 `tenacity.retry` 重试装饰器 +- 节点抛错时把已生成的部分塞进 progress_events + +### 风险 D:跨 worker 恢复时降级丢失 + +worker B 从 sqlite 恢复 worker A 的 session 时,以下数据回退到默认: + +| 数据 | 默认值 | 影响 | +|---|---|---| +| `_session_model_names` | `""` | 退回默认模型 | +| `_session_interventions` | `[]` | 本次 resume 之前的干预历史从内存里丢(MySQL 里有副本,但内存累计不全) | +| `_session_article_types` | `[]` | 没影响(graph 早过了 intent_parser) | + +**改进方向**:恢复时从 MySQL 反向填充内存元数据(约 20 行代码)。 + +--- + +## 六、Worker > 1 限制下的解决方案路径 + +### 三条路径对比 + +| 方案 | 是否引入新依赖 | 代码改动 | 运维改动 | 适用场景 | +|---|---|---|---|---| +| **B. Redis Pub/Sub 共享状态** | ⚠️ 需 Redis | ~150 行 | docker-compose 加 redis | 已有 Redis 或愿意上 | +| **C. AI 写作独立单 worker 进程** | 无 | ~30 行 | docker-compose 加一个容器 / 进程 + nginx 转发 | 不想上 Redis | +| **F. 多容器单 worker + nginx sticky** | 无 | ~5 行(nginx config) | 把 1 个多 worker 容器拆成 N 个单 worker 容器 | 已经在做容器编排 | + +### 方案 B:Redis Pub/Sub(最彻底) + +**改造点**: + +``` +之前(per-worker memory dict) 之后(Redis 共享) +───────────────────────── ───────────────────── +_event_queues[sid] = asyncio.Queue() → Redis Pub/Sub channel "ai_writing:events:{sid}" +_session_pending_pause → Redis Hash "ai_writing:pause" +_session_model_names → Redis Hash "ai_writing:meta:model" +_session_interventions → Redis Hash + JSON +_session_repos / _session_user_ids → Redis Hash +``` + +**关键设计**: +1. `/start` 在 worker A:graph 推进任务跑在 A,事件 publish 到 Redis channel +2. `/stream` 在 worker B:subscribe Redis channel,把消息转 SSE 推前端 +3. `/resume` 在 worker C:用 Redis 分布式锁保证「同一 session 同时只有一个 worker 驱动 graph」 + +**优点**:根治。任何 worker 数、任何部署形态都能跑。 + +**缺点**: +- 引入 Redis 依赖(运维负担) +- 代码改动较大(约 150 行) +- 需要处理 Redis 断连 / 分布式锁释放等边缘情况 + +### 方案 C:AI 写作独立服务(次推荐) + +**做法**:在同一个容器里多起一个 uvicorn 进程,专门处理 `/api/ai-writing/*`,**显式 `--workers 1`**。 + +```bash +# 主 gateway(多 worker,处理其他业务) +uvicorn app.gateway.app:app --host 0.0.0.0 --port 8001 --workers 4 + +# AI 写作专用单 worker +uvicorn app.gateway.app:app --host 0.0.0.0 --port 8002 --workers 1 +``` + +nginx 配置: +```nginx +location /api/ai-writing/ { + proxy_pass http://localhost:8002; + proxy_buffering off; + proxy_read_timeout 3600s; +} + +location / { + proxy_pass http://localhost:8001; +} +``` + +**优点**: +- 零代码改动 +- 不引入新依赖 +- AI 写作彻底单 worker,问题根除 + +**缺点**: +- 容器内多一个进程 +- 内存占用 +1 份 Python runtime(约 200 MB) + +### 方案 F:多容器单 worker + nginx sticky + +**做法**:把当前 1 个 4 worker 容器,拆成 4 个 1 worker 容器,nginx 用 `ip_hash` 做 sticky。 + +```yaml +# docker-compose.yml +services: + gateway-1: { ports: ["8001:8001"], command: ["...", "--workers", "1"] } + gateway-2: { ports: ["8002:8001"], command: ["...", "--workers", "1"] } + gateway-3: { ports: ["8003:8001"], command: ["...", "--workers", "1"] } + gateway-4: { ports: ["8004:8001"], command: ["...", "--workers", "1"] } + + nginx: + # upstream 配 ip_hash,让同一个客户端粘到同一个容器 +``` + +```nginx +upstream gateway_pool { + ip_hash; + server gateway-1:8001; + server gateway-2:8001; + server gateway-3:8001; + server gateway-4:8001; +} +``` + +**优点**:零代码改动,标准做法 + +**缺点**: +- 改动 docker-compose +- 客户端 IP 不均匀时负载不均衡(但 AI 写作低并发场景影响小) + +### 推荐顺序 + +1. **方案 C(AI 写作独立单 worker 进程)** — 改动最小、风险最低、立刻可上 +2. **方案 F(多容器 sticky)** — 如果你们已经在做多容器编排 +3. **方案 B(Redis)** — 长期方案,等真正需要水平扩容时再上 + +--- + +## 七、运维操作清单 + +### 部署前验证 + +```bash +# 1) 确认 .deer-flow 数据卷挂出来了(持久化的命脉) +docker exec ls -la /app/.deer-flow/data/ +# 应该能看到 checkpoints.db 文件(首次启动后会自动创建) + +# 2) 确认 alembic 迁移跑完了 +docker exec sh -c "cd /app/backend && PYTHONPATH=. uv run alembic current" +# 应该看到包含 20260519_02 的 head + +# 3) 确认 transcript 列已扩容 +docker exec mysql -uroot -p deerflow -e \ + "SHOW COLUMNS FROM ai_writing_sessions LIKE 'transcript';" +# Type 列应该是 longtext + +# 4) 查 worker 数 +docker exec ps -ef | grep -E 'uvicorn|gunicorn' +# 如果 > 1,且没做任何方案 C/F/B,启动日志会打 ERROR 警告 +``` + +### 启动时关注的日志 + +```bash +docker logs 2>&1 | head -100 | grep -iE 'ai_writing|checkpointer|worker' +``` + +期望看到: +- `ai_writing: 使用 AsyncSqliteSaver (xxx.db)` — 持久化生效 +- `ai_writing_cleanup: 已启动 cron=...` — 清理任务起来了 + +警告: +- `⚠ 检测到 WEB_CONCURRENCY=N` — 多 worker 配置警告 +- `ai_writing: 持久化 checkpointer 不可用,退回 InMemorySaver` — sqlite 初始化失败,需立刻排查 + +### 用户报告「会话丢失」时的排查 + +```bash +# 1) 查 sqlite 里这个 session 的快照 +docker exec sh -c "sqlite3 /app/.deer-flow/data/checkpoints.db \ + \"SELECT thread_id, checkpoint_ns FROM checkpoints WHERE thread_id='' LIMIT 5;\"" + +# 2) 查 MySQL 里这条会话 +docker exec mysql -uroot -p deerflow -e \ + "SELECT id, status, draft_title, updated_at FROM ai_writing_sessions WHERE id='';" + +# 3) 看 gateway 日志里这个 session 的活动轨迹 +docker logs 2>&1 | grep '' | tail -50 + +# 4) 看是否有 recover_session 调用 +docker logs 2>&1 | grep 'ai_writing: recovered' | tail -20 + +# 5) 看是否有节点异常 +docker logs 2>&1 | grep -E 'ai_writing graph segment failed|ai_writing.*error' | tail -20 +``` + +### 手动触发清理(admin only) + +```bash +# 用 admin 账号登录后,取 cookie,调清理接口 +curl -X POST 'http:///api/ai-writing/admin/cleanup' \ + -H 'Content-Type: application/json' \ + -H 'Cookie: ' \ + -d '{"retention_days": 7}' +``` + +返回: +```json +{ + "deleted_db": 12, + "deleted_checkpoints": 12, + "skipped_active": 1, + "elapsed_seconds": 0.234, + "params": { "retention_days": 7, "only_finished": false, "batch_size": 500 } +} +``` + +`skipped_active` > 0 表示有正在跑的会话被活跃保护拦下来,不会被误删。 + +--- + +## 八、相关文件索引 + +### 后端 +- `offline-backend-20260512/backend/app/gateway/routers/ai_writing.py` — 路由 + 内存状态字典 + drain +- `offline-backend-20260512/backend/app/gateway/ai_writing_cleanup.py` — 定期清理 +- `offline-backend-20260512/backend/app/gateway/app.py` — lifespan / 多 worker 检测 +- `offline-backend-20260512/backend/packages/harness/deerflow/agents/ai_writing/graph.py` — Graph 定义 + 持久化 checkpointer +- `offline-backend-20260512/backend/packages/harness/deerflow/persistence/ai_writing_sessions/` — MySQL repo +- `offline-backend-20260512/backend/packages/harness/deerflow/persistence/types.py` — `PortableLongText` +- `offline-backend-20260512/backend/packages/harness/deerflow/persistence/migrations/versions/20260519_02_*.py` — LONGTEXT 迁移 +- `offline-backend-20260512/config.yaml` — `checkpointer:` + `ai_writing.cleanup:` 配置 + +### 前端 +- `frontend-web/src/open-canvas/pages/AIWritingPage.tsx` — 页面顶部有快速排查注释 +- `frontend-web/src/open-canvas/api/ai-writing-sessions.ts` — transcript 保存(已做 fire-and-forget 容错) +- `frontend-web/src/open-canvas/api/ai-writing.ts` — resume / stream + +### 测试 +- `tests/test_ai_writing_cleanup.py` — 清理 + drain + checkpointer 降级 + SSE 序列化(21 个测试) +- `tests/test_ai_writing_session_recovery.py` — 跨 worker 会话恢复(8 个测试) +- `tests/test_ai_writing_session_repo.py` — Repo(7 个测试) +- `tests/test_ai_writing_transcript_endpoint.py` — transcript 兜底(5 个测试) +- `tests/test_portable_long_text.py` — LONGTEXT 列类型(5 个测试) + +总测试:**51 个,全过**。 + +--- + +## 九、变更历史 + +| 日期 | Commit | 内容 | +|---|---|---| +| 2026-05-19 | c3d0c82 | 报错兜底(transcript LONGTEXT 迁移 + 路由 try/except + session 恢复) | +| 2026-05-20 | (本次) | checkpointer 持久化 + 定期清理 + admin 接口 + 多层兜底 + drain + 多 worker 检测 | + +--- + +## 十、决策摘要(给未来的你/同事) + +> **如果你现在第一次接手这个问题**,按这个清单往下走: +> +> 1. ✅ **transcript 500 不用管** — 已彻底修复 +> 2. ⚠️ **resume 报错的根因是多 worker** — sqlite 持久化已减轻,但 SSE 事件流仍隔离 +> 3. ⚠️ **写作能跑完,但用户体验断断续续** — 草稿数据安全,刷新就能看到 +> 4. 🎯 **要彻底解决,从方案 C / F / B 选一个**(建议从 C 开始) +> 5. 🚨 **务必确认 `.deer-flow` 数据卷挂出来了** — 否则 sqlite 持久化白做 +> 6. 📊 **关注的日志关键字**:`ai_writing:`、`checkpointer`、`AsyncSqliteSaver`、`WEB_CONCURRENCY` + +如有疑问,先看本文件的「七、运维操作清单」做基础排查,再问当前维护者。 diff --git a/offline-backend-20260512/backend/docs/AI_WRITING_BACKEND_DEV_ZH.md b/offline-backend-20260512/backend/docs/AI_WRITING_BACKEND_DEV_ZH.md new file mode 100644 index 00000000..ab2f9732 --- /dev/null +++ b/offline-backend-20260512/backend/docs/AI_WRITING_BACKEND_DEV_ZH.md @@ -0,0 +1,845 @@ +# AI 写作 — 后端开发文档 + +> **本文档目的**:帮助后端开发者快速理解 AI 写作子系统的代码结构、Graph 拓扑、数据流、持久化、容错策略和扩展点。 +> +> **配套文档**: +> - 前端:`frontend-web/docs/ai-writing-frontend-dev.md` +> - 故障排查:`frontend-web/docs/ai-writing-说明.md` +> - 前端 API 字段索引:`frontend-web/docs/ai-writing-assistant-backend-api.md` + +--- + +## 一、产品定位与技术架构 + +### 产品定位 +**杂志社协作写作子系统**:由 4 个 AI Agent(**素材收集专家 / 作家 / 编辑 / 系统**)协作完成一篇文章,通过 4 个用户干预点参与决策。 + +### 技术栈 +- **编排框架**:LangGraph(独立 StateGraph,不复用主 lead agent 的 middleware 链) +- **节点驱动**:异步 LLM 调用(`langchain_openai` 等),支持流式 token 推送到 SSE +- **状态持久化**:LangGraph checkpointer(SQLite 文件,跨重启可恢复) +- **业务数据**:MySQL `ai_writing_sessions` 表 +- **协议**:SSE(Server-Sent Events)推进度 + REST 控制流 + +### 与主 Agent 的关系 +**完全独立**。AI 写作 graph 不通过 `make_lead_agent` 也不走主 agent 的 18 层 middleware。它有自己的 StateGraph、自己的 checkpointer、自己的路由前缀(`/api/ai-writing/*`),跟主 agent 解耦。 + +--- + +## 二、目录结构 + +``` +offline-backend-20260512/backend/ +├── app/gateway/ +│ ├── routers/ +│ │ └── ai_writing.py # FastAPI 路由 + 模块级状态字典 + SSE 生成器(1000+ 行,核心) +│ ├── ai_writing_cleanup.py # 定期清理 + cron 调度(独立模块) +│ └── app.py # lifespan 启动/停止清理任务、检测多 worker、关闭 checkpointer +│ +└── packages/harness/deerflow/ + ├── agents/ai_writing/ + │ ├── __init__.py # 暴露 AI_WRITING_GRAPH + 工厂函数 + │ ├── graph.py # ⭐ StateGraph 定义 + 4 个 pause 节点 + checkpointer 工厂(538 行) + │ ├── state.py # AIWritingState TypedDict(121 行) + │ │ + │ ├── nodes/ # ── Graph 节点 ── + │ │ ├── _utils.py # LLM 工具:parse_json, ThinkTagParser, llm_json + │ │ ├── intent_parser.py # 节点 1:意图解析 + │ │ ├── researcher.py # 节点 2:素材检索(支持流式 progress) + │ │ ├── writer_outline.py # 节点 3:大纲规划 + │ │ ├── writer_draft.py # 节点 4:草稿生成(按章节流式) + │ │ └── editor.py # 节点 5:编辑审核(3 维度并行流式) + │ │ + │ ├── prompts/ # ── Prompt 模板 ── + │ │ ├── __init__.py + │ │ ├── researcher_prompts.py + │ │ ├── writer_prompts.py + │ │ └── editor_prompts.py + │ │ + │ └── search/ # ── 素材检索 provider ── + │ ├── base.py # SearchResult / SearchProvider 接口 + │ ├── duckduckgo_search.py + │ ├── internet_search.py # 互联网检索(默认) + │ ├── intranet_stub.py # 内网检索占位(按部署接入) + │ └── demo_search.py # 离线 demo 数据 + │ + └── persistence/ai_writing_sessions/ + ├── __init__.py # AIWritingSessionRepository 工厂 + ├── model.py # ORM:AIWritingSessionRow + └── sql.py # CRUD 实现(含清理用的 list_older_than / delete_by_ids) +``` + +--- + +## 三、Graph 拓扑(最重要) + +### StateGraph 节点 + 边 + +``` + START + │ + ↓ + ┌─────────────────┐ + │ intent_parser │ 解析用户意图(确定 article_type / target_audience) + └────────┬────────┘ + ↓ + ┌─────────────────┐ + │ researcher │ ← (re_search 回到这里) + └────────┬────────┘ + ↓ (after_researcher: ⼟材 ready) + ┌──────────────────────┐ + │ pause_material │ ⚙️ PAUSE-1:素材确认 + └─┬──────────┬─────────┘ + │ confirm │ re_search + ↓ └→ researcher + ┌─────────────────┐ + │ writer_outline │ ← (re_outline 回到这里) + └────────┬────────┘ + ↓ + ┌──────────────────────┐ + │ pause_outline │ ⚙️ PAUSE-2:大纲确认 + └─┬──────────┬─────────┘ + │ confirm │ re_outline + ↓ └→ writer_outline + ┌─────────────────┐ + │ writer_draft │ ← (素材补充后回到这里) + └─┬───────────────┘ + ↓ (after_writer_draft) + ┌──────┴────────────────┐ + ↓ ↓ + ┌──────────────────┐ ┌────────────────────┐ + │ pause_section_ │ │ pause_draft │ ⚙️ PAUSE-3:草稿确认 + │ help(素材不足)│ │ │ + └─┬─────────────┬──┘ └─┬────────┬───┬─────┘ + │ supplement │ del/ │ │ │ to_editor + │↓ │ loose │ │ user_revise + research writer_ │ final ↓ + draft │ ize writer_draft + │↓ + done + │ + (to_editor)│ + ↓ + ┌─────────────────┐ + │ editor │ 3 维度并行流式审核 + └────────┬────────┘ + ↓ (after_review) + ┌──────────────────────┐ + │ pause_review │ ⚙️ PAUSE-4:审核确认 + └─┬──────────┬─────────┘ + │ confirm │ revise + │↓ └→ writer_draft + done +``` + +### 节点定义位置一览 + +| 节点名 | 实现文件 | 类型 | 流式? | +|---|---|---|---| +| `intent_parser` | `nodes/intent_parser.py` | async function | 否 | +| `researcher` | `nodes/researcher.py` | async function | 是(research_progress) | +| `pause_material` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `writer_outline` | `nodes/writer_outline.py` | async function | 是(outline_chunk) | +| `pause_outline` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `writer_draft` | `nodes/writer_draft.py` | async function | 是(draft_chunk,按章节) | +| `pause_section_help` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `pause_draft` | `graph.py` 顶部 | async, 调 `interrupt()` | — | +| `editor` | `nodes/editor.py` | async function | 是(review_chunk,3 路并行) | +| `pause_review` | `graph.py` 顶部 | async, 调 `interrupt()` | — | + +### 条件路由函数(graph.py) + +| 函数 | 在哪个节点之后 | 路由依据 | +|---|---|---| +| `after_researcher` | researcher | 是否找到素材 | +| `after_material_pause` | pause_material | `state["status"]` 是 researching 还是 writing_outline | +| `after_outline_pause` | pause_outline | 同上,writing_outline 还是 writing_draft | +| `after_writer_draft` | writer_draft | 是否有 blocked_sections | +| `after_section_help_pause` | pause_section_help | status | +| `after_draft_pause` | pause_draft | done / revising / reviewing | +| `after_review` | editor | review_result.verdict 是 pass 还是 reject + revision_count 上限 | +| `after_review_pause` | pause_review | done / writing_draft | + +--- + +## 四、State 定义(state.py) + +```python +class AIWritingState(TypedDict): + # —— 写作配置(启动时填充)—— + user_intent: str + article_type: str + target_audience: str + word_count_target: int + keyword_count: int # 1-10,默认 6 + writing_mode: str # "strict" | "loose" + + # —— Agent 产出 —— + material_package: Optional[MaterialPackage] # 素材收集专家 + current_outline: Optional[Outline] # 作家 + drafts: Annotated[List[DraftArticle], operator.add] # 累计所有版本 + current_draft: Optional[DraftArticle] # 最新草稿 + review_result: Optional[ReviewResult] # 编辑 + + # —— 用户干预 —— + last_user_intervention: Optional[UserInterventionPayload] + blocked_sections: Optional[List[BlockedSection]] # 素材不足的章节 + + # —— 流程控制 —— + revision_count: int + max_revisions: int # 默认 3 + status: WritingStatus + + # —— SSE 推送队列 —— + progress_events: Annotated[List[dict], operator.add] # 累加,不覆盖 +``` + +### State 字段的特殊语义 + +- **`Annotated[..., operator.add]`** — LangGraph 的累加 reducer,节点 return 时 `progress_events: [evt]` 会**追加**到现有列表,而不是替换 +- **TypedDict** — 运行时是普通 dict,类型注解仅供 IDE 提示 +- **节点输出** — 节点 return 的 dict 只需要包含**要更新的字段**,未指定的字段保持不变 + +--- + +## 五、路由层(app/gateway/routers/ai_writing.py) + +### 14 个接口一览 + +| HTTP | 路径 | 用途 | 鉴权 | +|---|---|---|---| +| 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` | 查 graph 状态快照(调试用) | 用户 | +| GET | `/api/ai-writing/sessions` | 列出当前用户的历史会话 | 用户 | +| GET | `/api/ai-writing/sessions/{id}` | 获取单个历史会话 | 用户 | +| PATCH | `/api/ai-writing/sessions/{id}` | 重命名会话 | 用户 | +| DELETE | `/api/ai-writing/sessions/{id}` | 删除会话 | 用户 | +| PUT | `/api/ai-writing/sessions/{id}/transcript` | 保存对话时间线(**fire-and-forget**) | 用户 | +| GET | `/api/ai-writing/article-types` | 列出文章类型 | 用户 | +| POST | `/api/ai-writing/article-types` | 新增文章类型 | 用户(建议改成 admin) | +| PATCH | `/api/ai-writing/article-types/{id}` | 更新文章类型 | 用户 | +| DELETE | `/api/ai-writing/article-types/{id}` | 删除文章类型 | 用户 | +| POST | `/api/ai-writing/admin/cleanup` | **手动触发清理** | admin only | + +### 模块级状态字典(关键设计 — 也是 worker 限制的根源) + +```python +# 每个 session_id 对应一份内存状态。多 worker 部署时每个 worker 一份,互不共享。 +_event_queues: dict[str, asyncio.Queue] # SSE 事件队列 +_session_pending_pause: dict[str, tuple] # 挂起的 await_user 事件(供重连重投) +_session_model_names: dict[str, str] # 用户选的模型 +_session_interventions: dict[str, list] # 累计干预历史 +_session_repos: dict[str, repo] # DB 仓库引用 +_session_user_ids: dict[str, str] # 用户 id +_session_article_types: dict[str, list] # 动态文章类型列表 +_session_error_flags: dict[str, bool] +_inflight_graph_tasks: set[asyncio.Task] # in-flight task 注册表(lifespan drain 用) +_recovery_lock: asyncio.Lock # 跨 worker 会话恢复的并发锁 +``` + +⚠️ **关键限制**:这些都是 per-process 内存。多 worker 部署下,worker B 看不到 worker A 的状态 → SSE 事件分发失败。详见 `frontend-web/docs/ai-writing-说明.md` 第五、六章。 + +### `_stream_graph_segment`(graph 推进函数) + +`/start` 和 `/resume` 都调用这个函数(用 `_spawn_graph_task` 包装),负责: +1. 调 `graph.astream(input_or_command, config=config, stream_mode="updates")` +2. 把每个 chunk 解析成 SSE 事件 put 进 `_event_queues[session_id]` +3. 遇到 `__interrupt__` → 推 `await_user` 事件 → 返回(不关闭 queue) +4. 遇到 END → 推 `done` 事件 → 关闭 queue(put None 哨兵) +5. 任何异常 → 推 `error` 事件 → 关闭 queue +6. finally 段:更新 DB 最终状态、清理所有模块字典 + +### SSE 生成器 `_sse_generator` + +`GET /stream` 的实现,从 `_event_queues[session_id]` 拿事件 → `_sse(event, data)` 序列化 → yield 给 fastapi `StreamingResponse`。 + +**关键逻辑**: +- 队列不存在时**等待 5 秒**,期间触发 `_recover_session`(跨 worker / 进程重启恢复) +- `_sse()` 三层兜底:`default=str` → 手写 notice 帧 → 永不抛 TypeError +- 收到 None 哨兵 → 关闭流 + +### `_recover_session`(跨 worker 恢复) + +当 worker 内存里没有 session_id 但 sqlite checkpointer 有快照时: +1. 从 checkpointer 读出 state snapshot +2. 重建 `_event_queues[sid]` 队列 +3. 从 snapshot 提取当前 interrupt 负载,put 进队列 +4. 同时尽量恢复 `_session_repos`、`_session_user_ids` 等 + +返回三种状态: +- `"interrupted"` — 图停在干预点,已重建队列 +- `"finished"` — 图已完成或中断在节点中间,推 done/error 事件后关闭 +- `"unrecoverable"` — sqlite 也没有这个 thread_id → 真的丢了 + +### `_spawn_graph_task` / `wait_for_inflight_graph_tasks`(优雅关闭) + +- `_spawn_graph_task(coro)` — 启 task 并自动登记到 `_inflight_graph_tasks`,完成时反登记 +- `wait_for_inflight_graph_tasks(timeout)` — lifespan shutdown 等所有 task 跑到下一个 interrupt / END(默认 30s),超时则 cancel 强制结束 + +--- + +## 六、Graph 与持久化(graph.py) + +### Lazy 单例 + 自动降级 + +```python +AI_WRITING_GRAPH = None # 模块级槽位(测试可 monkeypatch) +_async_checkpointer_singleton = None # checkpointer 单例 + +async def get_ai_writing_graph(): + """懒加载 graph。第一次调用时按 config.yaml 决定 checkpointer。""" + global AI_WRITING_GRAPH + if AI_WRITING_GRAPH is not None: + return AI_WRITING_GRAPH + checkpointer = await _get_async_checkpointer() + AI_WRITING_GRAPH = build_ai_writing_graph(checkpointer=checkpointer) + return AI_WRITING_GRAPH +``` + +### Checkpointer 优先级(_get_async_checkpointer) + +``` +1. config.yaml 的 checkpointer 段 + ├── type: sqlite → AsyncSqliteSaver(aiosqlite.connect(...)) + ├── type: postgres → AsyncPostgresSaver(AsyncConnectionPool(...)) + └── type: memory → InMemorySaver + +2. 任何初始化失败 → InMemorySaver(降级,打 warning) +3. 未配置 checkpointer 段 → InMemorySaver(降级,打 warning) +4. 未知 type(如 "mysql") → InMemorySaver(降级,打 warning) +``` + +**核心契约**:`_get_async_checkpointer()` **绝不抛异常**,永远返回一个可用的 saver。这是「写作流程能完整继续」契约的关键。 + +### Shutdown + +`shutdown_ai_writing_graph()` 在 lifespan shutdown 时调用: +- 关闭 `aiosqlite.Connection` / `AsyncConnectionPool` +- 清空 `AI_WRITING_GRAPH` / `_async_checkpointer_singleton` + +调用顺序(详见 `app/gateway/app.py` lifespan): +1. 停清理任务 +2. `wait_for_inflight_graph_tasks(30s)` +3. `shutdown_ai_writing_graph()` + +--- + +## 七、节点详解(nodes/) + +### `nodes/_utils.py`(公共工具) + +| 函数 / 类 | 用途 | +|---|---| +| `get_model(config, model_name)` | 从 config.configurable 拿 model_name 或回退默认 | +| `parse_json(text, fallback)` | 鲁棒 JSON 解析(支持代码块、不完整 JSON、转义错误) | +| `_extract_balanced(text)` | 找出第一对平衡的 `{...}` | +| `_repair_json(text)` | 修复常见的 JSON 错误(缺尾引号等) | +| `ThinkTagParser` | 流式解析 `...` 标签(思考 / 正文分流) | +| `llm_json(system, user, fallback, config)` | 调 LLM 并解析 JSON 输出 | +| `count_words(text)` | 中文 + 英文混合计字数 | + +### `nodes/intent_parser.py` — 意图解析 + +- 调 LLM 把用户意图(`user_intent` 字符串)解析成结构化字段:`article_type`、`target_audience`、`word_count_target` +- 用户已经在 `AIWritingRequest` 里指定的字段会**覆盖** LLM 推断 +- 输出:更新 state 的对应字段,推 `step_done` 事件 + +### `nodes/researcher.py` — 素材检索 + +- 根据 `user_intent` 用 LLM 生成 N 个关键词(N = `keyword_count`,默认 6) +- 用 `search/*` 的 provider 并行检索 +- 流式推 `research_progress` 事件("正在搜索关键词 X 中...") +- 输出:`material_package = {keywords, materials, summary}`,推 `materials_ready` + +### `nodes/writer_outline.py` — 大纲规划 + +- 输入:`user_intent` + `material_package` +- 调 LLM 流式生成大纲 JSON,期间推 `outline_chunk`(正文 + thinking 分流) +- 解析最终 JSON → `current_outline`,推 `outline_ready` + +### `nodes/writer_draft.py` — 草稿生成(最复杂) + +- 按大纲**逐章节**调 LLM 生成正文,每章节独立流式推 `draft_chunk` +- 检测 LLM 是否「拒答」(`_looks_like_refusal`),如果是则换提示重试 +- **严格模式下**:如果章节关联的素材不足,把该章节加入 `blocked_sections`,触发 `pause_section_help` +- 完成所有章节后拼成 `current_draft.full_markdown`,推 `draft_ready` + +### `nodes/editor.py` — 编辑审核(3 维度并行) + +- 3 个维度(**fact / logic / language**)并行调 LLM +- 每个维度独立流式推 `review_chunk`,完成后推 `review_item_done` +- 三个都完成后聚合成 `review_result`,推 `review_ready` +- 评分加权(fact 40% + logic 30% + language 30%),超过 `pass_threshold`(默认 80)verdict = `pass` + +--- + +## 八、持久化(persistence/ai_writing_sessions/) + +### `model.py` — ORM + +```python +class AIWritingSessionRow(Base): + __tablename__ = "ai_writing_sessions" + id: Mapped[str] # = session_id(uuid) + user_id: Mapped[str | None] # NULL = no-auth 模式 + title: Mapped[str] # 从 user_intent 截取前 50 字 + user_intent: Mapped[str | None] + status: Mapped[str] # in_progress / draft_ready / review_ready / done / error + draft_title: Mapped[str | None] + draft_markdown: Mapped[str | None] + completed_interventions: Mapped[str | None] # JSON 字符串 + review_result: Mapped[str | None] # JSON 字符串 + transcript: Mapped[str | None] # PortableLongText(MySQL→LONGTEXT,4GB) + created_at / updated_at: BeijingDateTime +``` + +### `sql.py` — Repository + +| 方法 | 用途 | +|---|---| +| `create(id, title, user_intent, user_id)` | 启动写作时建一行 | +| `get(session_id, user_id)` | 取单个(带 user_id 权限校验) | +| `list(user_id, limit)` | 列出当前用户的会话 | +| `update(session_id, user_id, ...)` | 更新指定字段(其他保持不变) | +| `delete(session_id, user_id)` | 删除(带权限校验) | +| `list_older_than(cutoff, only_finished, limit)` | 清理用:列出过期 session_id | +| `delete_by_ids(session_ids)` | 清理用:批量删除 | + +### Alembic 迁移 + +- `20260517_03_ai_writing_transcript.py` — 加 transcript 列(TEXT) +- `20260519_02_ai_writing_transcript_longtext.py` — MySQL 上把 transcript 改成 LONGTEXT(修 64KB 限制导致的 500) + +--- + +## 九、定期清理(ai_writing_cleanup.py) + +### 设计目标 +- 删超过 `retention_days`(默认 7)的旧会话 +- 联动清理:MySQL 业务行 + SQLite checkpointer 里的 thread 状态 +- **绝对不删正在写的会话**(活跃保护) + +### 关键函数 + +```python +async def run_cleanup_once(*, retention_days, only_finished=False, batch_size=500) -> dict +``` + +返回 `{deleted_db, deleted_checkpoints, skipped_active}`。 + +执行流程: +1. 调 `repo.list_older_than(cutoff, only_finished, batch_size)` 拿过期 id +2. 从 `_event_queues.keys()` 拿活跃 session_id 集合 → **过滤掉所有仍在跑的** +3. 对每个待删 id:调 `checkpointer.adelete_thread(id)` 删 checkpoint +4. 调 `repo.delete_by_ids(ids)` 批量删 MySQL +5. 循环直到本批 < batch_size + +### Cron 调度 + +`cleanup_scheduler_loop()` 常驻后台任务: +- 用 `croniter` 解析 `ai_writing.cleanup.cron`(默认 `"0 3 * * *"`,每天凌晨 3 点) +- `asyncio.sleep` 到下次触发时间 +- 多 worker 用文件锁(`runtime_home() / ".ai_writing_cleanup.lock"`)做 leader election + +### Admin 手动触发 + +`POST /api/ai-writing/admin/cleanup`: +- 鉴权:`system_role == "admin"` +- 可选覆盖 `retention_days` / `only_finished` / `batch_size` +- 即使传 `retention_days: 0` 也不会秒杀活跃会话(活跃保护硬保证) + +--- + +## 十、配置(config.yaml) + +```yaml +# Checkpointer:graph 状态持久化 +checkpointer: + type: sqlite # sqlite | postgres | memory + connection_string: .deer-flow/data/checkpoints.db + +# 清理调度 +ai_writing: + cleanup: + enabled: true + retention_days: 7 + cron: "0 3 * * *" # 每天凌晨 3 点 + delete_only_finished: false + batch_size: 500 + + # (可选)素材检索 provider 配置 + search: + provider: deerflow.agents.ai_writing.search.internet_search:WebSearchProvider + ... +``` + +`_ai_writing_config()` 函数从 `get_app_config().ai_writing` 里读 dict,节点通过 `config.configurable.ai_writing_config` 拿到。 + +--- + +## 十一、SSE 事件协议(后端 → 前端契约) + +后端通过 `_sse(event_type, data)` 推 `event: \ndata: \n\n` 格式。 + +### 推出事件的位置 + +| 事件 | 推出位置 | +|---|---| +| `connected` | `_sse_generator` 入口 | +| `step_start` / `step_done` | 节点内部主动 emit | +| `research_progress` | `researcher_node` 内的 `_emit` | +| `materials_ready` | `_stream_graph_segment` 检测到 `material_package` 字段 | +| `outline_chunk` | `writer_outline_node` 流式推(通过 `sse_queue`) | +| `outline_ready` | `_stream_graph_segment` 检测到 `current_outline` | +| `draft_chunk` | `writer_draft_node` 按章节流式推 | +| `draft_ready` | `_stream_graph_segment` 检测到 `current_draft` | +| `review_chunk` / `review_item_done` | `editor_node` 3 维度并行流式 | +| `review_ready` | `_stream_graph_segment` 检测到 `review_result` | +| `await_user` | `_stream_graph_segment` 捕获 `__interrupt__` | +| `resumed` | pause 节点 return 时塞进 `progress_events` | +| `notice` | 节点内手动推(非致命告警) | +| `done` | `_stream_graph_segment` 正常结束 | +| `error` | `_stream_graph_segment` except 分支 | + +### 节点如何流式推 chunk + +通过 `config.configurable.sse_queue`: + +```python +async def writer_outline_node(state, config): + sse_queue = config.get("configurable", {}).get("sse_queue") + sse_session_id = config.get("configurable", {}).get("sse_session_id") + + async for chunk in llm.astream(prompt): + if sse_queue: + await sse_queue.put(("outline_chunk", { + "session_id": sse_session_id, + "text": chunk.content, + "is_thinking": False, + })) + + return {"current_outline": parsed, ...} +``` + +`sse_queue` 是路由层在 `_make_config` 里注入的 `_event_queues[session_id]`。 + +--- + +## 十二、扩展指南:常见开发任务 + +### 任务 1:新增一个 Graph 节点 + +**例**:在 `editor` 之后加一个「术语统一」节点。 + +1. **写节点**:在 `nodes/` 新建 `terminology.py`: + ```python + async def terminology_node(state, config): + # ... 读 current_draft 改术语 + return { + "current_draft": new_draft, + "progress_events": [{"type": "step_done", "agent_name": "术语统一", "message": "..."}], + } + ``` + +2. **注册到 graph**:在 `graph.py` 的 `build_ai_writing_graph` 里: + ```python + from .nodes.terminology import terminology_node + g.add_node("terminology", terminology_node) + # 改 editor 后的边 + g.add_edge("editor", "terminology") + g.add_edge("terminology", "pause_review") + ``` + +3. **State 加字段**(如果需要):`state.py` 加新字段到 `AIWritingState` + +4. **前端事件支持**(如果要新事件类型):见前端文档「扩展指南」 + +5. **测试**:在 `tests/` 加单元测试,至少测节点能正常 return 和 graph 拓扑能编译 + +### 任务 2:新增一个用户干预点 + +1. **写 pause 节点**:在 `graph.py` 顶部加: + ```python + async def pause_xxx(state): + user_input = interrupt({ + "pause_point": "xxx_confirm", + "message": "...", + "context_field": state["..."], + }) + action = user_input.get("action", "confirm") + if action == "confirm": + return {"status": "next_status", "progress_events": [{"type": "resumed", ...}]} + # ... 其他分支 + ``` + +2. **路由层 ResumeRequest 加字段**(如果干预需要新字段): + ```python + class ResumeRequest(BaseModel): + # ... + new_field: Optional[str] = None + ``` + +3. **`resume_writing` 路由把字段塞进 Command(resume=...)**: + ```python + intervention = { + # ... + "newField": body.new_field, + } + ``` + +4. **graph 拓扑加 pause 节点的边**:在 build_ai_writing_graph 里 + +5. **前端配合**:见前端文档 + +### 任务 3:新增一种素材检索 provider + +1. 在 `agents/ai_writing/search/` 加新文件,实现 `SearchProvider` 接口(参考 `internet_search.py`): + ```python + class MyProvider(SearchProvider): + async def search(self, keyword: str, limit: int) -> list[SearchResult]: + ... + ``` + +2. 配置 `config.yaml`: + ```yaml + ai_writing: + search: + provider: my_pkg.search:MyProvider + # provider 特定的配置... + ``` + +3. `researcher_node` 通过 `config.configurable.ai_writing_config.search.provider` 用 `resolve_class` 动态加载 + +### 任务 4:修改 LLM Prompt + +所有 prompt 集中在 `agents/ai_writing/prompts/`: +- `researcher_prompts.py` — 检索专家 +- `writer_prompts.py` — 大纲 + 草稿 +- `editor_prompts.py` — 审核 + +改 prompt 后**强烈建议**: +1. 在 `tests/` 加针对该 prompt 输出的回归测试(用 LLM mock 或固定输入) +2. 跑现有测试确认拓扑兼容 + +### 任务 5:扩展持久化字段 + +例:给 `ai_writing_sessions` 加一个 `total_tokens` 列。 + +1. `model.py` 加列: + ```python + total_tokens: Mapped[int | None] = mapped_column(Integer, nullable=True) + ``` + +2. **必须**新建 alembic 迁移: + ```bash + PYTHONPATH=. uv run alembic revision -m "add total_tokens column" + ``` + 然后填 upgrade/downgrade 逻辑(参考 `20260519_02_ai_writing_transcript_longtext.py`) + +3. `sql.py` 的 `update()` / `to_dict()` 加字段处理 + +4. `routers/ai_writing.py` 的 `SessionResponse` 加字段 + +5. 测试:在 `tests/test_ai_writing_session_repo.py` 加新字段的 round-trip 测试 + +--- + +## 十三、容错策略(关键设计原则) + +### 「写作流程能完整继续」契约 + +后端承诺:**除非 Pydantic 校验失败(4xx),AI 写作的 4 个核心接口任何后端组件失败都不阻断当次写作的完整执行**。 + +### 落实位置 + +| 失败点 | 兜底位置 | 行为 | +|---|---|---| +| sqlite checkpointer 初始化失败 | `_get_async_checkpointer` | 自动降级 InMemorySaver | +| SSE JSON 序列化失败 | `_sse()` | 三层兜底(default=str → notice 帧) | +| DB `create()` session 行失败 | `start_writing` | try/except 吞掉,session_id 仍返回 | +| DB `update(draft)` 失败 | `_stream_graph_segment` | try/except 吞掉,graph 继续推 | +| DB `update(review)` 失败 | 同上 | 同上 | +| DB `update(transcript)` 失败 | `save_transcript` 路由 | 返回 `success=false`,不抛 5xx | +| 文章类型加载失败 | `start_writing` | 退回 `[]` | +| Graph 编译失败 | `start_writing` | 抛 503(极端情况,节点 import 错) | +| Graph 运行时节点抛错 | `_stream_graph_segment` | except 推 error 事件,session 结束 | +| Cleanup 误删活跃会话 | `run_cleanup_once` | 活跃保护硬过滤 | + +### 容器优雅关闭 + +lifespan shutdown 顺序: +1. 停 cleanup scheduler +2. `wait_for_inflight_graph_tasks(30s)` — 等所有 in-flight 写作跑到下一个 checkpoint +3. 超时未结束的 task `cancel()` + 2s ack 等待 +4. `shutdown_ai_writing_graph()` — 关闭 aiosqlite Connection / pg pool + +--- + +## 十四、测试覆盖 + +| 测试文件 | 数量 | 覆盖范围 | +|---|---|---| +| `tests/test_ai_writing_session_repo.py` | 7 | Repo CRUD + 清理用的 list_older_than / delete_by_ids | +| `tests/test_ai_writing_session_recovery.py` | 8 | `_recover_session` / `_extract_interrupt` / `resume_writing` 跨 worker 恢复 | +| `tests/test_ai_writing_transcript_endpoint.py` | 5 | transcript 保存兜底 | +| `tests/test_ai_writing_cleanup.py` | 26 | 清理 + drain + checkpointer 降级 + SSE 序列化 + admin 接口 + 多 worker 检测 | +| `tests/test_portable_long_text.py` | 5 | LONGTEXT 列类型 + 迁移链 | + +跑全部: +```bash +PYTHONPATH=. uv run pytest tests/test_ai_writing*.py tests/test_portable_long_text.py -v +``` + +期望 **51 个测试全过**。 + +--- + +## 十五、关键调试 / 排查命令 + +### 看 graph 当前状态 + +```python +# 进 python shell +from deerflow.agents.ai_writing.graph import get_ai_writing_graph +graph = await get_ai_writing_graph() +snap = await graph.aget_state({"configurable": {"thread_id": ""}}) +print(snap.values, snap.next, snap.tasks) +``` + +### 直接读 sqlite checkpointer + +```bash +sqlite3 /app/.deer-flow/data/checkpoints.db +.tables # 看表(checkpoints / checkpoint_writes 等) +SELECT thread_id, checkpoint_ns, checkpoint_id FROM checkpoints + WHERE thread_id = '' LIMIT 5; +``` + +### 查 MySQL 业务行 + +```sql +SELECT id, status, draft_title, updated_at, + JSON_LENGTH(transcript) AS transcript_size + FROM ai_writing_sessions + WHERE id = ''; +``` + +### 查内存中的活跃会话(调试时手动加) + +```python +# 临时在路由加一个 admin 接口 +@router.get("/admin/inflight") +async def admin_inflight(): + return { + "event_queues": list(_event_queues.keys()), + "inflight_tasks": len(_inflight_graph_tasks), + "pending_pause": list(_session_pending_pause.keys()), + } +``` + +### 日志关键字 + +```bash +docker logs 2>&1 | grep -E 'ai_writing|AsyncSqliteSaver|checkpointer|recovered' +``` + +期望看到(启动时): +- `ai_writing: 使用 AsyncSqliteSaver (xxx.db)` — 持久化生效 +- `ai_writing_cleanup: 已启动 cron=...` — 清理调度起来 + +警告: +- `ai_writing: 持久化 checkpointer 不可用,退回 InMemorySaver` — sqlite 故障 +- `⚠ 检测到 WEB_CONCURRENCY=N` — 多 worker 警告(详见 `ai-writing-说明.md`) + +--- + +## 十六、性能与限制 + +### 单 session 资源占用 +- 内存:~50KB(state + progress_events 累计) +- sqlite checkpoint:每次 state update 写一行(一篇完整文章约 20-50 行) +- LLM tokens:约 30K-100K(取决于素材量 + 字数 + 修订次数) + +### 并发上限 +- 单 worker 内:理论上百级 session 并发(IO bound),实际 LLM API rate limit 才是瓶颈 +- 多 worker:**当前 in-memory dict 隔离,详见 ai-writing-说明.md 第六章** + +### 长会话风险 +- progress_events 累加无界 → 长会话 state 可能膨胀到几 MB +- transcript JSON 同理 → MySQL 行可能上 MB(但 LONGTEXT 列上限 4GB,影响读取性能) + +### 优化方向 +- progress_events 节流(聚合 chunk 事件) +- transcript 增量保存(diff 而非全量覆盖) +- 节点级 LLM 调用加 retry(tenacity) + +--- + +## 十七、与主 Agent 的关系 + +| 维度 | 主 Agent | AI 写作 | +|---|---|---| +| Graph 入口 | `make_lead_agent` | `AI_WRITING_GRAPH` | +| Middleware | 18 层中间件 | 无(直接节点链) | +| State | `ThreadState` | `AIWritingState` | +| Checkpointer | `make_checkpointer()`(共享) | `_get_async_checkpointer()`(同一文件,不同 thread_id 命名空间) | +| 路由前缀 | `/api/langgraph/*` `/api/threads/*` | `/api/ai-writing/*` | +| 流式协议 | LangGraph stream_mode(messages-tuple/values) | 自定义 SSE 事件 | +| 用户干预 | 通过 `ask_clarification` 工具 | 通过 LangGraph `interrupt()` | +| 持久化业务表 | `threads_meta`、`runs` | `ai_writing_sessions` | + +二者**完全独立**,仅在「共享同一个 sqlite checkpointer 文件」这一点上有间接关联(也可以拆开,性能层面没区别)。 + +--- + +## 十八、未来 RFC 方向 + +| 主题 | 描述 | +|---|---| +| 多 worker 部署支持 | 把 `_event_queues` 等搬 Redis Pub/Sub,参考 `ai-writing-说明.md` 方案 B | +| Graph 拓扑可配置 | 现在节点链路写死,未来按 article_type 走不同拓扑(如「研报」vs「散文」) | +| 节点级重试 | LLM 抽风时自动重试 N 次,参考 `tenacity` | +| 素材跨 session 复用 | 同一用户多次写作可能用同一批素材 → 缓存层 | +| 多模态素材 | 支持图片素材输入和图文混合输出 | +| 中文标点 / 排版规则注入 | 在节点 prompt 里强化中文出版规范 | + +--- + +## 附录 A:相关文件速查 + +| 文件 | 行数 | 用途 | +|---|---|---| +| `routers/ai_writing.py` | 1006 | 路由 + 状态字典 + SSE + drain | +| `agents/ai_writing/graph.py` | 538 | StateGraph + checkpointer 工厂 | +| `agents/ai_writing/state.py` | 121 | AIWritingState 定义 | +| `agents/ai_writing/nodes/_utils.py` | 242 | LLM 工具 | +| `agents/ai_writing/nodes/intent_parser.py` | 131 | 节点:意图解析 | +| `agents/ai_writing/nodes/researcher.py` | 271 | 节点:素材检索 | +| `agents/ai_writing/nodes/writer_outline.py` | 147 | 节点:大纲规划 | +| `agents/ai_writing/nodes/writer_draft.py` | 318 | 节点:草稿生成 | +| `agents/ai_writing/nodes/editor.py` | 180 | 节点:编辑审核 | +| `app/gateway/ai_writing_cleanup.py` | ~280 | 清理 + 调度 | +| `persistence/ai_writing_sessions/sql.py` | ~200 | Repository | +| `persistence/ai_writing_sessions/model.py` | ~35 | ORM | + +--- + +## 附录 B:常见踩坑 + +| 现象 | 根因 | 解决 | +|---|---|---| +| `/start` 报 500 | graph 编译失败(多半是 import 错) | 看启动日志 stack trace | +| `/resume` 报 404 | session_id 不存在或 checkpointer 没快照 | 看 `ai_writing recover` 日志 | +| transcript 500 | DB 列容量 / 网络抖 | ✅ 已修,前端 fire-and-forget | +| SSE 卡住不动 | 多 worker SSE 隔离 / 反代超时 | 见 `ai-writing-说明.md` | +| `draft_chunk` 丢失内容 | `ThinkTagParser` 在 `` 标签边界处理 | 看 `_utils.py:ThinkTagParser.feed` 单测 | +| graph 卡在某节点不退出 | LLM API 卡住 / 无限重试 | 看节点日志,加 timeout | +| cleanup 没跑 | cron 表达式配错 / 文件锁被其他 worker 持有 | 看 `ai_writing_cleanup` 日志 | +| 草稿在前端看不到但 MySQL 有 | 多 worker 导致 SSE 事件丢失 | 同上 | +| 配 mysql 做 checkpointer 报错 | LangGraph 不支持 MySQL checkpointer | 用 sqlite 或 postgres | + +--- + +如需补充内容,请直接编辑本文件,并在 git commit message 里注明改动原因。