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。
128 lines
5.3 KiB
Python
128 lines
5.3 KiB
Python
"""`--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-on，B=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 owner（usage_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)
|