test(qa): craft 成稿质量前后对照评估回路(灵感 F2)

固定 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。
This commit is contained in:
Yaojia Wang
2026-07-06 17:57:31 +02:00
parent 3d2a8b8cee
commit 51bfb6e320
7 changed files with 970 additions and 0 deletions

127
tests/craft_eval/live.py Normal file
View File

@@ -0,0 +1,127 @@
"""`--live` 真实生成 + LLM 五维裁判(灵感 F2
**成本警告**:本模块**真实调用付费 LLM**每题材writer 生成 2 章 + analyst 裁判 1 次)。
仅供人工手动跑,**绝不进 pytest / CI**(故与 dry 校验分文件pytest 不导入本模块)。
凭据从 DB 读(`SqlCredentialStore`),路由从 `tier_routing` 取——与生产写章同一条网关路径。
用法见 `README.md``uv run python -m tests.craft_eval --live`。
"""
from __future__ import annotations
import dataclasses
from dataclasses import dataclass
from ww_api.services.credentials import STUB_OWNER_ID, SqlCredentialStore
from ww_api.services.project_deps import build_gateway_for_tier
from ww_db import get_sessionmaker
from ww_llm_gateway.types import Block, LlmRequest, Scope
from .harness import CaseReport, PromptPair
# 五维对比裁判(对齐 craft_ab 实验的口径)——只就文本本身逐维对比,不复述原文。
JUDGE_SYSTEM = """你是资深中文网络小说主编,阅稿无数,眼光毒辣、直言不讳。
现在给你同一章大纲下、由两套不同写作指令产出的两版正文A=craft-onB=craft-off
请仅就文本本身逐维度对比谁更好每个维度给出「A更好 / B更好 / 难分」并附一句依据:
① show-don't-tell情绪是否靠动作细节外化而非旁白直说
② 对话是否有角色声音/口癖
③ 章末钩子是否成立(是否让读者想追下一章)
④ 是否注水/AI 腔(空泛排比、翻译腔、无信息量的辞藻堆砌)
⑤ 开篇代入感(前几段是否立住冲突/身份钩子、抓人)
最后给出总判「A 更好 / B 更好 / 难分」,并用 1-2 句话说明理由。
保持简洁、务实,不要复述原文。"""
# 裁判每稿截断字数(省 token
_JUDGE_EXCERPT = 1500
@dataclass(frozen=True)
class LiveResult:
chapter_on: str
chapter_off: str
chars_on: int
chars_off: int
judge_text: str
served_by: str
def char_count(text: str) -> int:
"""中文成稿近似字数:去空白后的字符数。"""
return len("".join(text.split()))
async def _stream_text(gateway: object, req: LlmRequest) -> str:
parts: list[str] = []
async for delta in gateway.stream(req): # type: ignore[attr-defined]
parts.append(delta.text)
return "".join(parts)
def _judge_request(pair: PromptPair, text_on: str, text_off: str) -> LlmRequest:
body = (
"以下是两版正文(各截取前若干字以省 token\n\n"
f"===== A 稿 / craft-on{_JUDGE_EXCERPT} 字)=====\n"
+ text_on[:_JUDGE_EXCERPT]
+ f"\n\n===== B 稿 / craft-off{_JUDGE_EXCERPT} 字)=====\n"
+ text_off[:_JUDGE_EXCERPT]
)
return LlmRequest(
tier="analyst",
system=[Block(text=JUDGE_SYSTEM, cache=True)],
input=body,
stream=False,
scope=Scope(user_id=STUB_OWNER_ID, project_id=None),
)
async def run_live(reports: list[CaseReport]) -> list[CaseReport]:
"""对每个 report 的 craft-on/off 请求真实生成一章 + 五维裁判,返回附 live 的新 report 列表。
不可变更新:用 `dataclasses.replace` 产出新 `CaseReport`,不原地改冻结实例。
"""
sm = get_sessionmaker()
out: list[CaseReport] = []
async with sm() as session:
store = SqlCredentialStore(session)
# writer / analyst 各建一次网关,跨题材复用(省装配开销)。
writer_gw = await build_gateway_for_tier(session, store, "writer")
analyst_gw = await build_gateway_for_tier(session, store, "analyst")
for report in reports:
pair = report.pair
print(f"[live] {report.case.slug}:生成 craft-on……", flush=True)
text_on = await _stream_text(writer_gw, _as_scoped(pair.craft_on))
print(f"[live] {report.case.slug}:生成 craft-off……", flush=True)
text_off = await _stream_text(writer_gw, _as_scoped(pair.craft_off))
print(f"[live] {report.case.slug}analyst 五维裁判……", flush=True)
resp = await analyst_gw.run(_judge_request(pair, text_on, text_off))
result = LiveResult(
chapter_on=text_on,
chapter_off=text_off,
chars_on=char_count(text_on),
chars_off=char_count(text_off),
judge_text=resp.text,
served_by=f"{resp.served_by.provider}:{resp.served_by.model}"
+ (" (fell_back)" if resp.served_by.fell_back else ""),
)
out.append(dataclasses.replace(report, live=result))
return out
def _as_scoped(req: LlmRequest) -> LlmRequest:
"""把请求 scope 换成 stub ownerusage_ledger.owner_id FK 须指向已 seed 的用户行)。"""
scoped = Scope(user_id=STUB_OWNER_ID, project_id=None)
return req.model_copy(update={"scope": scoped})
def summarize(reports: list[CaseReport]) -> str:
"""live 结果的一行式摘要stdout 用)。"""
lines: list[str] = []
for r in reports:
if r.live is None: # pragma: no cover - live 路径必有结果
continue
lines.append(
f"{r.case.slug}: craft-on {r.live.chars_on}字 / craft-off {r.live.chars_off}"
)
return "\n".join(lines)