1501 lines
52 KiB
Python
1501 lines
52 KiB
Python
"""进化引擎辅助函数与验证逻辑。
|
||
|
||
验证（validate_skill / validate_system / validate_tool）、受保护区构建、
|
||
编辑预算退火、rank-and-clip 裁剪、格式化工具等纯/准纯函数。
|
||
Task 8 将在此基础上添加进化入口（evolve_skill / evolve_system / evolve_tool）。
|
||
|
||
不依赖 app/ 或 adapters/（LLMProvider 通过 core.protocols 注入）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from dataclasses import asdict, dataclass, field
|
||
from typing import TYPE_CHECKING, Any, Literal
|
||
|
||
from loguru import logger
|
||
|
||
from core.evolution.patch import (
|
||
APPENDIX_END,
|
||
APPENDIX_START,
|
||
append_to_appendix,
|
||
apply_patch_with_report,
|
||
extract_appendix_notes,
|
||
momentum_region_bounds,
|
||
replace_appendix_notes,
|
||
)
|
||
from core.evolution.types import EvolutionRecord
|
||
|
||
if TYPE_CHECKING:
|
||
from collections.abc import Awaitable, Callable
|
||
|
||
from core.evolution.protocols import PromptStore, SkillStore
|
||
from core.evolution.types import (
|
||
EvolvePrompts,
|
||
RejectedEdit,
|
||
SkillCasePack,
|
||
SystemCasePack,
|
||
ToolCasePack,
|
||
)
|
||
from core.protocols import LLMProvider
|
||
|
||
# =========================================================================
|
||
# 0. 局部类型
|
||
# =========================================================================
|
||
|
||
|
||
@dataclass
|
||
class ValidationResult:
|
||
"""格式验证的结果。
|
||
|
||
属性:
|
||
passed: 验证是否通过。
|
||
errors: 失败原因列表；passed=True 时为空。
|
||
"""
|
||
|
||
passed: bool
|
||
errors: list[str] = field(default_factory=list)
|
||
|
||
|
||
# =========================================================================
|
||
# A. 内部辅助函数
|
||
# =========================================================================
|
||
|
||
|
||
def _parse_frontmatter(text: str) -> dict[str, str] | None:
|
||
"""解析 YAML frontmatter，失败返回 None。
|
||
|
||
参数:
|
||
text: Markdown 文件全文。
|
||
|
||
返回:
|
||
frontmatter 字典，无有效 frontmatter 时返回 None。
|
||
"""
|
||
import yaml as _yaml
|
||
|
||
match = re.match(r"^---\n(.*?)\n---", text, re.DOTALL)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return _yaml.safe_load(match.group(1))
|
||
except _yaml.YAMLError:
|
||
return None
|
||
|
||
|
||
def _strip_appendix_region(text: str) -> str:
|
||
"""剥离 appendix 受保护区（含 marker），返回其余正文。
|
||
|
||
宽松语义：APPENDIX_START / APPENDIX_END 任一缺失即当作「无区」原样返回，不报错。
|
||
"""
|
||
if APPENDIX_START in text and APPENDIX_END in text:
|
||
head, rest = text.split(APPENDIX_START, 1)
|
||
_, tail = rest.split(APPENDIX_END, 1)
|
||
return head.rstrip() + tail
|
||
return text
|
||
|
||
|
||
def _strip_momentum_region(text: str) -> str:
|
||
"""剥离 momentum 受保护区（含 marker），返回其余正文。
|
||
|
||
严格语义：委托 momentum_region_bounds 做配对检测，
|
||
marker 损坏/不配对时 raise ValueError。
|
||
|
||
异常:
|
||
ValueError: momentum marker 损坏/不配对。
|
||
"""
|
||
bounds = momentum_region_bounds(text)
|
||
if bounds is None:
|
||
return text
|
||
start, end = bounds
|
||
return text[:start].rstrip() + text[end:]
|
||
|
||
|
||
def _strip_protected_regions(text: str) -> str:
|
||
"""剥离 appendix + momentum 两个受保护区，返回正文部分。
|
||
|
||
先 appendix（宽松），再 momentum（严格）——顺序有关：
|
||
appendix 宽松剥离不会误杀 momentum marker。
|
||
|
||
异常:
|
||
ValueError: momentum marker 损坏/不配对。
|
||
"""
|
||
text = _strip_appendix_region(text)
|
||
text = _strip_momentum_region(text)
|
||
return text
|
||
|
||
|
||
def _check_length(original: str, evolved: str) -> list[str]:
|
||
"""检查改写后长度是否在 [0.3x, 2.0x] 范围内。
|
||
|
||
仅比正文，剔除 appendix + momentum 两区。orig_len==0 时跳过。
|
||
|
||
参数:
|
||
original: 改写前全文。
|
||
evolved: 改写后全文。
|
||
|
||
返回:
|
||
错误消息列表（空列表表示通过）。
|
||
"""
|
||
errors: list[str] = []
|
||
orig_body = _strip_protected_regions(original)
|
||
evol_body = _strip_protected_regions(evolved)
|
||
orig_len = len(orig_body)
|
||
if orig_len == 0:
|
||
return errors
|
||
ratio = len(evol_body) / orig_len
|
||
evol_len = len(evol_body)
|
||
if ratio > 2.0:
|
||
errors.append(f"长度超限: {evol_len} 字符是原文 {orig_len} 的 {ratio:.1f} 倍 (上限 2.0)")
|
||
if ratio < 0.3:
|
||
errors.append(f"长度不足: {evol_len} 字符是原文 {orig_len} 的 {ratio:.1f} 倍 (下限 0.3)")
|
||
return errors
|
||
|
||
|
||
def _check_code_blocks(text: str) -> list[str]:
|
||
"""检查代码块是否闭合。
|
||
|
||
参数:
|
||
text: 待检查的文本。
|
||
|
||
返回:
|
||
错误消息列表（空列表表示通过）。
|
||
"""
|
||
count = text.count("```")
|
||
if count % 2 != 0:
|
||
return [f"Markdown 格式错误: 代码块未闭合 (``` 出现 {count} 次)"]
|
||
return []
|
||
|
||
|
||
def _extract_section(text: str, heading: str) -> str | None:
|
||
"""提取 ## heading 到下一个 ## 之间的文本。
|
||
|
||
参数:
|
||
text: Markdown 全文。
|
||
heading: 二级标题名。
|
||
|
||
返回:
|
||
该 section 的完整文本（含标题行），未找到时返回 None。
|
||
"""
|
||
pattern = rf"(## {re.escape(heading)}.*?)(?=\n## |\Z)"
|
||
match = re.search(pattern, text, re.DOTALL)
|
||
return match.group(1).strip() if match else None
|
||
|
||
|
||
# =========================================================================
|
||
# B. 受保护区构建
|
||
# =========================================================================
|
||
|
||
|
||
def _appendix_span(content: str) -> str:
|
||
"""返回 appendix 受保护区整段（含 marker）；不存在返回空串。
|
||
|
||
参数:
|
||
content: 文本全文。
|
||
|
||
返回:
|
||
appendix 区的完整文本（含 marker），或空串。
|
||
"""
|
||
if APPENDIX_START in content and APPENDIX_END in content:
|
||
start = content.index(APPENDIX_START)
|
||
end = content.index(APPENDIX_END) + len(APPENDIX_END)
|
||
return content[start:end]
|
||
return ""
|
||
|
||
|
||
def _momentum_span(content: str) -> str:
|
||
"""返回 momentum 受保护区整段（含 marker）；不存在返回空串。
|
||
|
||
委托 momentum_region_bounds 做配对检测：
|
||
marker 损坏/不配对时由其 raise ValueError。
|
||
|
||
参数:
|
||
content: 文本全文。
|
||
|
||
返回:
|
||
momentum 区的完整文本（含 marker），或空串。
|
||
|
||
异常:
|
||
ValueError: momentum marker 损坏/不配对。
|
||
"""
|
||
bounds = momentum_region_bounds(content)
|
||
if bounds is None:
|
||
return ""
|
||
start, end = bounds
|
||
return content[start:end]
|
||
|
||
|
||
def _skill_protected_spans(text: str) -> list[str]:
|
||
"""Skill 冻结块：frontmatter + appendix 区 + momentum 区（各项可选）。
|
||
|
||
参数:
|
||
text: Skill 文件全文。
|
||
|
||
返回:
|
||
冻结文本块列表。
|
||
"""
|
||
spans: list[str] = []
|
||
match = re.match(r"^---\n.*?\n---", text, re.DOTALL)
|
||
if match:
|
||
spans.append(match.group(0))
|
||
appendix = _appendix_span(text)
|
||
if appendix:
|
||
spans.append(appendix)
|
||
momentum = _momentum_span(text)
|
||
if momentum:
|
||
spans.append(momentum)
|
||
return spans
|
||
|
||
|
||
def _system_protected_spans(text: str) -> list[str]:
|
||
"""System Prompt 冻结块：能力边界 / 输出格式 / 视频树结构 三段 + appendix 区。
|
||
|
||
参数:
|
||
text: system.md 全文。
|
||
|
||
返回:
|
||
冻结文本块列表。
|
||
"""
|
||
spans: list[str] = [
|
||
section
|
||
for section in (
|
||
_extract_section(text, name) for name in ("能力边界", "输出格式", "视频树结构")
|
||
)
|
||
if section
|
||
]
|
||
appendix = _appendix_span(text)
|
||
if appendix:
|
||
spans.append(appendix)
|
||
return spans
|
||
|
||
|
||
def _tool_protected_spans(text: str) -> list[str]:
|
||
"""Tool Prompt 冻结块：输出格式段 + appendix 区。
|
||
|
||
参数:
|
||
text: Tool Prompt 全文。
|
||
|
||
返回:
|
||
冻结文本块列表。
|
||
"""
|
||
spans: list[str] = []
|
||
section = _extract_section(text, "输出格式")
|
||
if section:
|
||
spans.append(section)
|
||
appendix = _appendix_span(text)
|
||
if appendix:
|
||
spans.append(appendix)
|
||
return spans
|
||
|
||
|
||
# =========================================================================
|
||
# C. 验证函数
|
||
# =========================================================================
|
||
|
||
|
||
def validate_skill(original: str, evolved: str) -> ValidationResult:
|
||
"""校验 Skill 改写结果。
|
||
|
||
检查项: frontmatter 三字段保留（name / description / task_type）、
|
||
长度比在 [0.3, 2.0]、代码块闭合。
|
||
|
||
参数:
|
||
original: 改写前的 Skill 文件全文。
|
||
evolved: 改写后的 Skill 文件全文。
|
||
|
||
返回:
|
||
ValidationResult 实例。
|
||
"""
|
||
errors: list[str] = []
|
||
orig_fm = _parse_frontmatter(original)
|
||
evol_fm = _parse_frontmatter(evolved)
|
||
if orig_fm is None:
|
||
errors.append("原文缺少有效 frontmatter")
|
||
elif evol_fm is None:
|
||
errors.append("改写后缺少有效 frontmatter")
|
||
else:
|
||
for key in ("name", "description", "task_type"):
|
||
if orig_fm.get(key) != evol_fm.get(key):
|
||
errors.append(
|
||
f"frontmatter 字段 {key} 被修改: {orig_fm.get(key)!r} → {evol_fm.get(key)!r}"
|
||
)
|
||
errors.extend(_check_length(original, evolved))
|
||
errors.extend(_check_code_blocks(evolved))
|
||
return ValidationResult(passed=len(errors) == 0, errors=errors)
|
||
|
||
|
||
def validate_system(original: str, evolved: str) -> ValidationResult:
|
||
"""校验 System Prompt 改写结果。
|
||
|
||
检查项: 三个冻结区值比较（能力边界 / 输出格式 / 视频树结构）、
|
||
长度比在 [0.3, 2.0]、代码块闭合。
|
||
|
||
参数:
|
||
original: 改写前的 system.md 全文。
|
||
evolved: 改写后的 system.md 全文。
|
||
|
||
返回:
|
||
ValidationResult 实例。
|
||
"""
|
||
errors: list[str] = []
|
||
frozen_sections = ["能力边界", "输出格式", "视频树结构"]
|
||
for section_name in frozen_sections:
|
||
orig_section = _extract_section(original, section_name)
|
||
if orig_section is None:
|
||
continue
|
||
evol_section = _extract_section(evolved, section_name)
|
||
if evol_section is None:
|
||
errors.append(f"冻结区 '## {section_name}' 在改写后缺失")
|
||
elif orig_section != evol_section:
|
||
errors.append(f"冻结区 '## {section_name}' 在改写后被修改")
|
||
errors.extend(_check_length(original, evolved))
|
||
errors.extend(_check_code_blocks(evolved))
|
||
return ValidationResult(passed=len(errors) == 0, errors=errors)
|
||
|
||
|
||
def validate_tool(
|
||
orig_extract: str,
|
||
evol_extract: str,
|
||
orig_verify: str,
|
||
evol_verify: str,
|
||
) -> ValidationResult:
|
||
"""校验 Tool Prompt 改写结果。
|
||
|
||
检查项: 输出格式 section 保留（per file）、长度比在 [0.3, 2.0]。
|
||
与 skill / system 不同，**不检查代码块闭合**。
|
||
|
||
参数:
|
||
orig_extract: 改写前的 extract prompt。
|
||
evol_extract: 改写后的 extract prompt。
|
||
orig_verify: 改写前的 verify prompt。
|
||
evol_verify: 改写后的 verify prompt。
|
||
|
||
返回:
|
||
ValidationResult 实例。
|
||
"""
|
||
errors: list[str] = []
|
||
for label, orig, evol in [
|
||
("extract", orig_extract, evol_extract),
|
||
("verify", orig_verify, evol_verify),
|
||
]:
|
||
orig_fmt = _extract_section(orig, "输出格式")
|
||
if orig_fmt is not None:
|
||
evol_fmt = _extract_section(evol, "输出格式")
|
||
if evol_fmt is None:
|
||
errors.append(f"{label}: 冻结区 '## 输出格式' 在改写后缺失")
|
||
elif orig_fmt != evol_fmt:
|
||
errors.append(f"{label}: 冻结区 '## 输出格式' 在改写后被修改")
|
||
errors.extend(_check_length(orig, evol))
|
||
return ValidationResult(passed=len(errors) == 0, errors=errors)
|
||
|
||
|
||
# =========================================================================
|
||
# D. 纯数学
|
||
# =========================================================================
|
||
|
||
|
||
def edit_budget_at(global_step: int, total_steps: int, start: int, end: int) -> int:
|
||
"""按 global_step 线性退火的 per-target 编辑预算。
|
||
|
||
借鉴 SkillOpt LinearScheduler：在 [0, total_steps] 上把预算从 start
|
||
线性退火到 end。total_steps<=1 直接返回 start（避免单步取到最小值）。
|
||
round 用 Python banker's rounding，max(end, ...) 兜底硬下限。
|
||
|
||
参数:
|
||
global_step: 当前全局步（0-indexed）。
|
||
total_steps: 退火地平线（step 数），即分母。
|
||
start: 退火起点（需 >= end）。
|
||
end: 退火终点（亦为硬下限）。
|
||
|
||
返回:
|
||
当步 per-target 最大 edit 条数。
|
||
|
||
异常:
|
||
AssertionError: start < end。
|
||
"""
|
||
assert start >= end, f"edit_budget_at 要求 start >= end，实际 start={start}, end={end}"
|
||
if total_steps <= 1:
|
||
return start
|
||
t = min(global_step, total_steps) / total_steps
|
||
return max(end, round(start + (end - start) * t))
|
||
|
||
|
||
# =========================================================================
|
||
# E. JSON 解析（进化版本，不同于 metrics 版）
|
||
# =========================================================================
|
||
|
||
|
||
def _parse_llm_json(raw: str) -> dict | None:
|
||
"""从 LLM 响应中解析 JSON。
|
||
|
||
仅两种策略：(1) 提取 ```json 代码块；(2) 直接 json.loads。
|
||
不做 outermost braces 推断、不用 json_repair。失败返回 None。
|
||
|
||
参数:
|
||
raw: LLM 原始输出文本。
|
||
|
||
返回:
|
||
解析后的字典；失败或结果非 dict 时返回 None。
|
||
"""
|
||
text = raw.strip()
|
||
# 策略 1：提取 ```json ... ``` 代码块
|
||
code_block = re.search(r"```json\s*\n(.*?)```", text, re.DOTALL)
|
||
if code_block:
|
||
text = code_block.group(1).strip()
|
||
# 策略 2：直接解析
|
||
try:
|
||
result = json.loads(text)
|
||
if isinstance(result, dict):
|
||
return result
|
||
return None
|
||
except (json.JSONDecodeError, ValueError):
|
||
return None
|
||
|
||
|
||
# =========================================================================
|
||
# F. rank_and_clip（async）
|
||
# =========================================================================
|
||
|
||
|
||
def _select_top_edits(
|
||
indices: list[Any],
|
||
edits: list[dict[str, Any]],
|
||
max_edits: int,
|
||
) -> list[dict[str, Any]]:
|
||
"""按 rank LLM 给出的优先级索引筛选 edits。
|
||
|
||
依次保留首个 max_edits 条合法、不重复、在范围内的索引对应 edit。
|
||
用 type(idx) is int（非 isinstance）以排除 bool。
|
||
|
||
参数:
|
||
indices: rank LLM 返回的 0-based 优先级索引。
|
||
edits: 候选 edit 列表。
|
||
max_edits: 最多保留条数。
|
||
|
||
返回:
|
||
按优先级顺序保留的 edit 列表。
|
||
"""
|
||
selected: list[dict[str, Any]] = []
|
||
seen: set[int] = set()
|
||
for idx in indices:
|
||
if type(idx) is int and 0 <= idx < len(edits) and idx not in seen:
|
||
selected.append(edits[idx])
|
||
seen.add(idx)
|
||
if len(selected) >= max_edits:
|
||
break
|
||
return selected
|
||
|
||
|
||
async def _request_rank_indices(
|
||
llm: LLMProvider,
|
||
prompts: str,
|
||
original: str,
|
||
edits: list[dict[str, Any]],
|
||
max_edits: int,
|
||
label: str,
|
||
) -> list[int]:
|
||
"""调 rank LLM 取重要性降序的索引列表。
|
||
|
||
守 P5：响应无法解析或 selected_indices 非列表时直接 raise ValueError。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
prompts: evolve_rank 模板内容。
|
||
original: 当前 prompt 全文（排序上下文）。
|
||
edits: 候选 edit 列表。
|
||
max_edits: 本轮预算上限。
|
||
label: 目标标签（仅用于报错信息）。
|
||
|
||
返回:
|
||
rank LLM 返回的原始索引列表（尚未去重/越界过滤）。
|
||
|
||
异常:
|
||
ValueError: 响应无 selected_indices，或其值非列表。
|
||
"""
|
||
edits_desc = "\n".join(
|
||
f"[{i}] op={e.get('op')} support_count={e.get('support_count', 0)} "
|
||
f"target={str(e.get('target', ''))[:60]!r} "
|
||
f"content={str(e.get('content', ''))[:60]!r}"
|
||
for i, e in enumerate(edits)
|
||
)
|
||
user_msg = (
|
||
f"## 当前文件\n\n{original}\n\n"
|
||
f"## 候选 edits（{len(edits)} 条，预算 {max_edits} 条）\n\n{edits_desc}\n\n"
|
||
f"请选出最重要的 {max_edits} 条，返回其 0-based 索引（重要性降序）。"
|
||
)
|
||
response = await llm.chat(
|
||
[
|
||
{"role": "system", "content": prompts},
|
||
{"role": "user", "content": user_msg},
|
||
]
|
||
)
|
||
parsed = _parse_llm_json(response.content)
|
||
if not parsed or "selected_indices" not in parsed:
|
||
raise ValueError(f"{label} rank LLM 未返回 selected_indices，拒绝静默截断")
|
||
indices = parsed["selected_indices"]
|
||
if not isinstance(indices, list):
|
||
raise ValueError(f"{label} rank LLM selected_indices 非列表")
|
||
return indices
|
||
|
||
|
||
async def rank_and_clip(
|
||
llm: LLMProvider,
|
||
original_content: str,
|
||
edits: list[dict[str, Any]],
|
||
max_edits: int,
|
||
label: str,
|
||
*,
|
||
rank_prompt: str = "",
|
||
) -> tuple[list[dict[str, Any]], dict[str, Any]]:
|
||
"""超预算时调 rank LLM 排序取 top-L；未超则原样返回。
|
||
|
||
三级降级：LLM rank → _select_top_edits → empty → fallback to edits[:max_edits]。
|
||
对 rank LLM 的输出波动一律优雅降级而非中止。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
original_content: 当前 prompt 全文（排序上下文）。
|
||
edits: 候选 edit 列表。
|
||
max_edits: 本轮预算上限。
|
||
label: 目标标签（skill/system/tool，仅用于日志）。
|
||
rank_prompt: evolve_rank 模板内容。
|
||
|
||
返回:
|
||
(裁剪后 edits, {"triggered": bool, "clipped": int})。
|
||
"""
|
||
if len(edits) <= max_edits:
|
||
return edits, {"triggered": False, "clipped": 0}
|
||
|
||
try:
|
||
indices = await _request_rank_indices(
|
||
llm, rank_prompt, original_content, edits, max_edits, label
|
||
)
|
||
except Exception as exc:
|
||
logger.warning(
|
||
"{} rank LLM 排序不可用（{}）；退化为按原序取前 {} 条",
|
||
label,
|
||
exc,
|
||
max_edits,
|
||
)
|
||
indices = []
|
||
|
||
selected = _select_top_edits(indices, edits, max_edits)
|
||
if not selected:
|
||
selected = edits[:max_edits]
|
||
logger.warning("{} rank 有效索引为 0；退化为按原序取前 {} 条", label, max_edits)
|
||
elif len(selected) < max_edits:
|
||
logger.warning(
|
||
"{} rank 仅得 {} 条有效(<预算 {})；按更保守的条数应用",
|
||
label,
|
||
len(selected),
|
||
max_edits,
|
||
)
|
||
logger.info("{} edits 超预算裁剪 {}->{}", label, len(edits), len(selected))
|
||
return selected, {"triggered": True, "clipped": len(edits) - len(selected)}
|
||
|
||
|
||
# =========================================================================
|
||
# G. resolve_skill_file
|
||
# =========================================================================
|
||
|
||
|
||
def resolve_skill_file(skill_store: SkillStore, task_type: str) -> str:
|
||
"""按运行时规则解析 task_type 对应的 skill 文件名。
|
||
|
||
转换规则：小写 + 空格替换为短横线 + .md 后缀。
|
||
若 store 中不存在匹配文件，退化到 default-strategy.md。
|
||
|
||
参数:
|
||
skill_store: 版本化技能读取端口。
|
||
task_type: 题目任务类型（如 "Action Reasoning"）。
|
||
|
||
返回:
|
||
匹配的 skill 文件名。
|
||
"""
|
||
file_name = f"{task_type.lower().replace(' ', '-')}.md"
|
||
available = skill_store.list_skill_files()
|
||
if file_name in available:
|
||
return file_name
|
||
return "default-strategy.md"
|
||
|
||
|
||
# =========================================================================
|
||
# H. 格式化辅助
|
||
# =========================================================================
|
||
|
||
|
||
def _format_case_samples(cases: list[Any]) -> str:
|
||
"""将 CaseSample 列表格式化为 LLM 可读文本。
|
||
|
||
对 trace 中的 tool_output 截断到 500 字符。
|
||
|
||
参数:
|
||
cases: CaseSample 实例列表（也兼容 dict）。
|
||
|
||
返回:
|
||
格式化后的多行文本。
|
||
"""
|
||
lines: list[str] = []
|
||
for case in cases:
|
||
if not isinstance(case, dict):
|
||
case = asdict(case)
|
||
lines.append(f"### {case.get('question_id', 'unknown')}")
|
||
lines.append(f"- question: {case.get('question', '')}")
|
||
options = case.get("options", [])
|
||
if options:
|
||
lines.append(f"- options: {json.dumps(options, ensure_ascii=False)}")
|
||
lines.append(f"- answer: {case.get('answer', '')}")
|
||
lines.append(f"- prediction: {case.get('prediction', '')}")
|
||
lines.append(f"- error_type: {case.get('error_type', '')}")
|
||
lines.append(f"- selection_reason: {case.get('selection_reason', '')}")
|
||
trace = case.get("trace", [])
|
||
if trace:
|
||
lines.append("- trace:")
|
||
for step in trace:
|
||
output_text = str(step.get("tool_output", ""))
|
||
if len(output_text) > 500:
|
||
output_text = output_text[:500] + "..."
|
||
lines.append(
|
||
f" - step {step.get('step', '?')}: "
|
||
f"tool={step.get('tool_name', '')} "
|
||
f"args={json.dumps(step.get('tool_args', {}), ensure_ascii=False)} "
|
||
f"output={output_text}"
|
||
)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _format_spans(spans: list[dict[str, Any]]) -> str:
|
||
"""将工具 span 字典列表格式化为 LLM 可读文本。
|
||
|
||
对 tool_output 截断到 500 字符。
|
||
|
||
参数:
|
||
spans: span 字典列表，每个包含 step / tool_name / tool_args 等字段。
|
||
|
||
返回:
|
||
格式化后的多行文本。
|
||
"""
|
||
lines: list[str] = []
|
||
for span in spans:
|
||
lines.append(f"### step {span.get('step', '?')}")
|
||
lines.append(f"- tool_name: {span.get('tool_name', '')}")
|
||
lines.append(f"- tool_args: {json.dumps(span.get('tool_args', {}), ensure_ascii=False)}")
|
||
output_text = str(span.get("tool_output", ""))
|
||
if len(output_text) > 500:
|
||
output_text = output_text[:500] + "..."
|
||
lines.append(f"- tool_output: {output_text}")
|
||
lines.append(f"- extraction_completeness: {span.get('extraction_completeness', '')}")
|
||
lines.append(f"- hallucination_rate: {span.get('hallucination_rate', '')}")
|
||
missed = span.get("missed_info_tags", [])
|
||
if missed:
|
||
lines.append(f"- missed_info_tags: {json.dumps(missed, ensure_ascii=False)}")
|
||
hall_tags = span.get("hallucination_tags", [])
|
||
if hall_tags:
|
||
lines.append(f"- hallucination_tags: {json.dumps(hall_tags, ensure_ascii=False)}")
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def _format_rejected_edits(rejected: list[RejectedEdit]) -> str:
|
||
"""将已验证无效的改法列表格式化为 LLM 可读文本。
|
||
|
||
gate 证据格式：W=... L=... E={:.2f} delta_hat={:+.3f}。
|
||
|
||
参数:
|
||
rejected: RejectedEdit 实例列表。
|
||
|
||
返回:
|
||
格式化后的多行文本。
|
||
"""
|
||
lines: list[str] = []
|
||
for edit in rejected:
|
||
lines.append(f"### {edit.target_file} | delta {edit.delta:+.2f}")
|
||
lines.append(f"- 已验证无效的改法: {edit.change_summary}")
|
||
if edit.gate_e_value is not None:
|
||
lines.append(
|
||
f"- 已验证无效: W={edit.gate_w} L={edit.gate_l} "
|
||
f"E={edit.gate_e_value:.2f} δ̂={edit.gate_delta_shrunk:+.3f}"
|
||
)
|
||
lines.append("")
|
||
return "\n".join(lines)
|
||
|
||
|
||
# =========================================================================
|
||
# I. 进化循环内部类型
|
||
# =========================================================================
|
||
|
||
|
||
@dataclass
|
||
class _PatchEvolutionAttempt:
|
||
"""单次补丁式进化尝试的中间结果。
|
||
|
||
属性:
|
||
evolved_content: 改写后内容。
|
||
validation: 校验结果。
|
||
suggestions: LLM 输出的改动建议列表。
|
||
edits: LLM 输出的补丁列表。
|
||
apply_report: 补丁逐条应用状态。
|
||
clip_info: 超预算裁剪信息。
|
||
"""
|
||
|
||
evolved_content: str
|
||
validation: ValidationResult
|
||
suggestions: list[dict[str, Any]] = field(default_factory=list)
|
||
edits: list[dict[str, Any]] = field(default_factory=list)
|
||
apply_report: list[dict[str, Any]] = field(default_factory=list)
|
||
clip_info: dict[str, Any] = field(default_factory=lambda: {"triggered": False, "clipped": 0})
|
||
|
||
|
||
# =========================================================================
|
||
# J. 进化循环辅助函数
|
||
# =========================================================================
|
||
|
||
|
||
def _count_applied_reports(reports: list[dict[str, Any]]) -> int:
|
||
"""统计补丁报告中成功应用的条数。
|
||
|
||
以 ``status`` 前缀 ``"applied"`` 为判据，涵盖 applied_append /
|
||
applied_replace / applied_rewrite 等所有成功状态。
|
||
|
||
参数:
|
||
reports: apply_patch_with_report 或合成报告的列表。
|
||
|
||
返回:
|
||
成功应用的条数。
|
||
"""
|
||
return sum(1 for r in reports if r["status"].startswith("applied"))
|
||
|
||
|
||
def _with_report_source(reports: list[dict[str, Any]], source: str) -> list[dict[str, Any]]:
|
||
"""给补丁报告补上来源字段（extract / verify 标注）。
|
||
|
||
参数:
|
||
reports: 原始补丁报告列表。
|
||
source: 来源标签（"extract" / "verify"）。
|
||
|
||
返回:
|
||
每条追加 ``"source"`` 字段的新列表。
|
||
"""
|
||
return [{**r, "source": source} for r in reports]
|
||
|
||
|
||
def _append_retry_messages(
|
||
messages: list[dict[str, Any]],
|
||
raw_content: str,
|
||
feedback: str,
|
||
) -> None:
|
||
"""向对话中追加一次失败后的重试反馈（assistant + user）。
|
||
|
||
参数:
|
||
messages: 当前对话消息列表（原地修改）。
|
||
raw_content: 上一轮 LLM 原始输出。
|
||
feedback: 给 LLM 的纠正反馈。
|
||
"""
|
||
messages.append({"role": "assistant", "content": raw_content})
|
||
messages.append({"role": "user", "content": feedback})
|
||
|
||
|
||
# =========================================================================
|
||
# K. 补丁进化循环
|
||
# =========================================================================
|
||
|
||
|
||
async def _run_patch_evolution_loop(
|
||
*,
|
||
llm: LLMProvider,
|
||
messages: list[dict[str, Any]],
|
||
attempts: list[dict[str, Any]],
|
||
target_file: str,
|
||
target_type: str,
|
||
original_content: str,
|
||
source_version: str,
|
||
log_target: str,
|
||
attempt_builder: Callable[[dict[str, Any]], Awaitable[_PatchEvolutionAttempt]],
|
||
) -> EvolutionRecord:
|
||
"""执行带补丁应用与 no-op 重试的两轮进化循环。
|
||
|
||
恰好 2 次尝试（range(2)），三种失败模式各有对应重试提示：
|
||
1. JSON 解析失败 → "你的输出不是合法 JSON，请重新输出"
|
||
2. 0 条 applied 补丁 → "你的 edit 的 target 都没在原文中匹配到…"
|
||
3. 校验失败 → 具体校验错误文本
|
||
|
||
首轮失败追加重试提示继续第二轮；第二轮失败直接 reject。
|
||
校验通过立即返回 accepted。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
messages: 对话消息列表（原地修改，追加重试上下文）。
|
||
attempts: 尝试摘要列表（原地追加）。
|
||
target_file: 目标文件名。
|
||
target_type: 目标类型（skill / system / tool）。
|
||
original_content: 改写前原文。
|
||
source_version: 改写前版本号。
|
||
log_target: 日志标签。
|
||
attempt_builder: 异步构建尝试的回调，接收 parsed JSON dict。
|
||
|
||
返回:
|
||
EvolutionRecord 实例。
|
||
"""
|
||
for attempt_idx in range(2):
|
||
response = await llm.chat(messages)
|
||
raw_content = response.content
|
||
attempts.append({"attempt": attempt_idx + 1, "raw_length": len(raw_content)})
|
||
parsed = _parse_llm_json(raw_content)
|
||
|
||
# 失败模式 1：JSON 解析失败
|
||
if parsed is None:
|
||
logger.warning("{} 进化 LLM 响应 JSON 解析失败: {}", target_type, log_target)
|
||
if attempt_idx == 0:
|
||
_append_retry_messages(
|
||
messages,
|
||
raw_content,
|
||
"你的输出不是合法 JSON，请重新输出。",
|
||
)
|
||
continue
|
||
return EvolutionRecord(
|
||
target_file=target_file,
|
||
target_type=target_type,
|
||
original_content=original_content,
|
||
evolved_content=original_content,
|
||
reason="LLM 响应 JSON 解析失败",
|
||
status="rejected",
|
||
source_version=source_version,
|
||
attempts=attempts,
|
||
validation_errors=["JSON 解析失败"],
|
||
)
|
||
|
||
attempt = await attempt_builder(parsed)
|
||
|
||
# 失败模式 2：0 条 applied 补丁
|
||
if _count_applied_reports(attempt.apply_report) == 0:
|
||
if attempt_idx == 0:
|
||
_append_retry_messages(
|
||
messages,
|
||
raw_content,
|
||
"你的 edit 的 target 都没在原文中匹配到，请逐字摘抄原文锚点后重输。",
|
||
)
|
||
continue
|
||
return EvolutionRecord(
|
||
target_file=target_file,
|
||
target_type=target_type,
|
||
original_content=original_content,
|
||
evolved_content=original_content,
|
||
reason="补丁无有效改动（target 全未匹配）",
|
||
status="rejected",
|
||
source_version=source_version,
|
||
suggestions=attempt.suggestions,
|
||
attempts=attempts,
|
||
edits=attempt.edits,
|
||
apply_report=attempt.apply_report,
|
||
clip_info=attempt.clip_info,
|
||
)
|
||
|
||
# 成功：校验通过
|
||
if attempt.validation.passed:
|
||
return EvolutionRecord(
|
||
target_file=target_file,
|
||
target_type=target_type,
|
||
original_content=original_content,
|
||
evolved_content=attempt.evolved_content,
|
||
reason="验证通过",
|
||
status="accepted",
|
||
source_version=source_version,
|
||
suggestions=attempt.suggestions,
|
||
attempts=attempts,
|
||
edits=attempt.edits,
|
||
apply_report=attempt.apply_report,
|
||
clip_info=attempt.clip_info,
|
||
)
|
||
|
||
# 失败模式 3：校验失败
|
||
error_feedback = "\n".join(attempt.validation.errors)
|
||
if attempt_idx == 0:
|
||
_append_retry_messages(
|
||
messages,
|
||
raw_content,
|
||
f"验证失败，请修正后重新输出：\n{error_feedback}",
|
||
)
|
||
continue
|
||
return EvolutionRecord(
|
||
target_file=target_file,
|
||
target_type=target_type,
|
||
original_content=original_content,
|
||
evolved_content=original_content,
|
||
reason="验证失败（重试后仍未通过）",
|
||
status="rejected",
|
||
source_version=source_version,
|
||
suggestions=attempt.suggestions,
|
||
attempts=attempts,
|
||
validation_errors=attempt.validation.errors,
|
||
edits=attempt.edits,
|
||
apply_report=attempt.apply_report,
|
||
clip_info=attempt.clip_info,
|
||
)
|
||
|
||
# 兜底（正常流程不可达）
|
||
return EvolutionRecord(
|
||
target_file=target_file,
|
||
target_type=target_type,
|
||
original_content=original_content,
|
||
evolved_content=original_content,
|
||
reason="未知错误",
|
||
status="rejected",
|
||
source_version=source_version,
|
||
attempts=attempts,
|
||
)
|
||
|
||
|
||
# =========================================================================
|
||
# L. Lapse-only 尝试构建
|
||
# =========================================================================
|
||
|
||
|
||
def _build_lapse_only_attempt(
|
||
original_content: str,
|
||
lapse_notes: list[str],
|
||
) -> _PatchEvolutionAttempt:
|
||
"""构造「仅 appendix 更新」尝试：无 defect edit，只把 lapse 提醒落进受保护区。
|
||
|
||
LLM 没给 defect edit 但有 lapse 提醒时，正常补丁路径会因 0 条 applied 报告被
|
||
``_run_patch_evolution_loop`` 判为 no-op 而丢弃。这里给出一条 ``applied_append``
|
||
合成报告（前缀 ``applied`` 使 ``_count_applied_reports > 0``），令该记录走
|
||
accepted 路径、appendix 真正落盘。
|
||
|
||
参数:
|
||
original_content: 改写前全文。
|
||
lapse_notes: 待落 appendix 的 LAPSE 提醒。
|
||
|
||
返回:
|
||
含 appendix 更新内容与合成 apply_report 的 _PatchEvolutionAttempt。
|
||
"""
|
||
evolved_content = append_to_appendix(original_content, lapse_notes)
|
||
apply_report = [
|
||
{
|
||
"op": "append",
|
||
"target": "",
|
||
"content_preview": "appendix LAPSE 提醒",
|
||
"status": "applied_append",
|
||
"index": 1,
|
||
}
|
||
]
|
||
return _PatchEvolutionAttempt(
|
||
evolved_content=evolved_content,
|
||
validation=validate_skill(original_content, evolved_content),
|
||
suggestions=[],
|
||
edits=[],
|
||
apply_report=apply_report,
|
||
clip_info={"triggered": False, "clipped": 0},
|
||
)
|
||
|
||
|
||
# =========================================================================
|
||
# M. Appendix consolidation
|
||
# =========================================================================
|
||
|
||
|
||
_CONSOLIDATE_SYSTEM = (
|
||
"你在压缩一个 agent skill 的「执行提醒 appendix」。每条提醒都重申一条 skill 已有"
|
||
"规则、是 agent 没遵循的点。你的任务是周期性压缩：去重、合并近义、精简措辞，但"
|
||
"保留每条的可执行性。禁止发明新规则；禁止写入任何具体题目/选项/实体名等案例事实。"
|
||
"只返回 JSON。"
|
||
)
|
||
|
||
|
||
async def consolidate_appendix(llm: LLMProvider, notes: list[str]) -> list[str]:
|
||
"""LLM 压缩 appendix notes（去重/合并/精简），失败永不丢内容。
|
||
|
||
四关守卫（对标 TRM4 consolidate_appendix）：
|
||
G1. clean 后 <2 条直接短路返回（无需压缩，不调 LLM）。
|
||
G2. 只接受「非空且 len(compacted) <= len(clean)」的压缩结果。
|
||
G3. 任何异常（解析/空/网络）→ 返回 clean（绝不丢 appendix）。
|
||
G4. 在调用方 _append_lapse_with_consolidation 中：
|
||
len(compacted) >= len(notes) → 拒绝等长压缩。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
notes: 待压缩的 appendix 提醒列表。
|
||
|
||
返回:
|
||
压缩后的提醒列表；任何守卫未通过时返回 clean 后的原 notes。
|
||
"""
|
||
# G1：clean 后不足 2 条，直接返回
|
||
clean = [str(n).strip() for n in (notes or []) if str(n).strip()]
|
||
if len(clean) < 2:
|
||
return clean
|
||
|
||
numbered = "\n".join(f"{i}. {n}" for i, n in enumerate(clean, 1))
|
||
user = (
|
||
f"## 当前执行提醒（共 {len(clean)} 条）\n{numbered}\n\n"
|
||
"压缩为更短的列表，不丢失可执行信息；合并重复与近义；保持每条简短具体可复用。"
|
||
'只返回 JSON：{ "appendix_notes": ["压缩后提醒1", "压缩后提醒2"] }'
|
||
)
|
||
try:
|
||
response = await llm.chat(
|
||
[
|
||
{"role": "system", "content": _CONSOLIDATE_SYSTEM},
|
||
{"role": "user", "content": user},
|
||
]
|
||
)
|
||
parsed = _parse_llm_json(response.content)
|
||
compacted = [
|
||
str(n).strip() for n in (parsed or {}).get("appendix_notes", []) if str(n).strip()
|
||
]
|
||
# G2：非空且确实压缩了
|
||
if compacted and len(compacted) <= len(clean):
|
||
return compacted
|
||
except Exception as exc: # noqa: BLE001
|
||
# G3：任何失败降级为保留原 notes。设计授权的优雅降级（非 P5 违规）：
|
||
# consolidation 是纯优化，失败不应中断 evolve 或丢 appendix；记 warning 非静默。
|
||
logger.warning("appendix consolidation 失败，保留原 notes：{}", exc)
|
||
return clean
|
||
|
||
|
||
async def _append_lapse_with_consolidation(
|
||
text: str,
|
||
lapse_notes: list[str],
|
||
llm: LLMProvider,
|
||
consolidate_threshold: int,
|
||
) -> str:
|
||
"""把 lapse 提醒追加进 appendix，超阈值时触发 LLM consolidation。
|
||
|
||
回写侧二确认——即便 consolidate_appendix 守卫已保证 <=，这里再校验「确实
|
||
变短」才 replace，避免等长压缩带来无意义改写抖动（守卫 G4）。
|
||
|
||
参数:
|
||
text: 待追加的 skill 全文（正文已改完）。
|
||
lapse_notes: 本轮待落 appendix 的 LAPSE 提醒。
|
||
llm: LLM 调用端口，供 consolidation 使用。
|
||
consolidate_threshold: appendix note 条数 >= 此值时触发压缩。
|
||
|
||
返回:
|
||
追加（必要时压缩）后的 skill 全文。
|
||
"""
|
||
after = append_to_appendix(text, lapse_notes)
|
||
notes = extract_appendix_notes(after)
|
||
if len(notes) >= consolidate_threshold:
|
||
compacted = await consolidate_appendix(llm, notes)
|
||
# G4：压缩结果必须严格变短才替换
|
||
if len(compacted) < len(notes):
|
||
after = replace_appendix_notes(after, compacted)
|
||
return after
|
||
|
||
|
||
# =========================================================================
|
||
# N. 整篇重写
|
||
# =========================================================================
|
||
|
||
|
||
_REWRITE_SYSTEM = (
|
||
"你负责根据改动建议整篇重写 Agent Skill 文件。保留 frontmatter（---...---）中的 "
|
||
"name / description / task_type 不变。保持文件精简，重写后长度不得超过原文。"
|
||
'只返回 JSON：{ "rewritten": "重写后的完整文件内容" }'
|
||
)
|
||
|
||
|
||
async def rewrite_from_suggestions(
|
||
llm: LLMProvider,
|
||
original: str,
|
||
suggestions: list[dict[str, Any]],
|
||
) -> str:
|
||
"""从抽象 suggestion 整篇重写 Skill；校验失败回退原文（skill 不变）。
|
||
|
||
硬约束由系统提示下达 + 本函数校验双重保证。三条拒绝条件任一触发
|
||
即返回原文（保守不改）：
|
||
1. 解析失败（JSON / rewritten 字段缺失或非字符串）
|
||
2. 重写后长度 > 原文
|
||
3. validate_skill 校验不过
|
||
|
||
仅捕获 ValueError / KeyError / TypeError / AttributeError；API 错误向上传播。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
original: 改写前 Skill 文件全文。
|
||
suggestions: 抽象改动建议列表。
|
||
|
||
返回:
|
||
校验通过的重写全文；任一守卫未通过时返回 original。
|
||
"""
|
||
sugg_text = "\n".join(f"- {s.get('change', '')}" for s in (suggestions or []))
|
||
user_msg = f"## 当前 Skill 文件\n\n{original}\n\n## 改动建议\n\n{sugg_text or '（无）'}"
|
||
try:
|
||
response = await llm.chat(
|
||
[
|
||
{"role": "system", "content": _REWRITE_SYSTEM},
|
||
{"role": "user", "content": user_msg},
|
||
]
|
||
)
|
||
parsed = _parse_llm_json(response.content)
|
||
rewritten = (parsed or {}).get("rewritten")
|
||
if not isinstance(rewritten, str) or not rewritten.strip():
|
||
raise ValueError("rewrite 未返回非空 rewritten 字符串")
|
||
except (ValueError, KeyError, TypeError, AttributeError):
|
||
logger.warning("rewrite 解析失败，回退原文")
|
||
return original
|
||
|
||
# 拒绝条件 2：不许变长
|
||
if len(rewritten) > len(original):
|
||
logger.warning("rewrite 变长（{}->{}），回退原文", len(original), len(rewritten))
|
||
return original
|
||
# 拒绝条件 3：冻结区/格式校验
|
||
if not validate_skill(original, rewritten).passed:
|
||
logger.warning("rewrite 校验未过（冻结区/格式），回退原文")
|
||
return original
|
||
return rewritten
|
||
|
||
|
||
# =========================================================================
|
||
# O. 单目标进化函数
|
||
# =========================================================================
|
||
|
||
|
||
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:
|
||
"""进化单个 Skill 文件。
|
||
|
||
三分支构建：
|
||
A. lapse-only：无 defect edit + 有 lapse_notes → 仅 appendix 更新。
|
||
B. rewrite：mode="rewrite" + 有 edit → 整篇重写；失败回退 A 或 no-op。
|
||
C. patch（默认）：rank_and_clip → apply_patch_with_report。
|
||
分支 B/C 完成后，若有 lapse_notes，追加 appendix（超阈值 consolidation）。
|
||
|
||
用户消息结构（按顺序）：
|
||
1. （可选）黑名单
|
||
2. 当前 Skill 文件原文
|
||
3. 聚合统计 JSON
|
||
4. 失败案例
|
||
5. 成功案例
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
pack: 该题型的案例包。
|
||
skill_store: 版本化技能读取端口。
|
||
prompts: 进化模板束。
|
||
source_version: 改写前版本号。
|
||
edit_budget: per-target 编辑预算上限。
|
||
consolidate_threshold: appendix note 条数 >= 此值触发 consolidation。
|
||
skill_update_mode: 正文更新模式，"patch"（局部 edit）/ "rewrite"（整篇重写）。
|
||
rejected: 已验证无效的历史改法列表。
|
||
|
||
返回:
|
||
EvolutionRecord 实例。
|
||
"""
|
||
target_file = pack.target_file
|
||
original_content = skill_store.read_skill(target_file)
|
||
|
||
# 构建用户消息
|
||
stats_json = json.dumps(pack.stats, ensure_ascii=False, indent=2)
|
||
user_msg = (
|
||
f"## 当前 Skill 文件\n\n{original_content}\n\n"
|
||
f"## 聚合统计\n\n```json\n{stats_json}\n```\n\n"
|
||
f"## 失败案例\n\n{_format_case_samples(pack.failure_cases)}\n\n"
|
||
f"## 成功案例\n\n{_format_case_samples(pack.success_cases)}"
|
||
)
|
||
rejected = rejected or []
|
||
if rejected:
|
||
user_msg = (
|
||
"## 已验证无效的改法（黑名单，勿重复）\n\n"
|
||
+ _format_rejected_edits(rejected)
|
||
+ "\n\n"
|
||
+ user_msg
|
||
)
|
||
|
||
messages: list[dict[str, Any]] = [
|
||
{"role": "system", "content": prompts.evolve_skill},
|
||
{"role": "user", "content": user_msg},
|
||
]
|
||
attempts: list[dict[str, Any]] = []
|
||
|
||
async def _build_attempt(parsed: dict[str, Any]) -> _PatchEvolutionAttempt:
|
||
suggestions = parsed.get("suggestions", [])
|
||
edits = parsed.get("edits", [])
|
||
|
||
# 分支 A：lapse-only（无 defect edit、仅有 lapse 提醒）
|
||
if not edits and pack.lapse_notes:
|
||
return _build_lapse_only_attempt(original_content, pack.lapse_notes)
|
||
|
||
# 分支 B：rewrite 模式（有 defect edits 时整篇重写）
|
||
if skill_update_mode == "rewrite" and edits:
|
||
rewritten = await rewrite_from_suggestions(llm, original_content, suggestions)
|
||
if rewritten == original_content:
|
||
# 重写校验失败/变长/解析失败 → 正文无改动
|
||
if pack.lapse_notes:
|
||
return _build_lapse_only_attempt(original_content, pack.lapse_notes)
|
||
return _PatchEvolutionAttempt(
|
||
evolved_content=original_content,
|
||
validation=validate_skill(original_content, original_content),
|
||
suggestions=suggestions,
|
||
edits=[],
|
||
apply_report=[],
|
||
clip_info={"triggered": False, "clipped": 0},
|
||
)
|
||
evolved_content = rewritten
|
||
apply_report = [
|
||
{
|
||
"op": "rewrite",
|
||
"target": "",
|
||
"content_preview": "整篇重写",
|
||
"status": "applied_rewrite",
|
||
"index": 1,
|
||
}
|
||
]
|
||
clip_info: dict[str, Any] = {"triggered": False, "clipped": 0}
|
||
|
||
# 分支 C：patch 模式（默认）
|
||
else:
|
||
edits, clip_info = await rank_and_clip(
|
||
llm,
|
||
original_content,
|
||
edits,
|
||
edit_budget,
|
||
"skill",
|
||
rank_prompt=prompts.evolve_rank,
|
||
)
|
||
evolved_content, apply_report = apply_patch_with_report(
|
||
original_content,
|
||
edits,
|
||
protected_spans=_skill_protected_spans(original_content),
|
||
)
|
||
|
||
# 分支 B/C 完成后：lapse 提醒追加 + consolidation
|
||
if pack.lapse_notes:
|
||
evolved_content = await _append_lapse_with_consolidation(
|
||
evolved_content,
|
||
pack.lapse_notes,
|
||
llm,
|
||
consolidate_threshold,
|
||
)
|
||
|
||
return _PatchEvolutionAttempt(
|
||
evolved_content=evolved_content,
|
||
validation=validate_skill(original_content, evolved_content),
|
||
suggestions=suggestions,
|
||
edits=edits,
|
||
apply_report=apply_report,
|
||
clip_info=clip_info,
|
||
)
|
||
|
||
return await _run_patch_evolution_loop(
|
||
llm=llm,
|
||
messages=messages,
|
||
attempts=attempts,
|
||
target_file=target_file,
|
||
target_type="skill",
|
||
original_content=original_content,
|
||
source_version=source_version,
|
||
log_target=target_file,
|
||
attempt_builder=_build_attempt,
|
||
)
|
||
|
||
|
||
async def evolve_system_prompt(
|
||
llm: LLMProvider,
|
||
pack: SystemCasePack,
|
||
prompt_store: PromptStore,
|
||
prompts: EvolvePrompts,
|
||
source_version: str,
|
||
edit_budget: int,
|
||
) -> EvolutionRecord:
|
||
"""进化 System Prompt。
|
||
|
||
无 lapse notes、无 appendix consolidation、无 rewrite 模式。
|
||
使用 system protected spans 保护冻结区。
|
||
用户消息中统计标题为「D5 行为模式统计」。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
pack: 跨题型行为模式案例包。
|
||
prompt_store: 版本化提示词读取端口。
|
||
prompts: 进化模板束。
|
||
source_version: 改写前版本号。
|
||
edit_budget: per-target 编辑预算上限。
|
||
|
||
返回:
|
||
EvolutionRecord 实例。
|
||
"""
|
||
target_file = "system.md"
|
||
original_content = prompt_store.read_prompt(target_file)
|
||
|
||
stats_json = json.dumps(pack.stats, ensure_ascii=False, indent=2)
|
||
user_msg = (
|
||
f"## 当前 System Prompt\n\n{original_content}\n\n"
|
||
f"## D5 行为模式统计\n\n```json\n{stats_json}\n```\n\n"
|
||
f"## 失败案例\n\n{_format_case_samples(pack.failure_cases)}\n\n"
|
||
f"## 成功案例\n\n{_format_case_samples(pack.success_cases)}"
|
||
)
|
||
|
||
messages: list[dict[str, Any]] = [
|
||
{"role": "system", "content": prompts.evolve_system},
|
||
{"role": "user", "content": user_msg},
|
||
]
|
||
attempts: list[dict[str, Any]] = []
|
||
|
||
async def _build_attempt(parsed: dict[str, Any]) -> _PatchEvolutionAttempt:
|
||
suggestions = parsed.get("suggestions", [])
|
||
edits = parsed.get("edits", [])
|
||
edits, clip_info = await rank_and_clip(
|
||
llm,
|
||
original_content,
|
||
edits,
|
||
edit_budget,
|
||
"system",
|
||
rank_prompt=prompts.evolve_rank,
|
||
)
|
||
evolved_content, apply_report = apply_patch_with_report(
|
||
original_content,
|
||
edits,
|
||
protected_spans=_system_protected_spans(original_content),
|
||
)
|
||
return _PatchEvolutionAttempt(
|
||
evolved_content=evolved_content,
|
||
validation=validate_system(original_content, evolved_content),
|
||
suggestions=suggestions,
|
||
edits=edits,
|
||
apply_report=apply_report,
|
||
clip_info=clip_info,
|
||
)
|
||
|
||
return await _run_patch_evolution_loop(
|
||
llm=llm,
|
||
messages=messages,
|
||
attempts=attempts,
|
||
target_file=target_file,
|
||
target_type="system",
|
||
original_content=original_content,
|
||
source_version=source_version,
|
||
log_target=target_file,
|
||
attempt_builder=_build_attempt,
|
||
)
|
||
|
||
|
||
async def evolve_single_tool(
|
||
llm: LLMProvider,
|
||
pack: ToolCasePack,
|
||
prompt_store: PromptStore,
|
||
prompts: EvolvePrompts,
|
||
source_version: str,
|
||
edit_budget: int,
|
||
) -> EvolutionRecord:
|
||
"""进化单个工具的 extract + verify prompt。
|
||
|
||
extract 与 verify 的 edits 合并到 SHARED 预算池（打 ``_src`` 标签），
|
||
整体 rank_and_clip 到 edit_budget 后按 ``_src`` 拆回各自文件应用。
|
||
``evolved_content`` 以 ``json.dumps({"extract": ..., "verify": ...})`` 存储。
|
||
``target_file`` 固定为 ``{tool_name}_extract.md``。
|
||
apply_report 每条带 ``"source"`` 注解（"extract" / "verify"）。
|
||
|
||
参数:
|
||
llm: LLM 调用端口。
|
||
pack: 该工具的案例包。
|
||
prompt_store: 版本化提示词读取端口。
|
||
prompts: 进化模板束。
|
||
source_version: 改写前版本号。
|
||
edit_budget: per-target 编辑预算上限（extract + verify 共享）。
|
||
|
||
返回:
|
||
EvolutionRecord 实例。
|
||
"""
|
||
tool_name = pack.tool_name
|
||
target_file = f"{tool_name}_extract.md"
|
||
orig_extract = prompt_store.read_prompt(f"{tool_name}_extract.md")
|
||
orig_verify = prompt_store.read_prompt(f"{tool_name}_verify.md")
|
||
original_combined = json.dumps(
|
||
{"extract": orig_extract, "verify": orig_verify},
|
||
ensure_ascii=False,
|
||
)
|
||
|
||
stats_json = json.dumps(pack.stats, ensure_ascii=False, indent=2)
|
||
user_msg = (
|
||
f"## 当前 extract prompt\n\n{orig_extract}\n\n"
|
||
f"## 当前 verify prompt\n\n{orig_verify}\n\n"
|
||
f"## 工具质量统计\n\n```json\n{stats_json}\n```\n\n"
|
||
f"## 失败 span 案例\n\n{_format_spans(pack.failure_spans)}\n\n"
|
||
f"## 成功 span 案例\n\n{_format_spans(pack.success_spans)}"
|
||
)
|
||
|
||
messages: list[dict[str, Any]] = [
|
||
{"role": "system", "content": prompts.evolve_tool},
|
||
{"role": "user", "content": user_msg},
|
||
]
|
||
attempts: list[dict[str, Any]] = []
|
||
|
||
async def _build_attempt(parsed: dict[str, Any]) -> _PatchEvolutionAttempt:
|
||
suggestions = parsed.get("suggestions", [])
|
||
edits_extract = parsed.get("edits_extract", [])
|
||
edits_verify = parsed.get("edits_verify", [])
|
||
|
||
# 合并打来源标记，对单 tool target 整体裁到 edit_budget
|
||
pool: list[dict[str, Any]] = [{**e, "_src": "extract"} for e in edits_extract] + [
|
||
{**e, "_src": "verify"} for e in edits_verify
|
||
]
|
||
|
||
pool, clip_info = await rank_and_clip(
|
||
llm,
|
||
original_combined,
|
||
pool,
|
||
edit_budget,
|
||
"tool",
|
||
rank_prompt=prompts.evolve_rank,
|
||
)
|
||
|
||
# 按 _src 拆回，并剥离 _src 字段
|
||
extract_kept = [
|
||
{k: v for k, v in e.items() if k != "_src"} for e in pool if e["_src"] == "extract"
|
||
]
|
||
verify_kept = [
|
||
{k: v for k, v in e.items() if k != "_src"} for e in pool if e["_src"] == "verify"
|
||
]
|
||
|
||
# 分别应用，使用各自的 tool protected spans
|
||
evolved_extract, extract_report = apply_patch_with_report(
|
||
orig_extract,
|
||
extract_kept,
|
||
protected_spans=_tool_protected_spans(orig_extract),
|
||
)
|
||
evolved_verify, verify_report = apply_patch_with_report(
|
||
orig_verify,
|
||
verify_kept,
|
||
protected_spans=_tool_protected_spans(orig_verify),
|
||
)
|
||
|
||
# 报告注解来源
|
||
apply_report = _with_report_source(extract_report, "extract") + _with_report_source(
|
||
verify_report, "verify"
|
||
)
|
||
|
||
evolved_combined = json.dumps(
|
||
{"extract": evolved_extract, "verify": evolved_verify},
|
||
ensure_ascii=False,
|
||
)
|
||
validation = validate_tool(orig_extract, evolved_extract, orig_verify, evolved_verify)
|
||
return _PatchEvolutionAttempt(
|
||
evolved_content=evolved_combined,
|
||
validation=validation,
|
||
suggestions=suggestions,
|
||
edits=extract_kept + verify_kept,
|
||
apply_report=apply_report,
|
||
clip_info=clip_info,
|
||
)
|
||
|
||
return await _run_patch_evolution_loop(
|
||
llm=llm,
|
||
messages=messages,
|
||
attempts=attempts,
|
||
target_file=target_file,
|
||
target_type="tool",
|
||
original_content=original_combined,
|
||
source_version=source_version,
|
||
log_target=tool_name,
|
||
attempt_builder=_build_attempt,
|
||
)
|