awoooi/apps/api/src/agents/base.py

"""
Base Agent - 專家 Agent 基礎類別
================================

定義所有專家 Agent 的共用介面和工具

使用 claude-agent-sdk 的 AgentDefinition
符合 ADR-009 架構規範
"""

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any, Generic, TypeVar

import structlog

logger = structlog.get_logger(__name__)


# =============================================================================
# Agent Result Base
# =============================================================================


class AgentStatus(str, Enum):
    """Agent 執行狀態"""
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    TIMEOUT = "timeout"


@dataclass
class AgentResult:
    """
    Agent 執行結果基類

    所有專家 Agent 的輸出都必須包含:
    - agent_name: 識別哪個 Agent
    - status: 執行狀態
    - confidence: 信心分數 (0-1)
    - analysis: 分析摘要
    - latency_ms: 執行時間
    """
    agent_name: str
    status: AgentStatus
    confidence: float
    analysis: str
    latency_ms: int
    error: str | None = None
    raw_response: dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

    def to_dict(self) -> dict[str, Any]:
        """轉換為 dict (API 回傳用)"""
        return {
            "agent_name": self.agent_name,
            "status": self.status.value,
            "confidence": self.confidence,
            "analysis": self.analysis,
            "latency_ms": self.latency_ms,
            "error": self.error,
            "timestamp": self.timestamp.isoformat(),
        }


# =============================================================================
# Base Agent
# =============================================================================

T = TypeVar("T", bound=AgentResult)


class BaseAgent(ABC, Generic[T]):
    """
    專家 Agent 基礎類別

    所有專家 Agent 都繼承此類別，並實作:
    - analyze(): 核心分析邏輯
    - _build_prompt(): 建構 Prompt
    - _parse_response(): 解析回應

    使用方式:
    ```python
    agent = SecurityAgent()
    result = await agent.analyze(incident_context)
    ```
    """

    # Agent 識別資訊 (子類別覆寫)
    AGENT_NAME: str = "base"
    AGENT_DESCRIPTION: str = "Base Agent"
    AGENT_TOOLS: list[str] = ["Read", "Grep"]

    def __init__(self, timeout_sec: float = 30.0):
        """
        初始化 Agent

        Args:
            timeout_sec: 執行超時時間 (秒)
        """
        self.timeout_sec = timeout_sec
        self.logger = logger.bind(agent=self.AGENT_NAME)

    @abstractmethod
    async def analyze(self, context: dict[str, Any]) -> T:
        """
        執行分析 (子類別必須實作)

        Args:
            context: 分析上下文 (incident 資訊)

        Returns:
            AgentResult 子類別實例
        """
        pass

    @abstractmethod
    def _build_prompt(self, context: dict[str, Any]) -> str:
        """
        建構 Prompt (子類別必須實作)

        Args:
            context: 分析上下文

        Returns:
            給 LLM 的 Prompt
        """
        pass

    @abstractmethod
    def _parse_response(self, response: str) -> dict[str, Any]:
        """
        解析 LLM 回應 (子類別必須實作)

        Args:
            response: LLM 原始回應

        Returns:
            解析後的結構化資料
        """
        pass

    def _extract_json(self, text: str) -> dict[str, Any]:
        """
        從 LLM 回應中提取 JSON

        支援:
        - ```json ... ``` 區塊
        - 純 JSON 文字
        """
        import json
        import re

        # 嘗試 ```json ... ``` 格式
        match = re.search(r"```json\s*(.*?)\s*```", text, re.DOTALL)
        if match:
            try:
                return json.loads(match.group(1))
            except json.JSONDecodeError:
                pass

        # 嘗試從第一個 { 到最後一個 } 提取（支援巢狀 JSON）
        # Gate 2: 舊 r"\{[^{}]*\}" 會拒絕巢狀物件，造成所有 Agent LLM 回應解析失敗
        start = text.find("{")
        end = text.rfind("}")
        if start != -1 and end > start:
            try:
                return json.loads(text[start:end + 1])
            except json.JSONDecodeError:
                pass

        # 嘗試整段解析
        try:
            return json.loads(text)
        except json.JSONDecodeError:
            self.logger.warning("json_parse_failed", text=text[:200])
            return {}

    def _get_agent_definition(self) -> dict[str, Any]:
        """
        取得 Claude Agent SDK 的 AgentDefinition

        Returns:
            符合 SDK 規範的 AgentDefinition dict
        """
        return {
            "name": self.AGENT_NAME,
            "description": self.AGENT_DESCRIPTION,
            "tools": self.AGENT_TOOLS,
        }