LLM 기반 로그 분석 및 AIOps 실전 가이드

def analyze_logs(logs: list, time_range: str) -> dict:
    """로그 묶음을 LLM으로 분석"""
    log_text = "\n".join(logs[-100:])  # 최근 100줄
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=2048,
        messages=[{"role": "user", "content": f"""
다음 서버 로그를 분석하세요 (시간 범위: {time_range}).
 
```
{log_text}
```
 
JSON으로 분석 결과 반환:
{{
  "summary": "전체 상황 요약",
  "error_count": 숫자,
  "warning_count": 숫자,
  "patterns": ["반복되는 패턴 목록"],
  "anomalies": ["비정상적인 패턴"],
  "severity": "critical/warning/info",
  "possible_causes": ["추정 원인"],
  "recommended_actions": ["권장 조치"]
}}"""}]
    )
    return json.loads(response.content[0].text)

def root_cause_analysis(alert: dict, logs: str, metrics: dict, history: list) -> dict:
    """장애의 근본 원인을 분석"""
    # 유사 과거 사례 검색
    similar_incidents = search_incident_db(alert["description"])
    
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": f"""
장애 근본 원인 분석(RCA)을 수행하세요.
 
## 현재 알림
{json.dumps(alert, ensure_ascii=False)}
 
## 관련 로그 (최근 30분)
{logs}
 
## 시스템 메트릭
{json.dumps(metrics, ensure_ascii=False)}
 
## 유사 과거 사례
{json.dumps(similar_incidents, ensure_ascii=False)}
 
## 분석 요청
1. 근본 원인 (확률 높은 순, 최대 3개)
2. 원인별 근거 (로그/메트릭 증거)
3. 영향 범위
4. 즉시 조치 (긴급 대응)
5. 근본적 해결 방안 (재발 방지)
6. 과거 유사 사례와의 비교
 
JSON으로 반환하세요."""}]
    )
    return json.loads(response.content[0].text)

class AutoRemediationAgent:
    def __init__(self):
        self.approved_actions = {
            "restart_service": {"risk": "low", "auto_approve": True},
            "scale_up": {"risk": "low", "auto_approve": True},
            "clear_cache": {"risk": "low", "auto_approve": True},
            "rollback_deployment": {"risk": "medium", "auto_approve": False},
            "failover": {"risk": "high", "auto_approve": False},
        }
    
    def handle_alert(self, alert: dict):
        """알림 처리 파이프라인"""
        # 1. RCA 수행
        analysis = root_cause_analysis(alert, get_logs(), get_metrics(), get_history())
        
        # 2. 조치 결정
        actions = analysis["recommended_actions"]
        
        for action in actions:
            action_config = self.approved_actions.get(action["type"])
            
            if action_config and action_config["auto_approve"]:
                # 자동 승인된 조치 → 즉시 실행
                self.execute_action(action)
                self.notify_team(f"자동 조치 실행: {action['type']}")
            else:
                # 수동 승인 필요 → 에스컬레이션
                self.escalate(action, analysis)
        
        # 3. 장애 보고서 생성
        report = self.generate_incident_report(alert, analysis, actions)
        self.send_report(report)

def generate_incident_report(alert, analysis, actions_taken) -> str:
    """장애 보고서 자동 생성"""
    response = client.messages.create(
        model="claude-sonnet-4-6",
        max_tokens=4096,
        messages=[{"role": "user", "content": f"""
다음 정보를 바탕으로 장애 보고서를 작성하세요.
 
알림: {json.dumps(alert)}
분석 결과: {json.dumps(analysis)}
수행된 조치: {json.dumps(actions_taken)}
 
보고서 형식:
## 장애 보고서
### 1. 개요 (발생 시각, 영향 범위, 심각도)
### 2. 타임라인 (발생 → 감지 → 분석 → 조치 → 복구)
### 3. 근본 원인
### 4. 수행된 조치
### 5. 재발 방지 대책
### 6. 관련 지표 (MTTD, MTTR)"""}]
    )
    return response.content[0].text