From 7f3eaaba3d79c31babb959fcbd92737da060c798 Mon Sep 17 00:00:00 2001 From: Yaojia Wang Date: Tue, 23 Jun 2026 16:54:47 +0200 Subject: [PATCH] =?UTF-8?q?feat(orchestrator):=20C1=20=E5=A4=9A=E7=AB=A0?= =?UTF-8?q?=E5=B7=A5=E4=BD=9C=E6=B5=81=E9=93=BE=E5=9B=BE=20=E2=80=94=20wri?= =?UTF-8?q?te=E2=86=92review=E2=86=92decide=E2=86=92accept=20=E6=9C=89?= =?UTF-8?q?=E7=8E=AF=E5=9B=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 实现 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。 --- packages/core/tests/test_chain_graph.py | 375 ++++++++++++++++++ packages/core/tests/test_orchestrator.py | 34 +- .../core/ww_core/orchestrator/__init__.py | 4 +- .../ww_core/orchestrator/chain/__init__.py | 54 +++ .../core/ww_core/orchestrator/chain/graph.py | 170 ++++++++ .../core/ww_core/orchestrator/chain/nodes.py | 311 +++++++++++++++ .../core/ww_core/orchestrator/chain/state.py | 36 ++ packages/core/ww_core/orchestrator/graph.py | 24 +- 8 files changed, 951 insertions(+), 57 deletions(-) create mode 100644 packages/core/tests/test_chain_graph.py create mode 100644 packages/core/ww_core/orchestrator/chain/__init__.py create mode 100644 packages/core/ww_core/orchestrator/chain/graph.py create mode 100644 packages/core/ww_core/orchestrator/chain/nodes.py create mode 100644 packages/core/ww_core/orchestrator/chain/state.py diff --git a/packages/core/tests/test_chain_graph.py b/packages/core/tests/test_chain_graph.py new file mode 100644 index 0000000..e52d035 --- /dev/null +++ b/packages/core/tests/test_chain_graph.py @@ -0,0 +1,375 @@ +"""多章链图单测（chain-workflow 设计 §3 / §10）——节点纯逻辑 + 全图 MemorySaver 流程。 + +全部注入 mock 网关 + fake session_factory + fake repos + MemorySaver，无真 LLM、无真 +Postgres。验：① 无冲突全跑到 END 且 written 满；② 有冲突 → `__interrupt__`；③ resume +（`Command(resume=decisions)`）→ accept → END；④ `decide` 纯逻辑；⑤ accept 推进章号。 +asyncio_mode=auto。 +""" + +from __future__ import annotations + +import uuid +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from typing import Any + +from langchain_core.runnables import RunnableConfig +from langgraph.checkpoint.memory import MemorySaver +from langgraph.types import Command +from pydantic import BaseModel +from ww_agents import Conflict, ContinuityReview +from ww_core.orchestrator.chain import ( + accept_chapter, + build_chain_graph, + decide, + has_conflicts, + initial_chain_state, + write_chapter, +) +from ww_llm_gateway.types import LlmRequest, LlmResponse, ServedBy, Usage + +PROJECT = uuid.UUID("00000000-0000-0000-0000-000000000001") +USER = uuid.UUID("00000000-0000-0000-0000-0000000000aa") + + +# ---- 测试替身 ---- + + +def _usage() -> Usage: + return Usage( + provider="fake", + model="fake", + input_tokens=1, + output_tokens=1, + cost_minor=0, + currency="USD", + ) + + +class FakeChainGateway: + """链网关 mock：write（无 output_schema）回固定正文；审稿按 `output_schema` 回 parsed。 + + `conflicts`：注给 continuity 审的冲突清单（空 = 无冲突章）。其余审 schema 回各自空实例。 + """ + + def __init__( + self, + *, + draft_text: str, + conflicts: list[Conflict], + by_schema: dict[type, BaseModel], + ) -> None: + self._draft = draft_text + self._continuity = ContinuityReview(conflicts=conflicts) + self._by_schema = by_schema + self.write_calls = 0 + self.review_calls = 0 + + async def run(self, req: LlmRequest) -> LlmResponse: + schema: type[BaseModel] | None = req.output_schema + if schema is None: + self.write_calls += 1 + return LlmResponse( + text=self._draft, + parsed=None, + usage=_usage(), + served_by=ServedBy(provider="fake", model="fake"), + ) + self.review_calls += 1 + parsed: BaseModel = ( + self._continuity if schema is ContinuityReview else self._by_schema[schema] + ) + return LlmResponse( + text=parsed.model_dump_json(), + parsed=parsed, + usage=_usage(), + served_by=ServedBy(provider="fake", model="fake"), + ) + + +class FakeDraftView: + def __init__(self, content: str) -> None: + self.content = content + + +class FakeChapterRepo: + """内存章草稿 repo：save_draft 存正文，get_draft 回读。""" + + def __init__(self, store: dict[int, str]) -> None: + self._store = store + + async def save_draft( + self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1 + ) -> Any: + self._store[chapter_no] = text + return FakeDraftView(text) + + async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> Any: + content = self._store.get(chapter_no) + return FakeDraftView(content) if content is not None else None + + +class FakeReviewRecord: + def __init__(self, chapter_no: int, conflicts: list[dict[str, Any]]) -> None: + self.id = uuid.uuid4() + self.chapter_no = chapter_no + self.conflicts = conflicts + + +class FakeReviewRepo: + """内存审稿 repo：record 追加留痕，list_for_chapter 回新→旧。""" + + def __init__(self, records: dict[int, list[FakeReviewRecord]]) -> None: + self._records = records + + 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: + rec = FakeReviewRecord(chapter_no, list(conflicts)) + self._records.setdefault(chapter_no, []).insert(0, rec) + return rec + + async def list_for_chapter(self, project_id: uuid.UUID, chapter_no: int) -> list[Any]: + return list(self._records.get(chapter_no, [])) + + +class FakeSession: + def __init__(self) -> None: + self.commits = 0 + + async def commit(self) -> None: + self.commits += 1 + + +class FakeContext: + def __init__(self) -> None: + self.stable_core = "## 世界观硬规则\n灵气可凝丹" + self.volatile = "## 写作指令\n写第 N 章" + + +@asynccontextmanager +async def _session_factory() -> AsyncIterator[FakeSession]: + """fake 短事务 session（每次新建；满足 SessionFactory）。""" + yield FakeSession() + + +def _empty_by_schema() -> dict[type, BaseModel]: + """非 continuity 审（pace/style/foreshadow）的空 parsed——按各 spec 的 output_schema。""" + from ww_agents import foreshadow_spec, pace_spec, style_drift_spec + + out: dict[type, BaseModel] = {} + for spec in (foreshadow_spec, pace_spec, style_drift_spec): + schema = spec.output_schema + assert schema is not None + out[schema] = schema() + return out + + +def _make_harness(*, conflicts: list[Conflict]) -> dict[str, Any]: + """造一套共享内存态 + 工厂闭包 + accept 间谍，供图/节点测试复用。""" + draft_store: dict[int, str] = {} + review_store: dict[int, list[FakeReviewRecord]] = {} + accepted: list[dict[str, Any]] = [] + + gateway = FakeChainGateway( + draft_text="第 N 章正文。", conflicts=conflicts, by_schema=_empty_by_schema() + ) + + def memory_repos_factory(session: Any) -> Any: + return object() # assemble 被 fake 替换，不实际用 repos + + def chapter_repo_factory(session: Any) -> Any: + return FakeChapterRepo(draft_store) + + def review_repo_factory(session: Any) -> Any: + return FakeReviewRepo(review_store) + + async def assemble(repos: Any, project_id: uuid.UUID, chapter_no: int) -> Any: + return FakeContext() + + async def accept_op( + *, + project_id: uuid.UUID, + chapter_no: int, + user_id: uuid.UUID, + decisions: list[Any], + ) -> None: + accepted.append({"chapter_no": chapter_no, "decisions": list(decisions)}) + + return { + "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, + "accept_op": accept_op, + "accepted": accepted, + "draft_store": draft_store, + "review_store": review_store, + } + + +def _build(h: dict[str, Any], checkpointer: Any) -> Any: + return build_chain_graph( + h["gateway"], + session_factory=h["session_factory"], + memory_repos_factory=h["memory_repos_factory"], + chapter_repo_factory=h["chapter_repo_factory"], + review_repo_factory=h["review_repo_factory"], + assemble=h["assemble"], + accept_op=h["accept_op"], + checkpointer=checkpointer, + ) + + +# ---- decide 纯逻辑 ---- + + +async def test_decide_no_conflicts_sets_false() -> None: + review_store: dict[int, list[FakeReviewRecord]] = {1: [FakeReviewRecord(1, [])]} + + def review_repo_factory(session: Any) -> Any: + return FakeReviewRepo(review_store) + + out = await decide( + {"project_id": PROJECT, "user_id": USER, "current_chapter_no": 1}, + session_factory=_session_factory, + review_repo_factory=review_repo_factory, + ) + + assert out == {"has_conflicts": False} + + +async def test_decide_with_conflicts_sets_true() -> None: + review_store: dict[int, list[FakeReviewRecord]] = { + 1: [FakeReviewRecord(1, [{"kind": "x", "detail": "矛盾"}])] + } + + def review_repo_factory(session: Any) -> Any: + return FakeReviewRepo(review_store) + + out = await decide( + {"project_id": PROJECT, "user_id": USER, "current_chapter_no": 1}, + session_factory=_session_factory, + review_repo_factory=review_repo_factory, + ) + + assert out == {"has_conflicts": True} + + +def test_has_conflicts_reads_flag() -> None: + assert has_conflicts({"has_conflicts": True}) is True + assert has_conflicts({"has_conflicts": False}) is False + assert has_conflicts({}) is False + + +# ---- accept 节点：推进章号 + 追加 written + 透传裁决 ---- + + +async def test_accept_chapter_advances_and_records_decisions() -> None: + accepted: list[dict[str, Any]] = [] + + async def accept_op( + *, + project_id: uuid.UUID, + chapter_no: int, + user_id: uuid.UUID, + decisions: list[Any], + ) -> None: + accepted.append({"chapter_no": chapter_no, "decisions": list(decisions)}) + + out = await accept_chapter( + { + "project_id": PROJECT, + "user_id": USER, + "current_chapter_no": 3, + "written": [1, 2], + "last_chapter_no": 5, + }, + decisions=[{"conflict_index": 0, "verdict": "ignore"}], + accept_op=accept_op, + ) + + assert out["current_chapter_no"] == 4 + assert out["written"] == [1, 2, 3] + assert out["has_conflicts"] is False + assert accepted == [ + {"chapter_no": 3, "decisions": [{"conflict_index": 0, "verdict": "ignore"}]} + ] + + +# ---- write 节点：收集版（非流式）落稿 ---- + + +async def test_write_chapter_saves_collected_draft() -> None: + h = _make_harness(conflicts=[]) + state = initial_chain_state(project_id=PROJECT, user_id=USER, start_chapter_no=1, count=1) + + out = await write_chapter( + state, + gateway=h["gateway"], + session_factory=h["session_factory"], + memory_repos_factory=h["memory_repos_factory"], + chapter_repo_factory=h["chapter_repo_factory"], + assemble=h["assemble"], + ) + + assert out == {} # state 不变（正文在 DB） + assert h["draft_store"][1] == "第 N 章正文。" + # 收集版：write 走非流式 run + assert h["gateway"].write_calls == 1 + + +# ---- 全图（MemorySaver）：无冲突全自动跑满 ---- + + +async def test_chain_runs_all_chapters_when_no_conflicts() -> None: + h = _make_harness(conflicts=[]) + graph = _build(h, MemorySaver()) + config: RunnableConfig = {"configurable": {"thread_id": "chain-noconflict"}} + initial = initial_chain_state(project_id=PROJECT, user_id=USER, start_chapter_no=1, count=2) + + final = await graph.ainvoke(initial, config=config) + + assert "__interrupt__" not in final + assert final["written"] == [1, 2] + assert final["current_chapter_no"] == 3 # last(2) + 1，越界 → END + assert [a["chapter_no"] for a in h["accepted"]] == [1, 2] + assert h["accepted"][0]["decisions"] == [] # 无冲突章 decisions=[] + + +# ---- 全图：有冲突 → interrupt → resume → accept → END ---- + + +async def test_chain_interrupts_on_conflict_then_resumes() -> None: + conflict = Conflict(type="性格漂移", where="第3段", suggestion="统一主角姓名为张三") + h = _make_harness(conflicts=[conflict]) + saver = MemorySaver() # 同进程同 saver 跨两次 invoke → interrupt 可恢复 + graph = _build(h, saver) + config: RunnableConfig = {"configurable": {"thread_id": "chain-conflict"}} + initial = initial_chain_state(project_id=PROJECT, user_id=USER, start_chapter_no=1, count=1) + + # 初始 run：第 1 章四审报冲突 → interrupt 暂停。 + interrupted = await graph.ainvoke(initial, config=config) + + assert "__interrupt__" in interrupted + assert h["accepted"] == [] # 尚未验收 + + # resume：作者裁决该冲突 → 续跑 accept → END。 + decisions = [{"conflict_index": 0, "verdict": "ignore", "note": "笔误，后续统一"}] + final = await graph.ainvoke(Command(resume=decisions), config=config) + + assert "__interrupt__" not in final + assert final["written"] == [1] + assert [a["chapter_no"] for a in h["accepted"]] == [1] + assert h["accepted"][0]["decisions"] == decisions # 裁决透传给 accept_op diff --git a/packages/core/tests/test_orchestrator.py b/packages/core/tests/test_orchestrator.py index fe42035..3224e90 100644 --- a/packages/core/tests/test_orchestrator.py +++ b/packages/core/tests/test_orchestrator.py @@ -8,15 +8,11 @@ from __future__ import annotations import uuid from fakes_orchestrator import FakeStreamGateway, RaisingGateway -from langchain_core.runnables import RunnableConfig -from langgraph.checkpoint.memory import MemorySaver from ww_core.orchestrator import ( EVENT_DONE, EVENT_ERROR, EVENT_TOKEN, - WRITE_NODE, ChapterState, - build_write_graph, build_write_request, normalize_deltas, stream_chapter_draft, @@ -135,31 +131,5 @@ async def test_normalize_deltas_maps_unexpected_error_to_internal() -> None: assert events[0].data["code"] == ErrorCode.INTERNAL -# ---- 图编译：装节点 + checkpointer（MemorySaver，不连真 Postgres） ---- - - -def test_build_write_graph_compiles_with_checkpointer() -> None: - gateway = FakeStreamGateway(["x"]) - - graph = build_write_graph(gateway, checkpointer=MemorySaver()) - - assert WRITE_NODE in graph.get_graph().nodes - assert graph.checkpointer is not None - - -async def test_compiled_graph_runs_write_node_end_to_end() -> None: - gateway = FakeStreamGateway(["云", "深"]) - graph = build_write_graph(gateway, checkpointer=MemorySaver()) - config: RunnableConfig = {"configurable": {"thread_id": "t-1"}} - initial: ChapterState = { - "project_id": PROJECT, - "chapter_no": 1, - "user_id": USER, - "stable_core": STABLE, - "volatile": VOLATILE, - "draft": "", - } - - final = await graph.ainvoke(initial, config=config) - - assert final["draft"] == "云深" +# 单 `write` 节点图 `build_write_graph` 已删（链图取代，见 chain-workflow 设计 §3.3）； +# write 节点本体 `write_node` 仍由上面的 stream/run 用例直接覆盖（注入 mock 网关，无需图运行时）。 diff --git a/packages/core/ww_core/orchestrator/__init__.py b/packages/core/ww_core/orchestrator/__init__.py index 5bd7d6a..5436229 100644 --- a/packages/core/ww_core/orchestrator/__init__.py +++ b/packages/core/ww_core/orchestrator/__init__.py @@ -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", diff --git a/packages/core/ww_core/orchestrator/chain/__init__.py b/packages/core/ww_core/orchestrator/chain/__init__.py new file mode 100644 index 0000000..1b44738 --- /dev/null +++ b/packages/core/ww_core/orchestrator/chain/__init__.py @@ -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", +] diff --git a/packages/core/ww_core/orchestrator/chain/graph.py b/packages/core/ww_core/orchestrator/chain/graph.py new file mode 100644 index 0000000..40b0383 --- /dev/null +++ b/packages/core/ww_core/orchestrator/chain/graph.py @@ -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 后无论有无冲突都进 accept；interrupt（若有冲突）在 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=start，written=[]）。 + + 纯函数：服务层 / 测试构初值用，集中区间换算（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", +] diff --git a/packages/core/ww_core/orchestrator/chain/nodes.py b/packages/core/ww_core/orchestrator/chain/nodes.py new file mode 100644 index 0000000..2ac69ba --- /dev/null +++ b/packages/core/ww_core/orchestrator/chain/nodes.py @@ -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`）， + 守不变量 #9（stable_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", +] diff --git a/packages/core/ww_core/orchestrator/chain/state.py b/packages/core/ww_core/orchestrator/chain/state.py new file mode 100644 index 0000000..a89f4a3 --- /dev/null +++ b/packages/core/ww_core/orchestrator/chain/state.py @@ -0,0 +1,36 @@ +"""多章工作流链状态 `ChainState`（chain-workflow 设计 §3.1）。 + +不变量 #5：langgraph 检查点 = state 内容，**只放控制流**——章号/计数/已验收清单/ +冲突标志。草稿正文/上下文/冲突明文**不进** state（落 `chapters`/`chapter_reviews`， +resume 时从领域表重读为真相）。 + +`ChainState` 用 TypedDict（LangGraph 原生状态形态），字段全 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 diff --git a/packages/core/ww_core/orchestrator/graph.py b/packages/core/ww_core/orchestrator/graph.py index 3f1535c..b7447a0 100644 --- a/packages/core/ww_core/orchestrator/graph.py +++ b/packages/core/ww_core/orchestrator/graph.py @@ -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,