AI 애플리케이션 관측 가능성(Observability) 완전 가이드: 2026년 프로덕션 환경 AI 모니터링 실전

서론
2026년, AI 애플리케이션은 프로덕션(Production) 환경에 널리 적용되었습니다. 하지만 AI 애플리케이션은 다음과 같은 독특한 특성을 가집니다: 모델 출력의 불안정성, 높은 지연 시간(Latency), 예측하기 어려운 비용. 기존의 애플리케이션 모니터링(APM)으로는 AI 모니터링의 요구사항을 충족할 수 없습니다. 본문에서는 AI 애플리케이션 관측 가능성의 핵심 방법론을 소개합니다.

AI 관측 가능성이란 무엇인가

전통적 모니터링 vs AI 모니터링

차원	전통적 모니터링	AI 모니터링
지연 시간 (Latency)	HTTP 요청 소요 시간	API 호출 + 모델 추론(Inference) 소요 시간
에러율 (Error Rate)	4xx/5xx 상태 코드	거부(Rejection), 환각(Hallucination), 형식 오류
비용 (Cost)	고정된 클라우드 리소스	토큰(Token) 소모 변동성
품질 (Quality)	정확하게 측정 가능	추가적인 평가 필요

AI 관측 가능성의 4대 지주
├── 로깅 (Logging, AI 요청 로그)
├── 메트릭 (Metrics, 토큰 소모, 지연 시간, 비용)
├── 트레이싱 (Tracing, AI 호출 체인 추적)
└── 평가 (Evaluation, 출력 품질 평가)

핵심 지표 체계

지연 시간 지표

import time
from functools import wraps

class AILatencyTracker :
    def __init__ ( self ):
        self . latencies = []

    def track ( self , func ):
        """ 지연 시간을 추적하는 데코레이터(Decorator) """
        @wraps ( func )
        async def async_wrapper ( * args , ** kwargs ):
            start = time . time ()
            try:
                result = await func ( * args , ** kwargs )
                elapsed = time . time () - start
                self . record ( " success " , elapsed )
                return result
            except Exception as e:
                elapsed = time . time () - start
                self . record ( " error " , elapsed )
                raise e

        @wraps ( func )
        def sync_wrapper ( * args , ** kwargs ):
            start = time . time ()
            try:
                result = func ( * args , ** kwargs )
                elapsed = time . time () - start
                self . record ( " success " , elapsed )
                return result
            except Exception as e:
                elapsed = time . time () - start
                self . record ( " error " , elapsed )
                raise e

        import asyncio
        if asyncio . iscoroutinefunction ( func ):
            return async_wrapper
        return sync_wrapper

    def record ( self , status : str , latency : float ):
        self . latencies . append ({ " timestamp " : time . time (), " status " : status , " latency_ms " : latency * 1000 })

    def get_stats ( self ) -> dict :
        """ 통계 정보 가져오기 """
        if not self . latencies :
            return {}
        latencies = [ l [ " latency_ms " ] for l in self . latencies ]
        return {
            " count " : len ( latencies ),
            " avg_ms " : sum ( latencies ) / len ( latencies ),
            " p50_ms " : sorted ( latencies )[ len ( latencies ) // 2 ],
            " p95_ms " : sorted ( latencies )[ int ( len ( latencies ) * 0.95 )],
            " p99_ms " : sorted ( latencies )[ int ( len ( latencies ) * 0.99 )],
        }

토큰(Token) 소모 지표

class TokenTracker :
    def __init__ ( self ):
        self . records = []
        self .

total_input_tokens = 0 self . total_output_tokens = 0 def record ( self , model : str , input_tokens : int , output_tokens : int , cost : float ): """ 토큰 사용량 기록 """ self . total_input_tokens += input_tokens self . total_output_tokens += output_tokens self . records . append ({ " timestamp " : time . time (), " model " : model , " input_tokens " : input_tokens , " output_tokens " : output_tokens , " total_tokens " : input_tokens + output_tokens , " cost " : cost def get_daily_cost ( self ) -> dict : """ 일일 비용 가져오기 """ today = time . time () - 86400 # 24시간 전 recent = [ r for r in self . records if r [ " timestamp " ] > today ] total_cost = sum ( r [ " cost " ] for r in recent ) total_tokens = sum ( r [ " total_tokens " ] for r in recent ) " cost_today " : total_cost , " tokens_today " : total_tokens , " avg_cost_per_request " : total_cost / len ( recent ) if recent else 0 def get_model_breakdown ( self ) -> dict : """ 모델별 분류 통계 가져오기 """ breakdown = {} for r in self . records : model = r [ " model " ] if model not in breakdown : breakdown [ model ] = { " cost " : 0 , " tokens " : 0 , " count " : 0 } breakdown [ model ][ " cost " ] += r [ " cost " ] breakdown [ model ][ " tokens " ] += r [ " total_tokens " ] breakdown [ model ][ " count " ] += 1 return breakdown 3. 오류 분류 class AIErrorClassifier : ERROR_TYPES = { " rate_limit " : { " retry " : True , " severity " : " medium " }, " auth_error " : { " retry " : False , " severity " : " high " }, " model_error " : { " retry " : True , " severity " : " medium " }, " timeout " : { " retry " : True , " severity " : " low " }, " invalid_request " : { " retry " : False , " severity " : " high " }, " content_filtered " : { " retry " : False , " severity " : " medium " }, @classmethod def classify ( cls , error : Exception ) -> dict : """ 오류 유형 분류 """ error_str = str ( error ). lower () if " 429 " in error_str or " rate_limit " in error_str : return { " type " : " rate_limit " , ** cls .

elif " 400 " in error_str or " invalid " in error_str : return { " type " : " invalid_request " , ** cls . ERROR_TYPES [ " invalid_request " ]} elif " filtered " in error_str or " content " in error_str : return { " type " : " content_filtered " , ** cls . ERROR_TYPES [ " content_filtered " ]} return { " type " : " unknown " , " retry " : False , " severity " : " high " } @classmethod def should_retry ( cls , error : Exception ) -> bool : """ 판단할 오류를 재시도해야 하는지 확인합니다.""" classification = cls . classify ( error ) return classification . get ( " retry " , False ) 로그 시스템 구조화된 AI 로그 import json import logging from datetime import datetime class AILogger : def init ( self , log_file : str = " ai_logs.jsonl " ): self . log_file = log_file self . logger = logging . getLogger ( " ai " ) self . logger . setLevel ( logging . INFO ) handler = logging . FileHandler ( log_file ) handler . setFormatter ( logging . Formatter ( ' %(message)s ' )) self . logger . addHandler ( handler ) def log_request ( self , request_id : str , model : str , prompt : str , response : str = None , latency_ms : float = None , tokens_used : int = None , cost : float = None , error : str = None ): """ AI 요청을 기록합니다.""" log_entry = { " timestamp " : datetime . utcnow (). isoformat (), " type " : " ai_request " , " request_id " : request_id , " model " : model , " prompt_length " : len ( prompt ), " response_length " : len ( response ) if response else None , " latency_ms " : latency_ms , " tokens_used " : tokens_used , " cost " : cost , " error " : error , " success " : error is None self . logger . info ( json .

dumps(log_entry, ensure_ascii=False))
def log_evaluation(self, request_id: str, quality_score: float, categories: dict):
""" 기록 품질 평가 결과 """
log_entry = {
"timestamp": datetime.utcnow().isoformat(),
"type": "quality_evaluation",
"request_id": request_id,
"quality_score": quality_score,
"categories": categories
}
self.logger.info(json.dumps(log_entry, ensure_ascii=False))

로그 분석 쿼리 (日志分析查询)

import json
class LogAnalyzer:
def init(self, log_file: str):
self.log_file = log_file

def load_logs(self, limit: int = None):
    with open(self.log_file, 'r') as f:
        for i, line in enumerate(f):
            if limit and i >= limit:
                logs.append(json.loads(line))
    return logs

def get_error_rate(self, hours: int = 24) -> float:
    """ 오류율 계산 """
    cutoff = datetime.utcnow().timestamp() - hours * 3600
    logs = self.load_logs()
    recent = [l for l in logs if datetime.fromisoformat(l["timestamp"]).timestamp() > cutoff]
    if not recent:
        errors = sum(1 for l in recent if not l.get("success", True))
        return errors / len(recent)

def get_expensive_requests(self, top_n: int = 10) -> list:
    """ 가장 비용이 많이 드는 요청 가져오기 """
    logs = self.load_logs()
    sorted_logs = sorted([l for l in logs if l.get("cost")], key=lambda x: x.get("cost", 0), reverse=True)
    return sorted_logs[:top_n]

def get_slow_requests(self, threshold_ms: float = 5000) -> list:
    """ 느린 요청 가져오기 """
    logs = self.load_logs()
    return [l for l in logs if l.

get(" latency_ms " , 0 ) > threshold_ms ] 추적 링크 LangChain + OpenTelemetry from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor , ConsoleSpanExporter provider = TracerProvider () processor = BatchSpanProcessor ( ConsoleSpanExporter ()) provider . add_span_processor ( processor ) trace . set_tracer_provider ( provider ) tracer = trace . get_tracer ( name ) class AIServiceWithTracing : def init ( self ): self . llm = OpenAI () self . vector_db = VectorDB () @tracer.start_as_current_span ( " ai_request " ) async def process_request ( self , user_input : str , user_id : str ): span = trace . get_current_span () span . set_attribute ( " user_id " , user_id ) span . set_attribute ( " input_length " , len ( user_input )) # 1. 관련 문서 검색 with tracer . start_as_current_span ( " retrieve_context " ) as span : docs = self . vector_db . search ( user_input ) span . set_attribute ( " docs_retrieved " , len ( docs )) # 2. LLM 호출 with tracer . start_as_current_span ( " llm_call " ) as span : start = time . time () response = self . llm . generate ( user_input , docs ) span . set_attribute ( " model " , " gpt-5.4 " ) span . set_attribute ( " latency_ms " , ( time . time () - start ) * 1000 ) span . set_attribute ( " response_length " , len ( response )) span . set_attribute ( " success " , True ) return response except Exception as e : span . set_attribute ( " success " , False ) span . set_attribute ( " error " , str ( e )) 출력 품질 평가 자동 품질 평가 class AIOutputEvaluator : def init ( self ): self . llm = OpenAI () def evaluate ( self , prompt : str , response : str ) -> dict : """ 평가 출력 품질 """ evaluation_prompt = f """ 평가 다음 AI 출력의 품질: 사용자 입력: { prompt } AI 출력: { response } 평가 차원(각 항목 1-5점): 1. 관련성: 출력이 문제와 관련된가 2. 정확성: 정보가 올바른가 3. 완전성: 질문에 완전히 답변했는가 4. 명확도: 표현이 명확하고 읽기 쉬운가 5.

안전성(安全性): 부적절한 콘텐츠 포함 여부 "relevance" : 4, "accuracy" : 5, "completeness" : 4, "clarity" : 5, "safety" : 5, "overall_score" : 4.6, "issues" : [ "문제1" , "문제2" ], "suggestions" : [ "제안1" , "제안2" ] result = self.llm.generate(evaluation_prompt) return json.loads(result) return { " error " : "평가 분석 실패" , " raw " : result } def batch_evaluate(self, requests: list) -> list: results = [] for req in requests: evaluation = self.evaluate(req[ "prompt" ], req[ "response" ]) results.append( { "request_id" : req[ "id" ], ** evaluation return results def detect_hallucination ( self , response : str , context : str ) -> dict : detection_prompt = f """ 检测以下回答是否存在幻觉（编造不存在的信息）：上下文/背景： { context } AI 回答： { response } 1. 是否有具体事实（人名、日期、数字）需要验证 2. 这些事实是否在上下文中 3. 是否有明显编造的内容 " has_hallucination" : true / false , " confidence" : 0.85 , " risky_content" : [ "구체적 의심 내용" ], " reason" : "판단 이유" result = self . llm . generate ( detection_prompt ) return json . loads ( result ) return { " has_hallucination" : False , " confidence" : 0 } Prometheus 모니터링 대시보드 지표 내보내기 from prometheus_client import Counter , Histogram , Gauge , generate_latest REQUEST_COUNT = Counter ( ' ai_requests_total ' , ' 총 AI 요청 수 ' , [ ' model ' , ' status ' ] REQUEST_LATENCY = Histogram ( ' ai_request_latency_seconds ' , ' AI 요청 지연 시간 ' , TOKEN_USAGE = Counter ( ' ai_tokens_used_total ' , ' 사용된 총 토큰 수 ' , [ ' model ' , ' type ' ] # type: input/output COST_USAGE = Counter ( ' ai_cost_total ' , ' 총 API 비용 ' , ACTIVE_REQUESTS = Gauge ( ' ai_active_requests ' , ' 활성 요청 수 ' , @app.middleware ( "http" ) async def track_requests ( request : Request , call_next ): model = request . headers . get ( "X-Model" , "unknown" ) ACTIVE_REQUESTS . labels ( model = model ). inc () start = time . time () response = await call_next ( request ) latency = time . time () - start REQUEST_COUNT . labels ( model = model , status = response . status_code ).

경고 설정 핵심 경고 규칙 # alertmanager.yml 또는 모니터링 구성 - name : ai_application - alert : HighAIErrorRate sum(rate(ai_requests_total{status="error"}[5m])) sum(rate(ai_requests_total[5m])) > 0.05 severity : critical annotations : summary : " AI 요청 오류율이 5%를 초과함" - alert : HighAILatency histogram_quantile(0.95, sum(rate(ai_request_latency_seconds_bucket[5m])) by (le) severity : warning annotations : summary : " AI 요청 P95 지연 시간이 10초를 초과함" - alert : HighAICost increase(ai_cost_total[1h]) > 100 severity : warning annotations : summary : " AI 호출 비용 시간당 증가량이 $100을 초과함" - alert : AIRateLimit increase(ai_requests_total{status="429"}[5m]) > 10 severity : warning annotations : summary : " AI API 제한이 빈번하게 발생함" Grafana 대시보드 핵심 패널 ┌─────────────────────────────────────────────────────────────┐ │ AI Application Dashboard

AI 2026AI

요약

핵심 포인트

로그 분석 쿼리 (日志分析查询)

댓글