OSVauco ML Layer
ML-laget gir OPAX-agentene parallell kjøring, delt state-kontroll og strukturert
telemetri. Implementert i tre faser.
Arkitektur
OPAX /run-endepunkt
│
▼
build_agent_dag() ← Dask DAG (task_graph.py)
│
┌────┴────┬──────────────┐
▼ ▼ ▼
agent_A agent_B ... agent_N ← kjøres parallelt
│ │ │
└────┬────┴──────────────┘
▼
AgentStateStore ← Parameter Server state (state_store.py)
│
▼
telemetry JSONL ← rådatagrunnlag for ML-2 og ML-3
Filer
| Fil |
Ansvar |
task_graph.py |
Dask DAG — parallell agent-scheduling |
state_store.py |
Thread-safe PS-inspirert state store |
telemetry.py |
Strukturert JSONL-logging per agent-kall |
__init__.py |
Pakke-eksporter |
Scheduler-valg
| Miljø |
scheduler-parameter |
| Dev / unit-test |
'synchronous' |
| Lokal multi-tråd |
'threads' |
| Cloud Run (lett) |
'threads' |
| Full skala |
'distributed' (krever Dask cluster) |
Fase-oversikt
| Fase |
Innhold |
Status |
| ML-1 |
Dask DAG + PS state + telemetri |
✅ Implementert |
| ML-2 |
Inkrementell læring (dask-ml) + ASHA-tuning (Ray Tune) |
🔜 Neste |
| ML-3 |
XGBoost feature importance + PBT |
🔜 Fremtidig |
Bruk
from ml import build_agent_dag, execute_dag, get_store, log_agent_call
# Bygg og kjør DAG
tasks = build_agent_dag([billing_agent, rag_agent], payload)
results = execute_dag(tasks, scheduler="threads")
# State store
store = get_store()
store.push("billing-agent", "last_cost_usd", 12.5)
cost = store.pull("billing-agent", "last_cost_usd")
# Telemetri
log_agent_call(
agent_id="billing-agent",
input_payload=payload,
output=result,
model_used="gemini-2.0-flash",
mode="standard",
duration_s=0.4,
success=True,
)
