51bfb6e320
固定 3 题材金标准输入（仙侠/都市/悬疑，各含最小 spec+world+outline 及开篇章）， 经真实 assemble()+build_write_request() 组装 craft-on / craft-off 两版写章请求： - dry 模式（默认，零成本、免真实 LLM，可进 CI）：断言 craft 教条块、genre 片段、 开篇黄金三章 marker 确实按 fixture 注入正确内容，且 craft-off 基线不含这三者； 并把两版将发送给模型的 prompt 并排 dump 供人眼看。这些断言即 tests/test_craft_eval.py， 是 craft/审查注入路径的回归门禁。 - --live 模式（真实付费 LLM，绝不进 CI）：经 gateway 各生成一章 + analyst 五维裁判， 凭据读 DB、路由走 tier_routing，与生产写章同一网关路径。 入口 `uv run python -m tests.craft_eval [--live]`；用法/成本/门禁说明见 README.md。
364 lines
11 KiB
Python
364 lines
11 KiB
Python
"""craft A/B 评估回路核心（灵感 F2）——组装请求 + dry 确定性校验 + 报告渲染。
|
||
|
||
**零 LLM 依赖**：本模块只走确定性组装（真 `assemble()` + 真 `build_write_request()`）与
|
||
纯字符串校验，可安全进 CI。真实生成/裁判在 `live.py`（惰性导入，不进本模块）。
|
||
|
||
craft-on = 生产路径（`assemble` 注入 genre 片段进 stable_core + 开篇 marker 进 volatile，
|
||
`build_write_request` 把 craft 教条置为最前 system 块）。
|
||
craft-off = 同一 fixture 抹掉 craft 层的基线：剥掉 genre/开篇两节 + 不加教条块——
|
||
两版**唯一差异即 craft 贡献**，便于人眼并排 diff 与 LLM 裁判对比。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
from ww_core.domain.repositories import (
|
||
CharacterView,
|
||
DigestView,
|
||
ForeshadowView,
|
||
MemoryRepos,
|
||
OutlineView,
|
||
ProjectSpecView,
|
||
RuleView,
|
||
StyleView,
|
||
WorldEntityView,
|
||
)
|
||
from ww_core.memory.assemble import OPENING_CHAPTER_LIMIT, assemble
|
||
from ww_core.orchestrator.write_node import build_write_request
|
||
from ww_llm_gateway.types import Block, LlmRequest, Scope
|
||
|
||
from .fixtures import FixtureCase
|
||
|
||
# 单用户原型 stub（dry 路径不落库，取值无关紧要，仅为满足 Scope）。
|
||
PROJECT_ID = uuid.UUID("00000000-0000-0000-0000-0000000c7a17")
|
||
USER_ID = uuid.UUID(int=1)
|
||
|
||
# assemble 内联的 craft 节标题（assemble.py:97 `本作题材写法` / :141 `开篇黄金三章`）。
|
||
# craft-off 据此剥离；若上游改名，本表失配 → craft-off 仍含片段 → 校验断言会当场炸出漂移。
|
||
_GENRE_SECTION_TITLE = "本作题材写法"
|
||
_OPENING_SECTION_TITLE = "开篇黄金三章"
|
||
|
||
__all__ = [
|
||
"OPENING_CHAPTER_LIMIT",
|
||
"PROJECT_ID",
|
||
"USER_ID",
|
||
"CaseReport",
|
||
"Check",
|
||
"PromptPair",
|
||
"all_passed",
|
||
"build_prompt_pair",
|
||
"input_text",
|
||
"render_report",
|
||
"system_text",
|
||
"verify_case",
|
||
"write_report",
|
||
]
|
||
|
||
|
||
# ---- 内存 fake repositories（只喂 assemble 所需，其余返回空）----
|
||
|
||
|
||
@dataclass
|
||
class _FakeOutlineRepo:
|
||
"""按查询章号合成 OutlineView（复用同一份 beats），支持任意 chapter_no 的边界测试。"""
|
||
|
||
volume: int
|
||
beats: dict[str, Any]
|
||
|
||
async def get(self, project_id: uuid.UUID, chapter_no: int) -> OutlineView | None:
|
||
return OutlineView(volume=self.volume, chapter_no=chapter_no, beats=dict(self.beats))
|
||
|
||
async def list_for_project(
|
||
self, project_id: uuid.UUID
|
||
) -> list[OutlineView]: # pragma: no cover
|
||
return []
|
||
|
||
|
||
@dataclass
|
||
class _FakeCharacterRepo:
|
||
rows: tuple[CharacterView, ...]
|
||
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[CharacterView]:
|
||
return list(self.rows)
|
||
|
||
|
||
@dataclass
|
||
class _FakeWorldEntityRepo:
|
||
rows: tuple[WorldEntityView, ...]
|
||
|
||
async def list_for_project(self, project_id: uuid.UUID) -> list[WorldEntityView]:
|
||
return list(self.rows)
|
||
|
||
|
||
@dataclass
|
||
class _FakeDigestRepo:
|
||
async def recent(self, project_id: uuid.UUID, k: int) -> list[DigestView]:
|
||
return []
|
||
|
||
|
||
@dataclass
|
||
class _FakeForeshadowRepo:
|
||
async def list_for_codes(self, project_id: uuid.UUID, codes: list[str]) -> list[ForeshadowView]:
|
||
return []
|
||
|
||
|
||
@dataclass
|
||
class _FakeStyleRepo:
|
||
async def latest(self, project_id: uuid.UUID) -> StyleView | None:
|
||
return None
|
||
|
||
|
||
@dataclass
|
||
class _FakeRulesRepo:
|
||
async def all_for_project(self, project_id: uuid.UUID) -> list[RuleView]:
|
||
return []
|
||
|
||
|
||
@dataclass
|
||
class _FakeProjectSpecRepo:
|
||
row: ProjectSpecView
|
||
|
||
async def spec(self, project_id: uuid.UUID) -> ProjectSpecView | None:
|
||
return self.row
|
||
|
||
|
||
def _build_repos(case: FixtureCase) -> MemoryRepos:
|
||
volume = 1
|
||
return MemoryRepos(
|
||
outline=_FakeOutlineRepo(volume=volume, beats=case.beats),
|
||
character=_FakeCharacterRepo(case.characters),
|
||
world_entity=_FakeWorldEntityRepo(case.world),
|
||
digest=_FakeDigestRepo(),
|
||
foreshadow=_FakeForeshadowRepo(),
|
||
style=_FakeStyleRepo(),
|
||
rules=_FakeRulesRepo(),
|
||
project=_FakeProjectSpecRepo(case.spec),
|
||
review=None, # 冲突裁决反哺不在 craft A/B 范围；优雅降级不反哺
|
||
)
|
||
|
||
|
||
# ---- craft-on / craft-off 请求组装 ----
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PromptPair:
|
||
craft_on: LlmRequest
|
||
craft_off: LlmRequest
|
||
chapter_no: int
|
||
|
||
|
||
def _strip_section(text: str, title: str) -> str:
|
||
"""按 `## {title}` 剥掉一节（节以 "\\n\\n" 分隔，节体无空行——见 assemble._section）。"""
|
||
header = f"## {title}"
|
||
kept = [chunk for chunk in text.split("\n\n") if not chunk.startswith(header)]
|
||
return "\n\n".join(kept)
|
||
|
||
|
||
async def build_prompt_pair(case: FixtureCase, *, chapter_no: int | None = None) -> PromptPair:
|
||
"""据 fixture 组装 craft-on（生产路径）与 craft-off（抹除 craft 层的基线）两版请求。"""
|
||
ch = chapter_no if chapter_no is not None else case.chapter_no
|
||
ctx = await assemble(_build_repos(case), PROJECT_ID, ch)
|
||
|
||
craft_on = build_write_request(
|
||
stable_core=ctx.stable_core,
|
||
volatile=ctx.volatile,
|
||
user_id=USER_ID,
|
||
project_id=PROJECT_ID,
|
||
)
|
||
|
||
off_stable = _strip_section(ctx.stable_core, _GENRE_SECTION_TITLE)
|
||
off_volatile = _strip_section(ctx.volatile, _OPENING_SECTION_TITLE)
|
||
craft_off = LlmRequest(
|
||
tier="writer",
|
||
system=[Block(text=off_stable, cache=True)],
|
||
input=off_volatile,
|
||
stream=True,
|
||
scope=Scope(user_id=USER_ID, project_id=PROJECT_ID),
|
||
)
|
||
return PromptPair(craft_on=craft_on, craft_off=craft_off, chapter_no=ch)
|
||
|
||
|
||
# ---- prompt 文本抽取（供校验与报告）----
|
||
|
||
|
||
def system_text(req: LlmRequest) -> str:
|
||
return "\n\n".join(b.text for b in req.system)
|
||
|
||
|
||
def input_text(req: LlmRequest) -> str:
|
||
if isinstance(req.input, str):
|
||
return req.input
|
||
return "\n\n".join(b.text for b in req.input)
|
||
|
||
|
||
# ---- dry 确定性校验（无 LLM，可进 CI）----
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Check:
|
||
name: str
|
||
passed: bool
|
||
detail: str
|
||
|
||
|
||
def _load_doctrine() -> str:
|
||
# 惰性导入避免 fixtures 模块与本模块的导入顺序耦合。
|
||
from ww_agents import load_craft_doctrine
|
||
|
||
return load_craft_doctrine()
|
||
|
||
|
||
def _genre_fragment(genre: str) -> str:
|
||
from ww_config import genre_craft_fragment
|
||
|
||
fragment = genre_craft_fragment(genre)
|
||
if fragment is None: # pragma: no cover - fixtures 已保证命中
|
||
raise ValueError(f"题材 {genre!r} 无 GENRE_CRAFT 片段")
|
||
return fragment
|
||
|
||
|
||
def verify_case(case: FixtureCase, pair: PromptPair) -> list[Check]:
|
||
"""对一份 fixture 的 craft-on/off 请求做确定性断言（craft 改动的回归门禁）。"""
|
||
on, off = pair.craft_on, pair.craft_off
|
||
doctrine = _load_doctrine()
|
||
fragment = _genre_fragment(case.genre)
|
||
on_sys, on_in = system_text(on), input_text(on)
|
||
off_sys, off_in = system_text(off), input_text(off)
|
||
is_opening = pair.chapter_no <= OPENING_CHAPTER_LIMIT
|
||
|
||
checks: list[Check] = [
|
||
Check(
|
||
"craft_on_doctrine_first_block",
|
||
len(on.system) >= 2 and on.system[0].text == doctrine and on.system[0].cache,
|
||
"教条须为首个 system 块、字节等于 load_craft_doctrine()、cache=True（#9）",
|
||
),
|
||
Check(
|
||
"craft_on_genre_in_stable_not_doctrine_not_input",
|
||
fragment in on.system[1].text
|
||
and fragment not in on.system[0].text
|
||
and fragment not in on_in,
|
||
f"genre 片段须进 stable_core 缓存块、不进教条块/volatile（{case.genre}）",
|
||
),
|
||
Check(
|
||
"craft_on_opening_marker",
|
||
(_OPENING_SECTION_TITLE in on_in and _OPENING_SECTION_TITLE not in on_sys)
|
||
if is_opening
|
||
else (_OPENING_SECTION_TITLE not in on_in),
|
||
f"开篇 marker：第 {pair.chapter_no} 章"
|
||
+ ("须进 volatile、不进缓存 system" if is_opening else "非开篇章须不注入"),
|
||
),
|
||
Check(
|
||
"craft_on_tier_writer",
|
||
on.tier == "writer",
|
||
"不变量 #2：只声明 tier=writer，不写具体 model",
|
||
),
|
||
Check(
|
||
"craft_off_no_doctrine",
|
||
doctrine not in off_sys and doctrine not in off_in,
|
||
"craft-off 基线不含教条块",
|
||
),
|
||
Check(
|
||
"craft_off_no_genre_fragment",
|
||
fragment not in off_sys and fragment not in off_in,
|
||
"craft-off 基线不含 genre 片段",
|
||
),
|
||
Check(
|
||
"craft_off_no_opening_marker",
|
||
_OPENING_SECTION_TITLE not in off_in,
|
||
"craft-off 基线不含开篇 marker",
|
||
),
|
||
Check(
|
||
"craft_off_retains_body",
|
||
case.spec.title in off_sys and "请创作第" in off_in,
|
||
"剥离未误伤本体：书名与写作指令仍在",
|
||
),
|
||
]
|
||
return checks
|
||
|
||
|
||
def all_passed(checks: list[Check]) -> bool:
|
||
return all(c.passed for c in checks)
|
||
|
||
|
||
# ---- 报告渲染 ----
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class CaseReport:
|
||
case: FixtureCase
|
||
pair: PromptPair
|
||
checks: list[Check]
|
||
live: Any = None # live.LiveResult | None（惰性，避免 dry 路径导入 live）
|
||
|
||
|
||
def _fence(label: str, body: str) -> str:
|
||
return f"**{label}**\n\n```\n{body}\n```"
|
||
|
||
|
||
def _render_pair(pair: PromptPair) -> str:
|
||
on, off = pair.craft_on, pair.craft_off
|
||
return "\n\n".join(
|
||
[
|
||
"#### craft-on（发送给模型的 prompt）",
|
||
_fence("system（缓存前缀）", system_text(on)),
|
||
_fence("input（volatile）", input_text(on)),
|
||
"#### craft-off（基线 prompt）",
|
||
_fence("system（缓存前缀）", system_text(off)),
|
||
_fence("input（volatile）", input_text(off)),
|
||
]
|
||
)
|
||
|
||
|
||
def _render_live(live: Any) -> str:
|
||
return "\n\n".join(
|
||
[
|
||
"#### 生成结果（--live）",
|
||
f"served_by: `{live.served_by}`",
|
||
_fence(f"craft-on 成稿（{live.chars_on} 字）", live.chapter_on),
|
||
_fence(f"craft-off 成稿（{live.chars_off} 字）", live.chapter_off),
|
||
"#### LLM 五维裁判",
|
||
live.judge_text,
|
||
]
|
||
)
|
||
|
||
|
||
def render_report(reports: list[CaseReport], *, live: bool) -> str:
|
||
total_checks = sum(len(r.checks) for r in reports)
|
||
failed = sum(1 for r in reports for c in r.checks if not c.passed)
|
||
lines: list[str] = [
|
||
"# craft A/B 评估报告（灵感 F2）",
|
||
"",
|
||
f"模式：{'live（真实生成+裁判）' if live else 'dry（零成本确定性校验）'}",
|
||
f"题材：{len(reports)} 个 · 断言：{total_checks - failed}/{total_checks} 通过",
|
||
"",
|
||
]
|
||
for r in reports:
|
||
c = r.case
|
||
lines.append(f"## {c.slug} · {c.genre} · 《{c.spec.title}》 · 第 {r.pair.chapter_no} 章")
|
||
lines.append("")
|
||
lines.append("### dry 校验")
|
||
for chk in r.checks:
|
||
mark = "✅" if chk.passed else "❌"
|
||
lines.append(f"- {mark} `{chk.name}` — {chk.detail}")
|
||
lines.append("")
|
||
lines.append("### prompt 并排")
|
||
lines.append("")
|
||
lines.append(_render_pair(r.pair))
|
||
lines.append("")
|
||
if r.live is not None:
|
||
lines.append(_render_live(r.live))
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def write_report(text: str, path: Any) -> Any:
|
||
from pathlib import Path
|
||
|
||
out = Path(path)
|
||
out.parent.mkdir(parents=True, exist_ok=True)
|
||
out.write_text(text, encoding="utf-8")
|
||
return out
|