AI API를 활용한 서비스를 운영하다 보면 규정 준수(Compliance) 감사가 필수적으로 요구됩니다. 특히 금융, 의료, 교육 분야에서는 모델 사용 로그, 토큰 소비 내역, 응답 품질 지표 등을 체계적으로 기록하고 보고서를 생성해야 하는 경우가 많습니다. 이번 튜토리얼에서는 HolySheep AI를 활용하여 AI API 규정 준수 감사를 자동화하는 보고서 생성 도구를 처음부터 만들어보겠습니다.
왜 AI API 규정 준수가 중요한가?
저는 실무에서 여러 기업의 AI 시스템 도입을 지원하면서 가장 많이 받는 질문이 바로 "우리가 AI 모델을 사용한 기록을 제대로 남기고 있는가?"입니다. 규제 기관이나 내부 감사 부서에서 요청할 때마다 수동으로 로그를 정리하면 상당한 시간이 소요됩니다. HolySheep AI는 단일 API 키로 다양한 모델의 사용량을 통합 관리할 수 있어 이러한 감사를 효율적으로 수행할 수 있습니다.
프로젝트 개요 및 준비물
- 목표: API 호출 로그를 자동으로 수집하고 규정 준수 보고서를 생성
- 사용 도구: Python 3.8+, HolySheep AI API
- 출력물: JSON/CSV 형태의 감사 보고서 + 요약 대시보드
1단계: HolySheep AI API 키 발급
가장 먼저 HolySheep AI에 가입하여 API 키를 발급받아야 합니다. HolySheep AI는 해외 신용카드 없이 로컬 결제가 가능하며, 다양한 모델(GPT-4.1, Claude Sonnet, Gemini, DeepSeek 등)을 단일 키로 통합 관리할 수 있습니다. 특히 규정 준수 감사를 위해 여러 모델의 사용량을 한눈에 확인하고 싶다면 매우 편리합니다.
[스크린샷 위치: HolySheep AI 대시보드 → API Keys → Create New Key 버튼]
2단계: 프로젝트 구조 설정
규정 준수 감사 도구를 위한 프로젝트 구조를 만들겠습니다. 저는 실무에서 항상 로깅 시스템과 비즈니스 로직을 분리하여 유지보수성을 높이는 방식을 선호합니다.
# 프로젝트 디렉토리 구조
compliance-audit-tool/
├── config.py # 설정 파일
├── audit_logger.py # API 호출 로깅 모듈
├── report_generator.py # 보고서 생성 모듈
├── dashboard.py # 대시보드 생성 모듈
├── main.py # 메인 실행 파일
└── reports/ # 생성된 보고서 저장 폴더
├── logs/
└── output/
3단계: 설정 파일 작성
config.py 파일에 HolySheep AI API 키와 로깅 설정을 정의합니다. 실제 키를 사용할 때는 환경 변수에서 불러오는 것을 권장합니다.
# config.py
import os
from datetime import datetime
HolySheep AI 설정
⚠️ 실제 키는 환경 변수 또는 .env 파일에서 관리하세요
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
감사 설정
AUDIT_CONFIG = {
"report_title": "AI API 규정 준수 감사 보고서",
"organization": "your-company-name",
"audit_period_start": "2024-01-01",
"audit_period_end": "2024-12-31",
"compliance_frameworks": ["GDPR", "SOC2", "ISO27001"],
"models_to_audit": ["gpt-4.1", "claude-sonnet-4", "gemini-2.5-flash", "deepseek-v3.2"],
}
로깅 설정
LOG_CONFIG = {
"log_dir": "reports/logs",
"output_dir": "reports/output",
"log_format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
"max_log_size_mb": 50,
"backup_count": 5,
}
토큰 가격표 (HolySheep AI 공식 기준)
실제 가격은 HolySheep AI 대시보드에서 확인하세요
TOKEN_PRICING = {
"gpt-4.1": {"input": 8.00, "output": 8.00}, # $8/MTok
"claude-sonnet-4": {"input": 4.50, "output": 22.50}, # $4.50/$22.50/MTok
"gemini-2.5-flash": {"input": 1.25, "output": 5.00}, # $2.50 promo (input/output 합산 기준)
"deepseek-v3.2": {"input": 0.28, "output": 1.10}, # $0.42/MTok promo
}
이상 탐지 임계값
ANOMALY_THRESHOLDS = {
"max_requests_per_minute": 100,
"max_tokens_per_request": 100000,
"max_cost_per_day_usd": 500.00,
"max_failed_requests_percent": 5.0,
}
4단계: API 호출 로깅 모듈 구현
실제 AI API를 호출하고 그 과정을 자동으로 기록하는 로깅 모듈을 만들겠습니다. HolySheep AI는 단일 엔드포인트로 여러 모델에 접근할 수 있어 로깅 구조가 훨씬 단순해집니다.
# audit_logger.py
import json
import os
import logging
from datetime import datetime
from typing import Dict, List, Optional, Any
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL, LOG_CONFIG, TOKEN_PRICING
class APILogger:
"""AI API 호출을 로깅하는 클래스"""
def __init__(self, log_file_path: Optional[str] = None):
self.base_url = HOLYSHEEP_BASE_URL
self.api_key = HOLYSHEEP_API_KEY
self.audit_logs: List[Dict[str, Any]] = []
# 로그 디렉토리 생성
os.makedirs(LOG_CONFIG["log_dir"], exist_ok=True)
# 로그 파일 경로 설정
if log_file_path is None:
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
log_filename = f"api_audit_log_{timestamp}.jsonl"
self.log_file_path = os.path.join(LOG_CONFIG["log_dir"], log_filename)
else:
self.log_file_path = log_file_path
# 로거 설정
self.logger = logging.getLogger("APIAuditLogger")
self.logger.setLevel(logging.INFO)
# 파일 핸들러가 이미 있는지 확인
if not self.logger.handlers:
handler = logging.FileHandler(self.log_file_path, encoding='utf-8')
formatter = logging.Formatter(LOG_CONFIG["log_format"])
handler.setFormatter(formatter)
self.logger.addHandler(handler)
def log_api_call(
self,
model_name: str,
request_data: Dict[str, Any],
response_data: Optional[Dict[str, Any]] = None,
status: str = "success",
error_message: Optional[str] = None,
latency_ms: Optional[float] = None,
user_id: Optional[str] = None,
session_id: Optional[str] = None,
) -> Dict[str, Any]:
"""API 호출 로그를 기록하고 반환합니다"""
timestamp = datetime.utcnow().isoformat() + "Z"
# 토큰 사용량 계산
usage = {}
if response_data and "usage" in response_data:
usage = {
"input_tokens": response_data["usage"].get("prompt_tokens", 0),
"output_tokens": response_data["usage"].get("completion_tokens", 0),
"total_tokens": response_data["usage"].get("total_tokens", 0),
}
# 비용 계산
cost_usd = 0.0
if usage and model_name in TOKEN_PRICING:
pricing = TOKEN_PRICING[model_name]
cost_usd = (
(usage["input_tokens"] / 1_000_000) * pricing["input"] +
(usage["output_tokens"] / 1_000_000) * pricing["output"]
)
# 로그 엔트리 생성
log_entry = {
"timestamp": timestamp,
"event_type": "api_call",
"model": model_name,
"status": status,
"user_id": user_id,
"session_id": session_id,
"request": {
"messages": request_data.get("messages", []),
"max_tokens": request_data.get("max_tokens"),
"temperature": request_data.get("temperature"),
},
"response": {
"id": response_data.get("id") if response_data else None,
"model": response_data.get("model") if response_data else None,
"finish_reason": response_data.get("choices", [{}])[0].get("finish_reason") if response_data else None,
},
"usage": usage,
"cost_usd": round(cost_usd, 6),
"latency_ms": latency_ms,
"error_message": error_message,
}
self.audit_logs.append(log_entry)
# 파일에 즉시 기록 (안전성을 위해)
with open(self.log_file_path, 'a', encoding='utf-8') as f:
f.write(json.dumps(log_entry, ensure_ascii=False) + '\n')
self.logger.info(f"API 호출 로깅 완료: {model_name} | 상태: {status} | 비용: ${cost_usd:.6f}")
return log_entry
def log_compliance_event(
self,
event_type: str,
description: str,
severity: str = "info",
metadata: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""규정 준수 관련 이벤트를 로깅합니다"""
timestamp = datetime.utcnow().isoformat() + "Z"
event_entry = {
"timestamp": timestamp,
"event_type": f"compliance_{event_type}",
"severity": severity,
"description": description,
"metadata": metadata or {},
}
self.audit_logs.append(event_entry)
with open(self.log_file_path, 'a', encoding='utf-8') as f:
f.write(json.dumps(event_entry, ensure_ascii=False) + '\n')
self.logger.info(f"규정 준수 이벤트: {event_type} | 심각도: {severity}")
return event_entry
def get_all_logs(self) -> List[Dict[str, Any]]:
"""모든 로그를 반환합니다"""
return self.audit_logs.copy()
def get_logs_by_model(self, model_name: str) -> List[Dict[str, Any]]:
"""특정 모델의 로그만 필터링합니다"""
return [log for log in self.audit_logs if log.get("model") == model_name]
def get_logs_by_date_range(
self,
start_date: datetime,
end_date: datetime,
) -> List[Dict[str, Any]]:
"""날짜 범위로 로그를 필터링합니다"""
filtered = []
for log in self.audit_logs:
log_time = datetime.fromisoformat(log["timestamp"].replace("Z", "+00:00"))
if start_date <= log_time <= end_date:
filtered.append(log)
return filtered
테스트 실행
if __name__ == "__main__":
logger = APILogger()
# 테스트 API 호출 로깅
test_request = {
"messages": [{"role": "user", "content": "규정 준수 감사 보고서 생성"}],
"max_tokens": 1000,
"temperature": 0.7,
}
test_response = {
"id": "test-completion-id",
"model": "gpt-4.1",
"usage": {"prompt_tokens": 50, "completion_tokens": 150, "total_tokens": 200},
"choices": [{"finish_reason": "stop"}],
}
logger.log_api_call(
model_name="gpt-4.1",
request_data=test_request,
response_data=test_response,
status="success",
latency_ms=1250.5,
user_id="test-user-001",
)
logger.log_compliance_event(
event_type="access",
description="사용자 test-user-001이 AI API에 접근",
severity="info",
)
print(f"로그 파일 경로: {logger.log_file_path}")
print(f"총 로그 개수: {len(logger.get_all_logs())}")
5단계: HolySheep AI를 활용한 실제 API 호출
이제 HolySheep AI를 통해 실제 AI 모델을 호출하고 그 결과를 자동으로 로깅하는 함수를 만들겠습니다. HolySheep AI의 장점은 OpenAI 호환 API를 제공하므로 기존 OpenAI SDK를 그대로 사용할 수 있다는 점입니다.
# holy_sheep_client.py
import time
import logging
from typing import Dict, List, Any, Optional
from datetime import datetime
import requests # 또는 openai SDK 사용 가능
from audit_logger import APILogger
from config import HOLYSHEEP_API_KEY, HOLYSHEEP_BASE_URL
logger = logging.getLogger("APIAuditLogger")
class HolySheepAIClient:
"""HolySheep AI API 클라이언트 - 규정 준수 감사용"""
def __init__(self, api_key: str = HOLYSHEEP_API_KEY):
self.api_key = api_key
self.base_url = HOLYSHEEP_BASE_URL
self.headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
self.audit_logger = APILogger()
# 모델별 엔드포인트 매핑
self.model_endpoints = {
"gpt-4.1": "/chat/completions",
"claude-sonnet-4": "/chat/completions",
"gemini-2.5-flash": "/chat/completions",
"deepseek-v3.2": "/chat/completions",
}
def chat_completion(
self,
model: str,
messages: List[Dict[str, str]],
max_tokens: Optional[int] = 1000,
temperature: float = 0.7,
user_id: Optional[str] = None,
session_id: Optional[str] = None,
) -> Dict[str, Any]:
"""AI API를 호출하고 결과를 로깅합니다"""
start_time = time.time()
request_data = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"temperature": temperature,
}
try:
# HolySheep AI API 호출
endpoint = self.model_endpoints.get(model, "/chat/completions")
url = f"{self.base_url}{endpoint}"
response = requests.post(
url,
headers=self.headers,
json=request_data,
timeout=60,
)
latency_ms = (time.time() - start_time) * 1000
if response.status_code == 200:
response_data = response.json()
# API 호출 로깅
self.audit_logger.log_api_call(
model_name=model,
request_data=request_data,
response_data=response_data,
status="success",
latency_ms=latency_ms,
user_id=user_id,
session_id=session_id,
)
# 규정 준수 이벤트 로깅
self.audit_logger.log_compliance_event(
event_type="api_request_success",
description=f"모델 {model} API 호출 성공",
severity="info",
metadata={
"model": model,
"user_id": user_id,
"latency_ms": latency_ms,
},
)
return {
"success": True,
"data": response_data,
"latency_ms": latency_ms,
}
else:
error_message = f"API 오류: {response.status_code} - {response.text}"
latency_ms = (time.time() - start_time) * 1000
# 실패 로그 기록
self.audit_logger.log_api_call(
model_name=model,
request_data=request_data,
status="error",
error_message=error_message,
latency_ms=latency_ms,
user_id=user_id,
session_id=session_id,
)
self.audit_logger.log_compliance_event(
event_type="api_request_failed",
description=error_message,
severity="warning",
metadata={"model": model, "status_code": response.status_code},
)
return {
"success": False,
"error": error_message,
"latency_ms": latency_ms,
}
except requests.exceptions.Timeout:
latency_ms = (time.time() - start_time) * 1000
error_message = "API 요청 시간 초과 (60초)"
self.audit_logger.log_api_call(
model_name=model,
request_data=request_data,
status="timeout",
error_message=error_message,
latency_ms=latency_ms,
user_id=user_id,
session_id=session_id,
)
return {
"success": False,
"error": error_message,
"latency_ms": latency_ms,
}
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
error_message = f"예상치 못한 오류: {str(e)}"
self.audit_logger.log_api_call(
model_name=model,
request_data=request_data,
status="exception",
error_message=error_message,
latency_ms=latency_ms,
user_id=user_id,
session_id=session_id,
)
return {
"success": False,
"error": error_message,
"latency_ms": latency_ms,
}
사용 예시
if __name__ == "__main__":
client = HolySheepAIClient()
# 테스트 메시지
messages = [
{"role": "system", "content": "당신은 규정 준수 감사를 지원하는 어시스턴트입니다."},
{"role": "user", "content": "2024년 1분기 AI API 사용량의 총 토큰 사용량을 분석해주세요."},
]
# DeepSeek 모델로 테스트 (가장 저렴한 가격)
result = client.chat_completion(
model="deepseek-v3.2",
messages=messages,
max_tokens=500,
temperature=0.5,
user_id="audit-user-001",
)
if result["success"]:
print(f"API 호출 성공!")
print(f"응답: {result['data']['choices'][0]['message']['content'][:200]}...")
print(f"지연 시간: {result['latency_ms']:.2f}ms")
else:
print(f"API 호출 실패: {result['error']}")
6단계: 규정 준수 보고서 생성기
수집된 로그 데이터를 기반으로 규정 준수 감사 보고서를 생성하는 모듈을 만들겠습니다. 이 보고서는 규제 기관이나 내부 감사 부서에 제출할 수 있는 형식이어야 합니다.
# report_generator.py
import json
import csv
import os
from datetime import datetime
from typing import Dict, List, Any, Optional
from collections import defaultdict
from config import (
AUDIT_CONFIG,
LOG_CONFIG,
TOKEN_PRICING,
ANOMALY_THRESHOLDS,
)
from audit_logger import APILogger
class ComplianceReportGenerator:
"""규정 준수 감사 보고서를 생성하는 클래스"""
def __init__(self, audit_logger: APILogger):
self.audit_logger = audit_logger
self.config = AUDIT_CONFIG
self.thresholds = ANOMALY_THRESHOLDS
# 출력 디렉토리 생성
os.makedirs(LOG_CONFIG["output_dir"], exist_ok=True)
def generate_summary_statistics(self, logs: List[Dict[str, Any]]) -> Dict[str, Any]:
"""로그 데이터에서 요약 통계를 계산합니다"""
stats = {
"total_api_calls": 0,
"successful_calls": 0,
"failed_calls": 0,
"total_input_tokens": 0,
"total_output_tokens": 0,
"total_cost_usd": 0.0,
"average_latency_ms": 0.0,
"calls_by_model": defaultdict(int),
"cost_by_model": defaultdict(float),
"tokens_by_model": defaultdict(int),
"latencies_by_model": defaultdict(list),
"errors_by_type": defaultdict(int),
"compliance_events": [],
}
latencies = []
for log in logs:
if log.get("event_type") == "api_call":
stats["total_api_calls"] += 1
# 성공/실패 카운트
if log.get("status") == "success":
stats["successful_calls"] += 1
else:
stats["failed_calls"] += 1
stats["errors_by_type"][log.get("status")] += 1
# 토큰 및 비용
if "usage" in log:
stats["total_input_tokens"] += log["usage"].get("input_tokens", 0)
stats["total_output_tokens"] += log["usage"].get("output_tokens", 0)
stats["total_cost_usd"] += log.get("cost_usd", 0.0)
# 모델별 통계
model = log.get("model", "unknown")
stats["calls_by_model"][model] += 1
stats["cost_by_model"][model] += log.get("cost_usd", 0.0)
stats["tokens_by_model"][model] += log["usage"].get("total_tokens", 0) if "usage" in log else 0
# 지연 시간
if log.get("latency_ms"):
stats["latencies_by_model"][model].append(log["latency_ms"])
latencies.append(log["latency_ms"])
elif "compliance" in log.get("event_type", ""):
stats["compliance_events"].append(log)
# 평균 지연 시간 계산
if latencies:
stats["average_latency_ms"] = sum(latencies) / len(latencies)
# 모델별 평균 지연 시간
for model, model_latencies in stats["latencies_by_model"].items():
stats[f"avg_latency_{model}"] = sum(model_latencies) / len(model_latencies)
return stats
def detect_anomalies(self, logs: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
"""이상치를 탐지합니다"""
anomalies = []
# 일별 비용 분석
daily_costs = defaultdict(float)
for log in logs:
if log.get("event_type") == "api_call" and log.get("status") == "success":
timestamp = log.get("timestamp", "")[:10] # YYYY-MM-DD만 추출
daily_costs[timestamp] += log.get("cost_usd", 0.0)
for date, cost in daily_costs.items():
if cost > self.thresholds["max_cost_per_day_usd"]:
anomalies.append({
"type": "high_daily_cost",
"date": date,
"cost": cost,
"threshold": self.thresholds["max_cost_per_day_usd"],
"severity": "high",
"description": f"일일 비용 임계값 초과: ${cost:.2f} (제한: ${self.thresholds['max_cost_per_day_usd']})",
})
# 대량 토큰 사용 탐지
for log in logs:
if log.get("event_type") == "api_call":
if "usage" in log:
total_tokens = log["usage"].get("total_tokens", 0)
if total_tokens > self.thresholds["max_tokens_per_request"]:
anomalies.append({
"type": "excessive_tokens",
"timestamp": log.get("timestamp"),
"model": log.get("model"),
"tokens": total_tokens,
"threshold": self.thresholds["max_tokens_per_request"],
"severity": "medium",
"description": f"단일 요청 토큰 수 초과: {total_tokens} (제한: {self.thresholds['max_tokens_per_request']})",
})
return anomalies
def generate_text_report(self, stats: Dict[str, Any], anomalies: List[Dict[str, Any]]) -> str:
"""텍스트 형태의 규정 준수 보고서를 생성합니다"""
report_date = datetime.now().strftime("%Y-%m-%d %H:%M:%S UTC")
report = f"""
================================================================================
AI API 규정 준수 감사 보고서
================================================================================
생성 일시: {report_date}
조직: {self.config['organization']}
감사 기간: {self.config['audit_period_start']} ~ {self.config['audit_period_end']}
준수 프레임워크: {', '.join(self.config['compliance_frameworks'])}
--------------------------------------------------------------------------------
1. 개요 (Executive Summary)
--------------------------------------------------------------------------------
총 API 호출 횟수: {stats['total_api_calls']:,}회
성공률: {(stats['successful_calls'] / stats['total_api_calls'] * 100):.2f}% ({stats['successful_calls']:,}회)
실패율: {(stats['failed_calls'] / stats['total_api_calls'] * 100):.2f}% ({stats['failed_calls']:,}회)
--------------------------------------------------------------------------------
2. 비용 분석 (Cost Analysis)
--------------------------------------------------------------------------------
총 비용: ${stats['total_cost_usd']:.6f}
총 입력 토큰: {stats['total_input_tokens']:,} 토큰
총 출력 토큰: {stats['total_output_tokens']:,} 토큰
총 토큰: {stats['total_input_tokens'] + stats['total_output_tokens']:,} 토큰
모델별 비용 상세:
"""
for model, cost in stats['cost_by_model'].items():
if cost > 0:
report += f" - {model}: ${cost:.6f} ({stats['tokens_by_model'][model]:,} 토큰)\n"
report += f"""
--------------------------------------------------------------------------------
3. 성능 지표 (Performance Metrics)
--------------------------------------------------------------------------------
평균 응답 지연: {stats['average_latency_ms']:.2f}ms
모델별 평균 응답 시간:
"""
for model, latencies in stats['latencies_by_model'].items():
if latencies:
avg = sum(latencies) / len(latencies)
report += f" - {model}: {avg:.2f}ms (샘플 수: {len(latencies)})\n"
report += f"""
--------------------------------------------------------------------------------
4. 오류 분석 (Error Analysis)
--------------------------------------------------------------------------------
총 오류 횟수: {stats['failed_calls']}회
"""
if stats['errors_by_type']:
report += "오류 유형별 분포:\n"
for error_type, count in stats['errors_by_type'].items():
report += f" - {error_type}: {count}회\n"
else:
report += "오류 유형별 분포: 없음\n"
report += f"""
--------------------------------------------------------------------------------
5. 이상 탐지 (Anomaly Detection)
--------------------------------------------------------------------------------
탐지된 이상 항목 수: {len(anomalies)}건
"""
if anomalies:
for i, anomaly in enumerate(anomalies, 1):
report += f"""
[{i}] {anomaly['type']}
심각도: {anomaly['severity'].upper()}
설명: {anomaly['description']}
"""
else:
report += "탐지된 이상 항목 없음\n"
report += f"""
--------------------------------------------------------------------------------
6. 규정 준수 이벤트 로그
--------------------------------------------------------------------------------
"""
if stats['compliance_events']:
for event in stats['compliance_events'][-20:]: # 최근 20개만
report += f"[{event['timestamp']}] {event['event_type']} - {event['description']}\n"
else:
report += "규정 준수 이벤트 없음\n"
report += """
================================================================================
보고서 종료
================================================================================
"""
return report
def generate_json_report(
self,
stats: Dict[str, Any],
anomalies: List[Dict[str, Any]],
) -> Dict[str, Any]:
"""JSON 형태의 규정 준수 보고서를 생성합니다"""
return {
"report_metadata": {
"generated_at": datetime.utcnow().isoformat() + "Z",
"report_type": "AI_API_compliance_audit",
"organization": self.config["organization"],
"audit_period": {
"start": self.config["audit_period_start"],
"end": self.config["audit_period_end"],
},
"compliance_frameworks": self.config["compliance_frameworks"],
},
"summary": {
"total_api_calls": stats["total_api_calls"],
"successful_calls": stats["successful_calls"],
"failed_calls": stats["failed_calls"],
"success_rate_percent": round(
stats["successful_calls"] / stats["total_api_calls"] * 100, 2
) if stats["total_api_calls"] > 0 else 0,
},
"cost_analysis": {
"total_cost_usd": round(stats["total_cost_usd"], 6),
"total_input_tokens": stats["total_input_tokens"],
"total_output_tokens": stats["total_output_tokens"],
"total_tokens": stats["total_input_tokens"] + stats["total_output_tokens"],
"by_model": dict(stats["cost_by_model"]),
},
"performance_metrics": {
"average_latency_ms": round(stats["average_latency_ms"], 2),
"latencies_by_model": {
model: round(sum(lats) / len(lats), 2)
for model, lats in stats["latencies_by_model"].items()
},
},
"anomalies": anomalies,
"compliance_events": stats["compliance_events"],
}
def generate_csv_report(self, logs: List[Dict[str, Any]]) -> str:
"""CSV 형태의 상세 로그 보고서를 생성합니다"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
csv_filename = f"audit_log_detail_{timestamp}.csv"
csv_path = os.path.join(LOG_CONFIG["output_dir"], csv_filename)
with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
fieldnames = [
'timestamp', 'event_type', 'model', 'status',
'input_tokens', 'output_tokens', 'total_tokens',
'cost_usd', 'latency_ms', 'user_id', 'session_id', 'error_message',
]
writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
writer.writeheader()
for log in logs:
if log.get("event_type") == "api_call":
row = {
'timestamp': log.get('timestamp', ''),
'event_type': log.get('event_type', ''),
'model': log.get('model', ''),
'status': log.get('status', ''),
'input_tokens': log.get('usage', {}).get('input_tokens', ''),
'output_tokens': log.get('usage', {}).get('output_tokens', ''),
'total_tokens': log.get('usage', {}).get('total_tokens', ''),
'cost_usd': log.get('cost_usd', ''),
'latency_ms': log.get('latency_ms', ''),
'user_id': log.get('user_id', ''),
'session_id': log.get('session_id', ''),
'error_message': log.get('error_message', ''),
}
writer.writerow(row)
return csv_path
def generate_all_reports(self, logs: List[Dict[str, Any]]) -> Dict[str, str]:
"""모든 형태의 보고서를 생성합니다"""
# 통계 계산
stats = self.generate_summary_statistics(logs)
# 이상 탐지
anomalies = self.detect_anomalies(logs)
# 텍스트 보고서
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
text_report_path = os.path.join(
LOG_CONFIG["output_dir"],
f"compliance_report_{timestamp}.txt"
)
with open(text_report_path, 'w', encoding='utf-8') as f:
f.write(self.generate_text_report(stats, anomalies))
# JSON 보고서
json_report_path = os.path.join(
LOG_CONFIG["output_dir"],
f"compliance_report_{timestamp}.json"
)
json_report = self.generate_json_report(stats, anomalies)
with open(json_report_path, 'w', encoding='utf-8') as f:
json.dump(json_report, f, ensure_ascii=False, indent=2)
# CSV 상세 로그
csv_report_path = self.generate_csv_report(logs)
return {
"text_report": text_report_path,
"json_report": json_report_path,
"csv_report": csv_report_path,
"statistics": stats,
"anomalies": anomalies,
}
실행 예시
if __name__ == "__main__":
# 로거에서 로그 데이터 로드
audit_logger = APILogger()
# 기존 로그 파일이 있다면 로드
log_dir = LOG_CONFIG["log_dir"]
if os.path.exists(log_dir):
log_files = [f for f in os.listdir(log_dir) if f.endswith('.jsonl')]
if log_files:
latest_log = os.path.join(log_dir, sorted(log_files)[-1])
with open(latest_log, 'r', encoding='utf-8') as f:
for line in f:
try:
log_entry = json.loads(line.strip())
audit_logger.audit_logs.append(log_entry)
except json.JSONDecodeError:
pass
# 보고서 생성
generator = ComplianceReportGenerator(audit_logger)
logs = audit_logger.get_all_logs()
if logs:
reports = generator.generate_all_reports(logs)
print("보고서 생성 완료!")
print(f"텍스트 보고서: {reports['text_report']}")
print(f"JSON 보고서: {reports['json_report']}")
print(f"CSV 보고서: {reports['csv_report']}")
print(f"\n총 로그 수: {len(logs)}")
print(f"탐지된 이상 항목: {len(reports['anomalies'])}건")
else:
print("로깅된 데이터가 없습니다. 먼저 API 호출을 실행해주세요.")
7단계: 대시보드 생성
터미널에서 확인할 수 있는 간단한 대시보드를 만들어보겠습니다. 실무에서는 이 데이터를 Grafana나 PowerBI 등에 연동하여 시각화합니다.
# dashboard.py
from typing import Dict, List, Any
from datetime import datetime
def print_dashboard(stats: Dict[str, Any], anomalies: List[Dict[str, Any]]):
"""터미널용 규정 준수 대시보드를 출력합니다"""
# ANSI 색상 코드
GREEN = '\033[92m'
RED = '\033[91m'
YELLOW = '\033[93m'
BLUE = '\033[94m'
BOLD = '\033[1m'
RESET = '\033[0m