Upload aco/optimizer.py with huggingface_hub

aco/optimizer.py

@@ -12,6 +12,7 @@ from .verifier_budgeter import VerifierBudgeter, VerifierDecision
 from .retry_optimizer import RetryOptimizer, RecoveryAction
 from .meta_tool_miner import MetaToolMiner, MacroTool
 from .doom_detector import DoomDetector, DoomAssessment
+from .execution_feedback import ExecutionFeedbackRouter, CascadeResult, FeedbackSignal
 
 class ACOOptimizer:
     def __init__(self, config: ACOConfig = None):
@@ -33,6 +34,10 @@ class ACOOptimizer:
         )
         self.meta_tool_miner = MetaToolMiner()
         self.doom_detector = DoomDetector()
+        self.execution_feedback = ExecutionFeedbackRouter(
+            tier_costs=self.config.tier_costs,
+            task_floors=self.config.task_floors,
+        )
         self._current_trace: Optional[AgentTrace] = None
         self._step_num = 0
         self._traces: List[AgentTrace] = []
@@ -123,6 +128,30 @@ class ACOOptimizer:
         return self.tool_gate.gate(tool_name, args, self._current_trace.task_type,
                                    self._step_num, self._step_num + 1, 0.5)
 
+    def cascade_step(self, request: str, initial_tier: int,
+                     cheap_logprobs: List[float],
+                     cheap_response: str,
+                     strong_response: str = "",
+                     task_type: str = None) -> CascadeResult:
+        """Execution-feedback cascade: use cheap model output to decide escalation."""
+        if not self._current_trace:
+            task_type = task_type or "unknown_ambiguous"
+        else:
+            task_type = task_type or self._current_trace.task_type
+        floor = self.config.task_floors.get(task_type, 1)
+        return self.execution_feedback.cascade(
+            request, initial_tier, cheap_logprobs,
+            cheap_response, strong_response,
+            task_type=task_type, task_floor=floor,
+        )
+
+    def analyze_output_confidence(self, token_logprobs: List[float],
+                                   task_type: str = "unknown",
+                                   current_tier: int = 2) -> FeedbackSignal:
+        """Analyze model output confidence for routing decisions."""
+        return self.execution_feedback.analyze_output(
+            token_logprobs, task_type=task_type, current_tier=current_tier)
+
     def get_recovery(self, failure_tag: str, current_tier: int,
                      retry_num: int, previous_actions: List[str] = None,
                      run_cost: float = 0.0) -> RecoveryAction: