Spaces:
Running
Running
File size: 4,060 Bytes
28a08e7 | 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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """backend/api/skills.py — P17-B2: Endpoint /api/skills/patterns.
Espone i pattern di tool-usage aggregati dal SkillTracker per sync frontend.
GET /api/skills/patterns → aggregato globale (tutte le sessioni)
GET /api/skills/patterns/{session_id} → singola sessione
fix(backend): corregge IndentationError presente al commit 1e1fa65
"""
from __future__ import annotations
import logging
import math
from fastapi import APIRouter, Depends, Query
from .auth_guard import require_role, AuthRole
_logger = logging.getLogger("agente_ai.api.skills")
router = APIRouter(prefix="/api/skills", tags=["skills"], dependencies=[Depends(require_role(AuthRole.MACHINE))]) # GAP-1-fix: router-level auth
def _wilson(success: int, total: int) -> float:
if total == 0:
return 0.5
p, z = success / total, 1.96
num = p + z*z/(2*total) - z*math.sqrt((p*(1-p) + z*z/(4*total))/total)
return max(0.0, min(1.0, num / (1 + z*z/total)))
@router.get("/patterns")
async def get_skill_patterns(
limit: int = Query(default=50, ge=1, le=200),
min_uses: int = Query(default=1, ge=1, description="Minimo utilizzi totali"),
):
"""Pattern aggregati da tutte le sessioni in memoria, ordinati per Wilson score."""
try:
from agents.skill_tracker import get_skill_tracker
tracker = get_skill_tracker()
agg: dict[str, dict] = {}
for _sid, sess in tracker._stats.items():
for tool, stats in sess.items():
if tool not in agg:
agg[tool] = {"s": 0, "f": 0, "lat": 0.0, "sessions": 0, "last": 0.0}
agg[tool]["s"] += stats.success_count
agg[tool]["f"] += stats.fail_count
agg[tool]["lat"] += stats.total_latency_ms
agg[tool]["sessions"] += 1
agg[tool]["last"] = max(agg[tool]["last"], stats.last_used)
out = []
for tool, a in agg.items():
total = a["s"] + a["f"]
if total < min_uses:
continue
out.append({
"tool": tool,
"total": total,
"success": a["s"],
"fail": a["f"],
"success_rate": round(a["s"] / total if total else 1.0, 4),
"wilson_score": round(_wilson(a["s"], total), 4),
"avg_latency_ms": round(a["lat"] / total if total else 0.0, 1),
"sessions": a["sessions"],
"last_used": a["last"],
})
out.sort(key=lambda x: x["wilson_score"], reverse=True)
return {"patterns": out[:limit], "total": len(out)}
except Exception as exc:
_logger.error("[skills/patterns] %s", exc)
return {"patterns": [], "total": 0, "error": str(exc)[:200]}
@router.get("/patterns/{session_id}")
async def get_session_patterns(session_id: str):
"""Pattern per una singola sessione."""
try:
from agents.skill_tracker import get_skill_tracker
tracker = get_skill_tracker()
sess = tracker._stats.get(session_id, {})
out = []
for tool, stats in sess.items():
total = stats.total_count
out.append({
"tool": tool,
"total": total,
"success": stats.success_count,
"fail": stats.fail_count,
"success_rate": round(stats.success_rate, 4),
"wilson_score": round(stats.wilson_score(), 4),
"avg_latency_ms": round(stats.avg_latency_ms, 1),
"last_used": stats.last_used,
})
out.sort(key=lambda x: x["wilson_score"], reverse=True)
return {"session_id": session_id, "patterns": out, "total": len(out)}
except Exception as exc:
_logger.error("[skills/patterns/%s] %s", session_id, exc)
return {"session_id": session_id, "patterns": [], "total": 0, "error": str(exc)[:200]}
|