추론 추적과 자동 대응

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)