#!/usr/bin/env python3 """ Agentic System Prompting — Offline Trace Analysis Feeds pre-recorded agent traces through TraceAnalyser (offline mode) to extract reusable strategies into a skillbook. Each trace file is loaded as a traces-format dict and passed directly to RRStep via a thin adapter step, so the sandbox receives the full conversation data. TraceAnalyser handles the rest of the learning-tail pipeline: [RRTraceStep] → UpdateStep (the agentic SkillManager mutates directly) Usage: python recursive_agentic_system_prompting.py /path/to/traces python recursive_agentic_system_prompting.py /path/to/traces --model gpt-4o python recursive_agentic_system_prompting.py /path/to/traces --input-skillbook existing.json python recursive_agentic_system_prompting.py /path/to/traces --epochs 2 Options: traces_dir Path to directory containing .json, .md, or .toon trace files --model, -m LLM model for analysis (default: bedrock/us.anthropic.claude-sonnet-4-6) --threshold, -t Deduplication similarity threshold 0.0-1.0 (default: 0.7) --epochs, -e Number of passes over all traces (default: 1) --input-skillbook, -i Path to existing skillbook to continue from --output-dir, -o Output directory for results (default: script directory) """ import argparse import json import logging import os from datetime import datetime from itertools import groupby from pathlib import Path from typing import Any, Dict, List from dotenv import load_dotenv, find_dotenv load_dotenv(find_dotenv()) # Show RR iteration progress _handler = logging.StreamHandler() _handler.setFormatter( logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%H:%M:%S") ) _logger = logging.getLogger("ace.steps.rr") _logger.setLevel(logging.DEBUG) _logger.addHandler(_handler) from pipeline import Pipeline from ace import TraceAnalyser, SkillManager, Skillbook from ace.steps.rr_step import RRStep, RRConfig from ace.core.context import ACEStepContext from ace.deduplication import DeduplicationManager from ace.protocols.deduplication import DeduplicationConfig from ace.implementations.prompts import wrap_skillbook_for_external_agent from ace.steps import UpdateStep, DeduplicateStep from ace.implementations.rr.prompts import REFLECTOR_RECURSIVE_PROMPT # --------------------------------------------------------------------------- # Adapter step: normalises raw traces into the dict format RRStep expects. # --------------------------------------------------------------------------- class RRTraceStep: """Bridge between TraceAnalyser's per-trace context and RRStep. TraceAnalyser places the raw trace on ``ctx.trace``. RRStep.__call__ expects a traces-format dict with a ``steps`` key. This adapter normalises the trace and delegates to ``RRStep.__call__``. """ requires = frozenset({"trace", "skillbook"}) provides = frozenset({"reflection"}) def __init__(self, rr: RRStep) -> None: self.rr = rr def __call__(self, ctx: ACEStepContext) -> ACEStepContext: trace = ctx.trace # If the trace is already a traces-format dict, pass it through. # Otherwise wrap it so the sandbox can access it via traces["steps"]. if isinstance(trace, dict) and "steps" in trace: traces_dict = trace else: traces_dict = { "question": str(trace.get("id", "")) if isinstance(trace, dict) else "", "steps": [trace], } return self.rr(ctx.replace(trace=traces_dict)) def load_traces(traces_dir: Path) -> Dict[str, Any]: """Load all trace files into a single batch trace dict. All files are combined into one traces-format dict so the REPL agent receives every conversation at once and can analyze cross-trace patterns. """ if not traces_dir.exists(): print(f"Directory not found: {traces_dir}") return {} steps: List[Dict[str, Any]] = [] for ext in ("*.json", "*.md", "*.toon"): for file_path in sorted(traces_dir.glob(ext)): try: raw = file_path.read_text(encoding="utf-8") content = json.loads(raw) if file_path.suffix == ".json" else raw steps.append( { "role": "conversation", "id": file_path.name, "content": content, } ) except Exception as e: print(f"Error reading {file_path.name}: {e}") print(f"Loaded {len(steps)} traces") if not steps: return {} return { "question": f"Analyze {len(steps)} conversation traces", "ground_truth": None, "feedback": None, "steps": steps, } def main(): parser = argparse.ArgumentParser( description="Offline trace analysis — extract strategies into a skillbook" ) parser.add_argument( "traces_dir", type=Path, help="Directory containing trace files" ) parser.add_argument( "-m", "--model", default="bedrock/eu.anthropic.claude-sonnet-4-6", help="LLM model for analysis", ) parser.add_argument( "-t", "--threshold", type=float, default=0.7, help="Deduplication similarity threshold (0.0-1.0)", ) parser.add_argument( "-e", "--epochs", type=int, default=1, help="Number of passes over all traces" ) parser.add_argument( "-i", "--input-skillbook", type=Path, default=None, help="Existing skillbook" ) parser.add_argument( "-o", "--output-dir", type=Path, default=None, help="Output directory" ) args = parser.parse_args() if not os.getenv("OPENAI_API_KEY"): print("WARNING: OPENAI_API_KEY required for deduplication embeddings!") return # Load all traces into a single batch dict batch_trace = load_traces(args.traces_dir) if not batch_trace: print(f"\nAdd .json, .md, or .toon trace files to {args.traces_dir}/") return n_traces = len(batch_trace["steps"]) # Skillbook (existing or empty) skillbook = Skillbook() if args.input_skillbook and args.input_skillbook.exists(): skillbook = Skillbook.load_from_file(str(args.input_skillbook)) print(f"Loaded skillbook: {len(skillbook.skills())} skills") # Build PydanticAI-backed roles directly from model strings rr = RRStep( args.model, config=RRConfig( max_requests=60, ), prompt_template=REFLECTOR_RECURSIVE_PROMPT, ) skill_manager = SkillManager(args.model) dedup = DeduplicationManager( DeduplicationConfig( enabled=True, similarity_threshold=args.threshold, embedding_model="text-embedding-3-small", ) ) # Build pipeline: RRTraceStep → Update → Dedup # (SkillManager mutates the skillbook directly via its tools, so no # separate ApplyStep is needed.) steps: list[Any] = [RRTraceStep(rr)] steps.extend( [ UpdateStep(skill_manager, skillbook), DeduplicateStep(dedup, skillbook), ] ) analyser = TraceAnalyser(pipeline=Pipeline(steps), skillbook=skillbook) print( f"\nStarting analysis: {n_traces} traces (single batch), " f"epochs={args.epochs}, model={args.model}" ) start = datetime.now() # Run — single batch trace through the pipeline results = analyser.run([batch_trace], epochs=args.epochs) # Surface any pipeline errors (the pipeline catches exceptions silently) failed = [r for r in results if r.error is not None] if failed: print(f"\n{len(failed)}/{len(results)} traces FAILED:") for r in failed: print(f" - {r.failed_at}: {r.error}") duration = (datetime.now() - start).total_seconds() # Save results output_dir = args.output_dir or Path(__file__).parent timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") output_skillbook = output_dir / f"skillbook_{timestamp}.json" analyser.save(str(output_skillbook)) skills = analyser.skillbook.skills() print(f"\nCompleted in {duration:.1f}s") print(f"Analyzed: {n_traces} traces (single batch) × {args.epochs} epoch(s)") print(f"Generated: {len(skills)} skills") print(f"Saved to: {output_skillbook}") # Markdown export output_md = output_dir / f"skills_{timestamp}.md" with open(output_md, "w") as f: for section, section_skills in groupby( sorted(skills, key=lambda s: s.section), key=lambda s: s.section ): f.write(f"## {section}\n\n") for skill in section_skills: f.write(f"- {skill.content}\n") if skill.justification: f.write(f" Justification: {skill.justification}\n") if skill.evidence: f.write(f" Evidence: {skill.evidence}\n") f.write("\n") print(f"Skills: {output_md}") if skills: print("\nTop skills:") for i, skill in enumerate( sorted(skills, key=lambda s: s.helpful, reverse=True)[:5], 1 ): print(f" {i}. [{skill.section}] {skill.content[:80]}...") # External agent injection injection = wrap_skillbook_for_external_agent(analyser.skillbook) if injection: output_injection = output_dir / f"external_agent_injection_{timestamp}.txt" with open(output_injection, "w") as f: f.write(injection) print(f"External agent injection: {output_injection}") if __name__ == "__main__": main()