File size: 2,378 Bytes
943946b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 | """Trackio integration for live ACO monitoring.
Logs optimization decisions, costs, and quality metrics to Trackio dashboard.
"""
import os
from typing import Dict, Any, Optional
class ACOTrackioLogger:
"""Lightweight Trackio logger for Agent Cost Optimizer runs."""
def __init__(self, project: str = "aco", space_id: Optional[str] = None):
self.project = project
self.space_id = space_id or os.environ.get("TRACKIO_SPACE_ID")
self._enabled = False
try:
import trackio
self.api = trackio.Api()
self._enabled = True
except ImportError:
self.api = None
def log_decision(self, run_id: str, decision: Any, cost: float, success: bool) -> None:
"""Log an optimization decision to Trackio."""
if not self._enabled:
return
try:
self.api.log_metric(
project=self.project,
run=run_id,
key="optimization_cost",
value=cost,
)
self.api.log_metric(
project=self.project,
run=run_id,
key="task_success",
value=1.0 if success else 0.0,
)
if hasattr(decision, "routing_decision"):
self.api.log_param(
project=self.project,
run=run_id,
key="model_tier",
value=decision.routing_decision.tier,
)
except Exception:
pass
def alert(self, run_id: str, title: str, text: str, level: str = "INFO") -> None:
"""Send an alert to Trackio."""
if not self._enabled:
return
try:
self.api.alert(
project=self.project,
run=run_id,
title=title,
text=text,
level=level,
)
except Exception:
pass
def log_cost_adjusted_score(self, run_id: str, score: float) -> None:
if not self._enabled:
return
try:
self.api.log_metric(
project=self.project,
run=run_id,
key="cost_adjusted_score",
value=score,
)
except Exception:
pass
|