Video-Tree-TRM5/research-wiki/designs/2026-07-07-core-evolution-design.md

Design: core/evolution/ 可提取内核

日期 2026-07-07 · 状态 提案 · 范围 core/evolution/ 全部 7 个文件

1 定位

core/evolution/ 是自进化循环的决策内核——诊断、进化、门控、补丁。它只依赖 Protocol 接口和标准库，可搬到无 adapters 的环境用假实现原样运行。

与 app/harness/ 的分工：core/ 做决策（"候选好不好"），app/ 做编排（"跑推理、写版本、管缓存"）。

2 模块结构与依赖

core/evolution/
├── __init__.py
├── protocols.py     # SkillStore, PromptStore, RunLog
├── types.py         # ~18 个 dataclass
├── gate.py          # CE-Gate e-process（算法 #5）
├── patch.py         # 补丁引擎 + 冻结区（算法 #9 局部）
├── validate.py      # 块验证纯决策函数（算法 #7 局部）
├── diagnose.py      # 两阶段诊断管线（算法 #8）
└── evolve.py        # 进化引擎（算法 #9）

flowchart LR
    gate["gate.py\n纯数学"] ~~~ patch["patch.py\n纯文本"]
    types["types.py"] ~~~ protocols["protocols.py"]
    validate["validate.py"] --> gate
    validate --> types
    diagnose["diagnose.py"] --> types & protocols
    diagnose -.->|LLMProvider| CP["core/protocols.py"]
    evolve["evolve.py"] --> types & protocols & patch
    evolve -.->|LLMProvider| CP

依赖规则：core/evolution/ 不 import app/ 或 adapters/。LLM 调用通过已有 core.protocols.LLMProvider。

3 protocols.py

三个 Protocol 均只读——core/ 返回结果，app/ 负责持久化。

@runtime_checkable
class SkillStore(Protocol):
    """版本化技能读取端口。实现方解析 manifest 指针，core/ 不感知版本号。"""
    def read_skill(self, filename: str) -> str: ...
    def list_skill_files(self) -> list[str]: ...

@runtime_checkable
class PromptStore(Protocol):
    """版本化提示词读取端口。覆盖 system.md 和 tool extract/verify。"""
    def read_prompt(self, filename: str) -> str: ...
    def list_prompt_files(self) -> list[str]: ...

@runtime_checkable
class RunLog(Protocol):
    """实验日志查询端口。隔离 SQLite，core/ 不写 SQL。"""
    async def get_predictions(
        self, run_id: str, *, question_ids: list[str] | None = None,
    ) -> list[dict[str, Any]]: ...
    async def get_traces(
        self, run_id: str, *, question_ids: list[str] | None = None,
    ) -> list[dict[str, Any]]: ...

固定模板 prompt（诊断/进化用）不走 PromptStore，由调用方加载后以 frozen dataclass 束传入：

@dataclass(frozen=True)
class DiagnosePrompts:
    defect_vs_lapse: str; reasoning_sub: str
    span_eval_system: str; span_eval_user: str
    missed_nodes: str; skill_adherence: str
    confirmation_bias: str; evidence_sufficiency: str

@dataclass(frozen=True)
class EvolvePrompts:
    evolve_skill: str; evolve_system: str; evolve_tool: str
    evolve_rank: str; consolidate_system: str

决策	理由
Protocol 只读	core/ 纯输入→纯输出，易测试；写入是 app/ 职责
RunLog 用领域方法	避免 SQL 泄入 core/
SkillStore/PromptStore 同步	文件读取量小且快，无需 async
模板束 frozen dataclass	零 I/O + 类型安全，不增 Protocol

ARCHITECTURE.md §3.1 修订说明：ARCHITECTURE.md 定义的 SkillStore/PromptStore/RunLog 含 write_skill/write_prompt/insert 写方法。本设计有意精简为只读——core/ 返回结果 dataclass，写入由 app/harness/ 编排层执行。写方法保留在 app/ 侧的实现类中，不进 core/ Protocol。此为对 ARCHITECTURE.md 的细化，需同步更新 §3.1。

4 types.py

4.1 Gate 决策

@dataclass(frozen=True)
class GateParams:
    e_confirm: float; e_provisional: float; w_net_min: int
    delta_min: float; lambda_dir: float; e_rollback: float

@dataclass(frozen=True)
class GateVerdict:
    decision: str  # accept_confirmed | accept_provisional |
                   # reject_directional | reject_futility |
                   # reject_inertia | continue
    e_value: float; wald_lambda: float
    delta_hat: float; delta_shrunk: float

4.2 诊断

类型	frozen	用途
SpanMetrics	✓	单 span judge 结果（step, tool_name, completeness, hallucination, tags）
SkillStepAdherence	✓	skill 步骤遵循度
QuestionMetrics	✓	Stage 1 单题完整指标（7 规则 + 5 judge 类型：span/missed/adherence/bias/sufficiency）
ErrorAttribution	✓	D1 归因（error_type + cause_category + lapse_note）
CaseSample	✓	案例包单样本
SkillCasePack	✓	按题型（failure_cases + success_cases + lapse_notes）
SystemCasePack	✓	跨题型行为（行为模式 ≥ 3 次触发）
ToolCasePack	✓	按工具（failure_spans + success_spans）
DiagnosisResult	✓	管线最终输出

4.3 进化

类型	frozen	说明
EvolutionRecord	—	构建过程 mutable；tool 类的 evolved_content 为 JSON {"extract":..,"verify":..}
RejectedEdit	✓	黑名单条目；gate 字段全 Optional
EvolutionResult	✓	聚合输出（由 app/harness/ 编排层组装，非 core/ 函数返回）

4.4 验证辅助

@dataclass(frozen=True)
class PairResult:
    w: int; l: int
    observed: dict[str, tuple[bool, bool]]  # qid → (baseline, candidate)

@dataclass(frozen=True)
class QuadrantClassification:
    improvements: list[str]; regressions: list[str]
    persistent_fails: list[str]; stable_successes: list[str]

5 gate.py — CE-Gate（算法 #5，纯数学）

常量	值	来源
_WALD_WIN	ln(1.4) ≈ +0.3365	θ₁=0.70 → 2×0.70
_WALD_LOSS	ln(0.6) ≈ -0.5108	2×(1-0.70)；loss 步幅 > win（不对称）
_SHRINK_PSEUDO	4	Agresti-Coull 伪计数

compute_e_value(w, l)：截断 Beta 混合 E = 2^(W+L+1)·B(W+1,L+1)·I½(L+1,W+1)。log 空间计算；用对称性 I½(b,a) 代替 1-I½(a,b) 防灾难性消去。w<0/l<0 → ValueError；tail≤0 → 0.0；W=L=0 → 1.0。

gate_decision 四出口优先级：confirmed(E+δ) → directional(Wald) → futility(best-case) → exhaustion(provisional/inertia) → continue。Wald 从累积 W/L 重算（非增量，避免浮点漂移）。delta_shrunk 仅观测，不进决策。

probation_verdict(w, l)：双向非对称——confirm 用 E(w,l) ≥ e_confirm，rollback 用 E(l,w) ≥ e_rollback（参数交换）。e_rollback < e_confirm（回滚比确认更容易）。

6 patch.py — 补丁引擎（纯文本）

标记常量：APPENDIX_START/END、MOMENTUM_START/END（HTML 注释形式）、*_MAX_CHARS=2000。

区域解析：appendix_region_bounds() 和 momentum_region_bounds() 均严格——标记不配对（单标记/重复/逆序）抛 ValueError，双缺合法返回 None。宽容语义仅存在于 evolve.py 的包装函数 _strip_appendix_region（缺标记 = no-op）和 _appendix_span（缺标记 = 空串），不在 patch.py 本身。

apply_patch_with_report(content: str, edits: list[dict], protected_spans: list[str]) -> tuple[str, list[dict]]：

• edits 为 [{"op": str, "target": str, "content": str}, ...]（松类型 dict，保持 TRM4 格式）
• report 为 [{"index": int, "op": str, "status": str, "target": str, "content_preview": str}, ...]
• 4 种 op：append（最早非 frontmatter 冻结区前）、insert_after（三结果：成功/降级 append/skip）、replace、delete（均首次出现、count=1）
• 每条 edit 前重算 _protected_ranges（坐标偏移）
• target 不 strip，payload strip
• 冻结区坐标半开 [start, end)

replace_momentum(content, guidance)：guidance 含标记字面量 → ValueError（注入防护）。空 guidance 合法（清除旧动量，保留标题行）。

7 validate.py — 纯决策函数（算法 #7 局部）

三个公开函数：pair_block（逐题比对 W/L）、classify_quadrants（四组各 sorted）、compute_accuracy。

编排循环（materialize candidate → 双臂推理 → 缓存 → INFRA 护栏 → 块序贯 gate_decision）在 app/harness/。

8 diagnose.py — 诊断管线（算法 #8）

公开入口

async def run_diagnosis(
    run_id: str,
    questions: list[GeneratedQuestion],
    tree_data: dict[str, Any],      # video_id → 树 JSON
    llm: LLMProvider,
    run_log: RunLog,
    skill_store: SkillStore,
    prompts: DiagnosePrompts,
    *, concurrency: int,
    question_ids: list[str] | None = None,
    task_types: list[str] | None = None,
    only_incorrect: bool = False,
) -> DiagnosisResult:

流程

Stage 1（asyncio.gather + Semaphore，per-question）:
  7 规则指标（纯函数） + 5 LLM judge（span/missed/adherence/bias/sufficiency）
  → D1 归因瀑布 → defect/lapse 分类（LLM）
  → ValueError 降级：规则指标保留，judge 指标 None，degraded=True

独立串行 pass:
  reasoning_failure 子分类（仅对 error_type=reasoning_failure 的题）

Stage 2（纯逻辑）:
  D2 按工具聚合 → D3 按题型×正误 → D4 skill adherence → D5 跨题型行为
  → 三类案例包构建

关键保真规则

• 归因瀑布顺序：extraction(completeness<0.5∨hallucination>0.5) → search(missed_nodes) → reasoning(evidence_sufficient=True) → mixed
• defect_vs_lapse 分类：解析失败降级为 "lapse"（保护性，防错误改正文）
• single-failure fallback：某题型仅剩 1 个 defect → 降级为 lapse_note
• lapse_note 空白过滤：strip 后为空则丢弃
• SystemCasePack 触发：3 种行为模式各需 ≥ 3 次出现
• merge_system_packs stats 用 {"per_step": [...]} 包裹（不数值合并）
• trigram 相似度是字符级，取 max（非 mean）
• _call_judge 重试 3 次（仅 ValueError），API 错误直传

9 evolve.py — 进化引擎（算法 #9）

公开 API

函数	参数	返回

async def evolve_single_skill(
    llm: LLMProvider, pack: SkillCasePack,
    skill_store: SkillStore, prompts: EvolvePrompts,
    source_version: str, edit_budget: int,
    consolidate_threshold: int, *,
    skill_update_mode: Literal["patch", "rewrite"] = "patch",
    rejected: list[RejectedEdit] | None = None,
) -> EvolutionRecord: ...

async def evolve_system_prompt(
    llm: LLMProvider, pack: SystemCasePack,
    prompt_store: PromptStore, prompts: EvolvePrompts,
    source_version: str, edit_budget: int,
) -> EvolutionRecord: ...

async def evolve_single_tool(
    llm: LLMProvider, pack: ToolCasePack,
    prompt_store: PromptStore, prompts: EvolvePrompts,
    source_version: str, edit_budget: int,
) -> EvolutionRecord: ...

def edit_budget_at(
    global_step: int, total_steps: int,
    start: int, end: int,
) -> int: ...  # 纯数学

Skill 三分支

分支	条件	行为
A: Lapse-only	无 defect edits + 有 lapse_notes	合成 applied_append report，防循环误判 no-op
B: Rewrite	mode="rewrite" + 有 edits	整篇重写；失败降级：有 lapse 转 A，否则 no-op
C: Patch	默认	rank_clip → apply_patch → validate → 最多 2 轮重试

所有分支后：有 lapse_notes → appendix 追加（≥ threshold 则 consolidate）

rank_and_clip 三级降级

LLM 排序 → _select_top_edits（type(idx) is int 排除 bool + 范围 + 去重）→ 空则降级原序前 N。

Tool 共享预算池

extract+verify edits 合池（_src 标记）→ rank_clip → 按标记拆回。evolved_content 存 json.dumps({"extract":..,"verify":..})。

冻结区配置

目标	冻结区
Skill	frontmatter + appendix + momentum
System	3 个 ## section（能力边界/输出格式/视频树结构）+ appendix
Tool	输出格式 section + appendix

验证规则

检查	Skill	System	Tool
Frontmatter 三字段	✓	—	—
冻结 section 值相等	—	✓	✓
长度比 [0.3, 2.0]（去冻结区后）	✓	✓	✓ per file
代码块闭合	✓	✓	—

consolidate_appendix 四守卫

G1(<2直返) → G2(结果非空且≤输入) → G3(any Exception返原文) → G4(调用方：≥拒绝等长)

10 共享工具函数

resolve_skill_file(skills_dir, task_type) -> str（core/evolution/ 内部工具函数）：

resolve_skill_file(skill_store: SkillStore, task_type: str) -> str

task_type.lower().replace(' ', '-') + ".md"，若文件不在 skill_store.list_skill_files() 中则回退 "default-strategy.md"。diagnose（加载 skill 内容做 adherence 判定）和 evolve（定位进化目标文件）共用此约定。接受 SkillStore（非 Path），保持 core/ 不依赖文件系统。

11 TRM4 → TRM5 变更总表

项	TRM4	TRM5	理由
并发	ThreadPoolExecutor	asyncio.gather + Semaphore	TRM5 async-first
LLM	LLMClient.from_env() 每线程构造	共享 LLMProvider 注入	Protocol 化
DB	HarnessLog + raw SQL	RunLog Protocol	隔离实现
文件	Path.read_text 直读	SkillStore / PromptStore	可提取性
模板	_PROJECT_ROOT / "prompts" 硬编码	DiagnosePrompts / EvolvePrompts 束传入	零路径依赖
输出	写 JSON + DB + advance_version	纯返回 dataclass，app/ 持久化	无副作用
response 访问	response.choices[0].message.content	LLMResponse.content	已有统一类型
validate 编排	在 core/	在 app/harness/	Clean Architecture
run_evolution 编排	在 evolve.py	在 app/harness/	版本管理属 app/

12 迁移保真约束

本节列出 TRM4 中影响正确性的实现细节，实现时必须逐条比对。

12.1 JSON 解析策略差异

模块	函数	策略	失败行为
metrics.py	extract_json_from_response	三级：fenced code block → 最外层 {...} → json_repair	全失败抛 ValueError
metrics.py	_call_judge	包裹上述，max_retries=2（共 3 次），仅 ValueError 重试	API 错误直传
evolve.py	_parse_llm_json	两级：fenced code block → 原文 json.loads	失败返回 None（不抛）
metrics.py	_parse_json_object	两级：json.loads → json_repair	失败返回 None

所有解析器均拒绝非 dict 结果（list/str → 视为失败）。

12.2 关键常量

常量	值	位置	说明
_INFRA_STOP_REASONS	frozenset({"error", "parse_error"})	diagnose	INFRA 排除集
_SPAN_EVAL_TOOLS	{"view_node", "search_similar", "observe_frame"}	metrics	span judge + all_tool_outputs 范围
_MIN_PATTERN_COUNT	3	diagnose	SystemCasePack 触发阈值
_TOOL_TARGET_FILES	view_node→4 文件, search_similar→2, observe_frame→2	diagnose	工具→prompt 文件映射
truncation	thought[:100], tool_output[:200] (metrics); 不截断 (diagnose)	metrics/diagnose	_format_trace_text 两版本不同！
case_sample truncation	tool_output[:500]	evolve	_format_case_samples

12.3 案例包选择规则

包	failure 选择	success 选择
Skill	按 error_type 分组，各取 severity top-2	max(2, len(failures)//2)；acc≤0.3 按 budget 升序，否则按 (-adherence, budget)
System	3 种行为模式（early_submit/high_conf_wrong/confirmation_bias）各取 top-2	correct + calibrated + no_bias + 0.3≤budget≤0.8，按 abs(budget-0.5)
Tool	低 completeness top-2 + 高 hallucination top-2，去重，总数≤4	completeness≥0.9 且 hallucination==0.0

12.4 validate 编排守卫（app/harness/ 侧，非 core/）

• gate_run_prefix 必须含 "_gate_" 子串（防泄漏标记）
• ladder_items 空 → ValueError
• INFRA guard：累计两臂 error，分母≥10 且 error_rate > gate_guard_err → RuntimeError
• 基线缓存补齐后 assert all(v is not None)

12.5 evolve 重试与退火

• _run_patch_evolution_loop：range(2) 两轮，三种失败反馈（JSON/target 未匹配/验证错误）
• edit_budget_at：assert start >= end；total_steps ≤ 1 返 start；Python round（banker's rounding）
• rewrite_from_suggestions：重写不得长于原文；只捕 ValueError/KeyError/TypeError/AttributeError

12.6 范围说明

momentum.py 按 ARCHITECTURE.md §2.3 归属 app/harness/（非 core/evolution/），不在本设计范围内。其 LLM 调用、四类常量（IMPROVED/REGRESSED/PERSISTENT_FAIL/STABLE_SUCCESS）、_format_comparison_pairs 放在 try 外的设计意图、解析失败返回 prev_guidance 等规则将在 Design B（app/harness/）中覆盖。

13 被拒方案

方案 A（validate Protocol 回调）：给 validate 造 InferenceRunner Protocol 让编排留 core/。拒绝理由：leaky abstraction，Protocol 签名暴露 workspace/skills_dir 等外层概念，形式反转实质耦合。

方案 B（同步 + ThreadPoolExecutor）：保持 TRM4 同步。拒绝理由：TRM5 LLMProvider.chat() 已是 async，同步调用需 asyncio.run() 嵌套或线程桥接，增加复杂度。