Upload aco/meta_tool_miner.py with huggingface_hub

aco/meta_tool_miner.py

@@ -1,200 +1,86 @@
-"""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
-
+"""Meta-Tool Miner: Mine repeated successful traces into reusable workflows."""
+from typing import Dict, List, Optional, Tuple
+from dataclasses import dataclass
+import json, hashlib
 
 @dataclass
-class WorkflowPattern:
-    pattern_id: str
-    task_type: TaskType
-    tool_sequence: Tuple[str, ...]
-    frequency: int
-    success_count: int
-    failure_count: int
+class MacroTool:
+    name: str
+    steps: List[Dict]  # sequence of tool calls
+    trigger_pattern: str
+    success_rate: float
     avg_cost: float
     avg_latency_ms: float
-    compressed_script: Optional[str] = None
-    is_deterministic: bool = False
-
+    usage_count: int
+    last_used: str = ""
+
+COMMON_WORKFLOWS = [
+    {"name":"search_inspect_patch","steps":["code_search","file_read","file_write"],
+     "trigger":"fix.*bug|implement.*feature|refactor"},
+    {"name":"research_answer_verify","steps":["web_search","summarize","verify"],
+     "trigger":"research|investigate|compare"},
+    {"name":"read_repair_retry","steps":["file_read","code_execute","file_read"],
+     "trigger":"debug|fix.*error|repair"},
+    {"name":"retrieve_extract_answer","steps":["web_search","file_read","summarize"],
+     "trigger":"find|look up|search.*for"},
+    {"name":"draft_review","steps":["file_read","summarize","verify"],
+     "trigger":"draft|review|check"},
+]
 
 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]
-            ],
-        }
+    def __init__(self, min_occurrences: int = 3, min_success_rate: float = 0.7):
+        self.min_occurrences = min_occurrences
+        self.min_success_rate = min_success_rate
+        self.discovered_macros: Dict[str, MacroTool] = {}
+        self._trace_patterns: Dict[str, List[Dict]] = {}
+        # Pre-load common workflows
+        for wf in COMMON_WORKFLOWS:
+            self.discovered_macros[wf["name"]] = MacroTool(
+                name=wf["name"], steps=[{"tool":s} for s in wf["steps"]],
+                trigger_pattern=wf["trigger"], success_rate=0.8,
+                avg_cost=0.05, avg_latency_ms=5000, usage_count=0)
+
+    def mine_traces(self, traces: List[Dict]) -> List[MacroTool]:
+        for trace in traces:
+            if not trace.get("task_success", False): continue
+            tool_seq = []
+            for step in trace.get("steps", []):
+                for tc in step.get("tool_calls", []):
+                    tool_seq.append(tc.get("tool_name","unknown"))
+            if len(tool_seq) >= 2:
+                pattern_key = "->".join(tool_seq)
+                if pattern_key not in self._trace_patterns:
+                    self._trace_patterns[pattern_key] = []
+                self._trace_patterns[pattern_key].append(trace)
+        new_macros = []
+        for pattern, pattern_traces in self._trace_patterns.items():
+            if len(pattern_traces) < self.min_occurrences: continue
+            success_rate = sum(1 for t in pattern_traces if t.get("task_success",False)) / len(pattern_traces)
+            if success_rate < self.min_success_rate: continue
+            name = f"macro_{hashlib.md5(pattern.encode()).hexdigest()[:8]}"
+            if name in self.discovered_macros: continue
+            macro = MacroTool(
+                name=name,
+                steps=[{"tool":t} for t in pattern.split("->")],
+                trigger_pattern="",
+                success_rate=success_rate,
+                avg_cost=sum(t.get("total_cost",0) for t in pattern_traces) / len(pattern_traces),
+                avg_latency_ms=sum(t.get("latency_total_ms",0) for t in pattern_traces) / len(pattern_traces),
+                usage_count=len(pattern_traces))
+            self.discovered_macros[name] = macro
+            new_macros.append(macro)
+        return new_macros
+
+    def match_macro(self, request: str, task_type: str) -> Optional[MacroTool]:
+        import re
+        r = request.lower()
+        best = None
+        best_score = 0
+        for macro in self.discovered_macros.values():
+            if macro.trigger_pattern:
+                if re.search(macro.trigger_pattern, r):
+                    score = macro.success_rate * len(macro.steps)
+                    if score > best_score:
+                        best_score = score
+                        best = macro
+        return best