feat(writer): add WriterState (cross-chapter regeneration tracking)

This commit is contained in:
lhl
2026-08-13 09:46:38 +08:00
parent d237201407
commit 0a2dd17b0b
3 changed files with 62 additions and 1 deletions
+2 -1
View File
@@ -12,7 +12,7 @@ LLM 语义维度(内容准确性/幻觉/规则遵守)通过 llm_evaluators
from __future__ import annotations
import re
from dataclasses import dataclass
from dataclasses import dataclass, field
from genesis.data_models import StructuredSource
from genesis.parsers.resolver import validate_source_uris
@@ -41,6 +41,7 @@ class EvalReport:
dimensions: list[DimensionScore]
total_score: float
passed: bool
failed_chapters: list[str] = field(default_factory=list)
# 维度默认通过阈值
+28
View File
@@ -0,0 +1,28 @@
"""Writer 跨章状态(Phase 5)。追踪各章版本/内容/最近评估结果。"""
from __future__ import annotations
from genesis.writer.models import ChapterContent
from genesis.eval.scorer import EvalReport
class WriterState:
def __init__(self, chapter_order: list[str]) -> None:
self.versions: dict[str, int] = {cid: 0 for cid in chapter_order}
self.contents: dict[str, ChapterContent | None] = {cid: None for cid in chapter_order}
self.last_eval: dict[str, EvalReport | None] = {cid: None for cid in chapter_order}
def record_success(self, content: ChapterContent) -> None:
self.versions[content.chapter_id] += 1
self.contents[content.chapter_id] = content
def record_eval(self, cid: str, report: EvalReport) -> None:
self.last_eval[cid] = report
def needs_regeneration(self) -> list[str]:
out: list[str] = []
for cid, content in self.contents.items():
if content is None:
out.append(cid)
elif self.last_eval[cid] is not None and cid in self.last_eval[cid].failed_chapters:
out.append(cid)
return out
+32
View File
@@ -0,0 +1,32 @@
"""WriterState 测试(P5-T5)。"""
from genesis.writer.writer_state import WriterState
from genesis.writer.models import ChapterContent
from genesis.eval.scorer import EvalReport
def _content(cid, version=1):
return ChapterContent(chapter_id=cid, version=version, title="x", blocks=[])
def _report(failed):
return EvalReport(
dimensions=[], total_score=0.0, passed=False, failed_chapters=failed,
)
def test_initial_state_empty():
st = WriterState(["a", "b"])
assert st.needs_regeneration() == ["a", "b"]
def test_record_success_clears_need():
st = WriterState(["a"])
st.record_success(_content("a"))
assert st.needs_regeneration() == []
def test_failed_eval_marks_regeneration():
st = WriterState(["a"])
st.record_success(_content("a"))
st.record_eval("a", _report(["a"]))
assert st.needs_regeneration() == ["a"]