kai/writer-work-flow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(agents): prompt 外置方案A — 21 prompt 散文外迁 .md + SpecResolver

Browse Source 
把 21 个内置 agent 的 system_prompt 从 specs.py 的 Python 常量外置为
prompts/<spec.name>.md，import 期由 load_prompt 确定性加载；建立 SPECS 名册
+ SCHEMA_CATALOG（Pydantic 类型留 Python）+ 统一只读解析入口 SpecResolver。
纯重构、零功能/schema 变更，缓存断点前块字节级不变（不变量 #9）。

@llm packages/agents（步骤1-3）
- spec_model.py：抽出 AgentSpec（frozen，字段不变）
- prompt_loader.py：load_prompt = utf-8-sig 去BOM → LF 归一 → NFC → rstrip尾LF，
  内存缓存 + fail-fast（PromptNotFoundError），import 期确定性
- schema_catalog.py：SCHEMA_CATALOG[name]→output type 唯一真相源（refiner=None）
- prompts/*.md ×21：取常量「运行时值」程序化外迁（反斜杠折行已塌缩，
  物理换行≡运行时换行）；文件名按 spec.name 连字符（style.md/character-gen.md 等）
- specs.py：删 21 常量 + AgentSpec 类；system_prompt=load_prompt(name)、
  output_schema=SCHEMA_CATALOG[name]；建 SPECS + REVIEW_RESERVED_NAMES；
  *_spec 兼容期保留且 SPECS[name] is *_spec（同一实例）。804→337 行
- __init__.py：显式 __all__ 重导出（避 F401）

@backend packages/skills（步骤4-5）
- SpecResolver：内置 SPECS（纯内存、零 DB）+ 用户 SkillRegistry 统一 get；
  内置 name 永不触发 DB；output_schema_for 精确匹配
- skill_registry：保留命名空间守卫前移至入库校验，拒同名内置 → VALIDATION
- toolbox_registry：GeneratorTool.spec 改走 SPECS[...]，删 12 个 *_spec 直接 import

@devops repo-root
- .gitattributes：prompts/*.md text eol=lf（修正：须用完整嵌套路径才匹配）
- packages/agents/pyproject：hatchling artifacts 纳入 prompts/*.md 随 wheel/sdist 分发
- ci.yml：新增 build wheel → 裸装 → import ww_agents.SPECS 冒烟

TDD 全程 mock 网关；门禁绿：ruff/format clean · mypy 209 files · pytest 744 passed
（含金标准 sha256 回归 / md↔spec↔catalog 一一对应 / fail-fast / BOM+NFC /
内置守卫 / 同一实例 / 编排器无回归 / apps/api import-smoke / 打包冒烟）
This commit is contained in:
Yaojia Wang
2026-06-24 04:49:44 +02:00
parent f9ff89cd6b
commit dca4d45d4e
52 changed files with 1907 additions and 701 deletions
Download Patch File Download Diff File Expand all files Collapse all files

55
packages/agents/tests/_gen_golden.py Normal file
View File

@@ -0,0 +1,55 @@
"""一次性金标准生成器（Prompt 外置方案A · 步0）。
取每个内置 AgentSpec 的**运行时** `system_prompt` 值算 sha256，写入
`fixtures/prompt_hashes.json`。这是 prompt 外置「字节级零变化」的唯一金标准
（设计 docs/design/prompt-management.md §6 / 原则6）。
为何用 import 读 `spec.system_prompt` 而非源码文本：21 个 `*_SYSTEM_PROMPT`
常量大量使用反斜杠行延续（行尾 `\\`），源码物理换行 ≠ 运行时换行。Python 在
import 时已把 `\\<newline>` 求值塌缩——`spec.system_prompt` 即生产实际发送的
运行时字符串，等价于 AST literal_eval 且更直接。
用法（仅在步0 运行一次，之后**不要**再生成，否则会用新值覆盖金标准、掩盖漂移）：
uv run python packages/agents/tests/_gen_golden.py
漂移防护靠 test_prompt_loader.py 比对 load_prompt 输出 vs 本文件产出的 json，
而非重跑本脚本。
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import ww_agents
from ww_agents.specs import AgentSpec
FIXTURE = Path(__file__).parent / "fixtures" / "prompt_hashes.json"
def _sha256(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
def collect_builtin_specs() -> dict[str, AgentSpec]:
"""ww_agents 命名空间里导出的全部内置 AgentSpec 实例（按 spec.name 去重）。"""
found: dict[str, AgentSpec] = {}
for value in vars(ww_agents).values():
if isinstance(value, AgentSpec):
found[value.name] = value
return found
def main() -> None:
specs = collect_builtin_specs()
golden = {name: _sha256(spec.system_prompt) for name, spec in sorted(specs.items())}
FIXTURE.parent.mkdir(parents=True, exist_ok=True)
FIXTURE.write_text(
json.dumps(golden, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
print(f"wrote {len(golden)} golden hashes -> {FIXTURE}")
if __name__ == "__main__":
main()

23
packages/agents/tests/fixtures/prompt_hashes.json vendored Normal file
View File

@@ -0,0 +1,23 @@
{
"blurb": "c6f89e3de577775aae1ff747a103876664b3075bdcb2ed118697593891707544",
"book-title": "bb6f0c091fdfb4770c3e398a5f44bc97ec2603a923d4624d784f15916820d305",
"brainstorm": "cbdc8f88a4eed5c1d34db03b522a77cec5fb3c15a474fd191399d05f58f071b5",
"character-gen": "7ef3075c5339174c4a40c92995ec7c71fd058936efbaa468d24c697a7cc9ee0e",
"continue": "d140ed43e508531798e24e708d79171f4f9c7a14d92e6523b217d00985507858",
"continuity": "043a838cf32bfab047d3ea271114733eb3634f091448406259f6464de46f2897",
"de-ai": "c8ef0e34bdd9806786aa0af1ec25b6334f6a31211a51fb025730c66b7a462835",
"expand": "c83fcff4c5820c4d0213e1f63609f19a57959d0a3660df1cbda8f752db4e083a",
"fine-outline": "f2941ce4c1df036fd14e110767c2118c28b6f7ea7fa50e79568313a56dbb6937",
"foreshadow": "df830dd27a51c79789fe4b9f9bc6a6e2228670d198d6b330c7b68063a3817227",
"glossary": "6526746766a67ae7b9bd082856410fc4be04a1efd200254bcd1f855cad69c376",
"golden-finger": "20b5b2c14474ec460e6f89a6e5aa1332fa73e49594fa8e81f7383caded4302f8",
"name": "3e5f631fd518ca77ea9cda7f68afcfcfd9636440dc3633fc496e94d0b40eba05",
"opening": "fa777ab5ffc861a4a0da6ceff629573f7388340173706ebe8452041d16a893b5",
"outliner": "c1ad0527d291a0a3914340086d3429740a3cc6860927485da64cc597b8622549",
"pace": "a4c8011c27904a7e8d1bfa0639e2d5b83f900a14e50ccd0a6f3051ecf6310674",
"refiner": "1a31b355938d95e99a4d17a80b3149c2c42e7dd1d40b9939c64f014d7380bbd2",
"style": "a527a540accb5902b405ac4b0e3421d63a6da792439a1ed429c711b48188d780",
"style_extract": "a8fbfce6f7885a9257e9036aeb2803bb0c623c250a559ab6bb6f1f6407855854",
"teardown": "ddde94186beec6c7f3edcfa060d3471fe86c223d24fd01bf9d37d5c80ab8a9cf",
"worldbuilder": "2fee12f3f001a24c8c8c97bf308436d979993c7a02a519ef511085622747c61d"
}

4
packages/agents/tests/test_brainstorm_spec.py
View File

@@ -63,8 +63,8 @@ def test_brainstorm_spec_is_light_tier() -> None:
def test_brainstorm_spec_reads_projects_writes_nothing() -> None:
# 纯预览：读作品设定、不写任何业务表（不变量 #3）
assert brainstorm_spec.reads == ["projects"]
assert brainstorm_spec.writes == []
assert brainstorm_spec.reads == ("projects",)
assert brainstorm_spec.writes == ()
def test_brainstorm_spec_output_schema() -> None:

4
packages/agents/tests/test_competitor_specs.py
View File

@@ -51,8 +51,8 @@ def test_competitor_spec_declares_expected_contract(
) -> None:
assert spec.name == name
assert spec.tier == tier
assert spec.reads == reads
assert spec.writes == writes # 不变量 #3：续写/扩写/降AI 纯预览；拆书落 rules
assert list(spec.reads) == reads
assert list(spec.writes) == writes # 不变量 #3：续写/扩写/降AI 纯预览；拆书落 rules
assert spec.output_schema is output_schema
assert spec.scope == "builtin"
assert spec.input_schema is None # 注入材料为序列化文本，非结构化入参

8
packages/agents/tests/test_generation_specs.py
View File

@@ -129,8 +129,8 @@ def test_worldbuilder_spec_is_writer_tier() -> None:
def test_worldbuilder_spec_reads_projects_writes_world_entities() -> None:
assert worldbuilder_spec.reads == ["projects"]
assert worldbuilder_spec.writes == ["world_entities"]
assert worldbuilder_spec.reads == ("projects",)
assert worldbuilder_spec.writes == ("world_entities",)
def test_worldbuilder_spec_output_schema() -> None:
@@ -152,8 +152,8 @@ def test_character_gen_spec_is_writer_tier() -> None:
def test_character_gen_spec_reads_world_and_characters() -> None:
# reads=world_entities+characters（约束 + 防雷同对照），writes=characters
assert character_gen_spec.reads == ["world_entities", "characters"]
assert character_gen_spec.writes == ["characters"]
assert character_gen_spec.reads == ("world_entities", "characters")
assert character_gen_spec.writes == ("characters",)
def test_character_gen_spec_output_schema() -> None:

6
packages/agents/tests/test_outliner_spec.py
View File

@@ -87,17 +87,17 @@ def test_outliner_spec_is_analyst_tier() -> None:
def test_outliner_spec_declares_expected_reads() -> None:
assert outliner_spec.reads == [
assert outliner_spec.reads == (
"projects",
"foreshadow",
"characters",
"world_entities",
]
)
def test_outliner_spec_declares_outline_write() -> None:
# 声明式 writes（经验收/T3.5 才真写库，不变量 #3）
assert outliner_spec.writes == ["outline"]
assert outliner_spec.writes == ("outline",)
def test_outliner_spec_output_schema_is_outline_result() -> None:

198
packages/agents/tests/test_prompt_loader.py Normal file
View File

@@ -0,0 +1,198 @@
"""prompt_loader 单测（Prompt 外置方案A）。
步0 先落「金标准比对」用例（此刻 load_prompt 未实现 → RED）；步2 补齐缓存/
fail-fast/规整/目录完整性/文件层尾换行等用例。
金标准 = packages/agents/tests/fixtures/prompt_hashes.json，取自内置 spec 的
运行时 system_prompt（见 _gen_golden.py）。比对 load_prompt 输出的 sha256 == 金标准，
即可证明 prompt 外置到 .md 后字节级零变化（不变量 #9 / 设计 §6）。
"""
from __future__ import annotations
import hashlib
import json
from pathlib import Path
import pytest
from ww_agents import (
REVIEW_RESERVED_NAMES,
SCHEMA_CATALOG,
SPECS,
continuity_spec,
output_schema_for,
)
from ww_agents.prompt_loader import (
_CACHE,
PROMPTS_DIR,
PromptNotFoundError,
load_prompt,
)
FIXTURE = Path(__file__).parent / "fixtures" / "prompt_hashes.json"
def _golden() -> dict[str, str]:
data: dict[str, str] = json.loads(FIXTURE.read_text(encoding="utf-8"))
return data
def _sha256(text: str) -> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()
@pytest.mark.parametrize("name", sorted(_golden().keys()))
def test_load_prompt_matches_golden(name: str) -> None:
# Arrange
expected = _golden()[name]
# Act
actual = _sha256(load_prompt(name))
# Assert
assert actual == expected, f"prompt {name!r} 字节漂移：load_prompt 输出与金标准不符"
# ---- #1 注册表唯一性 ----
def test_specs_registry_len_is_21() -> None:
assert len(SPECS) == 21
# name 即 key，dict 已去重；逐项确认 key == spec.name（无错位）
assert all(key == spec.name for key, spec in SPECS.items())
# ---- #2 md ↔ spec ↔ catalog 一一对应（覆盖 style.md / character-gen.md 连字符）----
def test_md_spec_catalog_one_to_one() -> None:
md_stems = {p.stem for p in PROMPTS_DIR.glob("*.md")}
assert set(SPECS) == md_stems == set(SCHEMA_CATALOG)
# 易错连字符 / 非变量名命名显式覆盖
assert "style" in md_stems and "style_drift" not in md_stems
assert "character-gen" in md_stems
assert "golden-finger" in md_stems
assert "book-title" in md_stems
assert "fine-outline" in md_stems
assert "de-ai" in md_stems
# ---- #3 load_prompt 正常 + 缓存命中 ----
def test_load_prompt_returns_text_and_caches() -> None:
# Arrange
_CACHE.pop("continuity", None)
# Act
first = load_prompt("continuity")
assert "continuity" in _CACHE # 首次调用后已缓存
second = load_prompt("continuity")
# Assert
assert first and isinstance(first, str)
assert first is second # 二次调用命中 _CACHE，返回同一对象
# ---- #4 load_prompt fail-fast ----
def test_load_prompt_unknown_name_raises() -> None:
with pytest.raises(PromptNotFoundError):
load_prompt("does-not-exist-spec-name")
# ---- #6 行尾确定性（CRLF/CR → LF）----
def test_load_prompt_normalizes_line_endings(tmp_path: Path) -> None:
# Arrange — 同一文本的 LF / CRLF / CR 三版
body = "第一行\n第二行\n第三行"
(tmp_path / "lf.md").write_text(body + "\n", encoding="utf-8", newline="")
(tmp_path / "crlf.md").write_bytes((body + "\n").replace("\n", "\r\n").encode("utf-8"))
(tmp_path / "cr.md").write_bytes((body + "\n").replace("\n", "\r").encode("utf-8"))
def _load(stem: str) -> str:
_CACHE.pop(stem, None)
from ww_agents import prompt_loader
orig = prompt_loader.PROMPTS_DIR
prompt_loader.PROMPTS_DIR = tmp_path # type: ignore[misc]
try:
return prompt_loader.load_prompt(stem)
finally:
prompt_loader.PROMPTS_DIR = orig # type: ignore[misc]
# Act / Assert — 三版规整后 hash 一致
lf = _load("lf")
assert _sha256(lf) == _sha256(_load("crlf")) == _sha256(_load("cr"))
# ---- #6b BOM + NFC 规整 ----
def test_load_prompt_strips_bom_and_normalizes_nfc(tmp_path: Path) -> None:
import unicodedata
# Arrange — 带 BOM 的 NFD 全角文本 vs 干净 NFC
text_nfd = unicodedata.normalize("NFD", "全角：测试。")
(tmp_path / "bom.md").write_bytes("".encode() + (text_nfd + "\n").encode("utf-8"))
(tmp_path / "clean.md").write_text(
unicodedata.normalize("NFC", "全角：测试。") + "\n", encoding="utf-8"
)
from ww_agents import prompt_loader
orig = prompt_loader.PROMPTS_DIR
prompt_loader.PROMPTS_DIR = tmp_path # type: ignore[misc]
try:
_CACHE.pop("bom", None)
_CACHE.pop("clean", None)
bom = prompt_loader.load_prompt("bom")
clean = prompt_loader.load_prompt("clean")
finally:
prompt_loader.PROMPTS_DIR = orig # type: ignore[misc]
# Assert — 无 BOM 首字节，NFD → NFC 归一后与 clean 一致
assert not bom.startswith("")
assert _sha256(bom) == _sha256(clean)
# ---- #9 / #10 output_schema_for + refiner None ----
def test_output_schema_for_builtin_and_refiner() -> None:
assert output_schema_for("continuity") is SCHEMA_CATALOG["continuity"]
assert output_schema_for("refiner") is None
assert SCHEMA_CATALOG["refiner"] is None
# 精确匹配：拼错近似 name 不误命中连字符版（KeyError，调用方按需 catch）
with pytest.raises(KeyError):
output_schema_for("character_gen")
# ---- #13 文件层尾换行契约：每个 .md 磁盘尾部 LF ≤ 1 ----
@pytest.mark.parametrize("md", sorted(PROMPTS_DIR.glob("*.md")), ids=lambda p: p.stem)
def test_md_trailing_lf_at_most_one(md: Path) -> None:
raw = md.read_bytes()
trailing = len(raw) - len(raw.rstrip(b"\n"))
assert trailing <= 1, f"{md.name} 尾部 LF={trailing} > 1"
# ---- #14 同一实例不变量 + 四审保留名安全边界 ----
def test_specs_identity_and_review_reserved_names() -> None:
assert SPECS["continuity"] is continuity_spec # 同一实例（兼容期不双真相）
assert REVIEW_RESERVED_NAMES <= set(SPECS)
for name in REVIEW_RESERVED_NAMES:
spec = SPECS[name]
assert spec.scope == "builtin"
assert spec.writes == ()
# ---- #15 reads/writes 为不可变 tuple：四审 writes 不可被 .append 旁路（不变量 #3）----
def test_review_specs_writes_immutable_tuple() -> None:
for name in REVIEW_RESERVED_NAMES:
spec = SPECS[name]
assert isinstance(spec.writes, tuple)
assert isinstance(spec.reads, tuple)
# tuple 无 append/clear → 任何旁路 writes 守卫的尝试 fail-fast
with pytest.raises(AttributeError):
spec.writes.append("INJECTED") # type: ignore[attr-defined]
# ---- #16 SPECS 注册表运行时只读（MappingProxyType）：单一真相源不可被污染 ----
def test_specs_registry_is_read_only() -> None:
with pytest.raises(TypeError):
SPECS["injected"] = continuity_spec # type: ignore[index]
with pytest.raises(TypeError):
del SPECS["outliner"] # type: ignore[attr-defined]
# ---- #17 .gitattributes 锁 prompts/*.md LF（防 CRLF 漂移，设计 §4/§8）----
def test_gitattributes_locks_prompts_lf() -> None:
repo_root = Path(__file__).resolve().parents[3]
gitattributes = repo_root / ".gitattributes"
assert gitattributes.is_file(), ".gitattributes 缺失（防 CRLF 漂移守卫）"
content = gitattributes.read_text(encoding="utf-8")
assert "prompts/*.md text eol=lf" in content

8
packages/agents/tests/test_review_specs.py
View File

@@ -111,11 +111,11 @@ def test_foreshadow_spec_is_analyst_tier() -> None:
def test_foreshadow_spec_is_read_only() -> None:
# 不变量 #3：四审只读
assert foreshadow_spec.writes == []
assert foreshadow_spec.writes == ()
def test_foreshadow_spec_reads_foreshadow() -> None:
assert foreshadow_spec.reads == ["foreshadow"]
assert foreshadow_spec.reads == ("foreshadow",)
def test_foreshadow_spec_output_schema() -> None:
@@ -136,12 +136,12 @@ def test_pace_spec_is_light_tier() -> None:
def test_pace_spec_is_read_only() -> None:
assert pace_spec.writes == []
assert pace_spec.writes == ()
def test_pace_spec_reads_rules() -> None:
# genre 模板 DSL 经 rules（genre 级）注入
assert pace_spec.reads == ["rules"]
assert pace_spec.reads == ("rules",)
def test_pace_spec_output_schema() -> None:

4
packages/agents/tests/test_specs.py
View File

@@ -93,11 +93,11 @@ def test_continuity_spec_is_analyst_tier() -> None:
def test_continuity_spec_is_read_only() -> None:
# 不变量 #3：四审只读
assert continuity_spec.writes == []
assert continuity_spec.writes == ()
def test_continuity_spec_declares_expected_reads() -> None:
assert continuity_spec.reads == ["chapter_digests", "characters", "world_entities"]
assert continuity_spec.reads == ("chapter_digests", "characters", "world_entities")
def test_continuity_spec_output_schema_is_continuity_review() -> None:

12
packages/agents/tests/test_style_specs.py
View File

@@ -111,8 +111,8 @@ def test_style_extract_spec_is_analyst_tier() -> None:
def test_style_extract_spec_reads_and_writes_fingerprint() -> None:
assert style_extract_spec.reads == ["style_fingerprint"]
assert style_extract_spec.writes == ["style_fingerprint"]
assert style_extract_spec.reads == ("style_fingerprint",)
assert style_extract_spec.writes == ("style_fingerprint",)
def test_style_extract_spec_output_schema() -> None:
@@ -134,11 +134,11 @@ def test_style_drift_spec_is_light_tier_named_style() -> None:
def test_style_drift_spec_is_read_only() -> None:
# 不变量 #3：四审只读
assert style_drift_spec.writes == []
assert style_drift_spec.writes == ()
def test_style_drift_spec_reads_fingerprint() -> None:
assert style_drift_spec.reads == ["style_fingerprint"]
assert style_drift_spec.reads == ("style_fingerprint",)
def test_style_drift_spec_output_schema() -> None:
@@ -165,8 +165,8 @@ def test_refiner_spec_output_schema_is_none() -> None:
def test_refiner_spec_is_read_only_and_no_writes() -> None:
# 回炉非持久（不变量 #3）：不写库，作者采纳经既有 draft 自动保存合入
assert refiner_spec.reads == []
assert refiner_spec.writes == []
assert refiner_spec.reads == ()
assert refiner_spec.writes == ()
def test_refiner_spec_has_nonempty_system_prompt() -> None:

4
packages/agents/tests/test_toolbox_specs.py
View File

@@ -72,8 +72,8 @@ def test_toolbox_spec_declares_expected_contract(
) -> None:
assert spec.name == name
assert spec.tier == tier
assert spec.reads == reads
assert spec.writes == writes
assert list(spec.reads) == reads
assert list(spec.writes) == writes
assert spec.output_schema is output_schema
assert spec.scope == "builtin"
assert spec.input_schema is None # 注入材料为序列化文本，非结构化入参