327 lines
14 KiB
Python
327 lines
14 KiB
Python
"""多章链长任务 runner（chain-workflow 设计 §5）——`run_chain_job` 在 jobs 壳里跑链图。
|
||
|
||
复用 `run_job` 的生命周期心智（set_running → work → complete/fail），但链多一态：
|
||
**interrupt 命中 → `awaiting_input`**（等作者裁决续跑）。故不直接复用 `run_job`，而是
|
||
仿其壳自建独立 session 写 job 状态，work 本体 = 驱动 LangGraph 链图。
|
||
|
||
纪律（守不变量）：
|
||
- **#1/#5**：图节点各自短事务落领域表（`chapters`/`chapter_reviews`/`chapter_digests`）；
|
||
checkpointer 只存控制流；resume 重读领域表。本 runner 不持业务事务。
|
||
- **#3/#4**：`accept_op`（apps/api 注入）跑冲突 gate → 终稿提炼 digest → 单原子验收事务 →
|
||
伏笔到期扫描，唯一正文写经此。
|
||
- **token 不泄漏**：job result/status/日志只记章号/计数/标志，绝不含 prompt/正文/token。
|
||
- **错误脱敏**：复用 `_classify_job_error`（AppError 透传安全文案，其余通用文案，P0-3）。
|
||
|
||
可测性：`session_factory`/`gateway`/`checkpointer`/`accept_op` 全可注入——单测直接 `await`
|
||
它，注 fake session 工厂 + mock 网关 + MemorySaver + fake accept_op（绝不联网/真 DB/真 LLM）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from collections.abc import Awaitable, Callable, Sequence
|
||
from contextlib import AbstractAsyncContextManager
|
||
from typing import Any
|
||
|
||
import structlog
|
||
from langchain_core.runnables import RunnableConfig
|
||
from langgraph.types import Command
|
||
from ww_core.domain.chapter_repo import ChapterRepo, SqlChapterRepo
|
||
from ww_core.domain.digest_repo import SqlDigestAppendRepo
|
||
from ww_core.domain.job_repo import SqlJobRepo
|
||
from ww_core.domain.review_repo import SqlReviewRepo
|
||
from ww_core.memory import assemble
|
||
from ww_core.memory.sql_repositories import sql_memory_repos
|
||
from ww_core.orchestrator import REVIEW_SPECS
|
||
from ww_core.orchestrator.chain import build_chain_graph, initial_chain_state
|
||
from ww_llm_gateway import Gateway
|
||
|
||
from ww_api.schemas.projects import ConflictDecision
|
||
from ww_api.services.accept_service import (
|
||
assert_conflicts_resolved,
|
||
run_accept_transaction,
|
||
)
|
||
from ww_api.services.digest_extraction import extract_digest_facts
|
||
from ww_api.services.foreshadow_scan import SessionFactory, run_overdue_scan
|
||
from ww_api.services.job_runner import _classify_job_error
|
||
|
||
log = structlog.get_logger(__name__)
|
||
|
||
# 链 job 的 kind（jobs.kind 自由 Text 列，零迁移复用；§7）。
|
||
JOB_KIND_CHAIN = "chain"
|
||
|
||
# 每章在链图里的超级步数（write→review→decide→accept），用于算 recursion_limit 上界。
|
||
NODES_PER_CHAPTER = 4 # 每章 4 个超级步：write→review→decide→accept
|
||
# START 边 + interrupt/resume 重入的余量（防御性冗余，非精确计数）。
|
||
RECURSION_LIMIT_HEADROOM = 10 # START 边 + interrupt/resume 重入余量
|
||
|
||
|
||
# 「按 session 建 digest 网关」缝：accept 节点在自建短事务里需 light 档网关跑终稿提炼。
|
||
DigestGatewayBuilder = Callable[[Any], Awaitable[Gateway]]
|
||
|
||
# 「按 session 建链网关」缝：write/review 节点在自建短事务里现建网关（ledger 绑节点 session，
|
||
# 不变量 #1）。BackgroundTask 跑时请求 session 已关闭，故必须用节点 session 重建网关。
|
||
GatewayChainBuilder = Callable[[Any], Awaitable[Gateway]]
|
||
|
||
|
||
def build_accept_op(
|
||
*,
|
||
session_factory: SessionFactory,
|
||
digest_gateway_builder: DigestGatewayBuilder,
|
||
request_id: str | None = None,
|
||
) -> AcceptOp:
|
||
"""组装单章验收落库闭包（apps/api，守不变量 #3/#4）。
|
||
|
||
链 `accept_chapter` 节点经此闭包把作者裁决落库——每章**独立短事务**（自建 session）：
|
||
重读最近审稿 → 冲突 gate（缺判 → `CONFLICT_UNRESOLVED`，链失败）→ 终稿(=已存草稿)提炼
|
||
digest（事务外）→ `run_accept_transaction` 单原子提交 → 提交后伏笔到期扫描（自建 session）。
|
||
|
||
自动链无作者改稿：`final_text` = `chapters` 里该章草稿正文（write 节点所落）。
|
||
`decisions` 为 dict 清单（interrupt 恢复值 / 无冲突 []），此处校验为 `ConflictDecision`。
|
||
"""
|
||
|
||
async def accept_op(
|
||
*,
|
||
project_id: uuid.UUID,
|
||
chapter_no: int,
|
||
user_id: uuid.UUID,
|
||
decisions: list[Any],
|
||
) -> None:
|
||
parsed_decisions = [
|
||
d if isinstance(d, ConflictDecision) else ConflictDecision.model_validate(d)
|
||
for d in (decisions or [])
|
||
]
|
||
async with session_factory() as session:
|
||
chapter_repo: ChapterRepo = SqlChapterRepo(session)
|
||
review_repo = SqlReviewRepo(session)
|
||
digest_repo = SqlDigestAppendRepo(session)
|
||
|
||
history = await review_repo.list_for_chapter(project_id, chapter_no)
|
||
latest_review = history[0] if history else None
|
||
# 冲突 gate（R5，事务前拦截，不写库；缺判 → CONFLICT_UNRESOLVED 抛，链失败）。
|
||
assert_conflicts_resolved(latest_review, parsed_decisions)
|
||
|
||
draft = await chapter_repo.get_draft(project_id, chapter_no)
|
||
final_text = draft.content if draft is not None else ""
|
||
|
||
gateway = await digest_gateway_builder(session)
|
||
# R2：终稿 digest 提炼在开原子事务之前（别在持开事务里跨网络调 LLM）。
|
||
digest_facts = await extract_digest_facts(
|
||
gateway,
|
||
final_text=final_text,
|
||
user_id=user_id,
|
||
project_id=project_id,
|
||
chapter_no=chapter_no,
|
||
)
|
||
await run_accept_transaction(
|
||
session=session,
|
||
chapter_repo=chapter_repo,
|
||
digest_repo=digest_repo,
|
||
review_repo=review_repo,
|
||
project_id=project_id,
|
||
chapter_no=chapter_no,
|
||
final_text=final_text,
|
||
digest_facts=digest_facts,
|
||
latest_review=latest_review,
|
||
decisions=parsed_decisions,
|
||
)
|
||
# 验收提交后伏笔到期扫描（自建独立 session，§6.2；与单章端点同纪律）。
|
||
await run_overdue_scan(
|
||
session_factory,
|
||
project_id=project_id,
|
||
chapter_no=chapter_no,
|
||
request_id=request_id,
|
||
)
|
||
|
||
return accept_op
|
||
|
||
|
||
# `()` → async-CM，进入得一个 langgraph checkpointer（Postgres 运行时 / MemorySaver 测试）。
|
||
CheckpointerCtx = Callable[[], AbstractAsyncContextManager[Any]]
|
||
|
||
#: apps/api 注入的「按 session 建 chapter draft repo」工厂（review 节点重读草稿用）。
|
||
ChapterRepoFactory = Callable[[Any], Any]
|
||
#: apps/api 注入的「按 session 建审稿留痕 repo」工厂（review/decide 节点用）。
|
||
ReviewRepoFactory = Callable[[Any], Any]
|
||
#: 单章验收落库闭包（apps/api 组装：冲突 gate + digest + 验收事务 + 到期扫描）。
|
||
AcceptOp = Callable[..., Awaitable[None]]
|
||
|
||
|
||
def _chain_result(state: dict[str, Any], *, awaiting_chapter: int | None) -> dict[str, Any]:
|
||
"""从图 state 取非密进度摘要（绝不含正文/token；§5）。
|
||
|
||
`awaiting_chapter` 非 None = interrupt 命中该章待裁决；None 且 completed = 全跑完。
|
||
"""
|
||
written = list(state.get("written") or [])
|
||
return {
|
||
"chain_key": state.get("chain_key"),
|
||
"written": written,
|
||
"completed": awaiting_chapter is None,
|
||
"awaiting_chapter": awaiting_chapter,
|
||
}
|
||
|
||
|
||
def _extract_awaiting_chapter(interrupts: Sequence[Any]) -> int | None:
|
||
"""从 pending interrupts 判暂停章号：空 → None（跑完）；否则取首个 interrupt 载荷的章号。
|
||
|
||
interrupts 经**公开** `graph.aget_state(config).interrupts`（`StateSnapshot`）拿到，不再
|
||
反手 ainvoke 返回值里的私有 `__interrupt__` 键。载荷 = `{"chapter_no": n}`（见链图 `_accept`
|
||
节点）。容错读 `.value`/dict 两种形。
|
||
"""
|
||
if not interrupts:
|
||
return None
|
||
first = interrupts[0]
|
||
payload = getattr(first, "value", first)
|
||
if isinstance(payload, dict):
|
||
chapter_no = payload.get("chapter_no")
|
||
return int(chapter_no) if chapter_no is not None else None
|
||
return None
|
||
|
||
|
||
async def run_chain_job(
|
||
session_factory: SessionFactory,
|
||
job_id: uuid.UUID,
|
||
*,
|
||
project_id: uuid.UUID,
|
||
user_id: uuid.UUID,
|
||
chain_key: str,
|
||
start_chapter_no: int,
|
||
count: int,
|
||
chain_gateway_builder: GatewayChainBuilder,
|
||
checkpointer_ctx: CheckpointerCtx,
|
||
accept_op: AcceptOp,
|
||
chapter_repo_factory: ChapterRepoFactory,
|
||
review_repo_factory: ReviewRepoFactory,
|
||
resume_decisions: list[ConflictDecision] | None = None,
|
||
review_specs: Sequence[Any] = REVIEW_SPECS,
|
||
request_id: str | None = None,
|
||
) -> None:
|
||
"""跑/续一条多章链：set_running → 驱动链图 → 据返回置 done/awaiting_input/failed。
|
||
|
||
`resume_decisions is None` → 初始 run（`graph.ainvoke(initial_state, config)`）；
|
||
否则 resume（`graph.ainvoke(Command(resume=decisions), config)`），从 interrupt 续跑。
|
||
`thread_id = str(job_id)`：同 job 的初始/续跑落同一检查点 thread。
|
||
|
||
`chain_gateway_builder`（非具体网关实例）：write/review 节点在自建短事务的**新鲜** session
|
||
上现建网关，使 usage_ledger 绑节点 session（不变量 #1）。**绝不**把请求阶段建的网关传进来——
|
||
BackgroundTask 跑时请求 session 已关闭，其 ledger.record() 会落到死 session 上被静默丢弃。
|
||
|
||
checkpointer 上下文（Postgres 连接 / MemorySaver）横跨整次 invoke；本 runner 在 task
|
||
内打开/关闭它。异常一律被吞（后台任务边界），脱敏后置 job failed，不冒泡崩进程。
|
||
"""
|
||
try:
|
||
async with checkpointer_ctx() as checkpointer:
|
||
await _set_running(session_factory, job_id)
|
||
# 显式每链 recursion_limit 上界（防御硬化，CR-C2）：count 已被端点上限 MAX_CHAIN_COUNT
|
||
# 夹住，故 count*每章步数+余量 是安全且贴合的上界。**必须放 config 顶层**——放到
|
||
# `configurable` 下 langgraph 会静默忽略。单份 config 同时喂初始 run 与 resume 分支。
|
||
recursion_limit = count * NODES_PER_CHAPTER + RECURSION_LIMIT_HEADROOM
|
||
config: RunnableConfig = {
|
||
"configurable": {"thread_id": str(job_id)},
|
||
"recursion_limit": recursion_limit,
|
||
}
|
||
graph = build_chain_graph(
|
||
chain_gateway_builder,
|
||
session_factory=session_factory,
|
||
memory_repos_factory=sql_memory_repos,
|
||
chapter_repo_factory=chapter_repo_factory,
|
||
review_repo_factory=review_repo_factory,
|
||
assemble=assemble,
|
||
accept_op=accept_op,
|
||
review_specs=review_specs,
|
||
checkpointer=checkpointer,
|
||
)
|
||
if resume_decisions is None:
|
||
initial = initial_chain_state(
|
||
project_id=project_id,
|
||
user_id=user_id,
|
||
start_chapter_no=start_chapter_no,
|
||
count=count,
|
||
chain_key=chain_key,
|
||
)
|
||
raw_final = await graph.ainvoke(initial, config=config)
|
||
else:
|
||
resume_value = [d.model_dump() for d in resume_decisions]
|
||
raw_final = await graph.ainvoke(Command(resume=resume_value), config=config)
|
||
|
||
# 经公开 `StateSnapshot.interrupts` 判 pending interrupt（不反手私有 `__interrupt__`）；
|
||
# checkpointer 仍在 `checkpointer_ctx()` 上下文内，aget_state 可读同一 thread 检查点。
|
||
snapshot = await graph.aget_state(config)
|
||
awaiting_chapter = _extract_awaiting_chapter(snapshot.interrupts)
|
||
result = _chain_result(dict(raw_final), awaiting_chapter=awaiting_chapter)
|
||
await _finish(session_factory, job_id, result, awaiting=awaiting_chapter is not None)
|
||
log.info(
|
||
"chain_job_settled",
|
||
job_id=str(job_id),
|
||
request_id=request_id,
|
||
awaiting_chapter=awaiting_chapter,
|
||
written_count=len(result["written"]),
|
||
)
|
||
except Exception as exc: # noqa: BLE001 — 后台任务边界：记错误 + 置 job failed，不冒泡。
|
||
# 日志只记脱敏文案 + 异常类型；绝不记 str(exc)（可能含 API key/连接串/LLM 输出，设计 §5）。
|
||
stored_error = _classify_job_error(exc)
|
||
log.error(
|
||
"chain_job_failed",
|
||
job_id=str(job_id),
|
||
request_id=request_id,
|
||
exc_type=type(exc).__name__,
|
||
error=stored_error,
|
||
)
|
||
await _mark_failed(session_factory, job_id, stored_error, request_id)
|
||
|
||
|
||
async def _set_running(session_factory: SessionFactory, job_id: uuid.UUID) -> None:
|
||
"""独立短事务把 job 置 running（图驱动开始前）。"""
|
||
async with session_factory() as session:
|
||
await SqlJobRepo(session).set_running(job_id)
|
||
await session.commit()
|
||
|
||
|
||
async def _finish(
|
||
session_factory: SessionFactory,
|
||
job_id: uuid.UUID,
|
||
result: dict[str, Any],
|
||
*,
|
||
awaiting: bool,
|
||
) -> None:
|
||
"""据是否 interrupt 把 job 置 awaiting_input（待裁决）或 done（跑完），独立短事务。"""
|
||
async with session_factory() as session:
|
||
repo = SqlJobRepo(session)
|
||
if awaiting:
|
||
await repo.set_awaiting(job_id, result)
|
||
else:
|
||
await repo.complete(job_id, result)
|
||
await session.commit()
|
||
|
||
|
||
async def _mark_failed(
|
||
session_factory: SessionFactory,
|
||
job_id: uuid.UUID,
|
||
error: str,
|
||
request_id: str | None,
|
||
) -> None:
|
||
"""在**全新** session 里把 job 置 failed（前一事务已因异常作废）。"""
|
||
try:
|
||
async with session_factory() as session:
|
||
await SqlJobRepo(session).fail(job_id, error)
|
||
await session.commit()
|
||
except Exception as exc: # noqa: BLE001 — 置失败态本身再炸只记日志，不冒泡。
|
||
# 同 §5：不记 str(exc)，只记异常类型（避免泄露内部细节）。
|
||
log.error(
|
||
"chain_job_fail_mark_failed",
|
||
job_id=str(job_id),
|
||
request_id=request_id,
|
||
exc_type=type(exc).__name__,
|
||
)
|
||
|
||
|
||
__all__ = [
|
||
"JOB_KIND_CHAIN",
|
||
"AcceptOp",
|
||
"CheckpointerCtx",
|
||
"DigestGatewayBuilder",
|
||
"GatewayChainBuilder",
|
||
"build_accept_op",
|
||
"run_chain_job",
|
||
]
|