> ## Documentation Index > Fetch the complete documentation index at: https://docs.sifi.life/llms.txt > Use this file to discover all available pages before exploring further. # Trace 검색과 분석 패턴 > 이 문서는 **[머신러닝](/training/09-machine-learning/overview)** 섹션의 일부입니다. ## 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` 입니다. ```python theme={null} 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. 태그 기반 필터링 ```python theme={null} # 특정 태그 조건으로 필터링 # 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. 상태 및 시간 기반 필터링 ```python theme={null} # 에러 상태의 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 찾기 ```python theme={null} 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", "": "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 시스템 테이블) 로도 동일한 분석이 가능합니다. ```sql theme={null} -- 가장 느린 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 사용량 분석 ```python theme={null} # 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. 에러 패턴 분석 ```python theme={null} # 에러가 발생한 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)) ``` ***