86 lines
3.0 KiB
Python
86 lines
3.0 KiB
Python
"""Search Gateway Web UI dashboard model."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import asdict, dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Any
|
|
|
|
|
|
def utc_now() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SearchDashboardMetric:
|
|
key: str
|
|
title: str
|
|
value: Any
|
|
status: str = "ok"
|
|
detail: str = ""
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return asdict(self)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SearchDashboardPanel:
|
|
key: str
|
|
title: str
|
|
metrics: list[SearchDashboardMetric] = field(default_factory=list)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"key": self.key,
|
|
"title": self.title,
|
|
"metrics": [m.to_dict() for m in self.metrics],
|
|
}
|
|
|
|
|
|
class SearchGatewayDashboard:
|
|
def overview(self, search_runs: list[dict[str, Any]] | None = None, providers: list[dict[str, Any]] | None = None) -> dict[str, Any]:
|
|
search_runs = search_runs or []
|
|
providers = providers or []
|
|
blocked = [r for r in search_runs if r.get("status") == "blocked"]
|
|
not_found = [r for r in search_runs if r.get("status") == "not_found"]
|
|
provider_failures = [p for p in providers if p.get("status") == "failed"]
|
|
panels = [
|
|
SearchDashboardPanel(
|
|
"search_health",
|
|
"搜索健康度",
|
|
[
|
|
SearchDashboardMetric("total_runs", "搜索请求数", len(search_runs)),
|
|
SearchDashboardMetric("not_found", "未找到结果", len(not_found), "warn" if not_found else "ok"),
|
|
SearchDashboardMetric("blocked", "合规阻断", len(blocked), "warn" if blocked else "ok"),
|
|
],
|
|
),
|
|
SearchDashboardPanel(
|
|
"provider_health",
|
|
"Provider 可用性",
|
|
[
|
|
SearchDashboardMetric("providers", "Provider 数量", len(providers)),
|
|
SearchDashboardMetric("provider_failures", "Provider 失败", len(provider_failures), "warn" if provider_failures else "ok"),
|
|
],
|
|
),
|
|
SearchDashboardPanel(
|
|
"evidence_quality",
|
|
"证据质量",
|
|
[
|
|
SearchDashboardMetric("with_evidence", "有证据的回答", len([r for r in search_runs if r.get("evidence")])) ,
|
|
SearchDashboardMetric("fallback_required", "要求多源降级", True),
|
|
],
|
|
),
|
|
]
|
|
return {
|
|
"module": "search-gateway",
|
|
"route": "/admin/ai-evolution/search",
|
|
"title": "AIOS 联网搜索治理中心",
|
|
"status": "attention_required" if provider_failures else "healthy",
|
|
"generated_at": utc_now(),
|
|
"panels": [p.to_dict() for p in panels],
|
|
"recent_runs": search_runs[-20:],
|
|
}
|
|
|
|
|
|
__all__ = ["SearchDashboardMetric", "SearchDashboardPanel", "SearchGatewayDashboard"]
|