Skip to main content

​
1. 왜 Trace 분석과 평가가 필요한가

LLM (Large Language Model) 기반 애플리케이션과 에이전트 (Agent) 는 블랙박스 특성을 갖습니다. 동일한 입력에도 확률적으로 다른 출력이 나오며, 내부 추론 과정이 불투명합니다. 이런 특성은 다음 세 가지 문제를 만들어냅니다.

​
1-1. 블랙박스 문제

전통적인 소프트웨어는 단위 테스트 (Unit Test) 로 품질을 보장할 수 있습니다. 하지만 LLM 응답은 결정론적이지 않기 때문에, 실제 실행 흔적 (Trace) 을 기록하고 분석 해야만 문제의 원인을 추적할 수 있습니다.
일반 소프트웨어LLM 애플리케이션
입력 → 결정론적 출력입력 → 확률적 출력
단위 테스트로 검증 가능Trace 기반 분석 필요
코드 디버깅으로 원인 파악Span 단위 분석으로 원인 파악
배포 후 변경 없음프롬프트/모델/데이터 드리프트 발생

​
1-2. 프로덕션 품질 보장

개발 환경에서 잘 동작하던 앱이 프로덕션에서 품질이 저하되는 이유는 다양합니다.
  • 데이터 드리프트 (Data Drift): 사용자가 예상치 못한 질문 패턴 사용
  • 모델 드리프트 (Model Drift): LLM 공급자가 모델을 조용히 업데이트
  • 컨텍스트 오염 (Context Pollution): 검색된 문서의 품질 저하
  • 프롬프트 엣지 케이스 (Prompt Edge Case): 특정 입력에서만 실패하는 케이스
Trace 분석을 통해 이런 문제를 조기에 감지 하고 대응할 수 있습니다.

​
1-3. 비용 최적화

LLM API 호출 비용은 입력/출력 토큰 수에 비례합니다. Trace 데이터를 분석하면 다음을 최적화할 수 있습니다.
  • 불필요하게 긴 시스템 프롬프트 식별
  • 중복 LLM 호출 (캐싱으로 대체 가능한 구간) 발견
  • 응답 생성에 불필요한 토큰이 포함된 패턴 탐지
  • 고비용 모델 호출을 저비용 모델로 대체 가능한 케이스 분류

​
2. Trace 검색과 필터링

​
2-1. mlflow.search_traces() API

mlflow.search_traces() 는 기록된 Trace 를 Python 코드로 검색하는 핵심 API 입니다. 반환값은 pandas.DataFrame 입니다. 
import mlflow
import pandas as pd

# 기본 사용법: 특정 실험의 최근 Trace 조회
traces_df = mlflow.search_traces(
    experiment_ids=["123456789"],  # 실험 ID 목록
    max_results=100,               # 최대 반환 건수 (기본값: 100)
)

print(traces_df.columns.tolist())
# ['trace_id', 'experiment_id', 'timestamp_ms', 'execution_time_ms',
#  'status', 'request', 'response', 'tags', 'spans']

​
2-2. 태그 기반 필터링

# 특정 태그 조건으로 필터링
# filter_string은 SQL WHERE 절과 유사한 DSL을 사용합니다
traces_df = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string="tag.user_score >= '4' AND tag.environment = 'production'",
    max_results=500,
)

​
2-3. 상태 및 시간 기반 필터링

# 에러 상태의 Trace만 조회
error_traces = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string="status = 'ERROR'",
    max_results=200,
)

# 특정 기간의 Trace 조회 (timestamp_ms는 Unix 밀리초)
import time

seven_days_ago_ms = int((time.time() - 7 * 86400) * 1000)

recent_traces = mlflow.search_traces(
    experiment_ids=["123456789"],
    filter_string=f"timestamp_ms >= {seven_days_ago_ms}",
    order_by=["timestamp_ms DESC"],
    max_results=1000,
)

print(f"최근 7일 Trace 수: {len(recent_traces)}")

​
2-4. UI에서의 Trace 탐색

MLflow UI 에서 Trace 를 탐색할 때는 다음 경로를 사용합니다.
  1. Experiment 페이지 → Traces 탭 클릭
  2. 좌측 필터 패널에서 Status, 날짜 범위, 태그 필터 적용
  3. 특정 Trace 클릭 → Span 트리 (Span Tree) 뷰에서 단계별 실행 확인
  4. Span 클릭 시 입출력 (Input/Output), 속성 (Attributes), 이벤트 (Events) 상세 확인
Databricks 환경에서는 MLflow UI 가 Workspace UI 에 통합되어 있습니다. Experiment 페이지 접근 경로: Experiments → (실험명) → Traces 탭

​
3. Trace 분석 패턴

​
3-1. Latency 병목 Span 찾기

import mlflow
import pandas as pd

traces_df = mlflow.search_traces(
    experiment_ids=["123456789"],
    max_results=1000,
)

# Span 목록을 플랫하게 펼치기
span_records = []
for _, row in traces_df.iterrows():
    for span in row["spans"]:
        span_records.append({
            "trace_id": row["trace_id"],
            "span_name": span["name"],
            "span_type": span.get("span_type", "UNKNOWN"),
            "latency_ms": (span["end_time_ns"] - span["start_time_ns"]) / 1e6,
            "status": span["status"]["status_code"],
        })

spans_df = pd.DataFrame(span_records)

# Span 유형별 Latency 통계
latency_stats = (
    spans_df.groupby("span_type")["latency_ms"]
    .agg(["mean", "median", lambda x: x.quantile(0.95), "count"])
    .rename(columns={"mean": "avg_ms", "median": "p50_ms", "<lambda_0>": "p95_ms", "count": "n"})
    .sort_values("p95_ms", ascending=False)
)

print(latency_stats)
# span_type     avg_ms   p50_ms   p95_ms     n
# LLM           1200.0   1100.0   2500.0   890
# RETRIEVER       80.0     65.0    200.0   890
# EMBEDDING       25.0     20.0     50.0   890
# → LLM이 P95 기준 전체 지연의 90% 이상을 차지
SQL (Unity Catalog 시스템 테이블) 로도 동일한 분석이 가능합니다. 
-- 가장 느린 Span 유형 식별 (Unity Catalog 시스템 테이블 기준)
SELECT
    span_type,
    AVG(latency_ms)                       AS avg_latency,
    PERCENTILE(latency_ms, 0.50)          AS p50_latency,
    PERCENTILE(latency_ms, 0.95)          AS p95_latency,
    PERCENTILE(latency_ms, 0.99)          AS p99_latency,
    COUNT(*)                              AS span_count
FROM system.mlflow.trace_spans
WHERE timestamp >= CURRENT_DATE() - INTERVAL 7 DAYS
GROUP BY span_type
ORDER BY p95_latency DESC;

​
3-2. Token 사용량 분석

# Trace 에서 LLM Span의 토큰 사용량 집계
token_records = []
for _, row in traces_df.iterrows():
    for span in row["spans"]:
        if span.get("span_type") == "LLM":
            attrs = span.get("attributes", {})
            token_records.append({
                "trace_id": row["trace_id"],
                "input_tokens": attrs.get("llm.token_count.prompt", 0),
                "output_tokens": attrs.get("llm.token_count.completion", 0),
                "model": attrs.get("llm.model_name", "unknown"),
            })

tokens_df = pd.DataFrame(token_records)
tokens_df["cost_usd"] = (
    tokens_df["input_tokens"] * 0.000003 +   # GPT-4o 기준 예시
    tokens_df["output_tokens"] * 0.000015
)

print("모델별 일평균 비용 추정:")
print(tokens_df.groupby("model")["cost_usd"].sum())
print(f"\n총 예상 비용 (수집 기간): ${tokens_df['cost_usd'].sum():.2f}")

​
3-3. 에러 패턴 분석

# 에러가 발생한 Trace 비율 확인
total = len(traces_df)
error_count = (traces_df["status"] == "ERROR").sum()
print(f"에러율: {error_count}/{total} ({error_count/total*100:.1f}%)")

# 에러 Span 의 원인 분류
error_spans = []
for _, row in traces_df[traces_df["status"] == "ERROR"].iterrows():
    for span in row["spans"]:
        if span["status"]["status_code"] == "ERROR":
            events = span.get("events", [])
            for event in events:
                if event.get("name") == "exception":
                    error_spans.append({
                        "span_name": span["name"],
                        "error_type": event["attributes"].get("exception.type", "Unknown"),
                        "error_msg": event["attributes"].get("exception.message", "")[:80],
                    })

error_df = pd.DataFrame(error_spans)
print("\n에러 유형별 발생 횟수:")
print(error_df["error_type"].value_counts().head(10))