From 18aa87d751b96fbe410f65da6d67365f00df26ef Mon Sep 17 00:00:00 2001 From: Yaojia Wang Date: Tue, 23 Jun 2026 20:09:08 +0200 Subject: [PATCH] =?UTF-8?q?feat(chain):=20F2=20=E7=BB=AD=E5=86=99=E5=BC=8F?= =?UTF-8?q?=E9=93=BE=20continue=5Fvolume=E2=80=94=E2=80=94=E4=B8=8A?= =?UTF-8?q?=E4=B8=80=E7=AB=A0=20accepted=20=E6=AD=A3=E6=96=87=E4=BD=9C?= =?UTF-8?q?=E5=89=8D=E6=96=87=E5=BC=95=E5=AD=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - core write_chapter 支持续写模式：chain_key==continue_volume 时经注入的 chapter_repo.latest_accepted 读上一章已验收终稿，build_continuation_context 改写 volatile（仅断点后块，守不变量 #9；core 不 import apps/api）。 - ChapterDraftRepo 协议加 latest_accepted；draft_volume 行为不变（无回归）。 - apps/api SUPPORTED_CHAINS 加 continue_volume；run 端点透传 chain_key 选模式 （chain_runner 经 initial_chain_state 把 chain_key 落入 ChainState）。 - 单测（mock 网关+MemorySaver+fake session）：续写第二章请求含第一章正文； draft_volume 不注入前文；端点 202 透传 chain_key。守不变量 #1/#5。 --- apps/api/tests/test_chain.py | 22 ++++ apps/api/ww_api/routers/chain.py | 5 +- packages/core/tests/test_chain_graph.py | 101 +++++++++++++++++- .../core/ww_core/orchestrator/chain/nodes.py | 39 ++++++- 4 files changed, 157 insertions(+), 10 deletions(-) diff --git a/apps/api/tests/test_chain.py b/apps/api/tests/test_chain.py index 9cb2948..a2635c4 100644 --- a/apps/api/tests/test_chain.py +++ b/apps/api/tests/test_chain.py @@ -252,6 +252,28 @@ async def test_run_chain_returns_202_and_schedules_job( assert len(_noop_run_chain_job) == 1 +async def test_run_continue_volume_chain_returns_202_and_forwards_chain_key( + _noop_run_chain_job: list[dict[str, Any]], +) -> None: + """F2：continue_volume 链受支持 → 202，且 chain_key 透传给 run_chain_job（续写模式）。""" + project_repo = FakeProjectRepo() + job_repo = FakeJobRepo() + pid = await _seed_project(project_repo) + app = _app(project_repo, job_repo) + + async with _client(app) as client: + resp = await client.post( + f"/projects/{pid}/chains/continue_volume/run", + json={"start_chapter_no": 2, "count": 2}, + ) + + assert resp.status_code == 202 + body = resp.json() + assert body["chain_key"] == "continue_volume" + assert len(_noop_run_chain_job) == 1 + assert _noop_run_chain_job[0]["kwargs"]["chain_key"] == "continue_volume" + + async def test_run_chain_unknown_key_returns_404( _noop_run_chain_job: list[dict[str, Any]], ) -> None: diff --git a/apps/api/ww_api/routers/chain.py b/apps/api/ww_api/routers/chain.py index 73ceea3..cde2885 100644 --- a/apps/api/ww_api/routers/chain.py +++ b/apps/api/ww_api/routers/chain.py @@ -59,8 +59,9 @@ log = get_logger("ww.api.chain") router = APIRouter(prefix="/projects", tags=["chain"]) -# 本期唯一内置链（§1.3/§3）。未知 key → 404（系统边界，fail fast）。 -SUPPORTED_CHAINS = frozenset({"draft_volume"}) +# 内置链（§1.3/§3）：draft_volume=从 start 章按记忆量产；continue_volume=每章以上一章已验收 +# 正文末尾作前文引子续写（F2）。未知 key → 404（系统边界，fail fast）。 +SUPPORTED_CHAINS = frozenset({"draft_volume", "continue_volume"}) ProjectRepoDep = Annotated[ProjectRepo, Depends(get_project_repo)] JobRepoDep = Annotated[JobRepo, Depends(get_job_repo)] diff --git a/packages/core/tests/test_chain_graph.py b/packages/core/tests/test_chain_graph.py index f004e31..b4f6166 100644 --- a/packages/core/tests/test_chain_graph.py +++ b/packages/core/tests/test_chain_graph.py @@ -64,13 +64,17 @@ class FakeChainGateway: self._by_schema = by_schema self.write_calls = 0 self.review_calls = 0 + self.write_inputs: list[str] = [] async def run(self, req: LlmRequest) -> LlmResponse: schema: type[BaseModel] | None = req.output_schema if schema is None: self.write_calls += 1 + self.write_inputs.append(req.input if isinstance(req.input, str) else "") + # 每次写章产不同正文（便于断言下一章续写注入了上一章正文）。 + draft = f"{self._draft}#{self.write_calls}" return LlmResponse( - text=self._draft, + text=draft, parsed=None, usage=_usage(), served_by=ServedBy(provider="fake", model="fake"), @@ -93,10 +97,11 @@ class FakeDraftView: class FakeChapterRepo: - """内存章草稿 repo：save_draft 存正文，get_draft 回读。""" + """内存章草稿 repo：save_draft 存正文，get_draft 回读，latest_accepted 读已验收正文。""" - def __init__(self, store: dict[int, str]) -> None: + def __init__(self, store: dict[int, str], accepted: dict[int, str] | None = None) -> None: self._store = store + self._accepted = accepted if accepted is not None else {} async def save_draft( self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1 @@ -108,6 +113,10 @@ class FakeChapterRepo: content = self._store.get(chapter_no) return FakeDraftView(content) if content is not None else None + async def latest_accepted(self, project_id: uuid.UUID, chapter_no: int) -> Any: + content = self._accepted.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: @@ -177,6 +186,7 @@ def _empty_by_schema() -> dict[type, BaseModel]: def _make_harness(*, conflicts: list[Conflict]) -> dict[str, Any]: """造一套共享内存态 + 工厂闭包 + accept 间谍，供图/节点测试复用。""" draft_store: dict[int, str] = {} + accepted_store: dict[int, str] = {} review_store: dict[int, list[FakeReviewRecord]] = {} accepted: list[dict[str, Any]] = [] @@ -192,7 +202,7 @@ def _make_harness(*, conflicts: list[Conflict]) -> dict[str, Any]: return object() # assemble 被 fake 替换，不实际用 repos def chapter_repo_factory(session: Any) -> Any: - return FakeChapterRepo(draft_store) + return FakeChapterRepo(draft_store, accepted_store) def review_repo_factory(session: Any) -> Any: return FakeReviewRepo(review_store) @@ -208,6 +218,9 @@ def _make_harness(*, conflicts: list[Conflict]) -> dict[str, Any]: decisions: list[Any], ) -> None: accepted.append({"chapter_no": chapter_no, "decisions": list(decisions)}) + # 仿真验收落库：把本章草稿晋升为 accepted（供下一章续写读前文）。 + if chapter_no in draft_store: + accepted_store[chapter_no] = draft_store[chapter_no] return { "gateway": gateway, @@ -220,6 +233,7 @@ def _make_harness(*, conflicts: list[Conflict]) -> dict[str, Any]: "accept_op": accept_op, "accepted": accepted, "draft_store": draft_store, + "accepted_store": accepted_store, "review_store": review_store, } @@ -330,7 +344,7 @@ async def test_write_chapter_saves_collected_draft() -> None: ) assert out == {} # state 不变（正文在 DB） - assert h["draft_store"][1] == "第 N 章正文。" + assert h["draft_store"][1] == "第 N 章正文。#1" # 收集版：write 走非流式 run assert h["gateway"].write_calls == 1 @@ -378,3 +392,80 @@ async def test_chain_interrupts_on_conflict_then_resumes() -> None: assert final["written"] == [1] assert [a["chapter_no"] for a in h["accepted"]] == [1] assert h["accepted"][0]["decisions"] == decisions # 裁决透传给 accept_op + + +# ---- F2 续写式链 continue_volume ---- + + +async def test_write_chapter_draft_volume_omits_prior_text() -> None: + """draft_volume（默认）写章不注入前文——仍走 assemble 的 volatile（无回归）。""" + h = _make_harness(conflicts=[]) + h["accepted_store"][0] = "上一章正文末尾片段。" # 即使有前文也不应注入 + state = initial_chain_state( + project_id=PROJECT, user_id=USER, start_chapter_no=1, count=1 + ) # 默认 chain_key=draft_volume + + await write_chapter( + state, + gateway_builder=h["gateway_builder"], + session_factory=h["session_factory"], + memory_repos_factory=h["memory_repos_factory"], + chapter_repo_factory=h["chapter_repo_factory"], + assemble=h["assemble"], + ) + + write_input = h["gateway"].write_inputs[0] + assert write_input == "## 写作指令\n写第 N 章" # 原 assemble volatile，未改写 + assert "上一章正文末尾片段" not in write_input + + +async def test_write_chapter_continue_volume_injects_prior_accepted_text() -> None: + """continue_volume 写第 2 章时，注入第 1 章 accepted 正文（经 build_continuation_context）。""" + h = _make_harness(conflicts=[]) + h["accepted_store"][1] = "第一章已验收的正文末尾。" + state = initial_chain_state( + project_id=PROJECT, + user_id=USER, + start_chapter_no=2, + count=1, + chain_key="continue_volume", + ) + + await write_chapter( + state, + gateway_builder=h["gateway_builder"], + session_factory=h["session_factory"], + memory_repos_factory=h["memory_repos_factory"], + chapter_repo_factory=h["chapter_repo_factory"], + assemble=h["assemble"], + ) + + write_input = h["gateway"].write_inputs[0] + assert "第一章已验收的正文末尾。" in write_input + assert "前文正文" in write_input # build_continuation_context 的小节标题 + + +async def test_continue_volume_chain_second_chapter_sees_first_chapter_text() -> None: + """全图 E2E：continue_volume 两章——第 2 章请求上下文含第 1 章正文（不变量 #1/#5）。""" + h = _make_harness(conflicts=[]) + graph = _build(h, MemorySaver()) + config: RunnableConfig = {"configurable": {"thread_id": "chain-continue"}} + initial = initial_chain_state( + project_id=PROJECT, + user_id=USER, + start_chapter_no=1, + count=2, + chain_key="continue_volume", + ) + + final = await graph.ainvoke(initial, config=config) + + assert "__interrupt__" not in final + assert final["written"] == [1, 2] + # 第 1 章写出的正文（accept_op 晋升为 accepted）应出现在第 2 章的写章请求里。 + first_chapter_text = h["accepted_store"][1] + second_write_input = h["gateway"].write_inputs[1] + assert first_chapter_text in second_write_input + # 第 1 章请求无前文（前一章 0 不存在）→ 降级占位，不报错。 + first_write_input = h["gateway"].write_inputs[0] + assert "前文正文" in first_write_input diff --git a/packages/core/ww_core/orchestrator/chain/nodes.py b/packages/core/ww_core/orchestrator/chain/nodes.py index 2926c94..15bccf0 100644 --- a/packages/core/ww_core/orchestrator/chain/nodes.py +++ b/packages/core/ww_core/orchestrator/chain/nodes.py @@ -20,6 +20,7 @@ from ww_agents import AgentSpec from .._protocols import GatewayRun from ..collect import collect_reviews +from ..generation_node import build_continuation_context from ..review_node import build_review_context, run_review from ..state import ChapterState from ..write_node import build_write_request @@ -27,6 +28,9 @@ from .state import ChainState log = structlog.get_logger(__name__) +#: 续写式链 key：写章以上一章已验收正文末尾作前文引子（F2）。 +CHAIN_CONTINUE_VOLUME = "continue_volume" + # ---- 注入缝（图工厂经默认参绑定，单测直接注 fake）---- #: `session -> Repo`：节点自建短事务里从 session 造 repo（仿端点依赖工厂）。 @@ -62,7 +66,10 @@ class DraftView(Protocol): class ChapterDraftRepo(Protocol): - """章草稿 repo 最小依赖：write 落稿 `save_draft` + review 重读 `get_draft`（只 flush）。""" + """章草稿 repo 最小依赖：write 落稿 `save_draft` + review 重读 `get_draft`（只 flush）。 + + 续写式链（`continue_volume`）写章另需 `latest_accepted`——读上一章已验收终稿正文作前文引子。 + """ async def save_draft( self, project_id: uuid.UUID, chapter_no: int, *, text: str, volume: int = 1 @@ -70,6 +77,8 @@ class ChapterDraftRepo(Protocol): async def get_draft(self, project_id: uuid.UUID, chapter_no: int) -> DraftView | None: ... + async def latest_accepted(self, project_id: uuid.UUID, chapter_no: int) -> DraftView | None: ... + class ReviewRecordRepo(Protocol): """review/accept 节点对审稿 repo 的最小依赖：collect 留痕 + accept 重读最近审稿。""" @@ -126,20 +135,30 @@ async def write_chapter( project_id = state["project_id"] chapter_no = state["current_chapter_no"] user_id = state["user_id"] + is_continuation = state.get("chain_key") == CHAIN_CONTINUE_VOLUME async with session_factory() as session: gateway = await gateway_builder(session) repos = memory_repos_factory(session) context = await assemble(repos, project_id, chapter_no) + chapter_repo = chapter_repo_factory(session) + # 续写模式：volatile 改写为「设定 + 上一章已验收正文末尾」（不变量 #9：仅改断点后块， + # stable_core 仍进缓存前缀）。前文经注入的 chapter_repo 重读（core 不 import apps/api）。 + if is_continuation: + prior_text = await _read_prior_accepted(chapter_repo, project_id, chapter_no) + volatile = build_continuation_context( + project_context=context.volatile, prior_text=prior_text + ) + else: + volatile = context.volatile req = build_write_request( stable_core=context.stable_core, - volatile=context.volatile, + volatile=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( @@ -295,6 +314,19 @@ async def _read_draft( return view.content if view is not None else "" +async def _read_prior_accepted( + chapter_repo: ChapterDraftRepo, project_id: uuid.UUID, chapter_no: int +) -> str: + """续写式链：读上一章（chapter_no-1）已验收终稿正文作前文引子（不变量 #1/#4）。 + + 第一章（前一章号 < 1）或前一章未验收 → 空串，由 `build_continuation_context` 降级占位。 + """ + if chapter_no <= 1: + return "" + view = await chapter_repo.latest_accepted(project_id, chapter_no - 1) + return view.content if view is not None else "" + + async def _commit(session: CommitSession) -> None: """提交短事务（session 满足 `CommitSession`：只需 `commit()`）。""" await session.commit() @@ -307,6 +339,7 @@ def has_conflicts(state: ChainState) -> bool: __all__ = [ + "CHAIN_CONTINUE_VOLUME", "AcceptChapterOp", "AssembleContext", "AssembleFn",