"""Meta-Tool / Workflow Compression - Module 9. Mines repeated successful traces into reusable workflows. Compresses recurring workflows into: - deterministic scripts - macro tools - reusable skills - cached plans Metrics: - LLM calls saved - workflow success - bad automation rate - regression rate - latency saved """ import hashlib import json from typing import Dict, List, Tuple, Optional, Set from dataclasses import dataclass, field from collections import defaultdict from .trace_schema import AgentTrace, TraceStep, Outcome, TaskType from .config import ACOConfig @dataclass class WorkflowPattern: pattern_id: str task_type: TaskType tool_sequence: Tuple[str, ...] frequency: int success_count: int failure_count: int avg_cost: float avg_latency_ms: float compressed_script: Optional[str] = None is_deterministic: bool = False class MetaToolMiner: """Discovers and compresses repeated agent workflows into meta-tools.""" def __init__(self, config: Optional[ACOConfig] = None): self.config = config or ACOConfig() self.patterns: Dict[str, WorkflowPattern] = {} self.meta_tools: Dict[str, Dict] = {} # pattern_id -> meta_tool definition self.trace_buffer: List[AgentTrace] = [] def ingest_trace(self, trace: AgentTrace) -> None: """Add a completed trace for pattern mining.""" if trace.final_outcome not in (Outcome.SUCCESS, Outcome.PARTIAL_SUCCESS): return # Only mine successful workflows self.trace_buffer.append(trace) # Extract tool sequences tool_sequence = tuple( tc.tool_name for step in trace.steps for tc in step.tool_calls ) if len(tool_sequence) < 2: return # Hash the sequence seq_hash = hashlib.md5(json.dumps(tool_sequence).encode()).hexdigest()[:16] # Update or create pattern if seq_hash in self.patterns: pattern = self.patterns[seq_hash] pattern.frequency += 1 if trace.final_outcome == Outcome.SUCCESS: pattern.success_count += 1 else: pattern.failure_count += 1 pattern.avg_cost = ( pattern.avg_cost * (pattern.frequency - 1) + trace.total_cost_computed ) / pattern.frequency pattern.avg_latency_ms = ( pattern.avg_latency_ms * (pattern.frequency - 1) + trace.total_latency_ms ) / pattern.frequency else: self.patterns[seq_hash] = WorkflowPattern( pattern_id=seq_hash, task_type=trace.task_type, tool_sequence=tool_sequence, frequency=1, success_count=1 if trace.final_outcome == Outcome.SUCCESS else 0, failure_count=0 if trace.final_outcome == Outcome.SUCCESS else 1, avg_cost=trace.total_cost_computed, avg_latency_ms=trace.total_latency_ms, ) def extract_meta_tools(self) -> List[WorkflowPattern]: """Extract workflow patterns that meet meta-tool thresholds.""" qualified = [] for pattern in self.patterns.values(): success_rate = pattern.success_count / max(pattern.frequency, 1) if (pattern.frequency >= self.config.meta_tool_min_frequency and success_rate >= self.config.meta_tool_min_success_rate): # Generate deterministic script if possible if self._is_deterministic_sequence(pattern.tool_sequence): pattern.is_deterministic = True pattern.compressed_script = self._generate_script(pattern) qualified.append(pattern) return sorted(qualified, key=lambda p: p.frequency * p.avg_cost, reverse=True) def _is_deterministic_sequence(self, tool_sequence: Tuple[str, ...]) -> bool: """Check if a tool sequence can be made deterministic.""" # Simple heuristic: if all tools are read-only, it's likely deterministic read_only_tools = {"file_read", "search", "retrieve", "fetch", "calculator", "linter"} return all(t in read_only_tools for t in tool_sequence) def _generate_script(self, pattern: WorkflowPattern) -> str: """Generate a deterministic script for a workflow pattern.""" lines = [f"# Meta-tool: {pattern.pattern_id}"] lines.append(f"# Task type: {pattern.task_type.value}") lines.append(f"# Success rate: {pattern.success_count / max(pattern.frequency, 1):.1%}") lines.append(f"# Avg cost: ${pattern.avg_cost:.4f}") lines.append("") for i, tool in enumerate(pattern.tool_sequence): lines.append(f"step_{i+1} = execute_tool('{tool}', params=auto_resolve())") lines.append("return aggregate_results()") return "\n".join(lines) def match_and_compress( self, task_type: TaskType, planned_tools: List[str], ) -> Optional[Dict]: """Match current plan against known meta-tools and return compressed plan if found.""" planned_tuple = tuple(planned_tools) for pattern in self.patterns.values(): if pattern.task_type != task_type: continue if pattern.compressed_script is None: continue # Check if planned tools are a subset or match of pattern if self._sequence_match(planned_tuple, pattern.tool_sequence): success_rate = pattern.success_count / max(pattern.frequency, 1) return { "use_meta_tool": True, "meta_tool_id": pattern.pattern_id, "compressed_script": pattern.compressed_script, "estimated_cost_savings": pattern.avg_cost * 0.3, # meta-tools save ~30% "estimated_latency_savings_ms": pattern.avg_latency_ms * 0.3, "success_rate": success_rate, "fallback_tools": list(pattern.tool_sequence), } return None def _sequence_match(self, planned: Tuple[str, ...], pattern: Tuple[str, ...]) -> bool: """Check if planned sequence matches or is contained in pattern.""" if planned == pattern: return True # Allow prefix match if pattern is longer if len(pattern) >= len(planned) and pattern[:len(planned)] == planned: return True return False def get_stats(self) -> Dict: """Return meta-tool mining statistics.""" total_patterns = len(self.patterns) qualified = self.extract_meta_tools() total_traces = len(self.trace_buffer) total_llm_calls_saved = sum( p.frequency * len(p.tool_sequence) * 0.5 # each meta-tool saves ~50% LLM calls for p in qualified ) return { "total_patterns": total_patterns, "qualified_meta_tools": len(qualified), "total_traces_mined": total_traces, "estimated_llm_calls_saved": total_llm_calls_saved, "top_patterns": [ { "pattern_id": p.pattern_id, "task_type": p.task_type.value, "tool_sequence": p.tool_sequence, "frequency": p.frequency, "success_rate": p.success_count / max(p.frequency, 1), "avg_cost": p.avg_cost, } for p in qualified[:5] ], }