From 92ee07ad4b2bacb517b56855226c2007576e8234 Mon Sep 17 00:00:00 2001 From: OG T Date: Thu, 26 Mar 2026 10:02:31 +0800 Subject: [PATCH] =?UTF-8?q?refactor(api):=20Phase=2017=20agents.py=20Route?= =?UTF-8?q?r=20=E5=B1=A4=E9=81=95=E8=A6=8F=E4=BF=AE=E5=BE=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 建立 AgentService 封裝所有 Redis 操作 - 定義 IAgentTaskRepository Protocol 介面支援 DI - Router 層改用 AgentService，不再直接 get_redis() - 符合 leWOOOgo 積木化原則 (Router → Service → Repository) Co-Authored-By: Claude Opus 4.5 --- apps/api/src/api/v1/agents.py | 389 +++------------ apps/api/src/services/agent_service.py | 664 +++++++++++++++++++++++++ 2 files changed, 723 insertions(+), 330 deletions(-) create mode 100644 apps/api/src/services/agent_service.py diff --git a/apps/api/src/api/v1/agents.py b/apps/api/src/api/v1/agents.py index e7cde8df5..d9cf4fbaa 100644 --- a/apps/api/src/api/v1/agents.py +++ b/apps/api/src/api/v1/agents.py @@ -14,6 +14,11 @@ Phase 9.4-9.5 核心功能: 3. Redis Working Memory 儲存結果 4. SSE 推送即時進度 +Phase 17 技術債修復 (2026-03-26): +- Router 層不再直接存取 Redis +- 所有業務邏輯封裝至 AgentService +- 符合 leWOOOgo 積木化原則 + 統帥鐵律: - 所有分析任務必須可追蹤 (task_id) - 超過 60 秒的分析必須用 BackgroundTasks @@ -22,48 +27,24 @@ Phase 9.4-9.5 核心功能: import asyncio import json -from datetime import UTC, datetime -from enum import Enum from typing import Any -from uuid import uuid4 from fastapi import APIRouter, BackgroundTasks, HTTPException, status from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from src.core.logging import get_logger -from src.core.redis_client import get_redis -from src.core.sse import EventType, SSEEvent, get_publisher -from src.models.incident import Incident, IncidentStatus, Severity, Signal -from src.services.consensus_engine import ( - get_consensus_engine, +from src.core.sse import get_publisher +from src.services.agent_service import ( + AgentService, + TaskState, + get_agent_service, ) router = APIRouter(prefix="/agents", tags=["Agent Teams"]) logger = get_logger("awoooi.agents") -# ============================================================================= -# Constants -# ============================================================================= - -TASK_PREFIX = "agent_task:" -TASK_TTL = 604800 # 7 天 - - -# ============================================================================= -# Task States -# ============================================================================= - -class TaskState(str, Enum): - """分析任務狀態""" - PENDING = "pending" # 等待中 - ANALYZING = "analyzing" # 分析中 - CONSENSUS = "consensus" # 共識計算中 - COMPLETED = "completed" # 已完成 - FAILED = "failed" # 失敗 - - # ============================================================================= # Request/Response Models # ============================================================================= @@ -131,199 +112,25 @@ class TaskResultResponse(BaseModel): # ============================================================================= -# Background Task Handler +# Background Task Wrapper # ============================================================================= -async def run_agent_analysis( +async def _run_analysis_task( + service: AgentService, task_id: str, - incident: Incident, + incident_id: str, ) -> None: """ - 背景執行 Agent Teams 分析 + 背景任務包裝器 - 流程: - 1. 更新狀態為 ANALYZING - 2. 收集各專家意見 - 3. 計算共識 - 4. 儲存結果 - 5. 推送 SSE 通知 + 從 AgentService 取得 Incident 並執行分析 """ - redis_client = get_redis() - consensus_engine = get_consensus_engine() - task_key = f"{TASK_PREFIX}{task_id}" + incident = await service.get_incident(incident_id) + if incident is None: + logger.error("background_task_incident_not_found", incident_id=incident_id) + return - try: - # Step 1: 更新狀態 - await _update_task_state( - task_id, - TaskState.ANALYZING, - progress=10, - current_step="正在收集專家意見...", - ) - - # 推送 SSE 進度 - publisher = await get_publisher() - await publisher.publish(SSEEvent( - type=EventType.AI_THINKING, - data={ - "task_id": task_id, - "state": TaskState.ANALYZING.value, - "progress": 10, - "message": "Agent Teams 分析開始", - }, - )) - - # Step 2: 收集意見 (模擬進度) - opinions = await consensus_engine.gather_opinions(incident, timeout_sec=25.0) - - await _update_task_state( - task_id, - TaskState.CONSENSUS, - progress=60, - current_step="正在計算共識...", - agents_completed=len(opinions), - ) - - await publisher.publish(SSEEvent( - type=EventType.AI_THINKING, - data={ - "task_id": task_id, - "state": TaskState.CONSENSUS.value, - "progress": 60, - "message": f"已收集 {len(opinions)} 位專家意見", - }, - )) - - # Step 3: 計算共識 - consensus_score, recommended_action, dissenting = consensus_engine.calculate_consensus(opinions) - - await _update_task_state( - task_id, - TaskState.CONSENSUS, - progress=80, - current_step="正在產生最終決策...", - ) - - # Step 4: 產生最終決策 - result = await consensus_engine.generate_final_decision( - incident=incident, - opinions=opinions, - consensus_score=consensus_score, - recommended_action_type=recommended_action, - dissenting=dissenting, - ) - - # Step 5: 儲存完整結果 - task_data = { - "task_id": task_id, - "state": TaskState.COMPLETED.value, - "progress": 100, - "current_step": "分析完成", - "agents_completed": len(opinions), - "total_agents": 4, - "consensus_id": result.consensus_id, - "incident_id": incident.incident_id, - "consensus_score": result.consensus_score, - "recommended_action": result.recommended_action, - "recommended_kubectl": result.recommended_kubectl, - "risk_level": result.risk_level, - "final_reasoning": result.final_reasoning, - "opinions": [op.to_dict() for op in result.opinions], - "dissenting_opinions": result.dissenting_opinions, - "completed_at": datetime.now(UTC).isoformat(), - } - - await redis_client.set( - task_key, - json.dumps(task_data), - ex=TASK_TTL, - ) - - # 推送完成通知 - await publisher.publish(SSEEvent( - type=EventType.AI_THINKING, - data={ - "task_id": task_id, - "state": TaskState.COMPLETED.value, - "progress": 100, - "message": "分析完成", - "consensus_score": result.consensus_score, - "recommended_action": result.recommended_action, - }, - )) - - logger.info( - "agent_analysis_completed", - task_id=task_id, - consensus_id=result.consensus_id, - consensus_score=result.consensus_score, - ) - - except Exception as e: - logger.exception( - "agent_analysis_failed", - task_id=task_id, - error=str(e), - ) - - # 更新為失敗狀態 - task_data = { - "task_id": task_id, - "state": TaskState.FAILED.value, - "progress": 0, - "error": str(e), - "completed_at": datetime.now(UTC).isoformat(), - } - - await redis_client.set( - task_key, - json.dumps(task_data), - ex=TASK_TTL, - ) - - # 推送失敗通知 - publisher = await get_publisher() - await publisher.publish(SSEEvent( - type=EventType.ERROR, - data={ - "task_id": task_id, - "state": TaskState.FAILED.value, - "error": str(e), - }, - )) - - -async def _update_task_state( - task_id: str, - state: TaskState, - progress: int = 0, - current_step: str | None = None, - agents_completed: int = 0, -) -> None: - """更新任務狀態""" - redis_client = get_redis() - task_key = f"{TASK_PREFIX}{task_id}" - - # 讀取現有資料 - existing = await redis_client.get(task_key) - if existing: - task_data = json.loads(existing) - else: - task_data = {"task_id": task_id} - - # 更新欄位 - task_data.update({ - "state": state.value, - "progress": progress, - "current_step": current_step, - "agents_completed": agents_completed, - }) - - await redis_client.set( - task_key, - json.dumps(task_data), - ex=TASK_TTL, - ) + await service.run_analysis(task_id, incident) # ============================================================================= @@ -359,41 +166,25 @@ async def analyze( 返回 task_id 用於追蹤進度 """ - redis_client = get_redis() + service = get_agent_service() # 取得或建立 Incident - incident: Incident | None = None - if request.incident_id: # 從 Redis 讀取現有 Incident - key = f"incident:{request.incident_id}" - data = await redis_client.get(key) + incident = await service.get_incident(request.incident_id) - if not data: + if incident is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Incident not found: {request.incident_id}", ) - incident = Incident.model_validate_json(data) - elif request.severity and request.affected_services: # 建立臨時 Incident - signals = [] - if request.alert_names: - for alert_name in request.alert_names: - signals.append(Signal( - alert_name=alert_name, - severity=Severity(request.severity), - source="manual", - fired_at=datetime.now(UTC), - )) - - incident = Incident( - severity=Severity(request.severity), - status=IncidentStatus.INVESTIGATING, - signals=signals, + incident = service.create_adhoc_incident( + severity=request.severity, affected_services=request.affected_services, + alert_names=request.alert_names, ) else: @@ -403,28 +194,14 @@ async def analyze( ) # 建立任務 - task_id = f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}" - - # 初始化任務狀態 - task_data = { - "task_id": task_id, - "state": TaskState.PENDING.value, - "progress": 0, - "current_step": "任務已建立", - "agents_completed": 0, - "total_agents": 4, - "incident_id": incident.incident_id, - "started_at": datetime.now(UTC).isoformat(), - } - - await redis_client.set( - f"{TASK_PREFIX}{task_id}", - json.dumps(task_data), - ex=TASK_TTL, - ) + task_id = await service.create_analysis_task(incident, trigger="manual") # 加入背景任務 - background_tasks.add_task(run_agent_analysis, task_id, incident) + background_tasks.add_task( + service.run_analysis, + task_id, + incident, + ) logger.info( "agent_analysis_started", @@ -453,18 +230,15 @@ async def get_status(task_id: str) -> TaskStatusResponse: 返回進度百分比與目前步驟 """ - redis_client = get_redis() - task_key = f"{TASK_PREFIX}{task_id}" + service = get_agent_service() + task_data = await service.get_task_status(task_id) - data = await redis_client.get(task_key) - if not data: + if task_data is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Task not found: {task_id}", ) - task_data = json.loads(data) - return TaskStatusResponse( task_id=task_id, state=task_data.get("state", "unknown"), @@ -490,18 +264,15 @@ async def get_result(task_id: str) -> TaskResultResponse: 只有 COMPLETED 狀態才有完整結果 """ - redis_client = get_redis() - task_key = f"{TASK_PREFIX}{task_id}" + service = get_agent_service() + task_data = await service.get_task_result(task_id) - data = await redis_client.get(task_key) - if not data: + if task_data is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Task not found: {task_id}", ) - task_data = json.loads(data) - return TaskResultResponse( task_id=task_id, state=task_data.get("state", "unknown"), @@ -529,12 +300,11 @@ async def stream_progress(task_id: str) -> StreamingResponse: 客戶端可訂閱此端點接收即時更新 """ - redis_client = get_redis() - task_key = f"{TASK_PREFIX}{task_id}" + service = get_agent_service() # 驗證任務存在 - data = await redis_client.get(task_key) - if not data: + task_data = await service.get_task_status(task_id) + if task_data is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Task not found: {task_id}", @@ -550,20 +320,23 @@ async def stream_progress(task_id: str) -> StreamingResponse: try: # 發送初始狀態 - current_data = await redis_client.get(task_key) + current_data = await service.get_task_status(task_id) if current_data: - task_data = json.loads(current_data) - yield f"data: {json.dumps({'type': 'status', **task_data}, ensure_ascii=False)}\n\n" + payload = json.dumps( + {"type": "status", **current_data}, + ensure_ascii=False, + ) + yield f"data: {payload}\n\n" # 串流後續更新 async for event_str in publisher.stream(client): yield event_str # 檢查是否完成或失敗 - current_data = await redis_client.get(task_key) + current_data = await service.get_task_status(task_id) if current_data: - task_data = json.loads(current_data) - if task_data.get("state") in [TaskState.COMPLETED.value, TaskState.FAILED.value]: + final_states = [TaskState.COMPLETED.value, TaskState.FAILED.value] + if current_data.get("state") in final_states: break except asyncio.CancelledError: @@ -599,65 +372,21 @@ async def trigger_agent_analysis_for_incident( Returns: task_id if triggered, None if skipped """ - redis_client = get_redis() + service = get_agent_service() - # 讀取 Incident - key = f"incident:{incident_id}" - data = await redis_client.get(key) + task_id, incident = await service.trigger_for_incident(incident_id) - if not data: - logger.warning("trigger_agent_skipped_not_found", incident_id=incident_id) + if task_id is None: return None - incident = Incident.model_validate_json(data) - - # 判斷是否需要 Agent Teams (複雜決策條件) - should_trigger = ( - # P0/P1 緊急事件 - incident.severity in (Severity.P0, Severity.P1) - # 或多個服務受影響 - or len(incident.affected_services) > 2 - # 或多個告警 - or len(incident.signals) > 3 - ) - - if not should_trigger: - logger.debug( - "trigger_agent_skipped_simple_case", - incident_id=incident_id, - severity=incident.severity.value, - ) + if incident is None: return None - # 建立任務 - task_id = f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}" - - task_data = { - "task_id": task_id, - "state": TaskState.PENDING.value, - "progress": 0, - "current_step": "自動觸發 Agent Teams", - "agents_completed": 0, - "total_agents": 4, - "incident_id": incident_id, - "started_at": datetime.now(UTC).isoformat(), - "trigger": "auto", - } - - await redis_client.set( - f"{TASK_PREFIX}{task_id}", - json.dumps(task_data), - ex=TASK_TTL, - ) - # 加入背景任務 - background_tasks.add_task(run_agent_analysis, task_id, incident) - - logger.info( - "agent_analysis_auto_triggered", - task_id=task_id, - incident_id=incident_id, - severity=incident.severity.value, + background_tasks.add_task( + service.run_analysis, + task_id, + incident, ) return task_id diff --git a/apps/api/src/services/agent_service.py b/apps/api/src/services/agent_service.py new file mode 100644 index 000000000..8732fb462 --- /dev/null +++ b/apps/api/src/services/agent_service.py @@ -0,0 +1,664 @@ +""" +Agent Service - Phase 17 技術債修復 +=================================== +Phase 17 R4: 從 agents.py Router 抽離 Redis 操作 + +設計原則: +- Router 層只做 HTTP 路由，不直接存取 Redis +- Service 層封裝所有業務邏輯與 Redis 操作 +- 使用 Protocol 定義介面，支援依賴注入 + +版本: v1.0 +建立: 2026-03-26 (台北時區) +建立者: Claude Code (Phase 17 技術債修復) +""" + +import json +from datetime import UTC, datetime +from enum import Enum +from typing import Any, Protocol, runtime_checkable +from uuid import uuid4 + +from src.core.logging import get_logger +from src.core.redis_client import get_redis +from src.core.sse import EventType, SSEEvent, get_publisher +from src.models.incident import Incident, IncidentStatus, Severity, Signal +from src.services.consensus_engine import get_consensus_engine + +logger = get_logger("awoooi.agent_service") + + +# ============================================================================= +# Constants +# ============================================================================= + +TASK_PREFIX = "agent_task:" +INCIDENT_PREFIX = "incident:" +TASK_TTL = 604800 # 7 天 + + +# ============================================================================= +# Task States +# ============================================================================= + +class TaskState(str, Enum): + """分析任務狀態""" + PENDING = "pending" # 等待中 + ANALYZING = "analyzing" # 分析中 + CONSENSUS = "consensus" # 共識計算中 + COMPLETED = "completed" # 已完成 + FAILED = "failed" # 失敗 + + +# ============================================================================= +# Protocol Interface +# ============================================================================= + +@runtime_checkable +class IAgentTaskRepository(Protocol): + """ + Agent Task Repository Protocol + + 職責: Agent Task CRUD 操作 (Redis Working Memory) + """ + + async def create_task( + self, + task_id: str, + incident_id: str, + trigger: str = "manual", + ) -> dict[str, Any]: + """建立新的分析任務""" + ... + + async def get_task(self, task_id: str) -> dict[str, Any] | None: + """取得任務資料""" + ... + + async def update_task_state( + self, + task_id: str, + state: TaskState, + progress: int = 0, + current_step: str | None = None, + agents_completed: int = 0, + **extra_data: Any, + ) -> bool: + """更新任務狀態""" + ... + + async def save_task_result( + self, + task_id: str, + result_data: dict[str, Any], + ) -> bool: + """儲存任務結果""" + ... + + async def get_incident(self, incident_id: str) -> Incident | None: + """從 Redis 取得 Incident""" + ... + + +# ============================================================================= +# Repository Implementation +# ============================================================================= + +class AgentTaskRedisRepository: + """ + Agent Task Redis Repository + + 實作 IAgentTaskRepository Protocol + 所有 Redis 操作都封裝在此 + """ + + async def create_task( + self, + task_id: str, + incident_id: str, + trigger: str = "manual", + ) -> dict[str, Any]: + """ + 建立新的分析任務 + + Args: + task_id: 任務 ID + incident_id: 關聯的 Incident ID + trigger: 觸發來源 (manual/auto) + + Returns: + 建立的任務資料 + """ + redis_client = get_redis() + task_key = f"{TASK_PREFIX}{task_id}" + + task_data = { + "task_id": task_id, + "state": TaskState.PENDING.value, + "progress": 0, + "current_step": "任務已建立", + "agents_completed": 0, + "total_agents": 4, + "incident_id": incident_id, + "started_at": datetime.now(UTC).isoformat(), + "trigger": trigger, + } + + await redis_client.set( + task_key, + json.dumps(task_data), + ex=TASK_TTL, + ) + + logger.info( + "task_created", + task_id=task_id, + incident_id=incident_id, + trigger=trigger, + ) + + return task_data + + async def get_task(self, task_id: str) -> dict[str, Any] | None: + """ + 取得任務資料 + + Args: + task_id: 任務 ID + + Returns: + 任務資料字典，不存在則返回 None + """ + redis_client = get_redis() + task_key = f"{TASK_PREFIX}{task_id}" + + data = await redis_client.get(task_key) + if data is None: + return None + + return json.loads(data) + + async def update_task_state( + self, + task_id: str, + state: TaskState, + progress: int = 0, + current_step: str | None = None, + agents_completed: int = 0, + **extra_data: Any, + ) -> bool: + """ + 更新任務狀態 + + Args: + task_id: 任務 ID + state: 新狀態 + progress: 進度百分比 (0-100) + current_step: 目前步驟描述 + agents_completed: 已完成的 Agent 數量 + **extra_data: 額外資料 + + Returns: + 是否更新成功 + """ + redis_client = get_redis() + task_key = f"{TASK_PREFIX}{task_id}" + + try: + # 讀取現有資料 + existing = await redis_client.get(task_key) + if existing: + task_data = json.loads(existing) + else: + task_data = {"task_id": task_id} + + # 更新欄位 + task_data.update({ + "state": state.value, + "progress": progress, + "current_step": current_step, + "agents_completed": agents_completed, + **extra_data, + }) + + await redis_client.set( + task_key, + json.dumps(task_data), + ex=TASK_TTL, + ) + + return True + + except Exception as e: + logger.exception( + "update_task_state_error", + task_id=task_id, + error=str(e), + ) + return False + + async def save_task_result( + self, + task_id: str, + result_data: dict[str, Any], + ) -> bool: + """ + 儲存任務結果 + + Args: + task_id: 任務 ID + result_data: 完整結果資料 + + Returns: + 是否儲存成功 + """ + redis_client = get_redis() + task_key = f"{TASK_PREFIX}{task_id}" + + try: + await redis_client.set( + task_key, + json.dumps(result_data), + ex=TASK_TTL, + ) + + logger.info( + "task_result_saved", + task_id=task_id, + state=result_data.get("state"), + ) + return True + + except Exception as e: + logger.exception( + "save_task_result_error", + task_id=task_id, + error=str(e), + ) + return False + + async def get_incident(self, incident_id: str) -> Incident | None: + """ + 從 Redis 取得 Incident + + Args: + incident_id: Incident ID + + Returns: + Incident 物件，不存在則返回 None + """ + redis_client = get_redis() + key = f"{INCIDENT_PREFIX}{incident_id}" + + data = await redis_client.get(key) + if data is None: + return None + + return Incident.model_validate_json(data) + + +# ============================================================================= +# Agent Service +# ============================================================================= + +class AgentService: + """ + Agent Service - Agent Teams 業務邏輯 + + 職責: + 1. 任務生命週期管理 + 2. 協調 ConsensusEngine 執行分析 + 3. 推送 SSE 進度通知 + + 使用方式: + service = get_agent_service() + task_id = await service.create_analysis_task(incident_id) + """ + + def __init__( + self, + repository: IAgentTaskRepository | None = None, + ) -> None: + """ + 初始化 Agent Service + + Args: + repository: Task Repository (預設使用 Redis 實作) + """ + self._repository = repository or AgentTaskRedisRepository() + + # ========================================================================= + # Task Management + # ========================================================================= + + def generate_task_id(self) -> str: + """產生新的 Task ID""" + return f"TASK-{datetime.now(UTC).strftime('%Y%m%d')}-{uuid4().hex[:8].upper()}" + + async def create_analysis_task( + self, + incident: Incident, + trigger: str = "manual", + ) -> str: + """ + 建立分析任務 + + Args: + incident: 要分析的 Incident + trigger: 觸發來源 (manual/auto) + + Returns: + task_id + """ + task_id = self.generate_task_id() + + await self._repository.create_task( + task_id=task_id, + incident_id=incident.incident_id, + trigger=trigger, + ) + + logger.info( + "analysis_task_created", + task_id=task_id, + incident_id=incident.incident_id, + severity=incident.severity.value, + ) + + return task_id + + async def get_task_status(self, task_id: str) -> dict[str, Any] | None: + """取得任務狀態""" + return await self._repository.get_task(task_id) + + async def get_task_result(self, task_id: str) -> dict[str, Any] | None: + """取得任務結果""" + return await self._repository.get_task(task_id) + + async def get_incident(self, incident_id: str) -> Incident | None: + """取得 Incident""" + return await self._repository.get_incident(incident_id) + + # ========================================================================= + # Analysis Execution + # ========================================================================= + + async def run_analysis( + self, + task_id: str, + incident: Incident, + ) -> None: + """ + 執行 Agent Teams 分析 + + 流程: + 1. 更新狀態為 ANALYZING + 2. 收集各專家意見 + 3. 計算共識 + 4. 儲存結果 + 5. 推送 SSE 通知 + + Args: + task_id: 任務 ID + incident: 要分析的 Incident + """ + consensus_engine = get_consensus_engine() + + try: + # Step 1: 更新狀態 + await self._repository.update_task_state( + task_id, + TaskState.ANALYZING, + progress=10, + current_step="正在收集專家意見...", + ) + + # 推送 SSE 進度 + publisher = await get_publisher() + await publisher.publish(SSEEvent( + type=EventType.AI_THINKING, + data={ + "task_id": task_id, + "state": TaskState.ANALYZING.value, + "progress": 10, + "message": "Agent Teams 分析開始", + }, + )) + + # Step 2: 收集意見 + opinions = await consensus_engine.gather_opinions( + incident, timeout_sec=25.0 + ) + + await self._repository.update_task_state( + task_id, + TaskState.CONSENSUS, + progress=60, + current_step="正在計算共識...", + agents_completed=len(opinions), + ) + + await publisher.publish(SSEEvent( + type=EventType.AI_THINKING, + data={ + "task_id": task_id, + "state": TaskState.CONSENSUS.value, + "progress": 60, + "message": f"已收集 {len(opinions)} 位專家意見", + }, + )) + + # Step 3: 計算共識 + ( + consensus_score, + recommended_action, + dissenting, + ) = consensus_engine.calculate_consensus(opinions) + + await self._repository.update_task_state( + task_id, + TaskState.CONSENSUS, + progress=80, + current_step="正在產生最終決策...", + ) + + # Step 4: 產生最終決策 + result = await consensus_engine.generate_final_decision( + incident=incident, + opinions=opinions, + consensus_score=consensus_score, + recommended_action_type=recommended_action, + dissenting=dissenting, + ) + + # Step 5: 儲存完整結果 + task_data = { + "task_id": task_id, + "state": TaskState.COMPLETED.value, + "progress": 100, + "current_step": "分析完成", + "agents_completed": len(opinions), + "total_agents": 4, + "consensus_id": result.consensus_id, + "incident_id": incident.incident_id, + "consensus_score": result.consensus_score, + "recommended_action": result.recommended_action, + "recommended_kubectl": result.recommended_kubectl, + "risk_level": result.risk_level, + "final_reasoning": result.final_reasoning, + "opinions": [op.to_dict() for op in result.opinions], + "dissenting_opinions": result.dissenting_opinions, + "completed_at": datetime.now(UTC).isoformat(), + } + + await self._repository.save_task_result(task_id, task_data) + + # 推送完成通知 + await publisher.publish(SSEEvent( + type=EventType.AI_THINKING, + data={ + "task_id": task_id, + "state": TaskState.COMPLETED.value, + "progress": 100, + "message": "分析完成", + "consensus_score": result.consensus_score, + "recommended_action": result.recommended_action, + }, + )) + + logger.info( + "analysis_completed", + task_id=task_id, + consensus_id=result.consensus_id, + consensus_score=result.consensus_score, + ) + + except Exception as e: + logger.exception( + "analysis_failed", + task_id=task_id, + error=str(e), + ) + + # 更新為失敗狀態 + task_data = { + "task_id": task_id, + "state": TaskState.FAILED.value, + "progress": 0, + "error": str(e), + "completed_at": datetime.now(UTC).isoformat(), + } + + await self._repository.save_task_result(task_id, task_data) + + # 推送失敗通知 + publisher = await get_publisher() + await publisher.publish(SSEEvent( + type=EventType.ERROR, + data={ + "task_id": task_id, + "state": TaskState.FAILED.value, + "error": str(e), + }, + )) + + # ========================================================================= + # Incident Integration + # ========================================================================= + + def should_trigger_agent_analysis(self, incident: Incident) -> bool: + """ + 判斷是否需要觸發 Agent Teams 分析 + + 條件 (任一符合): + - P0/P1 緊急事件 + - 多個服務受影響 (>2) + - 多個告警 (>3) + + Args: + incident: 要判斷的 Incident + + Returns: + 是否應觸發分析 + """ + return ( + # P0/P1 緊急事件 + incident.severity in (Severity.P0, Severity.P1) + # 或多個服務受影響 + or len(incident.affected_services) > 2 + # 或多個告警 + or len(incident.signals) > 3 + ) + + async def trigger_for_incident( + self, + incident_id: str, + ) -> tuple[str | None, Incident | None]: + """ + 為 Incident 觸發 Agent Teams 分析 (如果符合條件) + + Args: + incident_id: Incident ID + + Returns: + (task_id, incident) - task_id 為 None 表示未觸發 + """ + # 讀取 Incident + incident = await self._repository.get_incident(incident_id) + if incident is None: + logger.warning("trigger_skipped_not_found", incident_id=incident_id) + return None, None + + # 判斷是否需要 Agent Teams + if not self.should_trigger_agent_analysis(incident): + logger.debug( + "trigger_skipped_simple_case", + incident_id=incident_id, + severity=incident.severity.value, + ) + return None, incident + + # 建立任務 + task_id = await self.create_analysis_task(incident, trigger="auto") + + logger.info( + "auto_trigger_success", + task_id=task_id, + incident_id=incident_id, + severity=incident.severity.value, + ) + + return task_id, incident + + # ========================================================================= + # Incident Factory + # ========================================================================= + + def create_adhoc_incident( + self, + severity: str, + affected_services: list[str], + alert_names: list[str] | None = None, + ) -> Incident: + """ + 建立臨時 Incident (用於直接分析請求) + + Args: + severity: 嚴重度 (P0/P1/P2/P3) + affected_services: 受影響服務列表 + alert_names: 告警名稱列表 + + Returns: + 建立的 Incident + """ + signals = [] + if alert_names: + for alert_name in alert_names: + signals.append(Signal( + alert_name=alert_name, + severity=Severity(severity), + source="manual", + fired_at=datetime.now(UTC), + )) + + return Incident( + severity=Severity(severity), + status=IncidentStatus.INVESTIGATING, + signals=signals, + affected_services=affected_services, + ) + + +# ============================================================================= +# Singleton +# ============================================================================= + +_agent_service: AgentService | None = None + + +def get_agent_service() -> AgentService: + """取得 Agent Service 實例 (Singleton)""" + global _agent_service + if _agent_service is None: + _agent_service = AgentService() + return _agent_service