Upload aco/meta_tool_miner.py

aco/meta_tool_miner.py

@@ -0,0 +1,200 @@
+"""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]
+            ],
+        }