Trace 검색과 분석 패턴

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']

# 특정 태그 조건으로 필터링
# 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,
)

# 에러 상태의 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)}")

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% 이상을 차지

-- 가장 느린 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;

# 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}")

# 에러가 발생한 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))