feat(harness): store.py — Store 版本操作 + Seed 管理

从 TRM4 core/workspace.py 拆出 Store + Seed 相关函数:
- _parse_version / list_versions / next_version / advance_version
- _write_meta / init_store
- init_seed / list_seeds / read_seed
- extract_run_db(保留原始 CREATE 语句重建主键约束)
- promote_to_seed(强校验版本一致 + 非 NULL + finally 清理)

26 个测试全部通过,radon 复杂度 A (2.83)。

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-07-07 12:36:36 -04:00
parent be3c176a46
commit b052c1f3ee
2 changed files with 838 additions and 0 deletions
+378
View File
@@ -0,0 +1,378 @@
"""Store 版本操作 + Seed 管理。
Store 存储版本化资源(视频、题目、Skill、Prompt),
通过版本号(v1, v2, ...)管理资源的演化历史。
Seed 是可复现的实验起点,包含权重快照 + baseline 数据库。
"""
from __future__ import annotations
import json
import re
import shutil
import sqlite3
from datetime import UTC, datetime
from typing import TYPE_CHECKING
from loguru import logger
if TYPE_CHECKING:
from pathlib import Path
def _now_iso() -> str:
"""返回当前 UTC 时间的 ISO 格式字符串。"""
return datetime.now(UTC).isoformat()
def _parse_version(name: str) -> int:
"""解析版本目录名 ``v\\d+`` 为整数。
参数:
name: 版本目录名,如 ``"v1"``、``"v10"``。
返回:
版本号整数。
异常:
ValueError: 版本目录名格式不合法(不匹配 ``v\\d+``)。
"""
match = re.match(r"v(\d+)$", name)
if not match:
raise ValueError(f"无效版本号: {name}")
return int(match.group(1))
def list_versions(store_dir: Path, resource_type: str) -> list[str]:
"""列出 Store 中某类资源的所有版本号,按数字值排序。
按数字排序保证 v10 排在 v2 后面(而非字典序 v10 < v2)。
参数:
store_dir: Store 根目录。
resource_type: 资源类型路径,如 ``"skills"``、``"questions/generated"``。
返回:
排序后的版本号列表,如 ``["v1", "v2", "v10"]``。
"""
resource_dir = store_dir / resource_type
if not resource_dir.is_dir():
return []
versions = []
for entry in resource_dir.iterdir():
if entry.is_dir() and re.match(r"v\d+$", entry.name):
versions.append(entry.name)
return sorted(versions, key=_parse_version)
def next_version(store_dir: Path, resource_type: str) -> str:
"""返回某类资源的下一个可用版本号。
参数:
store_dir: Store 根目录。
resource_type: 资源类型路径。
返回:
下一个版本号字符串,如 ``"v3"``。
"""
versions = list_versions(store_dir, resource_type)
if not versions:
return "v1"
latest = _parse_version(versions[-1])
return f"v{latest + 1}"
def _write_meta(target_dir: Path, version: str, source: str, **extra: str | None) -> None:
"""写入版本元数据文件 ``meta.json``。
参数:
target_dir: 版本目录。
version: 版本号。
source: 来源标识(``"manual"`` / ``"evolution"`` / ``"auto-gen"``)。
**extra: 额外字段(parent, trigger_run, trigger_workspace, description)。
"""
meta = {
"version": version,
"created_at": _now_iso(),
"parent": extra.get("parent"),
"source": source,
"trigger_run": extra.get("trigger_run"),
"trigger_workspace": extra.get("trigger_workspace"),
"description": extra.get("description", ""),
}
(target_dir / "meta.json").write_text(json.dumps(meta, ensure_ascii=False, indent=2))
def advance_version(
store_dir: Path,
resource_type: str,
source_dir: Path,
meta: dict,
) -> str:
"""将 source_dir 的内容写入 Store 的下一个版本目录,写入 meta.json。
参数:
store_dir: Store 根目录。
resource_type: 资源类型路径,如 ``"skills"``、``"questions/generated"``。
source_dir: 包含新版本资源文件的源目录。
meta: 元数据字典,至少包含 ``source`` 字段。
返回:
新版本号字符串,如 ``"v2"``。
"""
version = next_version(store_dir, resource_type)
target = store_dir / resource_type / version
shutil.copytree(source_dir, target)
_write_meta(
target,
version,
meta.get("source", "manual"),
parent=meta.get("parent"),
trigger_run=meta.get("trigger_run"),
trigger_workspace=meta.get("trigger_workspace"),
description=meta.get("description", ""),
)
logger.info("Store 版本推进: {}/{}", resource_type, version)
return version
def init_store(
store_dir: Path,
videos_source: Path,
skills_dir: Path,
prompts_dir: Path,
) -> None:
"""初始化 Store:拷贝视频数据,创建 skills/v1、prompts/v1 和 questions 目录。
参数:
store_dir: Store 目标路径(不得已存在)。
videos_source: 视频数据源目录。
skills_dir: 初始 Skill 文件目录。
prompts_dir: 初始 Prompt 文件目录。
异常:
FileExistsError: Store 目录已存在。
"""
if store_dir.exists():
raise FileExistsError(f"Store 已存在: {store_dir}")
store_dir.mkdir(parents=True)
shutil.copytree(videos_source, store_dir / "videos")
(store_dir / "questions" / "benchmarks").mkdir(parents=True)
(store_dir / "questions" / "generated").mkdir(parents=True)
shutil.copytree(skills_dir, store_dir / "skills" / "v1")
_write_meta(
store_dir / "skills" / "v1",
"v1",
"manual",
description="手工创建的初始版本",
)
shutil.copytree(prompts_dir, store_dir / "prompts" / "v1")
_write_meta(
store_dir / "prompts" / "v1",
"v1",
"manual",
description="手工创建的初始版本",
)
logger.info("Store 初始化完成: {}", store_dir)
# ---------------------------------------------------------------------------
# 种子库(Seed)函数
# ---------------------------------------------------------------------------
def init_seed(
store_dir: Path,
name: str,
skills_dir: Path,
prompts_dir: Path,
baseline_db: Path,
baseline_run_id: str,
parent: str | None,
description: str,
) -> Path:
"""在 store/seeds/<name> 写一个种子:权重 + baseline.db + seed.json。
参数:
store_dir: Store 根目录。
name: 种子名(如 ``'initial'``、``'from-evolve-v20'``)。
skills_dir: 该版本 Skill 权重源目录。
prompts_dir: 该版本 Prompt 权重源目录。
baseline_db: 该版本全量记录 db(含 _runs + predictions 行)。
baseline_run_id: 全量记录的 run_idfresh 时注入 build_pools。
parent: 来源(initial 为 None)。
description: 人类可读说明。
返回:
种子目录路径。
异常:
FileExistsError: 同名种子已存在(不覆盖)。
"""
seed_dir = store_dir / "seeds" / name
if seed_dir.exists():
raise FileExistsError(f"种子已存在,不覆盖: {seed_dir}")
seed_dir.mkdir(parents=True)
shutil.copytree(skills_dir, seed_dir / "skills")
shutil.copytree(prompts_dir, seed_dir / "prompts")
shutil.copy2(baseline_db, seed_dir / "baseline.db")
(seed_dir / "seed.json").write_text(
json.dumps(
{
"baseline_run_id": baseline_run_id,
"parent": parent,
"created_at": _now_iso(),
"description": description,
},
ensure_ascii=False,
indent=2,
)
)
logger.info("种子创建完成: {}", seed_dir)
return seed_dir
def list_seeds(store_dir: Path) -> list[str]:
"""列出 store/seeds 下所有种子名(按名排序)。
参数:
store_dir: Store 根目录。
返回:
种子名列表(仅含 seed.json 存在的目录),按名排序。
"""
seeds_root = store_dir / "seeds"
if not seeds_root.is_dir():
return []
return sorted(e.name for e in seeds_root.iterdir() if (e / "seed.json").exists())
def read_seed(store_dir: Path, name: str) -> dict:
"""读取种子 seed.json;不存在则报错。
参数:
store_dir: Store 根目录。
name: 种子名。
返回:
seed.json 解析后的字典。
异常:
FileNotFoundError: 该种子不存在。
"""
seed_json = store_dir / "seeds" / name / "seed.json"
if not seed_json.exists():
raise FileNotFoundError(f"种子不存在: {name}{seed_json}")
return json.loads(seed_json.read_text())
def extract_run_db(src_db: Path, dst_db: Path, run_id: str) -> None:
"""从 src_db 抽出某 run_id 的 _runs + predictions 行,写一个最小 db(种子 baseline.db)。
用源表的**原始 CREATE 语句**重建目标表,保留主键/列类型/约束——
``_runs.run_id TEXT PRIMARY KEY`` 是 HarnessLog ``INSERT OR IGNORE`` 去重的依据,
若 seed db 丢主键则续训/fresh-bootstrap 的去重失效。
参数:
src_db: 源 harness.db。
dst_db: 目标 db(不得已存在)。
run_id: 要抽取的 run。
异常:
RuntimeError: 源中无该表或无该 run 的行。
"""
src = sqlite3.connect(src_db)
dst = sqlite3.connect(dst_db)
try:
for table in ("_runs", "predictions"):
create_sql = src.execute(
"SELECT sql FROM sqlite_master WHERE type='table' AND name=?",
(table,),
).fetchone()
if create_sql is None or create_sql[0] is None:
raise RuntimeError(f"源 db 无表 {table}")
dst.execute(create_sql[0])
cols = [r[1] for r in src.execute(f"PRAGMA table_info({table})")]
col_sql = ", ".join(cols)
rows = src.execute(
f"SELECT {col_sql} FROM {table} WHERE run_id=?", (run_id,)
).fetchall()
if not rows:
raise RuntimeError(f"{table} 中无 run_id={run_id} 的行")
ph = ", ".join("?" * len(cols))
dst.executemany(f"INSERT INTO {table} ({col_sql}) VALUES ({ph})", rows)
dst.commit()
finally:
dst.close()
src.close()
def promote_to_seed(
workspace_dir: Path,
store_dir: Path,
version: str,
eval_run_id: str,
name: str,
description: str,
) -> Path:
"""把 workspace 的指定版本 + 配套 prompts + 指定 eval run 全量记录固化成新种子。
强校验 eval_run_id 对应的 _runs 行中 skills_version 必须与 version 一致,
且 skills_version/prompts_version 均不得为 NULL。
参数:
workspace_dir: 来源 workspace。
store_dir: Store 根目录。
version: skills 版本号。
eval_run_id: canonical eval run(其 _runs 行提供配套 prompts 版本与全量记录)。
name: 新种子名(冲突报错不覆盖)。
description: 说明。
返回:
新种子目录。
异常:
ValueError: eval_run_id 不存在,或其 skills_version 与 version 不符,或版本为 NULL。
FileExistsError: 同名种子已存在(由 init_seed 抛出)。
"""
con = sqlite3.connect(workspace_dir / "harness.db")
con.row_factory = sqlite3.Row
try:
row = con.execute(
"SELECT skills_version, prompts_version FROM _runs WHERE run_id=?",
(eval_run_id,),
).fetchone()
finally:
con.close()
if row is None:
raise ValueError(f"eval run 不存在: {eval_run_id}")
skills_v, prompts_v = row["skills_version"], row["prompts_version"]
# 强校验——eval run 的版本必须与 --version 一致,且不得为 NULL
if skills_v is None or prompts_v is None:
raise ValueError(f"eval run {eval_run_id} 的 _runs 版本对为 NULL(未回填?),无法 promote")
if skills_v != version:
raise ValueError(f"eval run {eval_run_id} 的版本 {skills_v} 与 --version {version} 不符")
tmp_db = workspace_dir / "_promote_tmp.db"
if tmp_db.exists():
tmp_db.unlink()
extract_run_db(workspace_dir / "harness.db", tmp_db, eval_run_id)
try:
seed_dir = init_seed(
store_dir,
name,
workspace_dir / "skills" / skills_v,
workspace_dir / "prompts" / prompts_v,
tmp_db,
baseline_run_id=eval_run_id,
parent=f"{workspace_dir.name}:{version}",
description=description,
)
finally:
tmp_db.unlink()
logger.info("Promote 完成: {} -> {}", workspace_dir.name, seed_dir)
return seed_dir