从 LangGraph 到小说 Agent Runtime：用 flashNovel 实现章节级工作流、上下文记忆与人工确认
本文以我的个人项目
flashNovel(https://github.com/CuSO41108/flashnovel)为例，记录一次从“单次 Prompt 生成小说”到“章节级 Agent Runtime”的工程化尝试。项目当前还不是完整的 multi-agent 系统，更准确地说，它是一个基于 LangGraph 的多阶段 LLM 工作流：把长篇小说创作拆成 context、plan、draft、extract、check、review、rewrite、commit、checkpoint 等节点，并通过结构化记忆维持章节之间的连续性。参考了GitHub上的LoreSmith项目
1. LangGraph 是什么
LangGraph 是 LangChain 生态里的一个图编排框架，适合构建有状态、可循环、可分支的 LLM workflow 或 agent runtime。
如果只写一个简单的 ChatGPT 调用，普通函数就够了：
用户输入 -> LLM -> 输出
但真实的 Agent 应用往往不是一次调用能完成的。它可能需要：
保存中间状态；
根据模型输出决定下一步；
多轮检查、修正、重试；
中间暂停，等待人工确认；
在多个工具或多个角色之间切换；
出错后从 checkpoint 恢复。
这类流程更接近状态机，而不是一条直线。LangGraph 的核心价值就是把这些节点、状态和分支显式表达出来。
官方文档中比较关键的几个能力包括：
StateGraph：把应用建模成状态图，每个节点读取和更新 state。
Conditional edges：根据 state 决定下一步走向。
Persistence / checkpointer：通过 checkpoint 保存图状态，支持恢复、人工介入、时间旅行等能力。
Interrupts：在图执行中暂停，等待用户输入后继续。
Multi-agent patterns：可以用 supervisor、handoff、orchestrator-worker 等模式组织多个 agent。
LangSmith / Studio：用于 trace、调试和观察 graph 执行过程。
参考资料：
LangGraph Workflows and Agents
LangGraph Persistence
LangGraph Interrupts
LangGraph Multi-agent
2. 为什么小说生成适合 graph
最开始我对“AI 写小说”的想象很简单：给一个故事设定，让模型直接生成下一章。
但真正尝试长篇连续创作时，会很快遇到问题：
前几章发生过什么？
人物关系有没有变化？
某个伏笔是否已经埋下或回收？
时间线有没有冲突？
本章是否完成了章节目标？
如果质量不够，是直接重写，还是只局部润色？
什么时候把这一章写入长期记忆？
如果每次都把全部历史正文塞进 Prompt，token 成本会越来越高，而且模型也不一定能稳定抓住重点。更合理的方式是把章节生成拆成一个工作流：
load_context -> plan -> draft -> extract -> check -> review -> rewrite -> commit -> checkpoint
在这个流程里，每一步都有相对清晰的职责：
| 节点 | 作用 |
|---|---|
load_context | 读取本章需要的上下文包 |
plan | 生成章节计划 |
draft | 根据计划和上下文写正文 |
extract | 从草稿中抽取候选记忆 |
check | 检查连续性和设定冲突 |
review | 像主编一样判断是否通过 |
rewrite | 根据 review 意见重写 |
commit | 提交最终章节，并把候选记忆写入长期记忆 |
checkpoint | 判断是否继续下一章，或暂停等待人工确认 |
这样做以后，小说生成就不再是一次 Prompt，而是一个可观察、可恢复、可迭代的章节级状态机。
3. flashNovel 当前怎么用 LangGraph
在flashNovel里，LangGraph 的入口是backend/app/graph/workflow.py。
核心代码大致如下：
from langgraph.graph import END, START, StateGraph graph = StateGraph(GraphState) graph.add_node("load_context", load_context_node(self.runtime)) graph.add_node("plan", plan_node(self.runtime)) graph.add_node("draft", draft_node(self.runtime)) graph.add_node("extract", extract_node(self.runtime)) graph.add_node("check", check_node(self.runtime)) graph.add_node("review", review_node(self.runtime)) graph.add_node("rewrite", rewrite_node(self.runtime)) graph.add_node("commit", commit_node(self.runtime)) graph.add_node("checkpoint", checkpoint_node(self.runtime)) graph.add_node("finish", finish_node(self.runtime)) graph.add_edge(START, "load_context") graph.add_edge("load_context", "plan") graph.add_edge("plan", "draft") graph.add_edge("draft", "extract") graph.add_edge("extract", "check") graph.add_edge("check", "review") graph.add_conditional_edges( "review", route_after_review, {"rewrite": "rewrite", "commit": "commit"}, ) graph.add_edge("rewrite", "extract") graph.add_edge("commit", "checkpoint") graph.add_conditional_edges( "checkpoint", route_after_checkpoint, {"load_context": "load_context", "finish": "finish"}, ) graph.add_edge("finish", END) self._compiled = graph.compile()这里最重要的是两个条件边。
第一个条件边发生在review之后：
review -> rewrite review -> commit
如果审稿节点认为当前章节需要修改，就进入rewrite；如果通过，就进入commit。
第二个条件边发生在checkpoint之后：
checkpoint -> load_context checkpoint -> finish
如果本批次还没写够指定章节数，就继续下一章；如果达到上限，就暂停等待人工确认。
3.1 GraphState：工作流共享状态
项目里的 state 定义在backend/app/graph/state.py：
class GraphState(TypedDict, total=False): run_id: str story_id: str seed_prompt: str chapter: int chapters_done_in_batch: int max_chapters: int context: dict[str, Any] plan: dict[str, Any] draft: str candidate_memory: dict[str, Any] consistency: dict[str, Any] review: dict[str, Any] rewrite_count: int status: str next_action: str error: str
每个节点都读取并更新这个 state。比如draft节点会写入state["draft"]，review节点会写入state["review"]和state["next_action"]。
这比散落在各个函数里的局部变量更清晰，因为整个章节生成过程的状态都被集中建模了。
3.2 rewrite loop：每章最多迭代修改
项目里当前设置了：
MAX_REWRITES = 2
review 节点会根据模型返回的 verdict 决定下一步：
if verdict in {"rewrite", "polish"} and rewrite_count < MAX_REWRITES: state["next_action"] = "rewrite" else: state["next_action"] = "commit"rewrite 完成后，并不是直接提交，而是回到：
rewrite -> extract -> check -> review
这样做的原因是：重写后的正文可能引入新的设定变化或连续性问题，所以需要重新抽取记忆、重新检查、重新 review。
3.3 checkpoint：每批章节后等待人工确认
项目默认每批最多生成 5 章，也可以在前端调整max_chapters。当本批次章节数达到上限时，checkpoint节点会把 run 状态改成：
awaiting_confirmation
这时前端可以显示“等待确认”，用户确认后再继续跑下一批章节。
这已经具备 Human-in-the-loop 的雏形，但它还不是 LangGraph 官方interrupt()机制，而是项目自己在 Runtime 层实现的暂停、确认和恢复。
4. 上下文工程：不是塞全文，而是结构化记忆加 token budget
长篇小说最难的地方不是生成一章，而是保持几十章、几百章之间的连续性。
如果每一章都把前文全文塞给模型，会有几个问题：
token 成本随章节数线性甚至指数增长；
模型容易被大量无关文本干扰；
很多信息其实不需要全文，只需要摘要、状态变化和约束；
草稿被重写前，不应该污染长期记忆。
所以flashNovel使用了结构化记忆。
当前记忆大致分为四类：
| 层级 | 内容 |
|---|---|
| Artifact | 草稿、最终章节、prompt 快照、原始输出 |
| Canon | 角色、世界规则、地点等稳定设定 |
| Episodic | 章节摘要、时间线事件 |
| Continuity | 人物关系、伏笔、状态变化、review issue |
生成下一章时，系统会调用get_writer_context()，组装出一个上下文包：
context = { "story": ..., "workspace": ..., "chapter": chapter, "chapter_plan": ..., "characters": ..., "world_rules": ..., "locations": ..., "recent_summaries": ..., "timeline": ..., "relationships": ..., "foreshadows": ..., "state_changes": ..., "review_reports": ..., "review_issues": ..., }这里的关键点是：系统读取的是“对下一章有用的信息”，而不是机械塞入全部正文。
同时，项目还实现了简单的 token budget 裁剪：
def apply_context_budget(context: dict[str, Any], token_budget: int) -> dict[str, Any]: if token_budget <= 0: return context list_keys = [ "recent_summaries", "review_issues", "foreshadows", "relationships", "state_changes", "timeline", "review_reports", ]
在预算有限时，系统会优先保留 story、characters、world_rules 等核心设定，然后裁剪列表型上下文。
4.1 什么是候选记忆
项目里有一个概念叫candidate_memory，可以理解为“候选记忆”。
它来自extract节点：模型读完当前草稿后，抽取出可能影响后续章节的信息，例如：
本章摘要；
关键事件；
时间线变化；
人物关系变化；
伏笔埋设或回收；
角色状态变化。
但这些信息不会立刻进入长期记忆。原因很简单：当前草稿可能会被 review 打回重写。
如果一个被废弃的草稿已经污染了长期记忆，后续章节就可能继承错误设定。
所以项目采用了这个策略：
draft -> extract candidate_memory -> check/review/rewrite -> commit -> persist_candidate_memory
也就是：先暂存为候选记忆，只有最终章节被 commit 后，才把候选记忆写入 SQLite 的章节摘要、时间线、人物关系、伏笔、状态变化和 review issue 表。
这是我觉得小说类 Agent 很需要的一层保护。
5. 和真正 multi-agent 的差距
虽然项目里有很多“角色”：
章节规划助手；
小说写作助手；
记忆抽取器；
连续性审校器；
小说主编；
改稿助手。
但当前它们只是不同的 prompt/template 和 tool，并不是独立 agent。
当前项目更准确的定义是：
单 Runtime + 多节点 + 多角色 Prompt 工具链
真正的 multi-agent 通常至少会有以下特征：
每个 agent 有自己的职责、状态和工具集合；
agent 之间有消息传递或 handoff；
有 supervisor/router 决定任务分配；
可能存在并行 agent；
不同 agent 可以使用不同模型、不同 memory、不同评估策略；
每个 agent 的结果可以被其他 agent 审核、引用或合并。
对比一下，flashNovel当前的工具链是由同一个RuntimeHost调度，同一个 LLM client 发起调用，同一个 store 持久化数据。
这不是坏事。对 V1 来说，先把 workflow、memory、checkpoint、SSE、前端工作台跑通，比一开始就拆成多个 agent 更重要。
什么时候需要拆成不同 agent？
我觉得有几个信号：
角色需要独立工具例如 ContinuityAgent 需要专门查记忆库，WriterAgent 只负责写正文。
角色需要不同模型例如写作使用强模型，审稿使用便宜快速模型，记忆抽取使用结构化输出能力更稳的模型。
角色需要独立记忆例如 EditorAgent 保存全局创作策略，MemoryAgent 保存长期事实。
角色之间需要协商或仲裁例如 WriterAgent 和 ReviewerAgent 对一章是否通过产生冲突，需要 EditorAgent 裁决。
部分任务可以并行例如连续性检查、节奏检查、人物动机检查可以并行跑，然后汇总成 review report。
到了这些阶段，再拆 multi-agent 就比较自然。
6. 当前不足
6.1 同步串行，吞吐有限
当前项目对用户接口来说是异步的：创建 run 后，后端 worker 在后台执行。
但 Agent Runtime 内部还是同步串行：
graph.invoke(state)是同步执行；工具层使用
complete_sync/stream_sync；每章的 plan、draft、extract、check、review 串行执行；
rewrite 后又会重新串行执行 extract、check、review。
如果一章没有重写，大约需要 5 次 LLM 调用。
如果最多重写 2 次，最坏可能达到：
plan + draft + extract + check + review + 2 * (rewrite + extract + check + review) = 13 次 LLM 调用
所以主要瓶颈是 LLM 延迟，而不是 SQLite 或文件写入。
6.2 没有接 LangGraph 原生 checkpointer
项目现在有自己的 run/checkpoint/event 持久化，但还没有接 LangGraph 官方 checkpointer。
也就是说，当前状态恢复是项目 Runtime 层自己管理的，而不是通过 LangGraph persistence 统一管理 graph state。
6.3 Human-in-the-loop 还不是 interrupt
项目目前的人工确认点是：
每批章节结束 -> awaiting_confirmation -> 用户确认 -> resume
这已经能满足“批次确认”，但还不是 LangGraph 官方interrupt()。
更理想的方式是，在关键节点中断，例如：
plan 后让用户确认本章方向；
review 后让用户选择 accept/rewrite/manual edit；
commit 前让用户确认是否写入长期记忆。
6.4 没有接 LangSmith
项目目前通过 SSE 和数据库 events 观察运行过程，但没有 LangSmith trace。
如果接入 LangSmith，可以更方便地看到：
每个节点耗时；
每次 LLM 输入输出；
条件边为什么走到 rewrite；
每章 rewrite 了几次；
哪些 prompt 成本最高；
失败发生在哪个节点。
6.5 没有生产级任务队列
当前 worker 是进程内 daemon thread，适合本地 MVP。
如果要支持多用户或高并发，需要引入生产级队列和 worker，例如：
Redis Queue / Celery / Dramatiq；
Kafka / RabbitMQ；
独立 Python Agent Runtime worker；
Java 平台层负责用户、权限、租户、审计、任务调度和限流。
7. 下一步优化方向
7.1 async runtime
第一步可以把 Runtime 从同步调用改成异步：
graph.invoke改为graph.ainvoke；LLM client 使用 async HTTP client；
工具层提供 async execute；
可并行的检查任务用
asyncio.gather；worker 支持并发执行多个 run。
这能改善吞吐，但要注意小说章节之间强依赖，不能盲目并行章节。
比较适合并行的是章节内部的检查项，例如：
draft -> continuity_check -> style_check -> character_motivation_check -> pacing_check -> aggregate_review
7.2 接入 LangGraph checkpointer
把项目自建 checkpoint 和 LangGraph persistence 对齐：
每个 run 对应 LangGraph thread；
graph state 通过 checkpointer 保存；
恢复时从 checkpoint 继续；
项目自己的 run/event 表保留为业务视图。
这样可以减少“业务 checkpoint”和“graph state checkpoint”之间的不一致。
7.3 使用 interrupt 做更细的 Human-in-the-loop
下一步可以把人工确认点放得更细：
章节计划生成后暂停，让用户确认方向；
review 结果为 rewrite 时暂停，让用户选择自动重写还是手动修改；
commit 前展示候选记忆，让用户确认哪些写入长期记忆。
这会比“每 5 章确认一次”更适合真实创作。
7.4 接入 LangSmith trace
接入 LangSmith 后，可以把项目从“能跑”推进到“可调试、可评估”。
尤其是 rewrite loop，一旦出现质量不稳定，就需要观察：
review prompt 是否太严格；
check prompt 是否误判；
rewrite 是否真的修复了问题；
token budget 裁剪是否丢掉了关键设定；
某个模型是否不适合做结构化抽取。
7.5 拆分真正的 Agent
当前 V1 不急着拆，但后续可以考虑：
| Agent | 职责 |
|---|---|
WriterAgent | 写章节正文 |
ContinuityAgent | 检查设定、时间线、人物状态 |
MemoryAgent | 抽取、筛选、合并长期记忆 |
EditorAgent | 判断章节质量和重写方向 |
PublisherAgent | 导出章节、生成阅读视图、整理发布格式 |
这时 LangGraph 的 multi-agent 模式会更有价值。
总结
我对这个项目目前的定位是：
不是完整 multi-agent 系统， 而是一个以 LangGraph 为核心的章节级 Agent Runtime MVP。
它已经具备几个 Agent 应用开发里很重要的工程要素：
状态机式 LLM workflow；
结构化上下文工程；
长期记忆与候选记忆；
review/rewrite 质量循环；
人工确认 checkpoint；
事件流和运行可观察性；
前后端工作台闭环。
但距离生产级 Agent 平台还有明显差距：
运行时仍是同步串行；
未接 LangGraph 原生 checkpointer；
Human-in-the-loop 还不是 interrupt；
未接 LangSmith trace；
没有生产级任务队列；
还不是真正的 multi-agent 协作。
这也是下一阶段最值得继续做的地方：先把 Runtime 做稳，再把角色拆成 agent；先把记忆和评估跑通，再谈更复杂的协作模式。