9장: AI 서비스 관측 가능성

import time
from opentelemetry import trace, metrics
from opentelemetry.trace import SpanKind
 
tracer = trace.get_tracer("ai-service.llm")
meter = metrics.get_meter("ai-service.llm")
 
# 메트릭 정의
token_counter = meter.create_counter(
    "llm.token.usage",
    description="Total tokens used in LLM calls",
    unit="tokens",
)
llm_duration = meter.create_histogram(
    "llm.request.duration",
    description="Duration of LLM API calls",
    unit="s",
)
llm_cost = meter.create_counter(
    "llm.request.cost",
    description="Estimated cost of LLM API calls",
    unit="USD",
)
 
# 모델별 토큰당 비용 (USD)
MODEL_PRICING = {
    "gpt-4o": {"input": 0.0025 / 1000, "output": 0.01 / 1000},
    "gpt-4o-mini": {"input": 0.00015 / 1000, "output": 0.0006 / 1000},
    "claude-sonnet-4-20250514": {"input": 0.003 / 1000, "output": 0.015 / 1000},
}
 
 
def traced_llm_call(client, model: str, messages: list, **kwargs):
    """OTel 계측이 적용된 LLM 호출 래퍼"""
    
    with tracer.start_as_current_span(
        "llm.chat.completion",
        kind=SpanKind.CLIENT,
    ) as span:
        # 요청 속성 설정
        span.set_attribute("llm.model", model)
        span.set_attribute("llm.provider", "openai")
        span.set_attribute("llm.request.temperature", kwargs.get("temperature", 1.0))
        span.set_attribute("llm.request.max_tokens", kwargs.get("max_tokens", 0))
        
        start_time = time.time()
        
        try:
            response = client.chat.completions.create(
                model=model,
                messages=messages,
                **kwargs,
            )
            duration = time.time() - start_time
            
            # 응답 속성 설정
            usage = response.usage
            span.set_attribute("llm.usage.input_tokens", usage.prompt_tokens)
            span.set_attribute("llm.usage.output_tokens", usage.completion_tokens)
            span.set_attribute("llm.usage.total_tokens", usage.total_tokens)
            span.set_attribute("llm.response.finish_reason", response.choices[0].finish_reason)
            
            # 메트릭 기록
            attrs = {"llm.model": model, "llm.provider": "openai"}
            
            token_counter.add(usage.prompt_tokens, {**attrs, "llm.token.type": "input"})
            token_counter.add(usage.completion_tokens, {**attrs, "llm.token.type": "output"})
            llm_duration.record(duration, attrs)
            
            # 비용 계산
            pricing = MODEL_PRICING.get(model, {"input": 0, "output": 0})
            cost = (
                usage.prompt_tokens * pricing["input"]
                + usage.completion_tokens * pricing["output"]
            )
            llm_cost.add(cost, attrs)
            span.set_attribute("llm.cost.estimated_usd", cost)
            
            return response
            
        except Exception as e:
            span.set_status(trace.StatusCode.ERROR, str(e))
            span.record_exception(e)
            raise

from opentelemetry import trace
 
tracer = trace.get_tracer("ai-service.agent")
 
 
class TracedAgent:
    def __init__(self, llm_client):
        self.llm = llm_client
        self.max_iterations = 10
    
    def run(self, user_query: str):
        with tracer.start_as_current_span("agent.run") as root_span:
            root_span.set_attribute("agent.query", user_query)
            root_span.set_attribute("agent.max_iterations", self.max_iterations)
            
            iteration = 0
            result = None
            
            while iteration < self.max_iterations:
                iteration += 1
                
                # 추론 단계
                with tracer.start_as_current_span(f"agent.think") as think_span:
                    think_span.set_attribute("agent.iteration", iteration)
                    action = self._plan_next_action(user_query, result)
                    think_span.set_attribute("agent.action.type", action["type"])
                
                if action["type"] == "final_answer":
                    root_span.set_attribute("agent.total_iterations", iteration)
                    root_span.set_attribute("agent.status", "completed")
                    return action["content"]
                
                # 도구 호출 단계
                with tracer.start_as_current_span("agent.tool_call") as tool_span:
                    tool_span.set_attribute("agent.tool.name", action["tool"])
                    tool_span.set_attribute("agent.tool.input", str(action["input"]))
                    
                    result = self._execute_tool(action["tool"], action["input"])
                    
                    tool_span.set_attribute("agent.tool.output_length", len(str(result)))
                    tool_span.add_event("tool-executed", {
                        "tool.name": action["tool"],
                        "tool.success": True,
                    })
            
            root_span.set_attribute("agent.status", "max_iterations_reached")
            root_span.set_status(trace.StatusCode.ERROR, "Max iterations reached")
            return None

import logging
import hashlib
 
logger = logging.getLogger("ai-service.prompts")
 
 
def log_llm_interaction(
    model: str,
    messages: list,
    response_text: str,
    log_content: bool = False,
):
    """LLM 상호작용 로깅 (개인정보 고려)"""
    
    log_data = {
        "llm.model": model,
        "llm.message_count": len(messages),
        "llm.response_length": len(response_text),
    }
    
    if log_content:
        # 개발/스테이징 환경에서만 프롬프트 내용 로깅
        log_data["llm.prompt"] = messages[-1]["content"][:1000]  # 최대 1000자
        log_data["llm.response"] = response_text[:1000]
    else:
        # 프로덕션 환경에서는 해시만 기록
        prompt_text = messages[-1]["content"]
        log_data["llm.prompt_hash"] = hashlib.sha256(prompt_text.encode()).hexdigest()[:16]
        log_data["llm.response_hash"] = hashlib.sha256(response_text.encode()).hexdigest()[:16]
    
    logger.info("LLM 호출 완료", extra=log_data)

meter = metrics.get_meter("ai-service.quality")
 
# 응답 길이 분포 -- 급격한 변화는 드리프트 의심
response_length = meter.create_histogram(
    "llm.response.length",
    description="Length of LLM responses in characters",
    unit="characters",
)
 
# 구조화 응답 파싱 성공률 -- JSON 파싱 실패율 증가는 드리프트 징후
parse_success = meter.create_counter(
    "llm.response.parse.count",
    description="Count of response parsing attempts",
)
 
# 사용자 피드백 기반 품질 점수
quality_score = meter.create_histogram(
    "llm.response.quality_score",
    description="Quality score of LLM responses (0-1)",
)
 
# 토큰 효율성 -- 같은 작업에 토큰 사용량이 증가하면 드리프트
token_efficiency = meter.create_histogram(
    "llm.token.efficiency",
    description="Tokens used per unit of useful output",
    unit="tokens/char",
)

groups:
  - name: ai-drift-detection
    rules:
      # 응답 길이가 지난 7일 평균 대비 50% 이상 변화
      - alert: LLMResponseLengthDrift
        expr: |
          abs(
            avg_over_time(llm_response_length_characters[1h])
            - avg_over_time(llm_response_length_characters[7d])
          ) / avg_over_time(llm_response_length_characters[7d]) > 0.5
        for: 30m
        labels:
          severity: warning
        annotations:
          summary: "LLM 응답 길이 드리프트 감지"
 
      # JSON 파싱 실패율이 5%를 초과
      - alert: LLMParseFailureRate
        expr: |
          rate(llm_response_parse_count_total{result="failure"}[5m])
          / rate(llm_response_parse_count_total[5m]) > 0.05
        for: 10m
        labels:
          severity: critical

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from opentelemetry.instrumentation.langchain import LangchainInstrumentor
 
# LangChain 자동 계측 활성화
LangchainInstrumentor().instrument()
 
# 일반적인 LangChain 코드 -- 자동으로 스팬 생성
prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 기술 문서 작성 전문가입니다."),
    ("human", "{topic}에 대해 설명해 주세요."),
])
 
model = ChatOpenAI(model="gpt-4o", temperature=0.3)
chain = prompt | model | StrOutputParser()
 
# 실행 시 자동으로 트레이스 생성
# - chain.invoke 스팬
#   - prompt.format 스팬
#   - llm.call 스팬 (토큰 사용량 포함)
#   - output_parser.parse 스팬
result = chain.invoke({"topic": "Kubernetes 네트워킹"})

from llama_index.core import VectorStoreIndex, SimpleDirectoryReader
from opentelemetry.instrumentation.llamaindex import LlamaIndexInstrumentor
 
# LlamaIndex 자동 계측
LlamaIndexInstrumentor().instrument()
 
# RAG 파이프라인 -- 각 단계가 자동 추적됨
documents = SimpleDirectoryReader("./data").load_data()
index = VectorStoreIndex.from_documents(documents)
 
# 쿼리 시 생성되는 트레이스:
# - query 스팬
#   - retrieval 스팬 (벡터 검색)
#     - embedding 스팬 (임베딩 생성)
#   - synthesis 스팬 (LLM 호출)
query_engine = index.as_query_engine()
response = query_engine.query("OpenTelemetry의 장점은 무엇인가요?")