버전 관리와 모니터링 (Versioning & Monitoring)

import mlflow

mlflow.set_registry_uri("databricks-uc")

with mlflow.start_run(run_name="customer-support-agent-v4"):
    model_info = mlflow.langchain.log_model(
        lc_model=agent_chain,
        artifact_path="agent",
        registered_model_name="catalog.schema.customer_support_agent",
        input_example={"messages": [{"role": "user", "content": "환불 방법을 알려주세요."}]},
    )
    print(f"등록된 버전: {model_info.registered_model_version}")

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# champion: 현재 프로덕션 버전
# challenger: 검증 중인 신규 버전
w.model_registry.set_registered_model_alias(
    full_name="catalog.schema.customer_support_agent",
    alias="champion",
    version_num=3
)
w.model_registry.set_registered_model_alias(
    full_name="catalog.schema.customer_support_agent",
    alias="challenger",
    version_num=4
)

def auto_promote_if_ready(challenger_metrics: dict, champion_metrics: dict) -> bool:
    """Challenger가 Champion 기준을 초과하면 자동 승격"""
    conditions = [
        challenger_metrics["accuracy"] >= champion_metrics["accuracy"] * 0.95,
        challenger_metrics["p95_latency_ms"] <= 3000,
        challenger_metrics["error_rate"] <= 0.01,
        challenger_metrics["positive_feedback_rate"] >= 0.70,
    ]
    return all(conditions)

from databricks.sdk.service.serving import (
    ServedEntityInput, TrafficConfig, Route, EndpointCoreConfigInput
)

# Canary: v4에 10% 트래픽 할당
w.serving_endpoints.update_config(
    name="customer-support-agent",
    served_entities=[
        ServedEntityInput(
            entity_name="catalog.schema.customer_support_agent",
            entity_version="3",
            name="v3-stable",
            workload_size="Small",
            scale_to_zero_enabled=False
        ),
        ServedEntityInput(
            entity_name="catalog.schema.customer_support_agent",
            entity_version="4",
            name="v4-canary",
            workload_size="Small",
            scale_to_zero_enabled=False
        ),
    ],
    traffic_config=TrafficConfig(
        routes=[
            Route(served_model_name="v3-stable", traffic_percentage=90),
            Route(served_model_name="v4-canary",  traffic_percentage=10),
        ]
    )
)

# Green 환경 검증 완료 후 100% 전환
w.serving_endpoints.update_config(
    name="customer-support-agent",
    served_entities=[
        ServedEntityInput(
            entity_name="catalog.schema.customer_support_agent",
            entity_version="4",  # Green → 전체 트래픽
            workload_size="Small",
            scale_to_zero_enabled=True
        )
    ]
)

# A/B 테스트: 50/50 분할
traffic_config = TrafficConfig(
    routes=[
        Route(served_model_name="variant-a", traffic_percentage=50),
        Route(served_model_name="variant-b", traffic_percentage=50),
    ]
)

-- LLM Judge를 활용한 응답 품질 점수 집계
SELECT
    DATE(timestamp)                                             AS date,
    AVG(quality_score)                                         AS avg_quality_score,
    AVG(relevance_score)                                       AS avg_relevance_score,
    AVG(groundedness_score)                                    AS avg_groundedness_score,
    COUNT(CASE WHEN quality_score < 3 THEN 1 END)              AS low_quality_count
FROM catalog.schema.agent_quality_assessments
GROUP BY DATE(timestamp)
ORDER BY date DESC;

import mlflow

# LangChain 자동 계측
mlflow.langchain.autolog()

# 또는 LlamaIndex 자동 계측
mlflow.llama_index.autolog()

import mlflow

@mlflow.trace(name="customer-support-agent", span_type="CHAIN")
def run_agent(user_query: str) -> str:

    # 검색 단계 추적
    with mlflow.start_span(name="retrieve_context", span_type="RETRIEVER") as span:
        context_docs = vector_store.similarity_search(user_query, k=5)
        span.set_attribute("num_retrieved_docs", len(context_docs))
        span.set_attribute("query", user_query)

    # LLM 호출 단계 추적
    with mlflow.start_span(name="generate_response", span_type="LLM") as span:
        response = llm.invoke(
            messages=[{"role": "user", "content": user_query}],
            context=context_docs
        )
        span.set_attribute("input_tokens",  response.usage.prompt_tokens)
        span.set_attribute("output_tokens", response.usage.completion_tokens)
        span.set_attribute("total_cost_usd", calculate_cost(response.usage))

    return response.content

import mlflow

# 최근 실패한 Trace 조회
failed_traces = mlflow.search_traces(
    filter_string="status = 'ERROR'",
    max_results=50,
    order_by=["timestamp_ms DESC"]
)

# 특정 Request ID로 Trace 조회 (Inference Table과 연계)
trace = mlflow.get_trace(request_id="tr-abc123")
print(trace.to_json(pretty=True))

from databricks.sdk.service.serving import AutoCaptureConfigInput

w.serving_endpoints.create(
    name="customer-support-agent",
    config=EndpointCoreConfigInput(
        served_entities=[...],
        auto_capture_config=AutoCaptureConfigInput(
            catalog_name="catalog",
            schema_name="schema",
            table_name_prefix="agent_logs",
            enabled=True
        )
    )
)

-- 최근 1시간 주요 지표 요약
SELECT
    COUNT(*)                                                            AS total_requests,
    ROUND(AVG(execution_time_ms), 0)                                   AS avg_latency_ms,
    PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY execution_time_ms)    AS p95_latency_ms,
    ROUND(SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END)
          * 100.0 / COUNT(*), 2)                                       AS error_rate_pct
FROM catalog.schema.agent_logs_payload
WHERE timestamp >= CURRENT_TIMESTAMP() - INTERVAL 1 HOUR;

-- 토큰 사용량 및 비용 추정 (일별)
SELECT
    DATE(timestamp)                                                     AS date,
    SUM(response:usage:prompt_tokens)                                  AS total_input_tokens,
    SUM(response:usage:completion_tokens)                              AS total_output_tokens,
    -- Claude 3.5 Sonnet 기준 예시 단가 적용
    ROUND(SUM(response:usage:prompt_tokens)    / 1e6 * 3.0 +
          SUM(response:usage:completion_tokens) / 1e6 * 15.0, 2)      AS estimated_cost_usd
FROM catalog.schema.agent_logs_payload
WHERE status_code = 200
GROUP BY DATE(timestamp)
ORDER BY date DESC;

-- 드리프트 감지: 응답 길이 분포 변화
SELECT
    DATE_TRUNC('hour', timestamp)                                       AS hour,
    AVG(LENGTH(response:choices[0]:message:content::STRING))           AS avg_response_length,
    STDDEV(LENGTH(response:choices[0]:message:content::STRING))        AS stddev_response_length
FROM catalog.schema.agent_logs_payload
GROUP BY DATE_TRUNC('hour', timestamp)
ORDER BY hour DESC;

# 부정 피드백이 달린 요청을 재훈련 후보 데이터셋으로 큐레이션
negative_samples = spark.sql("""
    SELECT
        logs.request_id,
        logs.request:messages                        AS messages,
        logs.response:choices[0]:message:content     AS response,
        feedback.rating,
        feedback.comment
    FROM catalog.schema.agent_logs_payload logs
    JOIN catalog.schema.user_feedback feedback
      ON logs.request_id = feedback.request_id
    WHERE feedback.rating <= 2
      AND logs.timestamp >= CURRENT_TIMESTAMP() - INTERVAL 7 DAY
""")
negative_samples.write.mode("append").saveAsTable("catalog.schema.retraining_candidates")

from databricks.sdk.service.catalog import MonitorInferenceLog, MonitorCronSchedule

w.quality_monitors.create(
    table_name="catalog.schema.agent_logs_payload",
    inference_log=MonitorInferenceLog(
        granularities=["1 hour", "1 day"],
        timestamp_col="timestamp",
        model_id_col="served_entity_name",
        prediction_col="response",
        problem_type="question_answering"
    ),
    schedule=MonitorCronSchedule(
        quartz_cron_expression="0 0 * * * ?",  # 매시간
        timezone_id="Asia/Seoul"
    ),
    notifications={
        "on_failure": {"email_addresses": ["ml-ops@company.com"]},
        "on_new_classification_tag_detected": {"email_addresses": ["ml-ops@company.com"]}
    },
    output_schema_name="catalog.schema"
)

def check_and_rollback(endpoint_name: str, stable_version: str, threshold: dict):
    """메트릭 임계값 초과 시 자동 롤백"""

    # 최근 15분 메트릭 조회
    metrics = spark.sql("""
        SELECT
            COUNT(*) AS total,
            SUM(CASE WHEN status_code != 200 THEN 1 ELSE 0 END)
                * 100.0 / COUNT(*) AS error_rate,
            PERCENTILE_CONT(0.95) WITHIN GROUP (ORDER BY execution_time_ms)
                AS p95_latency
        FROM catalog.schema.agent_logs_payload
        WHERE timestamp >= CURRENT_TIMESTAMP() - INTERVAL 15 MINUTE
    """).first()

    needs_rollback = (
        metrics["error_rate"] > threshold["max_error_rate"] or
        metrics["p95_latency"] > threshold["max_p95_latency_ms"]
    )

    if needs_rollback:
        print(f"임계값 초과 — 롤백 실행: 현재 버전 → v{stable_version}")
        w.serving_endpoints.update_config(
            name=endpoint_name,
            served_entities=[
                ServedEntityInput(
                    entity_name="catalog.schema.customer_support_agent",
                    entity_version=stable_version,
                    workload_size="Small",
                    scale_to_zero_enabled=True
                )
            ]
        )
        notify_team(
            f"[자동 롤백] {endpoint_name} → v{stable_version} "
            f"(error_rate={metrics['error_rate']:.1f}%)"
        )

# 임계값 설정 및 실행
THRESHOLDS = {"max_error_rate": 10.0, "max_p95_latency_ms": 8000}
check_and_rollback("customer-support-agent", stable_version="3", threshold=THRESHOLDS)