linkong/planet
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

feat: ship persistent ai playground and alerts foundation

Browse Source
This commit is contained in:
linkong
2026-04-10 15:57:34 +08:00
parent 60f5ff9bab
commit 89a71e6f29
31 changed files with 4754 additions and 669 deletions
Download Patch File Download Diff File Expand all files Collapse all files

138
backend/app/services/bgp_ai_brief.py
View File

@@ -1,6 +1,7 @@
from __future__ import annotations
from collections import Counter
from typing import Any
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
@@ -11,6 +12,7 @@ from app.models.bgp_incident import BGPIncident
from app.models.bgp_observation import BGPObservation
from app.schemas.ai import SituationalAnalysisRequest
from app.services.bgp_collectors import build_bgp_collector_coverage
from app.services.bgp_enrichment import lookup_prefix_geography
def _format_counter(counter: dict[str, int], empty_text: str = "无") -> str:
@@ -30,13 +32,59 @@ def _severity_rank(value: str | None) -> int:
return order.get((value or "").lower(), 99)
def _normalize_geo_key(country: str | None, city: str | None) -> str:
if city and country:
return f"{city}, {country}"
return city or country or "未知区域"
def _top_counter_items(counter: Counter[str], limit: int = 5) -> dict[str, int]:
return {name: count for name, count in counter.most_common(limit) if name}
def _collect_incident_regions(incidents: list[BGPIncident]) -> Counter[str]:
counter: Counter[str] = Counter()
for item in incidents:
for region in item.affected_regions or []:
if not isinstance(region, dict):
continue
counter[_normalize_geo_key(region.get("country"), region.get("city"))] += 1
return counter
def _collect_collector_regions(collectors: list[dict[str, Any]]) -> Counter[str]:
counter: Counter[str] = Counter()
for item in collectors:
counter[_normalize_geo_key(item.get("country"), item.get("city"))] += int(item.get("recent_24h_observation_count") or 0)
return counter
def _format_geo_evidence(prefix_geographies: dict[str, dict[str, Any]], limit: int = 6) -> str:
if not prefix_geographies:
return "没有命中 prefix geography 证据。"
rows = []
for prefix, item in list(prefix_geographies.items())[:limit]:
region = _normalize_geo_key(item.get("country"), item.get("city"))
source = item.get("source") or item.get("geography_mode") or "unknown"
as_hint = item.get("asn")
as_name = item.get("as_name")
as_text = ""
if as_hint:
as_text = f" / ASN AS{as_hint}"
if as_name:
as_text += f" ({as_name})"
rows.append(f"{prefix} -> {region} / 来源 {source}{as_text}")
return "；".join(rows)
async def build_bgp_brief_request(
db: AsyncSession,
*,
incident_limit: int = 5,
anomaly_limit: int = 6,
collector_limit: int = 5,
) -> SituationalAnalysisRequest:
) -> tuple[SituationalAnalysisRequest, list[str], dict[str, int | str | dict[str, int]]]:
incidents_result = await db.execute(
select(BGPIncident)
.order_by(BGPIncident.created_at.desc(), BGPIncident.id.desc())
@@ -68,6 +116,7 @@ async def build_bgp_brief_request(
incident_type_counts = Counter((item.incident_type or "unknown") for item in incidents)
anomaly_type_counts = Counter((item.anomaly_type or "unknown") for item in anomalies)
event_type_counts = Counter((item.event_type or "unknown") for item in observations)
incident_region_counts = _collect_incident_regions(incidents)
top_collectors = sorted(
active_collectors,
@@ -77,6 +126,29 @@ async def build_bgp_brief_request(
str(item["collector"]),
),
)[: max(collector_limit, 1)]
collector_region_counts = _collect_collector_regions(top_collectors)
prefix_candidates = sorted(
{
prefix
for item in incidents
for prefix in (item.affected_prefixes or [])
if prefix
}
| {item.prefix for item in anomalies if item.prefix}
)
prefix_geographies = await lookup_prefix_geography(db, prefix_candidates) if prefix_candidates else {}
geography_region_counts = Counter(
_normalize_geo_key(item.get("country"), item.get("city"))
for item in prefix_geographies.values()
if item.get("country") or item.get("city")
)
hotspot_region_counts = geography_region_counts + incident_region_counts
collector_bias_regions = [
region
for region, count in collector_region_counts.most_common(3)
if count > hotspot_region_counts.get(region, 0)
]
observations_lines: list[str] = [
f"当前共有 {total_incidents} 起 BGP incidents、{total_anomalies} 条 anomalies、{total_observations} 条原始观测事件。",
@@ -88,6 +160,27 @@ async def build_bgp_brief_request(
f"观测事件类型分布：{_format_counter(dict(event_type_counts.most_common(6)))}。",
]
if hotspot_region_counts:
observations_lines.append(
"区域热点事实层："
+ _format_counter(_top_counter_items(hotspot_region_counts, limit=5), empty_text="无明显区域聚集")
+ "。"
)
if prefix_geographies:
observations_lines.append("Prefix geography 证据：" + _format_geo_evidence(prefix_geographies))
if collector_bias_regions:
observations_lines.append(
"观测偏差提示：重点观测站最近 24h 活跃度更集中在 "
+ "、".join(collector_bias_regions)
+ "，这些区域的事件升温结论需要结合 prefix geography 与 affected regions 交叉验证。"
)
elif top_collectors:
observations_lines.append(
"观测偏差提示：当前未发现明显高于区域热点事实层的单一观测站集中区域，但仍需区分 collector coverage 与真实区域风险。"
)
if incidents:
observations_lines.append(
"最近 incident 摘要：" + "；".join(
@@ -124,24 +217,43 @@ async def build_bgp_brief_request(
)
)
context = {
"source": "bgp-overview",
"incident_total": total_incidents,
"anomaly_total": total_anomalies,
"observation_total": total_observations,
"active_collectors": len(active_collectors),
"top_incident_types": dict(incident_type_counts.most_common(5)),
"top_anomaly_types": dict(anomaly_type_counts.most_common(6)),
"top_event_types": dict(event_type_counts.most_common(6)),
"region_hotspots": _top_counter_items(hotspot_region_counts, limit=6),
"incident_regions": _top_counter_items(incident_region_counts, limit=6),
"collector_bias_regions": collector_bias_regions,
"prefix_geography_sources": dict(
Counter(str(item.get("source") or "unknown") for item in prefix_geographies.values()).most_common(5)
),
"prefix_geography_sample": {
prefix: {
"country": item.get("country"),
"city": item.get("city"),
"source": item.get("source"),
"asn": item.get("asn"),
"as_name": item.get("as_name"),
}
for prefix, item in list(prefix_geographies.items())[:8]
},
}
return SituationalAnalysisRequest(
title="BGP 态势 AI 简报",
objective="基于当前 BGP incidents、anomalies、原始观测事件与观测站覆盖情况，生成一份面向操作员的简明态势简报，突出当前风险、证据和优先动作。",
objective="基于当前 BGP incidents、anomalies、原始观测事件、观测站覆盖与 prefix geography 证据，生成一份面向操作员的简明态势简报，突出区域热点、观测偏差、当前风险、证据和优先动作。",
observations=observations_lines,
constraints=[
"明确区分事实、推断与建议。",
"优先指出需要立即关注的高严重度 incident 或异常模式。",
"需要单独指出哪些区域结论来自 prefix geography / affected regions，哪些可能受 collector coverage 偏差影响。",
"结论应服务值班排障，不要写成泛泛的模型演示文案。",
"如果证据不足，要明确指出缺失数据。",
],
context={
"source": "bgp-overview",
"incident_total": total_incidents,
"anomaly_total": total_anomalies,
"observation_total": total_observations,
"active_collectors": len(active_collectors),
"top_incident_types": dict(incident_type_counts.most_common(5)),
"top_anomaly_types": dict(anomaly_type_counts.most_common(6)),
"top_event_types": dict(event_type_counts.most_common(6)),
},
)
context=context,
), observations_lines, context