feat(orchestrator): C1 多章工作流链图 — write→review→decide→accept 有环图

实现 chain-workflow 设计 §3 的 LangGraph 链图,激活休眠资产
(checkpointer + interrupt-on-conflict)。

- packages/core/ww_core/orchestrator/chain/{state,nodes,graph}.py:
  - ChainState(§3.1):只放控制流(章号/计数/written/has_conflicts),
    正文/审稿/冲突明文落领域表,resume 重读(不变量 #1/#3/#5)。
  - 四节点 write_chapter/review_chapter/decide/accept_chapter(§3.2):
    各自短事务、独立 session;decide 纯逻辑读最近审稿留痕定 has_conflicts;
    accept 节点起始处遇冲突 interrupt() 暂停交人,恢复值=裁决清单。
  - build_chain_graph(§3.3 签名):节点 DB 依赖经 session_factory +
    repo 工厂 + assemble/accept_op 闭包默认参绑定(langgraph 不收 partial);
    条件边实现循环(current<=last 回 write)+ interrupt。
  - 验收事务(apps/api)经 accept_op 闭包注入,守目录所有权 + 不变量 #3/#4。
  - 收集版写章:复用 build_write_request 同一组装路径(stream=False → run)。
- 删除无人调用的 build_write_graph 工厂(write_node 本体保留,被链复用);
  清理其 __init__ 导出 + 旧单节点图单测(YAGNI,解决审计死代码)。
- 单测(mock gateway + fake session_factory + fake repos + MemorySaver):
  无冲突全跑到 END 且 written 满 / 有冲突→__interrupt__ / resume(Command)→
  accept→END / decide 纯逻辑 / accept 推进章号;零 token 零联网。

门禁:packages ruff/format/mypy 干净;pytest 314 passed。
This commit is contained in:
Yaojia Wang
2026-06-23 16:54:47 +02:00
parent 5ee9799885
commit 7f3eaaba3d
8 changed files with 951 additions and 57 deletions

View File

@@ -1,6 +1,6 @@
"""编排器C4 / ARCH §5.2)——写章图 + 审稿子图 + SSE 归一 + Postgres checkpointer。
- M1单 `write` 节点(`build_write_graph`)。
- write 节点本体 `write_node`(被多章链 `chain` 复用M1 单节点图 `build_write_graph` 已删)。
- M2并行审子图`build_review_graph`START→各审并行→collect 落 `chapter_reviews`+
审稿 SSE`section`/`conflict` 事件 + `normalize_review`)。
- accept验收事务/`interrupt_before`)不在本图——属 T2.4确定性代码读领域表R3
@@ -37,7 +37,6 @@ from .graph import (
REVIEW_SPECS,
WRITE_NODE,
build_review_graph,
build_write_graph,
setup_checkpointer,
)
from .outline_node import build_outline_request, run_outline
@@ -115,7 +114,6 @@ __all__ = [
"build_review_request",
"build_style_extract_request",
"build_worldbuilder_context",
"build_write_graph",
"build_write_request",
"collect_reviews",
"conflict_event",

View File

@@ -0,0 +1,54 @@
"""多章工作流链chain-workflow 设计)——`write → review → decide → accept` 有环图。
第一个真正需要 LangGraph 的场景(有环 + 可持久 + 可中断):批量量产 K 章,每章过
`assemble → write → 四审 → digest` 闭环,遇冲突 `interrupt()` 暂停交人裁决resume 续跑。
激活休眠资产checkpointer + interrupt。详见 docs/design/chain-workflow.md。
"""
from __future__ import annotations
from .graph import (
ACCEPT_CHAPTER,
DECIDE,
REVIEW_CHAPTER,
WRITE_CHAPTER,
build_chain_graph,
initial_chain_state,
)
from .nodes import (
AcceptChapterOp,
AssembleContext,
AssembleFn,
ChapterDraftRepo,
MemoryReposFactory,
ReviewRecordRepo,
SessionFactory,
accept_chapter,
decide,
has_conflicts,
review_chapter,
write_chapter,
)
from .state import ChainState
__all__ = [
"ACCEPT_CHAPTER",
"DECIDE",
"REVIEW_CHAPTER",
"WRITE_CHAPTER",
"AcceptChapterOp",
"AssembleContext",
"AssembleFn",
"ChainState",
"ChapterDraftRepo",
"MemoryReposFactory",
"ReviewRecordRepo",
"SessionFactory",
"accept_chapter",
"build_chain_graph",
"decide",
"has_conflicts",
"initial_chain_state",
"review_chapter",
"write_chapter",
]

View File

@@ -0,0 +1,170 @@
"""多章链图工厂 `build_chain_graph`chain-workflow 设计 §3.3)。
链 = LangGraph 有环图:`write → review → decide →(无冲突)→ accept →(还有章)→ write`
`decide →(有冲突)→ interrupt ⏸ → accept`,写满 K 章 → END。本图**激活休眠资产**
checkpointer持久控制流 + pending handle跨小时可恢复+ interrupt遇冲突暂停交人
依赖注入纪律与现有「gateway 闭包绑定」同模式):节点 DB 依赖经 `session_factory` +
`*_repo_factory` + `accept_op` 闭包绑入图节点。langgraph 不收 `functools.partial`,故用
**默认参绑定**(见 memory/gotchas.md
checkpointer 纪律CLAUDE.md「LangGraph」`setup()` 只在 migrations/CI 跑(`setup_checkpointer`
在 `orchestrator/graph.py`,迁移/CI 调一次build 函数绝不触碰 DDL单测注 `MemorySaver`。
"""
from __future__ import annotations
import uuid
from collections.abc import Callable, Sequence
from typing import TYPE_CHECKING, Any
from langgraph.graph import END, START, StateGraph
from langgraph.types import interrupt
from ww_agents import AgentSpec
from .._protocols import GatewayRun
from ..graph import REVIEW_SPECS
from . import nodes
from .nodes import (
AcceptChapterOp,
AssembleFn,
ChapterDraftRepo,
MemoryReposFactory,
ReviewRecordRepo,
SessionFactory,
)
from .state import ChainState
if TYPE_CHECKING:
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.graph.state import CompiledStateGraph
WRITE_CHAPTER = "write_chapter"
REVIEW_CHAPTER = "review_chapter"
DECIDE = "decide"
ACCEPT_CHAPTER = "accept_chapter"
def build_chain_graph(
gateway: GatewayRun,
*,
session_factory: SessionFactory,
memory_repos_factory: MemoryReposFactory,
chapter_repo_factory: Callable[[Any], ChapterDraftRepo],
review_repo_factory: Callable[[Any], ReviewRecordRepo],
assemble: AssembleFn,
accept_op: AcceptChapterOp,
review_specs: Sequence[AgentSpec] = REVIEW_SPECS,
checkpointer: BaseCheckpointSaver[Any] | None = None,
) -> CompiledStateGraph[ChainState, None, ChainState, ChainState]:
"""构建并编译 `draft_volume` 多章链图。
节点(`write_chapter`/`review_chapter`/`decide`/`accept_chapter`)经默认参绑定把
`gateway`/`session_factory`/各 repo 工厂/`assemble`/`accept_op` 闭包绑入。
条件边:`decide` 命中冲突 → `accept_chapter` 内 `interrupt()` 暂停交人;
`accept_chapter` 后判 `current_chapter_no <= last_chapter_no` 回 `write_chapter` 或 END。
`checkpointer`:运行时传 Postgres saver单测传 `MemorySaver`(同进程跨 invoke 可恢复)。
"""
async def _write(state: ChainState) -> dict[str, Any]:
return await nodes.write_chapter(
state,
gateway=gateway,
session_factory=session_factory,
memory_repos_factory=memory_repos_factory,
chapter_repo_factory=chapter_repo_factory,
assemble=assemble,
)
async def _review(state: ChainState) -> dict[str, Any]:
return await nodes.review_chapter(
state,
gateway=gateway,
session_factory=session_factory,
memory_repos_factory=memory_repos_factory,
chapter_repo_factory=chapter_repo_factory,
review_repo_factory=review_repo_factory,
assemble=assemble,
review_specs=review_specs,
)
async def _decide(state: ChainState) -> dict[str, Any]:
return await nodes.decide(
state,
session_factory=session_factory,
review_repo_factory=review_repo_factory,
)
async def _accept(state: ChainState) -> dict[str, Any]:
# interrupt 放节点**起始**CLAUDE.md「LangGraph」有冲突 → 暂停交人,恢复值 =
# 该章裁决清单;无冲突 → decisions=[],直接验收。恢复后 accept_op 落库(不变量 #3
decisions: list[Any] = []
if nodes.has_conflicts(state):
resumed = interrupt({"chapter_no": state["current_chapter_no"]})
decisions = list(resumed or [])
return await nodes.accept_chapter(state, decisions=decisions, accept_op=accept_op)
builder: StateGraph[ChainState, None, ChainState, ChainState] = StateGraph(ChainState)
builder.add_node(WRITE_CHAPTER, _write)
builder.add_node(REVIEW_CHAPTER, _review)
builder.add_node(DECIDE, _decide)
builder.add_node(ACCEPT_CHAPTER, _accept)
builder.add_edge(START, WRITE_CHAPTER)
builder.add_edge(WRITE_CHAPTER, REVIEW_CHAPTER)
builder.add_edge(REVIEW_CHAPTER, DECIDE)
# decide 后无论有无冲突都进 acceptinterrupt若有冲突在 accept 节点起始处暂停。
builder.add_edge(DECIDE, ACCEPT_CHAPTER)
# accept 后条件边:还有章 → 回 write 续跑;写满 K 章 → END。
builder.add_conditional_edges(
ACCEPT_CHAPTER,
_should_continue,
{WRITE_CHAPTER: WRITE_CHAPTER, END: END},
)
return builder.compile(checkpointer=checkpointer)
def _should_continue(state: ChainState) -> str:
"""accept 后路由:`current_chapter_no <= last_chapter_no` → 续写下一章;否则 END。
`accept_chapter` 已把 `current_chapter_no += 1`,故此处直接比较即判「是否还有章」。
纯函数,便于单测。
"""
if state["current_chapter_no"] <= state["last_chapter_no"]:
return WRITE_CHAPTER
return END
def initial_chain_state(
*,
project_id: uuid.UUID,
user_id: uuid.UUID,
start_chapter_no: int,
count: int,
chain_key: str = "draft_volume",
) -> ChainState:
"""构造初始 `ChainState`(区间 [start, start+count-1]current=startwritten=[])。
纯函数:服务层 / 测试构初值用集中区间换算last = start + count - 1
"""
return {
"project_id": project_id,
"user_id": user_id,
"chain_key": chain_key,
"start_chapter_no": start_chapter_no,
"last_chapter_no": start_chapter_no + count - 1,
"current_chapter_no": start_chapter_no,
"written": [],
"has_conflicts": False,
}
__all__ = [
"ACCEPT_CHAPTER",
"DECIDE",
"REVIEW_CHAPTER",
"WRITE_CHAPTER",
"build_chain_graph",
"initial_chain_state",
]

View File

@@ -0,0 +1,311 @@
"""多章链节点chain-workflow 设计 §3.2——write → review → decide → accept 循环。
节点是**小确定性单元**LLM 非确定性锁在网关后;`decide` 是纯逻辑。每个做 DB IO 的
节点用**短事务、独立 session**(每章 accept 原子提交,仿 `run_overdue_scan` 自建 session
checkpoint 只存控制流,正文/审稿落领域表——resume 时重读领域表为真相(不变量 #1/#3/#5
依赖注入:本模块的节点是裸异步函数(显式关键字依赖),供直接单测注入 fake图工厂
`graph.py`用默认参绑定把依赖闭包绑进图节点langgraph 不收 functools.partial见 gotcha
"""
from __future__ import annotations
import uuid
from collections.abc import Awaitable, Callable, Sequence
from contextlib import AbstractAsyncContextManager
from typing import Any, Protocol
import structlog
from ww_agents import AgentSpec
from .._protocols import GatewayRun
from ..collect import collect_reviews, extract_conflicts
from ..review_node import build_review_context, run_review
from ..state import ChapterState
from ..write_node import build_write_request
from .state import ChainState
log = structlog.get_logger(__name__)
# ---- 注入缝(图工厂经默认参绑定,单测直接注 fake----
#: `session -> Repo`:节点自建短事务里从 session 造 repo仿端点依赖工厂
SessionFactory = Callable[[], AbstractAsyncContextManager[Any]]
class AssembleContext(Protocol):
"""`assemble` 产出的最小读形——只需 stable_core/volatile构造写章/审稿请求)。"""
stable_core: str
volatile: str
#: `(repos, project_id, chapter_no) -> AssembledContext`:组装记忆(确定性,不变量 #6/#9
AssembleFn = Callable[[Any, uuid.UUID, int], Awaitable[AssembleContext]]
class MemoryReposFactory(Protocol):
"""`session -> MemoryRepos`assemble 所需 8 repo 捆绑)。"""
def __call__(self, session: Any) -> Any: ...
class DraftView(Protocol):
"""章草稿读形最小依赖——只需 `content`(重读正文构审稿上下文)。"""
content: str
class ChapterDraftRepo(Protocol):
"""章草稿 repo 最小依赖write 落稿 `save_draft` + review 重读 `get_draft`(只 flush"""
async def save_draft(
self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1
) -> Any: ...
async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> DraftView | None: ...
class ReviewRecordRepo(Protocol):
"""review/accept 节点对审稿 repo 的最小依赖collect 留痕 + accept 重读最近审稿。"""
async def record(
self,
project_id: uuid.UUID,
chapter_no: int,
*,
chapter_version: int | None = None,
conflicts: list[dict[str, Any]],
foreshadow_sug: list[dict[str, Any]] | None = None,
style: dict[str, Any] | None = None,
pace: dict[str, Any] | None = None,
health_score: int | None = None,
) -> Any: ...
async def list_for_chapter(self, project_id: uuid.UUID, chapter_no: int) -> list[Any]: ...
class CommitSession(Protocol):
"""节点短事务对 session 的最小依赖——只需 `commit()`write/review 节点末尾提交)。"""
async def commit(self) -> None: ...
#: accept 步:把作者裁决落库(晋升终稿 + digest + 裁决留痕 + 伏笔到期),各自独立 session。
#: 编排(链)在 core具体验收事务apps/api经此闭包注入守目录所有权 + 不变量 #3/#4。
#: 入参 `decisions` 为裁决清单(`accept_chapter` 节点经 interrupt 恢复值拿到,无冲突=[])。
AcceptChapterOp = Callable[..., Awaitable[None]]
# ---- 节点本体(裸函数,显式依赖;图工厂绑定闭包)----
async def write_chapter(
state: ChainState,
*,
gateway: GatewayRun,
session_factory: SessionFactory,
memory_repos_factory: MemoryReposFactory,
chapter_repo_factory: Callable[[Any], ChapterDraftRepo],
assemble: AssembleFn,
) -> dict[str, Any]:
"""write 节点:组装记忆 → 收集版写章(非 SSE→ 落 `chapters` draft → commit。
短事务、独立 session不跨章共享。state 不变(正文在 DB。返回空增量。
收集版写章复用 `build_write_request` 同一 prompt 组装路径stream=False → `gateway.run`
守不变量 #9stable_core 进缓存前缀);不确定性锁在网关后,注 mock 即可单测。
"""
project_id = state["project_id"]
chapter_no = state["current_chapter_no"]
user_id = state["user_id"]
async with session_factory() as session:
repos = memory_repos_factory(session)
context = await assemble(repos, project_id, chapter_no)
req = build_write_request(
stable_core=context.stable_core,
volatile=context.volatile,
user_id=user_id,
project_id=project_id,
)
# 收集版:非流式,一次 run 拿全文(链批量量产不走 SSE
collect_req = req.model_copy(update={"stream": False})
resp = await gateway.run(collect_req)
chapter_repo = chapter_repo_factory(session)
await chapter_repo.save_draft(project_id, chapter_no, text=resp.text)
await _commit(session)
log.info(
"chain_write_chapter",
project_id=str(project_id),
chapter_no=chapter_no,
draft_len=len(resp.text),
)
return {}
async def review_chapter(
state: ChainState,
*,
gateway: GatewayRun,
session_factory: SessionFactory,
memory_repos_factory: MemoryReposFactory,
chapter_repo_factory: Callable[[Any], ChapterDraftRepo],
review_repo_factory: Callable[[Any], ReviewRecordRepo],
assemble: AssembleFn,
review_specs: Sequence[AgentSpec],
) -> dict[str, Any]:
"""review 节点:复用四审 → 落 `chapter_reviews` 留痕(不变量 #3 只读、不写正文)→ commit。
复用现有 `run_review × specs` + `collect_reviews`(直接顺序跑各审,非起子图——链节点本身
已是图节点,无需嵌套图)。从 DB 重读本章草稿构审稿上下文(真相在表,不信 state
"""
project_id = state["project_id"]
chapter_no = state["current_chapter_no"]
user_id = state["user_id"]
async with session_factory() as session:
repos = memory_repos_factory(session)
context = await assemble(repos, project_id, chapter_no)
chapter_repo = chapter_repo_factory(session)
draft_view = await _read_draft(chapter_repo, project_id, chapter_no)
review_context = build_review_context(
draft=draft_view,
stable_core=context.stable_core,
volatile=context.volatile,
)
review_state: ChapterState = {
"project_id": project_id,
"chapter_no": chapter_no,
"user_id": user_id,
"review_context": review_context,
}
reviews: dict[str, Any] = {}
for spec in review_specs:
part = await run_review(spec, review_state, gateway=gateway)
reviews.update(part.get("reviews", {}))
review_state["reviews"] = reviews
review_repo = review_repo_factory(session)
await collect_reviews(review_state, review_repo=review_repo)
await _commit(session)
log.info(
"chain_review_chapter",
project_id=str(project_id),
chapter_no=chapter_no,
reviews=sorted(reviews.keys()),
)
return {}
async def decide(
state: ChainState,
*,
session_factory: SessionFactory,
review_repo_factory: Callable[[Any], ReviewRecordRepo],
) -> dict[str, Any]:
"""decide 节点(确定性纯逻辑,无 LLM读最近审稿留痕 → `has_conflicts`。
冲突明文**不入** state不变量 #5只写控制流标志accept 时再从领域表重读冲突。
"""
project_id = state["project_id"]
chapter_no = state["current_chapter_no"]
async with session_factory() as session:
review_repo = review_repo_factory(session)
history = await review_repo.list_for_chapter(project_id, chapter_no)
latest = history[0] if history else None
conflicts = _conflicts_of(latest)
has_conflicts = len(conflicts) > 0
log.info(
"chain_decide",
project_id=str(project_id),
chapter_no=chapter_no,
has_conflicts=has_conflicts,
conflict_count=len(conflicts),
)
return {"has_conflicts": has_conflicts}
async def accept_chapter(
state: ChainState,
*,
decisions: list[Any],
accept_op: AcceptChapterOp,
) -> dict[str, Any]:
"""accept 节点:用裁决落库(晋升终稿 + digest + 裁决留痕 + 伏笔到期),推进章号。
具体验收事务(重读 `chapter_reviews`、`assert_conflicts_resolved` gate、终稿提炼 digest、
`run_accept_transaction`、`run_overdue_scan`)属 apps/api经 `accept_op` 闭包注入——守目录
所有权 + 不变量 #3/#4缺判 → `CONFLICT_UNRESOLVED` 由 op 抛,链失败)。
`decisions`:无冲突章 = [];有冲突章 = interrupt 恢复值(作者裁决清单)。
返回 `current_chapter_no += 1` + `written` 追加本章——推进控制流(不可变增量)。
"""
project_id = state["project_id"]
chapter_no = state["current_chapter_no"]
user_id = state["user_id"]
await accept_op(
project_id=project_id,
chapter_no=chapter_no,
user_id=user_id,
decisions=decisions,
)
written = list(state.get("written") or [])
written.append(chapter_no)
log.info(
"chain_accept_chapter",
project_id=str(project_id),
chapter_no=chapter_no,
written_count=len(written),
)
return {
"current_chapter_no": chapter_no + 1,
"written": written,
"has_conflicts": False,
}
# ---- 辅助(纯函数 / 小工具)----
def _conflicts_of(latest_review: Any) -> list[dict[str, Any]]:
"""从最近审稿留痕取冲突清单(兼容 ReviewView Pydantic 与内存 fake dict"""
if latest_review is None:
return []
conflicts = getattr(latest_review, "conflicts", None)
if conflicts is None and isinstance(latest_review, dict):
conflicts = latest_review.get("conflicts")
return list(conflicts or [])
async def _read_draft(
chapter_repo: ChapterDraftRepo, project_id: uuid.UUID, chapter_no: int
) -> str:
"""从 repo 重读本章草稿正文(真相在表,不信 state"""
view = await chapter_repo.get_draft(project_id, chapter_no)
return view.content if view is not None else ""
async def _commit(session: CommitSession) -> None:
"""提交短事务session 满足 `CommitSession`:只需 `commit()`)。"""
await session.commit()
# decide 暴露给条件边的辅助(纯判定,便于单测)。
def has_conflicts(state: ChainState) -> bool:
"""条件边读 state 的冲突标志(缺省 False"""
return bool(state.get("has_conflicts"))
__all__ = [
"AcceptChapterOp",
"AssembleContext",
"AssembleFn",
"ChainState",
"ChapterDraftRepo",
"MemoryReposFactory",
"ReviewRecordRepo",
"SessionFactory",
"accept_chapter",
"decide",
"extract_conflicts",
"has_conflicts",
"review_chapter",
"write_chapter",
]

View File

@@ -0,0 +1,36 @@
"""多章工作流链状态 `ChainState`chain-workflow 设计 §3.1)。
不变量 #5langgraph 检查点 = state 内容,**只放控制流**——章号/计数/已验收清单/
冲突标志。草稿正文/上下文/冲突明文**不进** state落 `chapters`/`chapter_reviews`
resume 时从领域表重读为真相)。
`ChainState` 用 TypedDictLangGraph 原生状态形态),字段全 snake_case。
"""
from __future__ import annotations
import uuid
from typing import TypedDict
class ChainState(TypedDict, total=False):
"""`draft_volume` 链的控制流状态(检查点内容)。
- `project_id` / `user_id`:定位 + 调用作用域(原型单用户 stub
- `chain_key`:固定 `"draft_volume"`(本期唯一内置链)。
- `start_chapter_no` / `last_chapter_no`:本 run 区间 [start, last](含)。
- `current_chapter_no`:当前处理章;初值 = start每验收一章 +1。
- `written`:本 run 已验收章号进度来源resume 后累积)。
- `has_conflicts``decide` 节点写入的控制流标志(不存冲突明文,不变量 #5
`total=False`:初始 invoke 只需填区间字段,`written`/`has_conflicts` 由节点逐步填。
"""
project_id: uuid.UUID
user_id: uuid.UUID
chain_key: str
start_chapter_no: int
last_chapter_no: int
current_chapter_no: int
written: list[int]
has_conflicts: bool

View File

@@ -27,37 +27,17 @@ from ._protocols import GatewayRun
from .collect import ReviewRecorder, collect_reviews
from .review_node import run_review
from .state import ChapterState
from .write_node import GatewayStream, write_node
if TYPE_CHECKING:
from langgraph.checkpoint.base import BaseCheckpointSaver
from langgraph.graph.state import CompiledStateGraph
# write 节点名write_node 本体保留,被多章链 `chain.nodes.write_chapter` 复用M1 的
# `build_write_graph` 单节点图已删——链图取代之YAGNI见 chain-workflow 设计 §3.3)。
WRITE_NODE = "write"
COLLECT_NODE = "collect"
def build_write_graph(
gateway: GatewayStream,
*,
checkpointer: BaseCheckpointSaver[Any] | None = None,
) -> CompiledStateGraph[ChapterState, None, ChapterState, ChapterState]:
"""构建并编译单 `write` 节点图。
`gateway` 经闭包绑进节点(节点对图只暴露 `(state)->dict`,确定性可单测)。
`checkpointer`:运行时传 Postgres saver单测传 `MemorySaver` 或不传。
"""
async def _write(state: ChapterState) -> dict[str, str]:
return await write_node(state, gateway=gateway)
builder: StateGraph[ChapterState, None, ChapterState, ChapterState] = StateGraph(ChapterState)
builder.add_node(WRITE_NODE, _write)
builder.add_edge(START, WRITE_NODE)
builder.add_edge(WRITE_NODE, END)
return builder.compile(checkpointer=checkpointer)
# M4四审齐continuity + foreshadow + pace + style。第四审 = 文风漂移打分轨。
REVIEW_SPECS: tuple[AgentSpec, ...] = (
continuity_spec,