7e3cc8b3b0
Some checks failed
CD Pipeline / build-and-deploy (push) Has been cancelled
原設計 asyncio.wait_for(timeout_sec=25s) 是任意截斷，
只要 LLM 超過時限就降級為 confidence=20%，根本沒有分析。
正確做法：
- 移除所有 4 個 agent 的 asyncio.wait_for() 包裝
- 只留 except Exception 捕真實異常（連線失敗、模型崩潰）
- 全流程由 Orchestrator GLOBAL_TIMEOUT_SEC=90s 防掛死
- _PER_AGENT_TIMEOUT_SEC 常數廢棄移除
影響：LLM 推理多久就等多久，不再人工截斷，
deepseek-r1:14b 等模型得以完整輸出分析結果。
2026-04-16 ogt + Claude Sonnet 4.6
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
229 lines
8.0 KiB
Python
229 lines
8.0 KiB
Python
"""
|
||
AWOOOI AIOps Phase 2 — Critic Agent（質疑者）
|
||
=============================================
|
||
職責：刻意唱反調，防止幻覺與 echo chamber
|
||
|
||
輸入：DiagnosisReport + ActionPlan（兩者都看）
|
||
輸出：CriticReport（challenges[] 列表 + overall_assessment）
|
||
|
||
設計原則：
|
||
1. Critic 的工作是找漏洞，不是說好話（防 sycophancy）
|
||
2. prompt 強制要求批判性思維：「如果診斷是錯的，還有哪 3 種可能？」
|
||
3. challenge_count > 0 是 Phase 2 退出條件之一
|
||
4. Critic 連續 3 次找到 Diagnostician 嚴重漏洞 → 觸發 Diagnostician 狀態不穩（Phase 4 實作）
|
||
5. 熔斷降級：LLM 失敗 → 輸出空 challenges（不阻塞 Coordinator）
|
||
6. Critic 和 Reviewer 並行執行（都不阻塞對方）
|
||
|
||
ADR-082: Phase 2 多 Agent 協作
|
||
2026-04-15 ogt + Claude Sonnet 4.6（亞太）: Phase 2 初始建立
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import time
|
||
from typing import Any
|
||
|
||
import structlog
|
||
|
||
from src.agents.base import BaseAgent
|
||
from src.agents.protocol import (
|
||
ActionPlan,
|
||
AgentRole,
|
||
AgentVote,
|
||
Challenge,
|
||
CriticReport,
|
||
DiagnosisReport,
|
||
)
|
||
from src.services.sanitization_service import sanitize
|
||
|
||
logger = structlog.get_logger(__name__)
|
||
|
||
# Critic 挑戰數量上限（防止 LLM 生成無限質疑）
|
||
MAX_CHALLENGES = 5
|
||
|
||
|
||
class CriticAgent(BaseAgent):
|
||
"""
|
||
Critic Agent — 系統性懷疑論者
|
||
|
||
Usage:
|
||
agent = CriticAgent()
|
||
report = await agent.run(diagnosis, plan)
|
||
"""
|
||
|
||
AGENT_NAME = AgentRole.CRITIC.value
|
||
AGENT_DESCRIPTION = (
|
||
"Devil's advocate. Challenges diagnosis and proposed actions to prevent "
|
||
"hallucination and echo chamber effects."
|
||
)
|
||
|
||
async def run(
|
||
self,
|
||
diagnosis: DiagnosisReport,
|
||
plan: ActionPlan,
|
||
timeout_sec: float = 0.0, # noqa: ARG002 — 已廢棄，保留簽名相容性
|
||
) -> CriticReport:
|
||
"""
|
||
批判性審查診斷和方案。
|
||
|
||
Args:
|
||
diagnosis: Diagnostician 輸出
|
||
plan: Solver 輸出
|
||
timeout_sec: 已廢棄 (2026-04-16 ogt) — LLM 等完整回應，真實異常才降級
|
||
|
||
Returns:
|
||
CriticReport（真實異常時 degraded=True）
|
||
"""
|
||
start_ms = int(time.monotonic() * 1000)
|
||
|
||
try:
|
||
report = await self._critique(diagnosis, plan)
|
||
report.latency_ms = int(time.monotonic() * 1000) - start_ms
|
||
logger.info(
|
||
"critic_done",
|
||
challenges=report.challenge_count,
|
||
has_critical=report.has_critical_challenge,
|
||
vote=report.vote,
|
||
latency_ms=report.latency_ms,
|
||
)
|
||
return report
|
||
|
||
except Exception:
|
||
latency = int(time.monotonic() * 1000) - start_ms
|
||
logger.exception("critic_error")
|
||
return self._degraded_report(latency, "error")
|
||
|
||
async def _critique(
|
||
self,
|
||
diagnosis: DiagnosisReport,
|
||
plan: ActionPlan,
|
||
) -> CriticReport:
|
||
"""LLM 批判性推理。"""
|
||
top_hypothesis = diagnosis.top_hypothesis
|
||
top_candidate = plan.top_candidate
|
||
|
||
prompt = self._build_prompt({
|
||
"hypothesis": top_hypothesis.description if top_hypothesis else "（無假設）",
|
||
"action": top_candidate.action if top_candidate else "（無方案）",
|
||
"confidence": top_hypothesis.confidence if top_hypothesis else 0.0,
|
||
})
|
||
|
||
from src.services.openclaw import get_openclaw
|
||
openclaw = get_openclaw()
|
||
response_text, _provider, success = await openclaw.call(prompt)
|
||
|
||
if not success or not response_text:
|
||
return self._degraded_report(0, "llm_failed")
|
||
|
||
parsed = self._parse_response(sanitize(response_text, "critic_output"))
|
||
challenges = _extract_challenges(parsed)
|
||
|
||
# 有 critical challenge → vote = REJECT
|
||
vote = AgentVote.REJECT if any(c.severity == "critical" for c in challenges) else AgentVote.APPROVE
|
||
|
||
return CriticReport(
|
||
challenges=challenges,
|
||
overall_assessment=str(parsed.get("overall_assessment", ""))[:1000],
|
||
latency_ms=0,
|
||
vote=vote,
|
||
)
|
||
|
||
def _build_prompt(self, context: dict[str, Any]) -> str:
|
||
return f"""你是 AWOOOI SRE 系統的質疑者 Agent（Critic）。
|
||
|
||
你的工作是：找出診斷和方案的弱點。不是說好話，是找漏洞。
|
||
|
||
當前診斷：{context.get("hypothesis", "")}
|
||
當前方案：{context.get("action", "")}
|
||
診斷信心：{context.get("confidence", 0.0):.0%}
|
||
|
||
必須回答以下問題（每個問題產出一個 challenge）：
|
||
1. 如果這個診斷是錯的，還有哪些可能的根因？
|
||
2. 這個方案有什麼副作用或風險？
|
||
3. 是否有更好的替代方案被忽略了？
|
||
|
||
每個 challenge 標記嚴重度：
|
||
- "minor"：小瑕疵，不影響執行
|
||
- "major"：值得 Coordinator 考慮，但不是阻擋條件
|
||
- "critical"：嚴重邏輯漏洞，必須阻止此方案執行
|
||
|
||
以 JSON 回覆：
|
||
{{
|
||
"challenges": [
|
||
{{
|
||
"target": "diagnosis",
|
||
"argument": "可能是 OOM 但也可能是 code bug，需要看 GC logs 確認",
|
||
"severity": "major"
|
||
}}
|
||
],
|
||
"overall_assessment": "診斷可信但方案風險偏高"
|
||
}}"""
|
||
|
||
def _parse_response(self, response: str) -> dict[str, Any]:
|
||
return self._extract_json(response)
|
||
|
||
def analyze(self, context: dict[str, Any]) -> Any:
|
||
raise NotImplementedError("Use run() for Phase 2 agents")
|
||
|
||
def _degraded_report(
|
||
self,
|
||
latency_ms: int,
|
||
reason: str = "unknown",
|
||
) -> CriticReport:
|
||
"""熔斷降級：輸出空 challenges（不阻塞 Coordinator）"""
|
||
return CriticReport(
|
||
challenges=[],
|
||
overall_assessment=f"[降級] Critic LLM 失敗（{reason}），跳過批判性審查",
|
||
latency_ms=latency_ms,
|
||
vote=AgentVote.ABSTAIN,
|
||
degraded=True,
|
||
)
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Helpers
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
def _extract_challenges(parsed: dict[str, Any]) -> list[Challenge]:
|
||
"""從 LLM 解析結果提取 challenges（按嚴重度排序）。"""
|
||
raw = parsed.get("challenges", [])
|
||
challenges = []
|
||
severity_order = {"critical": 0, "major": 1, "minor": 2}
|
||
|
||
for item in raw:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
c = Challenge(
|
||
target=str(item.get("target", "unknown"))[:50],
|
||
argument=str(item.get("argument", ""))[:500],
|
||
severity=item.get("severity", "minor") if item.get("severity") in severity_order else "minor",
|
||
)
|
||
challenges.append(c)
|
||
|
||
challenges.sort(key=lambda c: severity_order.get(c.severity, 2))
|
||
return challenges[:MAX_CHALLENGES]
|
||
|
||
|
||
def compute_input_hash(diagnosis: DiagnosisReport, plan: ActionPlan) -> str:
|
||
key = diagnosis.evidence_snapshot_id + (
|
||
diagnosis.top_hypothesis.description if diagnosis.top_hypothesis else ""
|
||
) + (
|
||
plan.top_candidate.action if plan.top_candidate else ""
|
||
)
|
||
return hashlib.sha256(key.encode()).hexdigest()[:16]
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Singleton
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
|
||
_agent: CriticAgent | None = None
|
||
|
||
|
||
def get_critic_agent() -> CriticAgent:
|
||
global _agent
|
||
if _agent is None:
|
||
_agent = CriticAgent()
|
||
return _agent
|