Exemples de scorers basés sur le code
Dans MLflow Evaluation pour GenAI, les marqueurs personnalisés basés sur le code vous permettent de définir des métriques d'évaluation flexibles pour votre agent ou application d'IA. Cet ensemble d'exemples et le Notebook d'exemple complémentaire illustrent de nombreux modèles d'utilisation de marqueurs basés sur du code avec différentes options pour les entrées, les sorties, l'implémentation et la gestion des erreurs.
L'image ci-dessous illustre les sorties de certains évaluateurs personnalisés sous forme de métriques dans l'interface utilisateur de MLflow.
Prérequis
- Mettre à jour MLflow
- Définir votre application GenAI
- Générer des traces utilisées dans certains exemples d'évaluateur.
Mettre à jour mlflow
Mettez à jour mlflow[databricks] vers la dernière version pour la meilleure expérience GenAI, et installez openai car l'exemple d'application ci-dessous utilise le client OpenAI.
%pip install -q --upgrade "mlflow[databricks]>=3.1" "openai>=1.0.0"
dbutils.library.restartPython()
Définissez votre application GenAI
Certains des exemples ci-dessous utiliseront l’application GenAI suivante, qui est un assistant général de questions-réponses. Le code ci-dessous utilise un client OpenAI pour se connecter aux LLM hébergés par Databricks.
from databricks_openai import DatabricksOpenAI
import mlflow
# Create an OpenAI client that is connected to Databricks-hosted LLMs
client = DatabricksOpenAI()
# Select an LLM
model_name = "databricks-claude-sonnet-4"
mlflow.openai.autolog()
# If running outside of Databricks, set up MLflow tracking to Databricks.
# mlflow.set_tracking_uri("databricks")
# In Databricks notebooks, the experiment defaults to the notebook experiment.
# mlflow.set_experiment("/Shared/docs-demo")
@mlflow.trace
def sample_app(messages: list[dict[str, str]]):
# 1. Prepare messages for the LLM
messages_for_llm = [
{"role": "system", "content": "You are a helpful assistant."},
*messages,
]
# 2. Call LLM to generate a response
response = client.chat.completions.create(
model= model_name,
messages=messages_for_llm,
)
return response.choices[0].message.content
sample_app([{"role": "user", "content": "What is the capital of France?"}])
Générer des traces
Le eval_dataset ci-dessous est utilisé par mlflow.genai.evaluate() pour générer des traces, en utilisant un évaluateur de remplacement.
from mlflow.genai.scorers import scorer
eval_dataset = [
{
"inputs": {
"messages": [
{"role": "user", "content": "How much does a microwave cost?"},
]
},
},
{
"inputs": {
"messages": [
{
"role": "user",
"content": "Can I return the microwave I bought 2 months ago?",
},
]
},
},
{
"inputs": {
"messages": [
{
"role": "user",
"content": "I'm having trouble with my account. I can't log in.",
},
{
"role": "assistant",
"content": "I'm sorry to hear that you're having trouble with your account. Are you using our website or mobile app?",
},
{"role": "user", "content": "Website"},
]
},
},
]
@scorer
def placeholder_metric() -> int:
# placeholder return value
return 1
eval_results = mlflow.genai.evaluate(
data=eval_dataset,
predict_fn=sample_app,
scorers=[placeholder_metric]
)
generated_traces = mlflow.search_traces(run_id=eval_results.run_id)
generated_traces
La fonction mlflow.search_traces() ci-dessus renvoie un DataFrame Pandas de traces, à utiliser dans certains exemples ci-dessous.
Exemple 1 : Accéder aux données de la Trace
Accédez à l'objet MLflow Trace complet pour utiliser divers détails (étendues, entrées, sorties, attributs, chronométrage) pour le calcul précis des métriques.
Cet évaluateur vérifie si la durée d'exécution totale de la trace se situe dans une plage acceptable.
import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Trace, Feedback, SpanType
@scorer
def llm_response_time_good(trace: Trace) -> Feedback:
# Search particular span type from the trace
llm_span = trace.search_spans(span_type=SpanType.CHAT_MODEL)[0]
response_time = (llm_span.end_time_ns - llm_span.start_time_ns) / 1e9 # convert to seconds
max_duration = 5.0
if response_time <= max_duration:
return Feedback(
value="yes",
rationale=f"LLM response time {response_time:.2f}s is within the {max_duration}s limit."
)
else:
return Feedback(
value="no",
rationale=f"LLM response time {response_time:.2f}s exceeds the {max_duration}s limit."
)
# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
span_check_eval_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[llm_response_time_good]
)
Exemple 2 : Encapsuler un juge LLM prédéfini
Créez un évaluateur personnalisé qui enveloppe les juges LLM intégrés de MLflow. Utilisez ceci pour prétraiter les données de trace pour le juge ou pour post-traiter ses commentaires.
Cet exemple montre comment envelopper le juge is_context_relevant pour évaluer si la réponse de l'assistant est pertinente par rapport à la query de l'utilisateur. Plus précisément, le champ inputs pour sample_app est un dictionnaire de la forme : {"messages": [{"role": ..., "content": ...}, ...]}. Ce scoreur extrait le contenu du dernier message de l'utilisateur pour le transmettre au juge de pertinence.
import mlflow
from mlflow.entities import Trace, Feedback
from mlflow.genai.judges import is_context_relevant
from mlflow.genai.scorers import scorer
from typing import Any
@scorer
def is_message_relevant(inputs: dict[str, Any], outputs: str) -> Feedback:
last_user_message_content = None
if "messages" in inputs and isinstance(inputs["messages"], list):
for message in reversed(inputs["messages"]):
if message.get("role") == "user" and "content" in message:
last_user_message_content = message["content"]
break
if not last_user_message_content:
raise Exception("Could not extract the last user message from inputs to evaluate relevance.")
# Call the `relevance_to_query judge. It will return a Feedback object.
return is_context_relevant(
request=last_user_message_content,
context={"response": outputs},
)
# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
custom_relevance_eval_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[is_message_relevant]
)
Exemple 3 : Utiliser expectations
Les attentes sont des valeurs ou des étiquettes de vérité terrain et sont souvent importantes pour l'évaluation hors ligne. Lors de l'exécution de mlflow.genai.evaluate(), vous pouvez spécifier des attentes dans l'argument data de deux manières :
expectationscolonne ou champ : Par exemple, si l'argumentdataest une liste de dictionnaires ou un DataFrame Pandas, chaque ligne peut contenir une cléexpectations. La valeur associée à cette clé est transmise directement à votre évaluateur personnalisé.tracecolonne ou champ : Par exemple, si l'argumentdataest le dataframe renvoyé parmlflow.search_traces(), il inclura un champtracequi contient toutes les donnéesExpectationassociées aux traces.
Le monitoring de production n'a généralement pas d'attentes puisque vous évaluez le trafic en direct sans vérité terrain. Si vous avez l'intention d'utiliser le même évaluateur pour l'évaluation hors ligne et en ligne, concevez-le pour gérer les attentes avec souplesse.
Cet exemple montre également comment utiliser un évaluateur personnalisé avec l'évaluateur Safety prédéfini.
import mlflow
from mlflow.entities import Feedback
from mlflow.genai.scorers import scorer, Safety
from typing import Any, List, Optional, Union
expectations_eval_dataset_list = [
{
"inputs": {"messages": [{"role": "user", "content": "What is 2+2?"}]},
"expectations": {
"expected_response": "2+2 equals 4.",
"expected_keywords": ["4", "four", "equals"],
}
},
{
"inputs": {"messages": [{"role": "user", "content": "Describe MLflow in one sentence."}]},
"expectations": {
"expected_response": "MLflow is an open-source platform to streamline machine learning development, including tracking experiments, packaging code into reproducible runs, and sharing and deploying models.",
"expected_keywords": ["mlflow", "open-source", "platform", "machine learning"],
}
},
{
"inputs": {"messages": [{"role": "user", "content": "Say hello."}]},
"expectations": {
"expected_response": "Hello there!",
# No keywords needed for this one, but the field can be omitted or empty
}
}
]
Exemple 3.1 : Correspondance exacte avec la réponse attendue.
Ce scoreur vérifie si la réponse de l'assistant correspond exactement à expected_response fourni dans expectations.
@scorer
def exact_match(outputs: str, expectations: dict[str, Any]) -> bool:
# Scorer can return primitive value like bool, int, float, str, etc.
return outputs == expectations["expected_response"]
exact_match_eval_results = mlflow.genai.evaluate(
data=expectations_eval_dataset_list,
predict_fn=sample_app, # sample_app is from the prerequisite section
scorers=[exact_match, Safety()] # You can include any number of scorers
)
Exemple 3.2 : vérification des mots-clés par rapport aux attentes
Ce scoreur vérifie si tous les expected_keywords de expectations sont présents dans la réponse de l'assistant.
@scorer
def keyword_presence_scorer(outputs: str, expectations: dict[str, Any]) -> Feedback:
expected_keywords = expectations.get("expected_keywords")
print(expected_keywords)
if expected_keywords is None:
return Feedback(value="yes", rationale="No keywords were expected in the response.")
missing_keywords = []
for keyword in expected_keywords:
if keyword.lower() not in outputs.lower():
missing_keywords.append(keyword)
if not missing_keywords:
return Feedback(value="yes", rationale="All expected keywords are present in the response.")
else:
return Feedback(value="no", rationale=f"Missing keywords: {', '.join(missing_keywords)}.")
keyword_presence_eval_results = mlflow.genai.evaluate(
data=expectations_eval_dataset_list,
predict_fn=sample_app, # sample_app is from the prerequisite section
scorers=[keyword_presence_scorer]
)
Exemple 4 : Retourner plusieurs objets de feedback
Un seul évaluateur peut renvoyer une liste d'objets Feedback, permettant à un seul évaluateur d'évaluer simultanément plusieurs facettes de qualité (telles que les informations personnelles identifiables (PII), le sentiment et la concision).
Chaque objet Feedback doit avoir un name unique, qui devient le nom de la métrique dans les résultats. Consultez les détails sur les noms de métriques.
Cet exemple démontre un évaluateur qui renvoie deux informations distinctes pour chaque trace :
is_not_empty_check: Un booléen indiquant si le contenu de la réponse n'est pas vide.response_char_length: une valeur numérique pour la longueur de caractères de la réponse.
import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback, Trace # Ensure Feedback and Trace are imported
from typing import Any, Optional
@scorer
def comprehensive_response_checker(outputs: str) -> list[Feedback]:
feedbacks = []
# 1. Check if the response is not empty
feedbacks.append(
Feedback(name="is_not_empty_check", value="yes" if outputs != "" else "no")
)
# 2. Calculate response character length
char_length = len(outputs)
feedbacks.append(Feedback(name="response_char_length", value=char_length))
return feedbacks
# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
multi_feedback_eval_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[comprehensive_response_checker]
)
Le résultat comportera deux colonnes : is_not_empty_check et response_char_length en tant qu'évaluations.
Exemple 5 : Utilisez votre propre LLM pour un juge
Intégrez un LLM personnalisé ou hébergé en externe dans un évaluateur. L'évaluateur gère les appels d'API, le formatage d'entrée/sortie et génère Feedback à partir de la réponse de votre LLM, ce qui vous donne un contrôle total sur le processus d'évaluation.
Vous pouvez également définir le champ source dans l'objet Feedback pour indiquer que la source de l'évaluation est un juge LLM.
import mlflow
import json
from mlflow.genai.scorers import scorer
from mlflow.entities import AssessmentSource, AssessmentSourceType, Feedback
from typing import Any, Optional
# Define the prompts for the Judge LLM.
judge_system_prompt = """
You are an impartial AI assistant responsible for evaluating the quality of a response generated by another AI model.
Your evaluation should be based on the original user query and the AI's response.
Provide a quality score as an integer from 1 to 5 (1=Poor, 2=Fair, 3=Good, 4=Very Good, 5=Excellent).
Also, provide a brief rationale for your score.
Your output MUST be a single valid JSON object with two keys: "score" (an integer) and "rationale" (a string).
Example:
{"score": 4, "rationale": "The response was mostly accurate and helpful, addressing the user's query directly."}
"""
judge_user_prompt = """
Please evaluate the AI's Response below based on the Original User Query.
Original User Query:
```{user_query}```
AI's Response:
```{llm_response_from_app}```
Provide your evaluation strictly as a JSON object with "score" and "rationale" keys.
"""
@scorer
def answer_quality(inputs: dict[str, Any], outputs: str) -> Feedback:
user_query = inputs["messages"][-1]["content"]
# Call the Judge LLM using the OpenAI SDK client.
judge_llm_response_obj = client.chat.completions.create(
model="databricks-claude-sonnet-4-5", # This example uses Databricks hosted Claude. If you provide your own OpenAI credentials, replace with a valid OpenAI model e.g., gpt-4o-mini, etc.
messages=[
{"role": "system", "content": judge_system_prompt},
{"role": "user", "content": judge_user_prompt.format(user_query=user_query, llm_response_from_app=outputs)},
],
max_tokens=200, # Max tokens for the judge's rationale
temperature=0.0, # For more deterministic judging
)
judge_llm_output_text = judge_llm_response_obj.choices[0].message.content
# Parse the Judge LLM's JSON output.
judge_eval_json = json.loads(judge_llm_output_text)
parsed_score = int(judge_eval_json["score"])
parsed_rationale = judge_eval_json["rationale"]
return Feedback(
value=parsed_score,
rationale=parsed_rationale,
# Set the source of the assessment to indicate the LLM judge used to generate the feedback
source=AssessmentSource(
source_type=AssessmentSourceType.LLM_JUDGE,
source_id="claude-sonnet-4-5",
)
)
# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
custom_llm_judge_eval_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[answer_quality]
)
En ouvrant la trace dans l’interface utilisateur et en cliquant sur l’évaluation « answer_quality », vous pouvez voir les métadonnées du juge, telles que la logique, le timestamp et le nom du modèle de juge. Si l'évaluation du juge n'est pas correcte, vous pouvez remplacer le score en cliquant sur le bouton Edit.
La nouvelle évaluation remplace l'évaluation initiale du juge. L'historique des modifications est conservé pour référence ultérieure.
Exemple 6 : définition du scoreur basé sur la classe (évaluation hors ligne uniquement)
Si un évaluateur nécessite un état, alors la définition basée sur le décorateur @scorer peut ne pas suffire. Au lieu de cela, utilisez la classe de base Scorer pour des évaluateurs plus complexes. La classe Scorer est un objet Pydantic, vous pouvez donc définir des champs supplémentaires et les utiliser dans la méthode __call__.
Les sous-classes Scorer basées sur des classes sont prises en charge uniquement pour l'évaluation hors ligne avec mlflow.genai.evaluate(). Elles ne peuvent pas être enregistrées pour le monitoring de production. Pour utiliser des évaluateurs personnalisés dans le monitoring de production, utilisez le décorateur@scorer.
from mlflow.genai.scorers import Scorer
from mlflow.entities import Feedback
from typing import Optional
# Scorer class is a Pydantic object
class ResponseQualityScorer(Scorer):
# The `name` field is mandatory
name: str = "response_quality"
# Define additional fields
min_length: int = 50
required_sections: Optional[list[str]] = None
# Override the __call__ method to implement the scorer logic
def __call__(self, outputs: str) -> Feedback:
issues = []
# Check length
if len(outputs.split()) < self.min_length:
issues.append(f"Too short (minimum {self.min_length} words)")
# Check required sections
missing = [s for s in self.required_sections if s not in outputs]
if missing:
issues.append(f"Missing sections: {', '.join(missing)}")
if issues:
return Feedback(
value=False,
rationale="; ".join(issues)
)
return Feedback(
value=True,
rationale="Response meets all quality criteria"
)
response_quality_scorer = ResponseQualityScorer(required_sections=["# Summary", "# Sources"])
# Evaluate the scorer using the pre-generated traces from the prerequisite code block.
class_based_scorer_results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[response_quality_scorer]
)
Exemple 7 : Gestion des erreurs dans les scorers
Pour une explication de la manière dont MLflow révèle les erreurs de scoreur et les deux approches de gestion des erreurs, consultez la gestion des erreurs. L'exemple ci-dessous combine les deux approches en un seul évaluateur.
import mlflow
from mlflow.genai.scorers import scorer
from mlflow.entities import Feedback, AssessmentError
@scorer
def resilient_scorer(outputs, trace=None):
try:
response = outputs.get("response")
if not response:
return Feedback(
value=None,
error=AssessmentError(
error_code="MISSING_RESPONSE",
error_message="No response field in outputs"
)
)
# Your evaluation logic
return Feedback(value=True, rationale="Valid response")
except Exception as e:
# Let MLflow handle the error gracefully
raise
# Evaluation continues even if some scorers fail.
results = mlflow.genai.evaluate(
data=generated_traces,
scorers=[resilient_scorer]
)
Exemple 8 : Conventions de nommage dans les évaluateurs
Les exemples suivants illustrent le comportement de nommage des évaluateurs basés sur le code.
from mlflow.genai.scorers import Scorer
from mlflow.entities import Feedback
from typing import Optional, Any, List
# Primitive value or single `Feedback` without a name: The scorer function name becomes the metric name.
@scorer
def decorator_primitive(outputs: str) -> int:
# metric name = "decorator_primitive"
return 1
@scorer
def decorator_unnamed_feedback(outputs: Any) -> Feedback:
# metric name = "decorator_unnamed_feedback"
return Feedback(value=True, rationale="Good quality")
# Single `Feedback` with an explicit name: The name specified in the `Feedback` object is used as the metric name.
@scorer
def decorator_feedback_named(outputs: Any) -> Feedback:
# metric name = "decorator_named_feedback"
return Feedback(name="decorator_named_feedback", value=True, rationale="Factual accuracy is high")
# Multiple `Feedback` objects: Names specified in each `Feedback` object are preserved. You must specify a unique name for each `Feedback`.
@scorer
def decorator_named_feedbacks(outputs) -> list[Feedback]:
return [
Feedback(name="decorator_named_feedback_1", value=True, rationale="No errors"),
Feedback(name="decorator_named_feedback_2", value=0.9, rationale="Very clear"),
]
# Class returning primitive value
class ScorerPrimitive(Scorer):
# metric name = "scorer_primitive"
name: str = "scorer_primitive"
def __call__(self, outputs: str) -> int:
return 1
scorer_primitive = ScorerPrimitive()
# Class returning a Feedback object without a name
class ScorerFeedbackUnnamed(Scorer):
# metric name = "scorer_named_feedback"
name: str = "scorer_named_feedback"
def __call__(self, outputs: str) -> Feedback:
return Feedback(value=True, rationale="Good")
scorer_feedback_unnamed = ScorerFeedbackUnnamed()
# Class returning a Feedback object with a name
class ScorerFeedbackNamed(Scorer):
# metric name = "scorer_named_feedback"
name: str = "scorer_feedback_named"
def __call__(self, outputs: str) -> Feedback:
return Feedback(name="scorer_named_feedback", value=True, rationale="Good")
scorer_feedback_named = ScorerFeedbackNamed()
# Class returning multiple Feedback objects with names
class ScorerNamedFeedbacks(Scorer):
# metric names = ["scorer_named_feedback_1", "scorer_named_feedback_1"]
name: str = "scorer_named_feedbacks" # Not used
def __call__(self, outputs: str) -> List[Feedback]:
return [
Feedback(name="scorer_named_feedback_1", value=True, rationale="Good"),
Feedback(name="scorer_named_feedback_2", value=1, rationale="ok"),
]
scorer_named_feedbacks = ScorerNamedFeedbacks()
mlflow.genai.evaluate(
data=generated_traces,
scorers=[
decorator_primitive,
decorator_unnamed_feedback,
decorator_feedback_named,
decorator_named_feedbacks,
scorer_primitive,
scorer_feedback_unnamed,
scorer_feedback_named,
scorer_named_feedbacks,
],
)
Exemple 9 : enchaînement des résultats d'évaluation
Si un scoreur indique des problèmes avec un sous-ensemble de traces, vous pouvez collecter ce sous-ensemble de traces pour une itération ultérieure en utilisant mlflow.search_traces(). L'exemple ci-dessous identifie les défaillances générales de « Safety » et analyse ensuite le sous-ensemble de traces échouées à l'aide d'un évaluateur plus personnalisé (un exemple ludique d'évaluation utilisant un document de politique de contenu). Alternativement, vous pourriez utiliser le sous-ensemble de traces problématiques pour itérer sur votre application IA elle-même et améliorer ses performances sur les entrées difficiles.
from mlflow.genai.scorers import Safety, Guidelines
# Run initial evaluation
results1 = mlflow.genai.evaluate(
data=generated_traces,
scorers=[Safety()]
)
# Use results to create refined dataset
traces = mlflow.search_traces(run_id=results1.run_id)
# Filter to problematic traces
safety_failures = traces[traces['assessments'].apply(
lambda x: any(a['assessment_name'] == 'Safety' and a['feedback']['value'] == 'no' for a in x)
)]
# Updated app (not actually updated in this toy example)
updated_app = sample_app
# Re-evaluate with different scorers or updated app
if len(safety_failures) > 0:
results2 = mlflow.genai.evaluate(
data=safety_failures,
predict_fn=updated_app,
scorers=[
Guidelines(
name="content_policy",
guidelines="Response must follow our content policy"
)
]
)
Exemple 10 : Logique conditionnelle avec des lignes directrices
Vous pouvez intégrer les juges des directives dans des évaluateurs personnalisés basés sur du code pour appliquer différentes directives en fonction des attributs utilisateur ou d'un autre contexte.
from mlflow.genai.scorers import scorer, Guidelines
@scorer
def premium_service_validator(inputs, outputs, trace=None):
"""Custom scorer that applies different guidelines based on user tier"""
# Extract user tier from inputs (could also come from trace)
user_tier = inputs.get("user_tier", "standard")
# Apply different guidelines based on user attributes
if user_tier == "premium":
# Premium users get more personalized, detailed responses
premium_judge = Guidelines(
name="premium_experience",
guidelines=[
"The response must acknowledge the user's premium status",
"The response must provide detailed explanations with at least 3 specific examples",
"The response must offer priority support options (e.g., 'direct line' or 'dedicated agent')",
"The response must not include any upselling or promotional content"
]
)
return premium_judge(inputs=inputs, outputs=outputs)
else:
# Standard users get clear but concise responses
standard_judge = Guidelines(
name="standard_experience",
guidelines=[
"The response must be helpful and professional",
"The response must be concise (under 100 words)",
"The response may mention premium features as upgrade options"
]
)
return standard_judge(inputs=inputs, outputs=outputs)
# Example evaluation data
eval_data = [
{
"inputs": {
"question": "How do I export my data?",
"user_tier": "premium"
},
"outputs": {
"response": "As a premium member, you have access to advanced export options. You can export in 5 formats: CSV, Excel, JSON, XML, and PDF. Here's how: 1) Go to Settings > Export, 2) Choose your format and date range, 3) Click 'Export Now'. For immediate assistance, call your dedicated support line at 1-800-PREMIUM."
}
},
{
"inputs": {
"question": "How do I export my data?",
"user_tier": "standard"
},
"outputs": {
"response": "You can export your data as CSV from Settings > Export. Premium users can access additional formats like Excel and PDF."
}
}
]
# Run evaluation with the custom scorer
results = mlflow.genai.evaluate(
data=eval_data,
scorers=[premium_service_validator]
)
Exemple de Notebook
Le Notebook suivant inclut l'ensemble du code sur cette page.
Évaluateurs basés sur le code pour le notebook d'évaluation MLflow
Ressources supplémentaires
- Évaluateurs LLM personnalisés — Découvrez l’évaluation sémantique à l’aide des métriques LLM en tant que juge, qui peuvent être plus simples à définir que les évaluateurs basés sur le code.
- Exécutez les évaluateurs en production — Déployez vos évaluateurs pour un monitoring continu.
- Créez des datasets d'évaluation – Créez des données de test pour vos scoreurs.