自律型AIエージェントによるインフラ自己修復アーキテクチャの実装

import logging
import os
import json
from typing import Optional, Dict, Any
from langchain_openai import ChatOpenAI
from langchain.schema import HumanMessage, SystemMessage

# ロギングの設定
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

class SelfHealingAgent:
    def __init__(self, model_name: str = "gpt-4o", temperature: float = 0):
        """
        自己修復エージェントの初期化。
        Args:
            model_name: 使用するLLMモデル名
            temperature: 生成の多様性（0に近いほど決定的）
        """
        self.llm = ChatOpenAI(
            model=model_name,
            temperature=temperature,
            api_key=os.getenv("OPENAI_API_KEY")
        )
        logger.info(f"SelfHealingAgent initialized with model: {model_name}")

    def _construct_prompt(self, context: str) -> list:
        """
        LLMへのプロンプトを構築する。
        システムメッセージで役割と制約条件を厳密に定義する。
        """
        system_prompt = """
        あなたは熟練したSRE（サイト信頼性エンジニア）です。
        以下のコンテキストに基づき、システム障害の原因を特定し、解決策を提案してください。
        
        回答は必ず以下のJSON形式のみで出力してください。説明文は不要です。
        {
            "diagnosis": "障害原因の簡潔な説明",
            "action_type": "restart_pod | scale_up | rollback | ignore | manual_intervention",
            "command": "実行すべき具体的なコマンドまたはAPI操作内容",
            "confidence": 0.0から1.0までの信頼度
        }
        
        注意点：
        - 信頼度が0.7未満の場合は、action_typeを'manual_intervention'に設定してください。
        - データベースの削除など、破壊的な操作は絶対に提案しないでください。
        """
        
        return [SystemMessage(content=system_prompt), HumanMessage(content=context)]

    def analyze_and_heal(self, pod_name: str, namespace: str) -> Optional[Dict[str, Any]]:
        """
        障害分析と修復アクションを実行するメインメソッド。
        """
        try:
            logger.info(f"Analyzing failure for Pod: {pod_name} in Namespace: {namespace}")
            
            # 1. コンテキスト収集（擬似的な実装）
            logs = self._get_pod_logs(pod_name, namespace)
            metrics = self._get_pod_metrics(pod_name, namespace)
            
            context = f"""
            Pod Name: {pod_name}
            Namespace: {namespace}
            Status: CrashLoopBackOff
            
            Recent Logs:
            {logs}
            
            Metrics:
            {metrics}
            """

            # 2. LLMによる推論
            messages = self._construct_prompt(context)
            response = self.llm.invoke(messages)
            content = response.content
            
            logger.info(f"LLM Response received: {content}")

            # 3. レスポンスのパースと検証
            # 簡易的なJSONパース（実際にはより厳格なバリデーションが必要）
            try:
                # Markdownコードブロックなどが含まれる場合を考慮して前処理
                if "```json" in content:
                    content = content.split("```json")[1].split("```")[0]
                elif "```" in content:
                    content = content.split("```")[1].split("```")[0]
                
                decision = json.loads(content.strip())
            except json.JSONDecodeError as e:
                logger.error(f"Failed to parse LLM response as JSON: {e}")
                return None

            # 4. アクションの実行（ガードレール付き）
            if decision.get("confidence", 0) < 0.7:
                logger.warning("Confidence too low, escalating to human intervention.")
                self._notify_human(pod_name, decision)
                return decision

            return self._execute_action(pod_name, namespace, decision)

        except Exception as e:
            logger.error(f"Error during self-healing process: {e}", exc_info=True)
            self._notify_human(pod_name, {"error": str(e)})
            return None

    def _get_pod_logs(self, pod_name: str, namespace: str) -> str:
        # 実際にはKubernetes Python Clientを使用してログを取得
        logger.debug("Fetching pod logs...")
        return "Error: Unable to connect to database. Connection timeout after 30s."

    def _get_pod_metrics(self, pod_name: str, namespace: str) -> str:
        # 実際にはPrometheus APIなどからメトリクスを取得
        logger.debug("Fetching pod metrics...")
        return "CPU Usage: 5%, Memory Usage: 80%, Restart Count: 5"

    def _execute_action(self, pod_name: str, namespace: str, decision: Dict[str, Any]) -> Dict[str, Any]:
        action_type = decision.get("action_type")
        
        logger.info(f"Executing action: {action_type} for {pod_name}")
        
        if action_type == "restart_pod":
            # self._restart_pod(pod_name, namespace) # 実際のK8s API呼び出し
            logger.info(f"Pod {pod_name} restarted successfully.")
            decision["status"] = "executed"
        elif action_type == "manual_intervention":
            self._notify_human(pod_name, decision)
        else:
            logger.info(f"No automated action taken for type: {action_type}")
            
        return decision

    def _notify_human(self, pod_name: str, detail: Dict[str, Any]):
        # SlackやTeamsへの通知処理
        message = f"🚨 Self-Healing Agent requires help for {pod_name}. Detail: {json.dumps(detail)}"
        logger.warning(f"HUMAN NOTIFICATION: {message}")
        # send_to_slack(message)

if __name__ == "__main__":
    # 環境変数のチェック
    if not os.getenv("OPENAI_API_KEY"):
        logger.error("OPENAI_API_KEY is not set.")
    else:
        agent = SelfHealingAgent()
        result = agent.analyze_and_heal(pod_name="payment-service-xyz", namespace="production")
        print(json.dumps(result, indent=2))