버전 관리와 배포 전략

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))