393 lines
15 KiB
Python
393 lines
15 KiB
Python
"""
|
||
Hermes Rule Quality Advisor — ADR-090 § E3 AI 規則品質建議
|
||
==========================================================
|
||
每日 04:00 Taipei 分析 alert_rule_catalog,對 noise_rate > 0.7 的 rule 推 Telegram
|
||
建議 + 寫 aol(rule_rejected) 稽核,人工決策是否 deprecate.
|
||
|
||
職責邊界:
|
||
✅ 讀 alert_rule_catalog WHERE noise_rate >= 0.7
|
||
✅ 為每條寫 aol(rule_rejected) + proposed_action='review_or_deprecate'
|
||
✅ 推 Telegram 通知 SRE group (格式化清單)
|
||
⏳ 不自動改 review_status (統帥鐵律: AI 不做最終決策)
|
||
⏳ TODO: LLM 分析每條 rule 的假報真因 (下一階段)
|
||
|
||
統帥鐵律對齊:
|
||
- 禁止寫死規則做最終決策 → 本 agent 只推建議,人工決策
|
||
- 朝 AI 自主化方向 → aol 留 trail,未來可升級為 LLM 判斷
|
||
- noise_rate threshold 0.7 是「觸發討論」而非「自動動作」
|
||
|
||
排程:
|
||
- 首次延遲 420s
|
||
- 每日 04:00 Taipei
|
||
|
||
2026-04-19 ogt + Claude Opus 4.7 (1M context) Asia/Taipei
|
||
ADR-090 § E3 Hermes
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json as _json
|
||
import time as _time
|
||
from datetime import datetime, timedelta, timezone
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
_FIRST_DELAY_SEC = 420
|
||
_LOOP_BACKOFF_SEC = 1800
|
||
_DAILY_TRIGGER_HOUR_TAIPEI = 4
|
||
|
||
# 觸發討論的噪音閾值
|
||
_NOISE_THRESHOLD = 0.7
|
||
# 樣本不足不發建議 (避免只 fire 1 次就標為噪音)
|
||
_MIN_SAMPLE_SIZE = 5
|
||
|
||
|
||
async def run_hermes_rule_quality_loop() -> None:
|
||
"""每日 04:00 分析 rule 品質."""
|
||
logger.info("hermes_rule_quality_loop_started")
|
||
await asyncio.sleep(_FIRST_DELAY_SEC)
|
||
|
||
while True:
|
||
try:
|
||
await analyze_once()
|
||
except Exception as e:
|
||
logger.exception("hermes_rule_quality_loop_error", error=str(e))
|
||
await asyncio.sleep(_LOOP_BACKOFF_SEC)
|
||
continue
|
||
|
||
sleep_sec = _seconds_until_next_trigger()
|
||
logger.info("hermes_rule_quality_next_tick", sleep_sec=sleep_sec)
|
||
await asyncio.sleep(sleep_sec)
|
||
|
||
|
||
async def analyze_once() -> dict[str, int]:
|
||
"""一次分析: 找噪音 rule + LLM 分析真因 + 推建議 + aol 留痕.
|
||
|
||
2026-04-19 P0 修: 加 daily leader_lock 避免多 Pod 重複推.
|
||
"""
|
||
from src.services.ai_advisory_helpers import try_acquire_daily_lock
|
||
|
||
if not await try_acquire_daily_lock("hermes_rule_quality"):
|
||
logger.info("hermes_analyze_skipped_not_leader")
|
||
return {"skipped": "not_leader"}
|
||
|
||
started_ms = _time.time()
|
||
stats = {"noisy_rules": 0, "llm_analyzed": 0, "advisories_written": 0, "telegram_sent": 0}
|
||
error_msg: str | None = None
|
||
llm_analyses: dict[str, dict[str, Any]] = {}
|
||
|
||
try:
|
||
noisy = await _fetch_noisy_rules()
|
||
stats["noisy_rules"] = len(noisy)
|
||
|
||
# v2 升級: 對每條 noisy rule 跑 LLM 分析真因 + 具體建議
|
||
for r in noisy:
|
||
analysis = await _llm_analyze_noisy_rule(r)
|
||
if analysis:
|
||
llm_analyses[r["rule_name"]] = analysis
|
||
stats["llm_analyzed"] += 1
|
||
|
||
for r in noisy:
|
||
ok = await _write_advisory_aol(r, llm_analyses.get(r["rule_name"]))
|
||
if ok:
|
||
stats["advisories_written"] += 1
|
||
|
||
if noisy:
|
||
sent = await _send_telegram_summary(noisy, llm_analyses)
|
||
stats["telegram_sent"] = 1 if sent else 0
|
||
|
||
except Exception as e:
|
||
error_msg = f"{type(e).__name__}: {e}"[:1000]
|
||
logger.exception("hermes_analyze_once_failed", error=error_msg)
|
||
|
||
duration_ms = int((_time.time() - started_ms) * 1000)
|
||
logger.info(
|
||
"hermes_rule_quality_once_done",
|
||
noisy=stats["noisy_rules"],
|
||
llm_analyzed=stats["llm_analyzed"],
|
||
advisories=stats["advisories_written"],
|
||
telegram_sent=stats["telegram_sent"],
|
||
duration_ms=duration_ms,
|
||
)
|
||
return stats
|
||
|
||
|
||
# ============================================================================
|
||
# v2 LLM 分析 — 統帥鐵律「朝 AI 自主化方向」
|
||
# ============================================================================
|
||
|
||
_LLM_ANALYZE_PROMPT = """你是 AWOOOI SRE 告警規則品質分析專家。以下是一條 Prometheus alerting rule 過去 30 天的統計,請分析假報真因並提出具體改進建議。
|
||
|
||
## 告警規則
|
||
- rule_name: {rule_name}
|
||
- severity: {severity}
|
||
- expr: {expr}
|
||
- for: {duration_seconds}s
|
||
- labels: {labels}
|
||
- annotations: {annotations}
|
||
|
||
## 過去 30 天統計
|
||
- true_positive (確實解決的): {tp}
|
||
- false_positive (有破壞性動作但 EXPIRED 沒人理): {fp}
|
||
- noise_rate: {noise_rate}
|
||
|
||
## 輸出規格 (必須是合法 JSON,純 JSON 無前後文字)
|
||
{{
|
||
"probable_root_causes": ["3-4 個候選真因,繁中"],
|
||
"recommended_actions": [
|
||
{{"action": "adjust_threshold|add_for_duration|refine_labels|deprecate|split_rule|keep_as_is", "detail": "具體怎麼做,繁中一句話"}}
|
||
],
|
||
"confidence": 0.0-1.0,
|
||
"should_deprecate": true/false
|
||
}}
|
||
|
||
## 分析思路
|
||
1. 看 expr 是否過於敏感 (閾值太低 / 沒有 for: window)
|
||
2. 看 annotations 是否暗示「這是真實需要處理的問題」但被 AI 判 NO_ACTION → 可能是 action 流程問題而非規則問題
|
||
3. 考慮 severity warning/critical 是否合理
|
||
"""
|
||
|
||
|
||
async def _llm_analyze_noisy_rule(rule: dict[str, Any]) -> dict[str, Any] | None:
|
||
"""用 OpenClaw (多 provider) 分析噪音真因. 失敗回 None 不阻塞.
|
||
|
||
2026-04-19 P1.2 重構: 使用 llm_json_parser.parse_llm_json_response 共用 helper
|
||
(原 30 行重複 3-path parse 邏輯已抽出到 services/llm_json_parser.py).
|
||
"""
|
||
try:
|
||
import json as _j
|
||
from src.services.llm_json_parser import parse_llm_json_response
|
||
from src.services.openclaw import get_openclaw
|
||
|
||
prompt = _LLM_ANALYZE_PROMPT.format(
|
||
rule_name=rule["rule_name"],
|
||
severity=rule["severity"] or "-",
|
||
expr=(rule.get("expr") or "")[:500],
|
||
duration_seconds=rule.get("duration_seconds") or 0,
|
||
labels=_j.dumps(rule.get("labels", {}), ensure_ascii=False)[:300],
|
||
annotations=_j.dumps(rule.get("annotations", {}), ensure_ascii=False)[:300],
|
||
tp=rule["tp"],
|
||
fp=rule["fp"],
|
||
noise_rate=f"{rule['noise_rate']:.1%}",
|
||
)
|
||
openclaw = get_openclaw()
|
||
text, provider, success = await openclaw.call(prompt)
|
||
if not success or not text:
|
||
return None
|
||
|
||
parsed = parse_llm_json_response(
|
||
text,
|
||
required_key="recommended_actions",
|
||
logger_context=f"hermes:{rule['rule_name']}",
|
||
)
|
||
if parsed:
|
||
parsed["_llm_provider"] = provider
|
||
return parsed
|
||
except Exception as e:
|
||
logger.warning("hermes_llm_analyze_error", rule=rule["rule_name"], error=str(e))
|
||
return None
|
||
|
||
|
||
# ============================================================================
|
||
# 資料查詢
|
||
# ============================================================================
|
||
|
||
async def _fetch_noisy_rules() -> list[dict[str, Any]]:
|
||
"""撈 noise_rate >= 0.7 且樣本 >= 5 的 rules，或 AOL writeback 標記 draft 的 rules.
|
||
|
||
W2 PR-R2 2026-04-28 ogt + Claude Sonnet 4.6: 加 OR review_status = 'draft' 條件
|
||
讓 AOL writeback 觸發的 draft 規則能被 Hermes 自動推 Telegram 建議
|
||
（不再卡人工 SQL 才能觸發 advisory）
|
||
"""
|
||
from sqlalchemy import text as _sql
|
||
from src.db.base import get_db_context
|
||
|
||
try:
|
||
async with get_db_context() as db:
|
||
result = await db.execute(
|
||
_sql(f"""
|
||
SELECT
|
||
rule_id, rule_name, severity,
|
||
true_positive_count, false_positive_count, noise_rate,
|
||
last_fired_at, review_status
|
||
FROM alert_rule_catalog
|
||
WHERE (
|
||
(
|
||
noise_rate >= :thr
|
||
AND (true_positive_count + false_positive_count) >= :min_sample
|
||
AND (review_status IS NULL OR review_status = 'approved')
|
||
)
|
||
OR review_status = 'draft'
|
||
)
|
||
ORDER BY noise_rate DESC NULLS LAST,
|
||
(true_positive_count + false_positive_count) DESC
|
||
"""),
|
||
{"thr": _NOISE_THRESHOLD, "min_sample": _MIN_SAMPLE_SIZE},
|
||
)
|
||
return [
|
||
{
|
||
"rule_id": r.rule_id,
|
||
"rule_name": r.rule_name,
|
||
"severity": r.severity,
|
||
"tp": int(r.true_positive_count or 0),
|
||
"fp": int(r.false_positive_count or 0),
|
||
"noise_rate": float(r.noise_rate) if r.noise_rate else 0.0,
|
||
"last_fired_at": r.last_fired_at,
|
||
"review_status": r.review_status,
|
||
}
|
||
for r in result.fetchall()
|
||
]
|
||
except Exception as e:
|
||
logger.warning("fetch_noisy_rules_failed", error=str(e))
|
||
return []
|
||
|
||
|
||
# ============================================================================
|
||
# 建議寫入 (aol only,不改 rule 本身)
|
||
# ============================================================================
|
||
|
||
async def _write_advisory_aol(rule: dict[str, Any], llm_analysis: dict[str, Any] | None = None) -> bool:
|
||
"""寫 aol(rule_rejected) — 紀錄 AI 建議人工審查 + LLM 分析結果."""
|
||
try:
|
||
from sqlalchemy import text as _sql
|
||
from src.db.base import get_db_context
|
||
|
||
input_payload = {
|
||
"rule_name": rule["rule_name"],
|
||
"severity": rule["severity"],
|
||
"noise_rate": rule["noise_rate"],
|
||
"true_positive_count": rule["tp"],
|
||
"false_positive_count": rule["fp"],
|
||
}
|
||
output_payload: dict[str, Any] = {
|
||
"proposed_action": "review_or_deprecate",
|
||
"reason": (
|
||
f"過去 30d noise_rate {rule['noise_rate']:.1%} "
|
||
f"(tp={rule['tp']}, fp={rule['fp']}),"
|
||
f"假報過多應考慮 deprecate 或改進 expr"
|
||
),
|
||
"requires_human_decision": True,
|
||
}
|
||
if llm_analysis:
|
||
output_payload["llm_analysis"] = llm_analysis
|
||
|
||
async with get_db_context() as db:
|
||
await db.execute(
|
||
_sql("""
|
||
INSERT INTO automation_operation_log (
|
||
operation_type, actor, status,
|
||
input, output, tags
|
||
) VALUES (
|
||
'rule_rejected',
|
||
'hermes_rule_quality',
|
||
'success',
|
||
CAST(:input AS jsonb),
|
||
CAST(:output AS jsonb),
|
||
:tags
|
||
)
|
||
"""),
|
||
{
|
||
"input": _json.dumps(input_payload, ensure_ascii=False),
|
||
"output": _json.dumps(output_payload, ensure_ascii=False),
|
||
"tags": ["hermes", "rule_quality", "advisory"],
|
||
},
|
||
)
|
||
return True
|
||
except Exception as e:
|
||
logger.warning("write_advisory_aol_failed", rule=rule["rule_name"], error=str(e))
|
||
return False
|
||
|
||
|
||
# ============================================================================
|
||
# Telegram 推送
|
||
# ============================================================================
|
||
|
||
async def _send_telegram_summary(
|
||
noisy: list[dict[str, Any]],
|
||
llm_analyses: dict[str, dict[str, Any]] | None = None,
|
||
) -> bool:
|
||
"""推 Telegram 摘要訊息給 SRE group,含 LLM 分析結果 + 互動按鈕 (P0 修)."""
|
||
try:
|
||
import html
|
||
|
||
from src.services.ai_advisory_helpers import (
|
||
build_ai_advisory_keyboard,
|
||
is_snoozed,
|
||
)
|
||
from src.services.telegram_gateway import (
|
||
_telegram_send_delivery_succeeded,
|
||
get_telegram_gateway,
|
||
)
|
||
|
||
# Snooze check: 以第一條 noisy rule_name 為 key
|
||
primary_rule = noisy[0]["rule_name"] if noisy else "unknown"
|
||
if await is_snoozed("rule_quality", primary_rule):
|
||
logger.info("hermes_rule_snoozed", rule=primary_rule)
|
||
return False
|
||
|
||
llm_analyses = llm_analyses or {}
|
||
lines = [
|
||
"🔍 <b>Hermes 規則品質檢測 (AI 分析)</b>",
|
||
f"檢測到 {len(noisy)} 條規則噪音率 ≥ {_NOISE_THRESHOLD:.0%},請統帥審查:",
|
||
"",
|
||
]
|
||
for r in noisy[:8]: # LLM 分析含建議,單條訊息較長,只秀 8 條
|
||
safe_name = html.escape(r["rule_name"])
|
||
lines.append(
|
||
f"🟡 <code>{safe_name}</code> — noise {r['noise_rate']:.1%} (tp={r['tp']} fp={r['fp']})"
|
||
)
|
||
ai = llm_analyses.get(r["rule_name"])
|
||
if ai:
|
||
deprecate = ai.get("should_deprecate")
|
||
conf = ai.get("confidence", 0.0)
|
||
lines.append(f" AI 判定: should_deprecate={deprecate} confidence={conf:.0%}")
|
||
actions = ai.get("recommended_actions", []) or []
|
||
for act in actions[:2]: # 最多秀前 2 個建議
|
||
safe_detail = html.escape(str(act.get("detail", ""))[:120])
|
||
lines.append(f" ▸ <i>{html.escape(str(act.get('action', '')))}</i>: {safe_detail}")
|
||
else:
|
||
lines.append(" (LLM 分析不可用,僅依噪音率判斷)")
|
||
lines.append("")
|
||
if len(noisy) > 8:
|
||
lines.append(f"…還有 {len(noisy) - 8} 條,見 automation_operation_log")
|
||
lines.append("決策: 人工 UPDATE alert_rule_catalog SET review_status='deprecated' WHERE rule_name='...'")
|
||
|
||
msg = "\n".join(lines)
|
||
keyboard = build_ai_advisory_keyboard(
|
||
advisory_type="rule_quality",
|
||
advisory_id=primary_rule,
|
||
include_view=False,
|
||
include_produce_cmd=False,
|
||
)
|
||
|
||
tg = get_telegram_gateway()
|
||
result = await tg.send_canonical_message(
|
||
product_id="awoooi",
|
||
signal_family="product_raw_monitoring",
|
||
severity="P2",
|
||
text=msg,
|
||
parse_mode="HTML",
|
||
reply_markup=keyboard,
|
||
)
|
||
return _telegram_send_delivery_succeeded(result)
|
||
except Exception as e:
|
||
logger.warning("hermes_telegram_send_failed", error=str(e))
|
||
return False
|
||
|
||
|
||
# ============================================================================
|
||
# 時間
|
||
# ============================================================================
|
||
|
||
def _seconds_until_next_trigger() -> float:
|
||
tz_taipei = timezone(timedelta(hours=8))
|
||
now = datetime.now(tz_taipei)
|
||
today_trigger = now.replace(hour=_DAILY_TRIGGER_HOUR_TAIPEI, minute=0, second=0, microsecond=0)
|
||
if now >= today_trigger:
|
||
today_trigger = today_trigger + timedelta(days=1)
|
||
delta = (today_trigger - now).total_seconds()
|
||
return max(300.0, min(delta, 25 * 3600))
|