import inspect import os import sys from collections import Counter from enum import Enum from typing import TYPE_CHECKING, Any from urllib.parse import urlparse from mlflow.entities import Feedback from mlflow.entities.issue import IssueSeverity, IssueStatus from mlflow.environment_variables import MLFLOW_ENABLE_OTEL_GENAI_SEMCONV from mlflow.telemetry.constant import ( GENAI_MODULES, MODULES_TO_CHECK_IMPORT, ) if TYPE_CHECKING: from mlflow.genai.scorers.base import Scorer GENAI_EVALUATION_PATH = "mlflow/genai/evaluation/base" GENAI_SCORERS_PATH = "mlflow/genai/scorers/base" GENAI_EVALUATE_FUNCTION = "_run_harness" SCORER_RUN_FUNCTION = "run" def _get_scorer_class_name_for_tracking(scorer: "Scorer") -> str: from mlflow.genai.scorers.builtin_scorers import BuiltInScorer if isinstance(scorer, BuiltInScorer): return type(scorer).__name__ try: from mlflow.genai.scorers.deepeval import DeepEvalScorer if isinstance(scorer, DeepEvalScorer): return f"DeepEval:{scorer.name}" except ImportError: pass try: from mlflow.genai.scorers.ragas import RagasScorer if isinstance(scorer, RagasScorer): return f"Ragas:{scorer.name}" except ImportError: pass return "UserDefinedScorer" class Event: name: str @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: """ Parse the arguments and return the params. """ return None class CreateExperimentEvent(Event): name: str = "create_experiment" @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: # create_experiment API returns the experiment id return {"experiment_id": result} class CreatePromptEvent(Event): name: str = "create_prompt" class LoadPromptEvent(Event): name: str = "load_prompt" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: name_or_uri = arguments.get("name_or_uri", "") # Check if alias is used (format: "prompts:/name@alias") uses_alias = "@" in name_or_uri return {"uses_alias": uses_alias} class StartTraceEvent(Event): name: str = "start_trace" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: # Capture the set of currently imported packages at trace start time to # understand the flavor of the trace. return { "imports": [pkg for pkg in GENAI_MODULES if pkg in sys.modules], "format": "genai_semconv" if MLFLOW_ENABLE_OTEL_GENAI_SEMCONV.get() else "native", } class LogAssessmentEvent(Event): name: str = "log_assessment" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.entities.assessment import Expectation, Feedback assessment = arguments.get("assessment") if assessment is None: return None if isinstance(assessment, Expectation): return {"type": "expectation", "source_type": assessment.source.source_type} elif isinstance(assessment, Feedback): return {"type": "feedback", "source_type": assessment.source.source_type} class EvaluateEvent(Event): name: str = "evaluate" class GenAIEvaluateEvent(Event): name: str = "genai_evaluate" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.genai.scorers.base import Scorer record_params = {} # Track if predict_fn is provided record_params["predict_fn_provided"] = arguments.get("predict_fn") is not None # Track eval data type eval_data = arguments.get("data") if eval_data is not None: from mlflow.genai.evaluation.utils import _get_eval_data_type record_params["eval_data_type"] = _get_eval_data_type(eval_data) # Track scorer information scorers = arguments.get("scorers") or [] scorer_info = [ { "class": _get_scorer_class_name_for_tracking(scorer), "kind": scorer.kind.value, "scope": "session" if scorer.is_session_level_scorer else "trace", } for scorer in scorers if isinstance(scorer, Scorer) ] record_params["scorer_info"] = scorer_info return record_params @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: _, telemetry_data = result if not isinstance(telemetry_data, dict): return None return telemetry_data class CreateLoggedModelEvent(Event): name: str = "create_logged_model" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: data: dict[str, Any] = {} if flavor := arguments.get("flavor"): data["flavor"] = flavor.removeprefix("mlflow.") if serialization_format := arguments.get("serialization_format"): data["serialization_format"] = serialization_format if arguments.get("uses_uv"): data["uses_uv"] = True return data or None class GetLoggedModelEvent(Event): name: str = "get_logged_model" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "imports": [pkg for pkg in MODULES_TO_CHECK_IMPORT if pkg in sys.modules], } class CreateRegisteredModelEvent(Event): name: str = "create_registered_model" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: tags = arguments.get("tags") or {} return {"is_prompt": _is_prompt(tags)} class CreateRunEvent(Event): name: str = "create_run" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: # Capture the set of currently imported packages at run creation time to # understand how MLflow is used together with other libraries. Collecting # this data at run creation ensures accuracy and completeness. return { "imports": [pkg for pkg in MODULES_TO_CHECK_IMPORT if pkg in sys.modules], "experiment_id": arguments.get("experiment_id"), } class CreateModelVersionEvent(Event): name: str = "create_model_version" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: tags = arguments.get("tags") or {} return {"is_prompt": _is_prompt(tags)} class CreateDatasetEvent(Event): name: str = "create_dataset" class MergeRecordsEvent(Event): name: str = "merge_records" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.entities.evaluation_dataset import ( DatasetGranularity, EvaluationDataset, ) if arguments is None: return None records = arguments.get("records") if records is None: return None try: count = len(records) except TypeError: return None if count == 0: return None input_type = type(records).__name__.lower() input_keys: set[str] | None = None if "dataframe" in input_type: input_type = "pandas" try: if "inputs" in records.columns: if first_inputs := records.iloc[0].get("inputs", {}): input_keys = set(first_inputs.keys()) except Exception: pass elif isinstance(records, list): first_elem = records[0] if hasattr(first_elem, "__class__") and first_elem.__class__.__name__ == "Trace": input_type = "list[trace]" elif isinstance(first_elem, dict): input_type = "list[dict]" if first_inputs := first_elem.get("inputs", {}): input_keys = set(first_inputs.keys()) else: input_type = "list" else: input_type = "other" if input_type == "list[trace]": dataset_type = DatasetGranularity.TRACE elif input_keys: dataset_type = EvaluationDataset._classify_input_fields(input_keys) else: dataset_type = DatasetGranularity.UNKNOWN return { "record_count": count, "input_type": input_type, "dataset_type": dataset_type.value, } class DatasetToDataFrameEvent(Event): name: str = "dataset_to_df" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.entities.evaluation_dataset import EvaluationDataset dataset_instance = arguments.get("self") if not isinstance(dataset_instance, EvaluationDataset): return None callsite = "direct_call" frame = sys._getframe() for _ in range(10): if frame is None: break frame_filename = frame.f_code.co_filename.replace("\\", "/") if "mlflow/genai/evaluation" in frame_filename: callsite = "genai_evaluate" break if "mlflow/genai/simulators" in frame_filename: callsite = "conversation_simulator" break frame = frame.f_back granularity = dataset_instance._get_existing_granularity() return {"dataset_type": granularity.value, "callsite": callsite} @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: if result is None: return {"record_count": 0} return {"record_count": len(result)} def _is_prompt(tags: dict[str, str]) -> bool: try: from mlflow.prompt.constants import IS_PROMPT_TAG_KEY except ImportError: return False return tags.get(IS_PROMPT_TAG_KEY, "false").lower() == "true" class CreateWebhookEvent(Event): name: str = "create_webhook" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: events = arguments.get("events") or [] return {"events": [str(event) for event in events]} class PromptOptimizationEvent(Event): name: str = "prompt_optimization" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: result = {} # Track the optimizer type used if optimizer := arguments.get("optimizer"): result["optimizer_type"] = type(optimizer).__name__ else: result["optimizer_type"] = None # Track the number of prompts being optimized prompt_uris = arguments.get("prompt_uris") or [] try: result["prompt_count"] = len(prompt_uris) except TypeError: result["prompt_count"] = None # Track if custom scorers are provided and how many scorers = arguments.get("scorers") try: result["scorer_count"] = len(scorers) except TypeError: result["scorer_count"] = None # Track if custom aggregation is provided result["custom_aggregation"] = arguments.get("aggregation") is not None return result class LogDatasetEvent(Event): name: str = "log_dataset" class LogMetricEvent(Event): name: str = "log_metric" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return {"synchronous": arguments.get("synchronous")} class LogParamEvent(Event): name: str = "log_param" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return {"synchronous": arguments.get("synchronous")} class LogBatchEvent(Event): name: str = "log_batch" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "metrics": bool(arguments.get("metrics")), "params": bool(arguments.get("params")), "tags": bool(arguments.get("tags")), "synchronous": arguments.get("synchronous"), } class McpRunEvent(Event): name: str = "mcp_run" class TrackingServerStartEvent(Event): name: str = "tracking_server_start" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: backend_store_uri = arguments.get("backend_store_uri") or "" scheme = urlparse(backend_store_uri).scheme # Treat empty schemes (relative paths) and single-letter schemes # (Windows drive letters like C:\) as local file storage. # Strip SQLAlchemy driver suffixes (e.g. mysql+pymysql → mysql). backend_store_type = "file" if not scheme or len(scheme) == 1 else scheme.split("+")[0] app_name = arguments.get("app_name") return { "auth_enabled": app_name == "basic-auth", "app_name": app_name, "backend_store_type": backend_store_type, "serve_artifacts": bool(arguments.get("serve_artifacts")), "artifacts_only": bool(arguments.get("artifacts_only")), "expose_prometheus": arguments.get("expose_prometheus") is not None, "enable_workspaces": bool(arguments.get("enable_workspaces")), "workers": arguments.get("workers"), "dev": bool(arguments.get("dev")), } class GatewayStartEvent(Event): name: str = "gateway_start" # Gateway Resource CRUD Events class GatewayCreateEndpointEvent(Event): name: str = "gateway_create_endpoint" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "has_fallback_config": arguments.get("fallback_config") is not None, "routing_strategy": str(arguments.get("routing_strategy")) if arguments.get("routing_strategy") else None, "num_model_configs": len(arguments.get("model_configs") or []), "usage_tracking": arguments.get("usage_tracking"), } class GatewayUpdateEndpointEvent(Event): name: str = "gateway_update_endpoint" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "has_fallback_config": arguments.get("fallback_config") is not None, "routing_strategy": str(arguments.get("routing_strategy")) if arguments.get("routing_strategy") else None, "num_model_configs": len(arguments.get("model_configs")) if arguments.get("model_configs") is not None else None, "usage_tracking": arguments.get("usage_tracking"), } class GatewayDeleteEndpointEvent(Event): name: str = "gateway_delete_endpoint" class GatewayGetEndpointEvent(Event): name: str = "gateway_get_endpoint" class GatewayListEndpointsEvent(Event): name: str = "gateway_list_endpoints" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "filter_by_provider": arguments.get("provider") is not None, } class GatewayCreateModelDefinitionEvent(Event): name: str = "gateway_create_model_definition" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "model_name": arguments.get("model_name"), "provider": arguments.get("provider"), } # Gateway Budget Policy CRUD Events class GatewayCreateBudgetPolicyEvent(Event): name: str = "gateway_create_budget_policy" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: def _enum_str(val: Any) -> str | None: if val is None: return None return val.value if hasattr(val, "value") else str(val) duration = arguments.get("duration") return { "budget_unit": _enum_str(arguments.get("budget_unit")), "duration_unit": _enum_str(duration.unit if duration is not None else None), "target_scope": _enum_str(arguments.get("target_scope")), "budget_action": _enum_str(arguments.get("budget_action")), } class GatewayUpdateBudgetPolicyEvent(Event): name: str = "gateway_update_budget_policy" class GatewayDeleteBudgetPolicyEvent(Event): name: str = "gateway_delete_budget_policy" class GatewayListBudgetPoliciesEvent(Event): name: str = "gateway_list_budget_policies" # Gateway Guardrail CRUD Events class GatewayCreateGuardrailEvent(Event): name: str = "gateway_create_guardrail" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "stage": str(arguments.get("stage")) if arguments.get("stage") else None, "action": str(arguments.get("action")) if arguments.get("action") else None, } class GatewayUpdateGuardrailEvent(Event): name: str = "gateway_update_guardrail" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "stage": str(arguments.get("stage")) if arguments.get("stage") else None, "action": str(arguments.get("action")) if arguments.get("action") else None, } class GatewayDeleteGuardrailEvent(Event): name: str = "gateway_delete_guardrail" # Gateway Secret CRUD Events class GatewayCreateSecretEvent(Event): name: str = "gateway_create_secret" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "provider": arguments.get("provider"), } class GatewayUpdateSecretEvent(Event): name: str = "gateway_update_secret" class GatewayDeleteSecretEvent(Event): name: str = "gateway_delete_secret" class GatewayListSecretsEvent(Event): name: str = "gateway_list_secrets" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "filter_by_provider": arguments.get("provider") is not None, } # Gateway Invocation Events class GatewayInvocationType(str, Enum): """Type of gateway invocation endpoint.""" MLFLOW_INVOCATIONS = "mlflow_invocations" MLFLOW_CHAT_COMPLETIONS = "mlflow_chat_completions" OPENAI_PASSTHROUGH_CHAT = "openai_passthrough_chat" OPENAI_PASSTHROUGH_EMBEDDINGS = "openai_passthrough_embeddings" OPENAI_PASSTHROUGH_RESPONSES = "openai_passthrough_responses" ANTHROPIC_PASSTHROUGH_MESSAGES = "anthropic_passthrough_messages" GEMINI_PASSTHROUGH_GENERATE_CONTENT = "gemini_passthrough_generate_content" GEMINI_PASSTHROUGH_STREAM_GENERATE_CONTENT = "gemini_passthrough_stream_generate_content" RAW_PROXY = "raw_proxy" class GatewayInvocationEvent(Event): name: str = "gateway_invocation" class AiCommandRunEvent(Event): name: str = "ai_command_run" class AgentSetupEvent(Event): name: str = "agent_setup" class MlflowTestEvent(Event): name: str = "mlflow_test" class TracingContextPropagation(Event): name: str = "tracing_context_propagation" class GitModelVersioningEvent(Event): name: str = "git_model_versioning" class InvokeCustomJudgeModelEvent(Event): name: str = "invoke_custom_judge_model" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.metrics.genai.model_utils import _parse_model_uri model_uri = arguments.get("model_uri") if not model_uri: return {"model_provider": None} model_provider, _ = _parse_model_uri(model_uri) return {"model_provider": model_provider} class MakeJudgeEvent(Event): name: str = "make_judge" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: model = arguments.get("model") if model and isinstance(model, str): model_provider = model.split(":")[0] if ":" in model else None return {"model_provider": model_provider} return {"model_provider": None} class AlignJudgeEvent(Event): name: str = "align_judge" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: result = {} if (traces := arguments.get("traces")) is not None: try: result["trace_count"] = len(traces) except TypeError: result["trace_count"] = None if optimizer := arguments.get("optimizer"): result["optimizer_type"] = type(optimizer).__name__ else: result["optimizer_type"] = "default" return result class AutologgingEvent(Event): name: str = "autologging" class TraceAttachmentsEvent(Event): name: str = "trace_attachments" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: if attachments := arguments.get("attachments"): content_types = Counter(att.content_type for att in attachments.values()) return {"content_types": dict(content_types)} return None class TraceSource(str, Enum): """Source of a trace received by the MLflow server.""" MLFLOW_PYTHON_CLIENT = "MLFLOW_PYTHON_CLIENT" EXTERNAL_OTEL_CLIENT = "EXTERNAL_OTEL_CLIENT" UNKNOWN = "UNKNOWN" class TracesReceivedByServerEvent(Event): name: str = "traces_received_by_server" class SimulateConversationEvent(Event): name: str = "simulate_conversation" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: callsite = "conversation_simulator" for frame_info in inspect.stack()[:10]: frame_filename = frame_info.filename frame_function = frame_info.function if ( GENAI_EVALUATION_PATH in frame_filename.replace("\\", "/") and frame_function == GENAI_EVALUATE_FUNCTION ): callsite = "genai_evaluate" break return {"callsite": callsite} @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: return { "simulated_conversation_info": [ {"turn_count": len(conversation)} for conversation in result ] } class OptimizePromptsJobEvent(Event): name: str = "optimize_prompts_job" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: result = {} if optimizer_type := arguments.get("optimizer_type"): result["optimizer_type"] = optimizer_type if "scorer_names" in arguments: scorer_names = arguments["scorer_names"] # `scorer_count` is useful for indicating zero-shot vs few-shot optimization, and to # track the pattern of how users use prompt optimization. result["scorer_count"] = len(scorer_names) return result or None class ScorerCallEvent(Event): name: str = "scorer_call" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: from mlflow.genai.scorers.base import Scorer scorer_instance = arguments.get("self") if not isinstance(scorer_instance, Scorer): return None # Check if running inside an online scoring job # Import here to avoid circular imports from mlflow.genai.scorers.job import ( ONLINE_SESSION_SCORER_JOB_NAME, ONLINE_TRACE_SCORER_JOB_NAME, ) from mlflow.server.jobs.utils import MLFLOW_SERVER_JOB_NAME_ENV_VAR job_name = os.environ.get(MLFLOW_SERVER_JOB_NAME_ENV_VAR) if job_name in (ONLINE_TRACE_SCORER_JOB_NAME, ONLINE_SESSION_SCORER_JOB_NAME): callsite = "online_scoring" else: callsite = "direct_scorer_call" for frame_info in inspect.stack()[:10]: frame_filename = frame_info.filename frame_function = frame_info.function if ( GENAI_SCORERS_PATH in frame_filename.replace("\\", "/") and frame_function == SCORER_RUN_FUNCTION ): callsite = "genai_evaluate" break return { "scorer_class": _get_scorer_class_name_for_tracking(scorer_instance), "scorer_kind": scorer_instance.kind.value, "scope": "session" if scorer_instance.is_session_level_scorer else "trace", "callsite": callsite, } @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: if isinstance(result, Feedback): return {"has_feedback_error": result.error is not None} if isinstance(result, list) and result and all(isinstance(f, Feedback) for f in result): return {"has_feedback_error": any(f.error is not None for f in result)} return {"has_feedback_error": False} class DiscoverIssuesEvent(Event): name: str = "discover_issues" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: return { "model": arguments.get("model"), "trace_count": len(arguments.get("traces") or []), "categories": arguments.get("categories"), "source_run_id": arguments.get("run_id"), } @classmethod def parse_result(cls, result: Any) -> dict[str, Any] | None: return { "issue_count": len(result.issues), "total_traces_analyzed": result.total_traces_analyzed, "total_cost_usd": result.total_cost_usd, "triage_run_id": result.triage_run_id, } class UpdateIssueEvent(Event): name: str = "update_issue" @classmethod def parse(cls, arguments: dict[str, Any]) -> dict[str, Any] | None: status = arguments.get("status") if isinstance(status, IssueStatus): status = status.value severity = arguments.get("severity") if isinstance(severity, IssueSeverity): severity = severity.value return { "status": status, "has_name": arguments.get("name") is not None, "has_description": arguments.get("description") is not None, "severity": severity, } @classmethod def parse_result(cls, result: Any) -> dict[str, Any]: return {"source_run_id": result.source_run_id} if result else {}