diff --git a/configs/_base_/default.yaml b/configs/_base_/default.yaml index eb2d58d..fcfdd71 100644 --- a/configs/_base_/default.yaml +++ b/configs/_base_/default.yaml @@ -81,6 +81,9 @@ optimizer: slow_update_gate_with_selection: false longitudinal_pair_policy: mixed # mixed / changed / unchanged use_meta_skill: true + use_skill_aware_reflection: false # EmbodiSkill: split failures into SKILL_DEFECT (edit body) vs EXECUTION_LAPSE (protected appendix) + skill_aware_appendix_source: both # both = success+failure emit appendix notes; failure_only = only EXECUTION_LAPSE (paper-faithful) + skill_aware_consolidate_threshold: 0 # 0 = off; >0 = LLM-consolidate the appendix when its note count exceeds N evaluation: use_gate: true diff --git a/scripts/train.py b/scripts/train.py index c16474b..5c0621a 100644 --- a/scripts/train.py +++ b/scripts/train.py @@ -245,6 +245,10 @@ def parse_args() -> argparse.Namespace: p.add_argument("--longitudinal_pair_policy", type=str, choices=["mixed", "changed", "unchanged"]) p.add_argument("--use_meta_skill", type=_BOOL) + p.add_argument("--use_skill_aware_reflection", type=_BOOL) + p.add_argument("--skill_aware_appendix_source", type=str, + choices=["both", "failure_only"]) + p.add_argument("--skill_aware_consolidate_threshold", type=int) p.add_argument("--data_path", type=str) p.add_argument("--split_mode", type=str, choices=["ratio", "split_dir"]) @@ -360,6 +364,9 @@ _LEGACY_TO_STRUCTURED: dict[str, str] = { "slow_update_samples": "optimizer.slow_update_samples", "longitudinal_pair_policy": "optimizer.longitudinal_pair_policy", "use_meta_skill": "optimizer.use_meta_skill", + "use_skill_aware_reflection": "optimizer.use_skill_aware_reflection", + "skill_aware_appendix_source": "optimizer.skill_aware_appendix_source", + "skill_aware_consolidate_threshold": "optimizer.skill_aware_consolidate_threshold", "use_gate": "evaluation.use_gate", "sel_env_num": "evaluation.sel_env_num", "test_env_num": "evaluation.test_env_num", @@ -527,6 +534,7 @@ def main() -> None: print(f" minibatch_size: {cfg.get('minibatch_size')}") print(f" seed: {cfg.get('seed')}") print(f" meta_skill: {cfg.get('use_meta_skill', False)}") + print(f" skill_aware_reflection: {cfg.get('use_skill_aware_reflection', False)}") print(f" slow_update: {cfg.get('use_slow_update', False)}") print(f" out_root: {cfg.get('out_root')}") print(f"{'='*60}\n") diff --git a/skillopt/config.py b/skillopt/config.py index fe74462..e7dbb83 100644 --- a/skillopt/config.py +++ b/skillopt/config.py @@ -119,6 +119,9 @@ _FLATTEN_MAP: dict[str, str] = { "optimizer.slow_update_gate_with_selection": "slow_update_gate_with_selection", "optimizer.longitudinal_pair_policy": "longitudinal_pair_policy", "optimizer.use_meta_skill": "use_meta_skill", + "optimizer.use_skill_aware_reflection": "use_skill_aware_reflection", + "optimizer.skill_aware_appendix_source": "skill_aware_appendix_source", + "optimizer.skill_aware_consolidate_threshold": "skill_aware_consolidate_threshold", "evaluation.use_gate": "use_gate", "evaluation.gate_metric": "gate_metric", "evaluation.gate_mixed_weight": "gate_mixed_weight", diff --git a/skillopt/engine/trainer.py b/skillopt/engine/trainer.py index 6e3b402..414773f 100644 --- a/skillopt/engine/trainer.py +++ b/skillopt/engine/trainer.py @@ -32,6 +32,17 @@ from skillopt.optimizer.lr_autonomous import decide_autonomous_learning_rate from skillopt.optimizer.rewrite import rewrite_skill_from_suggestions from skillopt.optimizer.scheduler import build_scheduler from skillopt.optimizer.skill import apply_patch_with_report +from skillopt.optimizer.appendix import ( + append_to_appendix_field, + extract_appendix_notes as extract_appendix_notes_from_skill, + inject_empty_appendix_field, + _strip_all_appendix_fields, +) +from skillopt.optimizer.skill_aware import ( + configure_skill_aware_reflection, + consolidate_appendix_notes, + extract_appendix_notes as extract_appendix_notes_from_result, +) from skillopt.optimizer.slow_update import ( build_comparison_pairs, extract_slow_update_field, @@ -48,6 +59,7 @@ from skillopt.optimizer.update_modes import ( short_item_summary, ) from skillopt.model import ( + chat_optimizer, configure_azure_openai, configure_claude_code_exec, configure_codex_exec, @@ -838,6 +850,20 @@ class ReflACTTrainer: _save_skill(out_root, 0, skill_init) + # ── Skill-aware reflection: ensure the protected appendix (S_app) + # region exists on the working skill. Only current_skill carries the + # appendix; best_skill stays a faithful val-best snapshot (same policy + # as slow_update). No-op when the region already exists (resume-safe). + use_skill_aware = cfg.get("use_skill_aware_reflection", False) + # Publish the toggle process-wide so run_minibatch_reflect resolves it + # from config for EVERY env adapter — no per-benchmark wiring needed. + configure_skill_aware_reflection( + use_skill_aware, + cfg.get("skill_aware_appendix_source", "both"), + ) + if use_skill_aware: + current_skill = inject_empty_appendix_field(current_skill) + def _persist_runtime_state(last_completed_step: int) -> None: _save_runtime_state( out_root, @@ -1389,6 +1415,62 @@ class ReflACTTrainer: ): best_origin = current_origin + # ── Skill-aware reflection: flush execution-lapse reminders ── + # After the gate has settled current_skill, append this step's + # EXECUTION_LAPSE notes into the protected appendix (S_app). + # This bypasses the gate by design (the paper writes appendix + # reminders directly) and only touches current_skill, never + # best_skill. Body candidate evaluation already happened above + # and is unaffected. + if use_skill_aware: + step_appendix_notes: list[str] = [] + for rp in all_raw_patches: + if isinstance(rp, dict): + step_appendix_notes.extend(extract_appendix_notes_from_result(rp)) + if step_appendix_notes: + before_notes = extract_appendix_notes_from_skill(current_skill) + current_skill = append_to_appendix_field( + current_skill, step_appendix_notes, + ) + after_notes = extract_appendix_notes_from_skill(current_skill) + n_added = len(after_notes) - len(before_notes) + step_rec["n_execution_lapse_notes"] = len(step_appendix_notes) + step_rec["n_appendix_notes_added"] = n_added + step_rec["n_appendix_notes_total"] = len(after_notes) + with open(os.path.join(step_dir, "appendix_notes.json"), "w") as f: + json.dump( + { + "step_notes": step_appendix_notes, + "appendix_after": after_notes, + }, + f, indent=2, ensure_ascii=False, + ) + print( + f" [skill-aware] +{n_added} appendix note(s) " + f"(total {len(after_notes)}) from {len(step_appendix_notes)} lapse signal(s)" + ) + # Threshold-gated LLM consolidation (paper Eq.11): when the + # appendix grows past N notes, compact it with one optimizer + # call (dedupe / merge / shorten). 0 disables it. Any failure + # leaves the appendix unchanged. + consolidate_threshold = int( + cfg.get("skill_aware_consolidate_threshold", 0) or 0 + ) + if consolidate_threshold > 0 and len(after_notes) > consolidate_threshold: + compacted = consolidate_appendix_notes( + after_notes, chat_fn=chat_optimizer, + ) + if compacted and len(compacted) < len(after_notes): + current_skill = append_to_appendix_field( + _strip_all_appendix_fields(current_skill), compacted, + ) + step_rec["n_appendix_notes_consolidated"] = len(compacted) + step_rec["n_appendix_notes_total"] = len(compacted) + print( + f" [skill-aware] consolidated appendix " + f"{len(after_notes)} -> {len(compacted)} notes" + ) + if gate_metric == "hard": score_label = f"hard={cand_hard:.4f}" elif gate_metric == "soft": diff --git a/skillopt/gradient/reflect.py b/skillopt/gradient/reflect.py index 4e6395e..8078f85 100644 --- a/skillopt/gradient/reflect.py +++ b/skillopt/gradient/reflect.py @@ -29,6 +29,13 @@ from concurrent.futures import ThreadPoolExecutor, as_completed from skillopt.model import chat_optimizer from skillopt.optimizer.meta_skill import format_meta_skill_context +from skillopt.optimizer.skill_aware import ( + augment_error_prompt, + augment_success_prompt, + extract_appendix_notes, + get_skill_aware_appendix_source, + is_skill_aware_enabled, +) from skillopt.optimizer.update_modes import ( get_payload_items, is_full_rewrite_minibatch_mode, @@ -258,6 +265,7 @@ def run_error_analyst_minibatch( step_buffer_context: str = "", meta_skill_context: str = "", update_mode: str = "patch", + skill_aware_reflection: bool = False, ) -> dict | None: """Analyze a minibatch of failed trajectories in one optimizer call. @@ -287,6 +295,11 @@ def run_error_analyst_minibatch( """ mode = normalize_update_mode(update_mode) actual_system = _resolve_prompt(system_prompt, "analyst_error", mode) + # Skill-aware reflection: augment the resolved prompt at runtime so both + # env-specific and generic analyst prompts get the defect/lapse instruction. + # When the toggle is off this is a no-op (prompt byte-identical to baseline). + if skill_aware_reflection and not is_full_rewrite_minibatch_mode(mode): + actual_system = augment_error_prompt(actual_system) trajectories_text = fmt_minibatch_trajectories(items, prediction_dir) if not trajectories_text.strip(): @@ -325,11 +338,26 @@ def run_error_analyst_minibatch( stage="analyst", ) result = extract_json(response) - if result and "patch" in result: + if not result: + return None + notes = extract_appendix_notes(result) if skill_aware_reflection else [] + if "patch" in result: result["source_type"] = "failure" if not is_full_rewrite_minibatch_mode(mode): truncate_payload(result["patch"], edit_budget, mode) + if skill_aware_reflection: + result["appendix_notes"] = notes return result + # Skill-aware: a batch may legitimately yield ONLY execution-lapse notes + # (no body edit). Return a no-op patch so the notes still reach the + # trainer via all_raw_patches; empty edits are dropped from the body + # pipeline by _normalise_patches, so body behavior is unchanged. + if skill_aware_reflection and notes: + return { + "source_type": "failure", + "patch": {"reasoning": "execution-lapse only", "edits": []}, + "appendix_notes": notes, + } except Exception: # noqa: BLE001 traceback.print_exc() return None @@ -346,6 +374,8 @@ def run_success_analyst_minibatch( step_buffer_context: str = "", meta_skill_context: str = "", update_mode: str = "patch", + skill_aware_reflection: bool = False, + emit_appendix_notes: bool = True, ) -> dict | None: """Analyze a minibatch of successful trajectories in one optimizer call. @@ -365,6 +395,11 @@ def run_success_analyst_minibatch( """ mode = normalize_update_mode(update_mode) actual_system = _resolve_prompt(system_prompt, "analyst_success", mode) + # Only augment + parse appendix notes on the success side when allowed. + # failure_only mode (paper-faithful S_app) suppresses success-side notes. + sa_emit = skill_aware_reflection and emit_appendix_notes + if sa_emit and not is_full_rewrite_minibatch_mode(mode): + actual_system = augment_success_prompt(actual_system) trajectories_text = fmt_minibatch_trajectories(items, prediction_dir) if not trajectories_text.strip(): @@ -404,6 +439,8 @@ def run_success_analyst_minibatch( result["source_type"] = "success" if not is_full_rewrite_minibatch_mode(mode): truncate_payload(result["patch"], edit_budget, mode) + if sa_emit: + result["appendix_notes"] = extract_appendix_notes(result) return result except Exception: # noqa: BLE001 traceback.print_exc() @@ -450,6 +487,8 @@ def run_minibatch_reflect( step_buffer_context: str = "", meta_skill_context: str = "", update_mode: str = "patch", + skill_aware_reflection: bool | None = None, + skill_aware_appendix_source: str | None = None, ) -> list[dict | None]: """Full minibatch reflect stage: group → parallel optimizer calls → patches. @@ -484,6 +523,14 @@ def run_minibatch_reflect( list[dict | None] Patch dicts (with ``source_type`` "failure" or "success"). """ + # Resolve the skill-aware toggle: explicit kwargs win; otherwise fall back + # to the process-wide config switch set by the trainer, so the feature is + # env-independent and adapters need no per-benchmark wiring. + if skill_aware_reflection is None: + skill_aware_reflection = is_skill_aware_enabled() + if skill_aware_appendix_source is None: + skill_aware_appendix_source = get_skill_aware_appendix_source() + os.makedirs(patches_dir, exist_ok=True) # Separate failure / success @@ -539,6 +586,7 @@ def run_minibatch_reflect( trajectory_memory_context=trajectory_memory_context, meta_skill_context=meta_skill_context, update_mode=update_mode, + skill_aware_reflection=skill_aware_reflection, ) return f"minibatch_fail_{idx:03d}", patch @@ -551,6 +599,8 @@ def run_minibatch_reflect( trajectory_memory_context=trajectory_memory_context, meta_skill_context=meta_skill_context, update_mode=update_mode, + skill_aware_reflection=skill_aware_reflection, + emit_appendix_notes=(skill_aware_appendix_source != "failure_only"), ) return f"minibatch_succ_{idx:03d}", patch diff --git a/skillopt/optimizer/appendix.py b/skillopt/optimizer/appendix.py new file mode 100644 index 0000000..2509260 --- /dev/null +++ b/skillopt/optimizer/appendix.py @@ -0,0 +1,156 @@ +"""Skill-Aware Reflection — protected appendix field (EmbodiSkill S_app). + +EmbodiSkill (paper 2605.10332v1) splits a skill into ``S = (S_body, S_app)``: +the body holds the main prescriptive rules; the appendix only *emphasizes* +existing valid rules that the executor failed to follow (EXECUTION_LAPSE), and +**never introduces new rules**. + +This module owns the appendix region of the skill document. It mirrors the +protected-field pattern of :mod:`skillopt.optimizer.slow_update`, with two +differences: + +1. **Append semantics** (not replace): execution-lapse reminders accumulate + across steps within a run, so new notes are merged into the existing + appendix rather than overwriting it. +2. **Lightweight dedup**: near-duplicate reminders are collapsed (inspired by + GMemory's ``_dedupe_preserve_order``) so the appendix stays compact. + +The appendix lives **inside** the skill markdown, between dedicated markers, so +it is persisted by the normal ``_save_skill`` path and is resume-safe. Step-level +analyst edits cannot modify it (enforced by the shared protected-region check in +:mod:`skillopt.optimizer.skill`). + +Public API +---------- +- :func:`has_appendix_field` — check if markers are present +- :func:`inject_empty_appendix_field` — add empty placeholder (skill init) +- :func:`extract_appendix_notes` — read current notes as a list +- :func:`append_to_appendix_field` — merge new notes (dedup) into the region +""" +from __future__ import annotations + +import re + +# ── Protected field markers ───────────────────────────────────────────────── + +APPENDIX_START = "" +APPENDIX_END = "" + +# Heading shown inside the rendered appendix block (human-readable only). +APPENDIX_HEADING = "## Execution Notes Appendix" + +# Each note is rendered as a markdown bullet so the target model reads it as +# ordinary guidance. +_NOTE_BULLET_PREFIX = "- " + + +# ── Dedup helpers ─────────────────────────────────────────────────────────── + + +def _canonicalize(text: str) -> str: + """Normalize a note for duplicate detection (whitespace/punct/case-insensitive).""" + normalized = re.sub(r"\s+", " ", str(text or "").strip()) + normalized = normalized.rstrip(" .;:,_-") + return normalized.casefold() + + +def _dedupe_preserve_order(notes: list[str]) -> list[str]: + """Drop blanks and near-duplicates, preserving first-seen order.""" + seen: set[str] = set() + deduped: list[str] = [] + for note in notes: + text = re.sub(r"\s+", " ", str(note).strip()) + if not text: + continue + key = _canonicalize(text) + if not key or key in seen: + continue + seen.add(key) + deduped.append(text) + return deduped + + +# ── Field manipulation ────────────────────────────────────────────────────── + + +def has_appendix_field(skill: str) -> bool: + return APPENDIX_START in skill and APPENDIX_END in skill + + +def _render_block(notes: list[str]) -> str: + """Render the full marker-delimited appendix block for *notes*.""" + lines = [APPENDIX_START, APPENDIX_HEADING] + for note in notes: + lines.append(f"{_NOTE_BULLET_PREFIX}{note}") + lines.append(APPENDIX_END) + return "\n".join(lines) + + +def inject_empty_appendix_field(skill: str) -> str: + """Add an empty appendix placeholder at the end of *skill* (idempotent). + + Mirrors ``inject_empty_slow_update_field``: called once at skill init so the + protected region exists before any note is written. + """ + if has_appendix_field(skill): + return skill + block = f"\n\n{APPENDIX_START}\n{APPENDIX_HEADING}\n{APPENDIX_END}\n" + return skill.rstrip() + block + + +def extract_appendix_notes(skill: str) -> list[str]: + """Return the current appendix notes as a list of strings (no markers/heading).""" + start = skill.find(APPENDIX_START) + end = skill.find(APPENDIX_END) + if start == -1 or end == -1: + return [] + inner = skill[start + len(APPENDIX_START):end].strip() + notes: list[str] = [] + for raw_line in inner.splitlines(): + line = raw_line.strip() + if not line: + continue + if line == APPENDIX_HEADING or line.lstrip("#").strip() == APPENDIX_HEADING.lstrip("#").strip(): + continue + if line.startswith(_NOTE_BULLET_PREFIX): + line = line[len(_NOTE_BULLET_PREFIX):].strip() + elif line.startswith("-") or line.startswith("*"): + line = line[1:].strip() + if line: + notes.append(line) + return notes + + +def _strip_all_appendix_fields(skill: str) -> str: + """Remove every appendix marker pair (and content between) from *skill*.""" + while True: + start = skill.find(APPENDIX_START) + if start == -1: + break + end = skill.find(APPENDIX_END, start) + if end == -1: + skill = skill[:start] + skill[start + len(APPENDIX_START):] + break + skill = skill[:end + len(APPENDIX_END)].rsplit(APPENDIX_START, 1)[0] + skill[end + len(APPENDIX_END):] + skill = skill.replace(APPENDIX_END, "") + while "\n\n\n" in skill: + skill = skill.replace("\n\n\n", "\n\n") + return skill.rstrip() + + +def append_to_appendix_field(skill: str, new_notes: list[str]) -> str: + """Merge *new_notes* into the appendix region (dedup), returning updated skill. + + - If no appendix region exists yet, one is created. + - Existing notes are preserved; new ones are appended after dedup against the + combined set, so order is stable and duplicates are dropped. + - Empty / whitespace-only notes are ignored. If the merged set is empty, an + empty placeholder region is still ensured. + """ + incoming = _dedupe_preserve_order(list(new_notes or [])) + existing = extract_appendix_notes(skill) + merged = _dedupe_preserve_order(existing + incoming) + + base = _strip_all_appendix_fields(skill) + block = _render_block(merged) + return f"{base}\n\n{block}\n" diff --git a/skillopt/optimizer/skill.py b/skillopt/optimizer/skill.py index 0a8855f..65d5741 100644 --- a/skillopt/optimizer/skill.py +++ b/skillopt/optimizer/skill.py @@ -14,25 +14,62 @@ if TYPE_CHECKING: SLOW_UPDATE_START = "" SLOW_UPDATE_END = "" +# Skill-aware reflection (EmbodiSkill S_app) appendix region. Like the slow +# update region, it is protected: step-level analyst edits must not modify it. +APPENDIX_START = "" +APPENDIX_END = "" -def _is_in_slow_update_region(skill: str, target: str) -> bool: - """Check if *target* text falls within the protected slow update region.""" - start_idx = skill.find(SLOW_UPDATE_START) - end_idx = skill.find(SLOW_UPDATE_END) - if start_idx == -1 or end_idx == -1: +# All protected (start, end) marker pairs. Step-level edits cannot target text +# inside any of these regions, and `append` / `insert_after`-fallback ops are +# inserted before the earliest-occurring region so protected blocks stay at the +# document tail. With only the slow-update region present, every helper reduces +# to the original slow-update-only behavior (byte-identical skill output). +_PROTECTED_REGIONS: tuple[tuple[str, str], ...] = ( + (SLOW_UPDATE_START, SLOW_UPDATE_END), + (APPENDIX_START, APPENDIX_END), +) + + +def _earliest_protected_start(skill: str) -> int: + """Index of the earliest protected-region start marker, or -1 if none.""" + positions = [ + idx + for idx in (skill.find(start) for start, _ in _PROTECTED_REGIONS) + if idx != -1 + ] + return min(positions) if positions else -1 + + +def _is_in_protected_region(skill: str, target: str) -> bool: + """Check if *target* text falls within any protected region.""" + if not target: return False target_idx = skill.find(target) if target_idx == -1: return False - region_end = end_idx + len(SLOW_UPDATE_END) - return start_idx <= target_idx < region_end + for start_marker, end_marker in _PROTECTED_REGIONS: + start_idx = skill.find(start_marker) + end_idx = skill.find(end_marker) + if start_idx == -1 or end_idx == -1: + continue + region_end = end_idx + len(end_marker) + if start_idx <= target_idx < region_end: + return True + return False + + +def _is_in_slow_update_region(skill: str, target: str) -> bool: + """Backward-compatible alias kept for any external callers/tests.""" + return _is_in_protected_region(skill, target) def _strip_slow_update_markers(text: str) -> str: - """Remove any SLOW_UPDATE markers from edit content to prevent duplication.""" + """Remove any protected-region markers from edit content to prevent duplication.""" return ( text.replace(SLOW_UPDATE_START, "") .replace(SLOW_UPDATE_END, "") + .replace(APPENDIX_START, "") + .replace(APPENDIX_END, "") ) @@ -54,27 +91,27 @@ def _apply_edit_with_report(skill: str, edit: EditType | dict) -> tuple[str, dic "status": "unknown", } - if target and _is_in_slow_update_region(skill, target): - report["status"] = "skipped_protected_slow_update_region" + if target and _is_in_protected_region(skill, target): + report["status"] = "skipped_protected_region" return skill, report if op == "append": - su_start = skill.find(SLOW_UPDATE_START) - if su_start != -1: - before = skill[:su_start].rstrip() - after = skill[su_start:] - report["status"] = "applied_append_before_slow_update" + prot_start = _earliest_protected_start(skill) + if prot_start != -1: + before = skill[:prot_start].rstrip() + after = skill[prot_start:] + report["status"] = "applied_append_before_protected_region" return before + "\n\n" + content + "\n\n" + after, report report["status"] = "applied_append" return skill.rstrip() + "\n\n" + content + "\n", report if op == "insert_after": if not target or target not in skill: - su_start = skill.find(SLOW_UPDATE_START) - if su_start != -1: - before = skill[:su_start].rstrip() - after = skill[su_start:] - report["status"] = "applied_insert_after_fallback_before_slow_update" + prot_start = _earliest_protected_start(skill) + if prot_start != -1: + before = skill[:prot_start].rstrip() + after = skill[prot_start:] + report["status"] = "applied_insert_after_fallback_before_protected_region" return before + "\n\n" + content + "\n\n" + after, report report["status"] = "applied_insert_after_fallback_append" return skill.rstrip() + "\n\n" + content + "\n", report diff --git a/skillopt/optimizer/skill_aware.py b/skillopt/optimizer/skill_aware.py new file mode 100644 index 0000000..de39427 --- /dev/null +++ b/skillopt/optimizer/skill_aware.py @@ -0,0 +1,206 @@ +"""Skill-Aware Reflection — analyst prompt augmentation (EmbodiSkill). + +When ``use_skill_aware_reflection`` is enabled, the failure/success analysts are +asked to additionally classify each reflection by EmbodiSkill type and to route +**EXECUTION_LAPSE** reflections (the skill rule is correct, the executor just +failed to follow it) into a separate ``appendix_notes`` list instead of the body +patch. This module owns: + +1. the instruction text appended to the resolved analyst system prompt, and +2. extraction of ``appendix_notes`` from the analyst JSON response. + +Design notes +------------ +- The suffix is appended **at runtime, gated by the toggle**, so env-specific and + generic analyst prompts are augmented uniformly and — when the toggle is off — + remain byte-identical to baseline. +- Discrimination follows the paper / GMemory: ``SKILL_DEFECT`` = the skill rule is + wrong / missing / underspecified (→ body edit); ``EXECUTION_LAPSE`` = the rule + is valid but the agent didn't follow it (→ appendix reminder, body untouched). + **When unsure, default to EXECUTION_LAPSE** (protect the body — never delete a + valid rule over a one-off execution slip). +- Success reflections are labeled DISCOVERY / OPTIMIZATION for logging only; their + edit behavior is unchanged. +""" +from __future__ import annotations + + +# ── Runtime switch (config-driven, env-independent) ───────────────────────── +# +# The trainer calls :func:`configure_skill_aware_reflection` once at startup +# from the resolved config. ``run_minibatch_reflect`` then picks these values +# up automatically, so env adapters never need to thread the toggle through — +# the feature is controlled purely by ``optimizer.use_skill_aware_reflection`` +# regardless of benchmark. Mirrors the ``configure_azure_openai`` pattern in +# :mod:`skillopt.model`. Explicit kwargs at a call site still take precedence +# (backward compatible). + +_RUNTIME: dict = {"enabled": False, "appendix_source": "both"} + + +def configure_skill_aware_reflection( + enabled: bool, + appendix_source: str = "both", +) -> None: + """Set the process-wide skill-aware reflection switch from config.""" + _RUNTIME["enabled"] = bool(enabled) + _RUNTIME["appendix_source"] = str(appendix_source or "both") + + +def is_skill_aware_enabled() -> bool: + return bool(_RUNTIME["enabled"]) + + +def get_skill_aware_appendix_source() -> str: + return str(_RUNTIME["appendix_source"]) + + +# ── Prompt suffixes ───────────────────────────────────────────────────────── + +# Appended to the FAILURE analyst system prompt when the toggle is on. +ERROR_SUFFIX = """ + +## Skill-Aware Reflection (EmbodiSkill) + +Before proposing body edits, classify EACH failure pattern as one of: + +- **SKILL_DEFECT**: the current skill is wrong, missing, or underspecified for + this situation — i.e. an agent that *followed the skill* would still fail, or + the skill gives no relevant guidance. These become normal body `edits`. +- **EXECUTION_LAPSE**: the skill ALREADY contains a relevant, correct rule that + would have avoided the failure, but the agent did not follow it (e.g. ignored a + rule, malformed output, copied the feedback text verbatim, emitted a non-action + token like "stop", or otherwise broke execution unrelated to skill content). + +Discrimination test: "Is there a rule in the current skill that, if followed, +prevents this failure?" If yes → EXECUTION_LAPSE. If no (rule absent/wrong) → +SKILL_DEFECT. **When genuinely unsure, choose EXECUTION_LAPSE** — do not edit or +delete a valid rule over a one-off execution slip. + +Routing: +- SKILL_DEFECT → put the fix in `patch.edits` (body), as usual. +- EXECUTION_LAPSE → put a concise reminder in `appendix_notes` (a flat list of + strings). DO NOT add a body edit for it. Each note should re-emphasize the + existing valid rule the agent failed to follow; it must NOT introduce a new + rule. Keep notes short, concrete, and reusable. + +Add `appendix_notes` as a TOP-LEVEL key of your JSON output (a sibling of +`patch`), e.g. `"appendix_notes": ["Follow the existing X rule before Y."]`. +Use `[]` when there is no execution lapse. Body edits and appendix notes are +independent: a batch may yield only edits, only notes, both, or neither. +""" + +# Appended to the SUCCESS analyst system prompt when the toggle is on. +SUCCESS_SUFFIX = """ + +## Skill-Aware Reflection (EmbodiSkill) + +For each proposed edit, optionally label its `reflection_type` for logging: +- **DISCOVERY**: a useful new rule not yet in the skill (typically an `append`). +- **OPTIMIZATION**: a better way to perform an existing rule (typically a + `replace` of that rule). + +This labeling does not change edit behavior. You may also add a top-level +`appendix_notes` list (flat strings) if a successful trajectory reveals an +existing valid rule worth re-emphasizing; otherwise use `[]`. +""" + + +def augment_error_prompt(system_prompt: str) -> str: + """Append the failure-analyst skill-aware instruction.""" + return system_prompt.rstrip() + "\n" + ERROR_SUFFIX + + +def augment_success_prompt(system_prompt: str) -> str: + """Append the success-analyst skill-aware instruction.""" + return system_prompt.rstrip() + "\n" + SUCCESS_SUFFIX + + +# ── Response parsing ──────────────────────────────────────────────────────── + + +def extract_appendix_notes(result: dict | None) -> list[str]: + """Pull a clean list of appendix-note strings from an analyst JSON result. + + Tolerant of shape: accepts a top-level ``appendix_notes`` list, a single + string, or items wrapped in dicts with a ``note``/``content`` field. Returns + ``[]`` for anything missing or malformed (so a non-compliant model degrades + gracefully to baseline body-only behavior). + """ + if not isinstance(result, dict): + return [] + raw = result.get("appendix_notes") + if raw is None: + return [] + if isinstance(raw, str): + raw = [raw] + if not isinstance(raw, list): + return [] + notes: list[str] = [] + for item in raw: + if isinstance(item, str): + text = item.strip() + elif isinstance(item, dict): + text = str(item.get("note") or item.get("content") or "").strip() + else: + text = "" + if text: + notes.append(text) + return notes + + +# ── Appendix consolidation (threshold-gated, paper Eq.11 UpdateSkillAppendix) ── + +_CONSOLIDATE_SYSTEM = ( + "You compact the Execution Notes Appendix of an agent skill. Each note " + "re-emphasizes an existing skill rule the agent failed to follow. Your job " + "is a periodic compaction pass: remove duplicates and redundant overlap, " + "merge near-identical reminders into one, and simplify phrasing while keeping " + "each note concrete and operational. Do not invent new rules. Preserve the " + "distinct actionable content. Return valid JSON only." +) + + +def consolidate_appendix_notes( + notes: list[str], + *, + chat_fn, + max_completion_tokens: int = 4096, +) -> list[str]: + """LLM-consolidate appendix notes: dedupe / merge / compact. + + Mirrors GMemory ``_maybe_refactor_execution_notes`` and paper Eq.11. ``chat_fn`` + is the optimizer chat callable ``(system, user, max_completion_tokens, retries, + stage) -> (text, meta)``. On ANY failure (parse, empty, exception) the original + notes are returned unchanged, so consolidation can never lose the appendix. + """ + from skillopt.utils import extract_json # local import to avoid cycles + + 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"## Current Execution Notes ({len(clean)} total)\n{numbered}\n\n" + "Compact these into a shorter list without losing distinct actionable " + "information. Merge duplicates and near-duplicates; keep each note short, " + "concrete, and reusable. Return valid JSON only with this schema:\n" + '{ "appendix_notes": ["compacted note 1", "compacted note 2"] }' + ) + try: + response, _ = chat_fn( + system=_CONSOLIDATE_SYSTEM, + user=user, + max_completion_tokens=max_completion_tokens, + retries=2, + stage="appendix_consolidate", + ) + result = extract_json(response) + compacted = extract_appendix_notes(result) + # Guard: only accept a non-empty result that actually shrinks the set. + if compacted and len(compacted) <= len(clean): + return compacted + except Exception: # noqa: BLE001 + pass + return clean diff --git a/tests/test_skill_aware_reflection.py b/tests/test_skill_aware_reflection.py new file mode 100644 index 0000000..68d3533 --- /dev/null +++ b/tests/test_skill_aware_reflection.py @@ -0,0 +1,274 @@ +"""Standalone regression + function tests for skill-aware reflection. + +Run directly (no pytest needed): + python tests/test_skill_aware_reflection.py + +Covers: +1. Toggle-OFF byte-identical guarantee for skill.py edit application + (slow-update-only behavior must be unchanged). +2. Appendix module: inject / append / dedup / extract / accumulate. +3. Appendix-region protection from step-level edits. +4. Coexistence of appendix + slow_update regions. +5. reflect.py prompt augmentation + appendix_notes parsing (no LLM call). +""" +from __future__ import annotations + +import os +import sys + +# Ensure THIS repo's skillopt is imported (not an installed copy) when the +# file is run directly: script mode puts tests/ on sys.path, not the repo root. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + + +def _reference_old_apply(skill: str, edit: dict) -> str: + """Reproduce the ORIGINAL slow-update-only edit behavior inline.""" + SU_START = "" + SU_END = "" + op = edit.get("op", "") + content = edit.get("content", "").strip().replace(SU_START, "").replace(SU_END, "") + target = edit.get("target", "") + si = skill.find(SU_START) + ei = skill.find(SU_END) + + def in_su(t: str) -> bool: + if si == -1 or ei == -1: + return False + ti = skill.find(t) + if ti == -1: + return False + return si <= ti < ei + len(SU_END) + + if target and in_su(target): + return skill + if op == "append": + s = skill.find(SU_START) + if s != -1: + return skill[:s].rstrip() + "\n\n" + content + "\n\n" + skill[s:] + return skill.rstrip() + "\n\n" + content + "\n" + if op == "insert_after": + if not target or target not in skill: + s = skill.find(SU_START) + if s != -1: + return skill[:s].rstrip() + "\n\n" + content + "\n\n" + skill[s:] + return skill.rstrip() + "\n\n" + content + "\n" + idx = skill.index(target) + len(target) + nl = skill.find("\n", idx) + at = nl + 1 if nl != -1 else len(skill) + return skill[:at] + "\n" + content + "\n" + skill[at:] + if op == "replace": + if not target or target not in skill: + return skill + return skill.replace(target, content, 1) + if op == "delete": + if not target or target not in skill: + return skill + return skill.replace(target, "", 1) + return skill + + +def test_toggle_off_byte_identical() -> None: + from skillopt.optimizer.skill import _apply_edit_with_report + + SU_START = "" + SU_END = "" + skill = ( + "# QA Skill\n\n## Rules\n- Prefer shortest answer span.\n" + "- Use clue wording to constrain answer type.\n\n" + f"{SU_START}\nSome slow update guidance here.\n{SU_END}\n" + ) + edits = [ + {"op": "append", "content": "- New rule appended."}, + {"op": "insert_after", "target": "## Rules", "content": "- Inserted rule."}, + {"op": "insert_after", "target": "NONEXISTENT", "content": "- Fallback rule."}, + {"op": "replace", "target": "Prefer shortest answer span.", "content": "Prefer the exact minimal span."}, + {"op": "delete", "target": "- Use clue wording to constrain answer type."}, + {"op": "replace", "target": "Some slow update guidance here.", "content": "HACKED"}, + {"op": "delete", "target": "Some slow update guidance here."}, + ] + for e in edits: + new_skill, _ = _apply_edit_with_report(skill, e) + old_skill = _reference_old_apply(skill, e) + assert new_skill == old_skill, f"byte mismatch for {e['op']}" + print("PASS test_toggle_off_byte_identical") + + +def test_appendix_module() -> None: + from skillopt.optimizer.appendix import ( + has_appendix_field, inject_empty_appendix_field, + extract_appendix_notes, append_to_appendix_field, APPENDIX_START, + ) + skill = "# QA Skill\n\n- Prefer shortest answer span." + s1 = inject_empty_appendix_field(skill) + assert has_appendix_field(s1) and extract_appendix_notes(s1) == [] + assert inject_empty_appendix_field(s1) == s1 # idempotent + s2 = append_to_appendix_field(s1, ["Go to fridge for ice water.", "No stop token."]) + assert extract_appendix_notes(s2) == ["Go to fridge for ice water.", "No stop token."] + s3 = append_to_appendix_field(s2, ["go to fridge for ice water", "Check sheet range."]) + assert extract_appendix_notes(s3) == [ + "Go to fridge for ice water.", "No stop token.", "Check sheet range.", + ], "near-duplicate must be dropped" + assert s3.count(APPENDIX_START) == 1, "exactly one region after accumulation" + assert "# QA Skill" in s3 and "Prefer shortest answer span" in s3 + assert extract_appendix_notes(append_to_appendix_field(s1, [" ", "", "real"])) == ["real"] + print("PASS test_appendix_module") + + +def test_appendix_protection() -> None: + from skillopt.optimizer.skill import _apply_edit_with_report + from skillopt.optimizer.appendix import append_to_appendix_field, inject_empty_appendix_field + + skill = inject_empty_appendix_field("# QA Skill\n\n- Rule one.") + skill = append_to_appendix_field(skill, ["Follow rule one before acting."]) + for e in ( + {"op": "delete", "target": "Follow rule one before acting."}, + {"op": "replace", "target": "Follow rule one before acting.", "content": "HACK"}, + ): + new, rep = _apply_edit_with_report(skill, e) + assert new == skill, f"appendix must be protected from {e['op']}" + assert rep["status"] == "skipped_protected_region" + new, rep = _apply_edit_with_report(skill, {"op": "replace", "target": "Rule one.", "content": "Rule 1."}) + assert "Rule 1." in new and "Follow rule one before acting." in new + print("PASS test_appendix_protection") + + +def test_coexistence_with_slow_update() -> None: + from skillopt.optimizer.skill import _apply_edit_with_report + from skillopt.optimizer.appendix import ( + inject_empty_appendix_field, append_to_appendix_field, extract_appendix_notes, + ) + from skillopt.optimizer.slow_update import ( + inject_empty_slow_update_field, replace_slow_update_field, extract_slow_update_field, + ) + skill = inject_empty_appendix_field("# QA Skill\n\n- Rule one.") + skill = append_to_appendix_field(skill, ["Follow rule one."]) + skill = inject_empty_slow_update_field(skill) + skill = replace_slow_update_field(skill, "Slow guidance v2.") + assert extract_appendix_notes(skill) == ["Follow rule one."] + assert extract_slow_update_field(skill) == "Slow guidance v2." + # both regions protected + n1, r1 = _apply_edit_with_report(skill, {"op": "delete", "target": "Follow rule one."}) + n2, r2 = _apply_edit_with_report(skill, {"op": "replace", "target": "Slow guidance v2.", "content": "X"}) + assert n1 == skill and n2 == skill + # append lands before both regions (body stays at top) + n3, _ = _apply_edit_with_report(skill, {"op": "append", "content": "- Rule two."}) + assert n3.find("- Rule two.") < n3.find("") + assert n3.find("- Rule two.") < n3.find("") + print("PASS test_coexistence_with_slow_update") + + +def test_reflect_parsing_and_augment() -> None: + import inspect + import skillopt.gradient.reflect as R + from skillopt.optimizer.skill_aware import extract_appendix_notes, augment_error_prompt + + for fn in ("run_error_analyst_minibatch", "run_success_analyst_minibatch"): + sig = inspect.signature(getattr(R, fn)) + assert "skill_aware_reflection" in sig.parameters + assert sig.parameters["skill_aware_reflection"].default is False, f"{fn} default must be False" + # run_minibatch_reflect uses a None sentinel: explicit kwarg wins, else the + # process-wide config switch (configure_skill_aware_reflection) decides. + sig = inspect.signature(R.run_minibatch_reflect) + assert sig.parameters["skill_aware_reflection"].default is None + assert sig.parameters["skill_aware_appendix_source"].default is None + assert extract_appendix_notes({"appendix_notes": ["a", "b"]}) == ["a", "b"] + assert extract_appendix_notes({"appendix_notes": "x"}) == ["x"] + assert extract_appendix_notes({"appendix_notes": [{"note": "n"}, {"content": "c"}, {}]}) == ["n", "c"] + assert extract_appendix_notes({}) == [] and extract_appendix_notes(None) == [] + aug = augment_error_prompt("ORIG") + assert aug.startswith("ORIG") and "SKILL_DEFECT" in aug and "EXECUTION_LAPSE" in aug + print("PASS test_reflect_parsing_and_augment") + + +def test_global_switch_env_independent() -> None: + """The config switch alone must drive SAR for ANY env adapter (no kwargs).""" + from unittest import mock + import skillopt.gradient.reflect as R + from skillopt.optimizer.skill_aware import ( + configure_skill_aware_reflection, + get_skill_aware_appendix_source, + is_skill_aware_enabled, + ) + + # configure() round-trip. + configure_skill_aware_reflection(True, "failure_only") + assert is_skill_aware_enabled() and get_skill_aware_appendix_source() == "failure_only" + configure_skill_aware_reflection(False) + assert not is_skill_aware_enabled() and get_skill_aware_appendix_source() == "both" + + # run_minibatch_reflect with NO skill-aware kwargs (adapter-style call): + # capture what it forwards to the analyst workers under each switch state. + import tempfile + captured: dict = {} + + def fake_error_analyst(*args, **kwargs): + captured["skill_aware_reflection"] = kwargs.get("skill_aware_reflection") + return None + + def run_once() -> None: + captured.clear() + with mock.patch.object(R, "run_error_analyst_minibatch", fake_error_analyst), \ + tempfile.TemporaryDirectory() as tmp: + R.run_minibatch_reflect( + results=[{"id": "t1", "hard": 0, "soft": 0.0}], + skill_content="# Skill", + prediction_dir=tmp, + patches_dir=tmp, + workers=1, + failure_only=True, + minibatch_size=8, + ) + + try: + configure_skill_aware_reflection(True, "both") + run_once() + assert captured.get("skill_aware_reflection") is True, \ + "switch ON must reach the analyst without adapter wiring" + + configure_skill_aware_reflection(False) + run_once() + assert captured.get("skill_aware_reflection") is False, \ + "switch OFF must keep the analyst at baseline" + + # Explicit kwarg still overrides the global switch (backward compat). + captured.clear() + with mock.patch.object(R, "run_error_analyst_minibatch", fake_error_analyst), \ + tempfile.TemporaryDirectory() as tmp: + R.run_minibatch_reflect( + results=[{"id": "t1", "hard": 0, "soft": 0.0}], + skill_content="# Skill", + prediction_dir=tmp, + patches_dir=tmp, + workers=1, + failure_only=True, + minibatch_size=8, + skill_aware_reflection=True, + ) + assert captured.get("skill_aware_reflection") is True + finally: + configure_skill_aware_reflection(False) + print("PASS test_global_switch_env_independent") + + +def main() -> int: + tests = [ + test_toggle_off_byte_identical, + test_appendix_module, + test_appendix_protection, + test_coexistence_with_slow_update, + test_reflect_parsing_and_augment, + test_global_switch_env_independent, + ] + failed = 0 + for t in tests: + try: + t() + except AssertionError as exc: + failed += 1 + print(f"FAIL {t.__name__}: {exc}") + print(f"\n{len(tests) - failed}/{len(tests)} passed") + return 1 if failed else 0 + + +if __name__ == "__main__": + sys.exit(main())