Monitorování GenAI v produkčním prostředí

Důležité

Tato funkce je v beta verzi. Správci pracovního prostoru můžou řídit přístup k této funkci ze stránky Previews . Podívejte se na Správa náhledů Azure Databricks.

Provozní monitorování GenAI na Azure Databricks umožňuje automaticky spouštět hodnotící moduly MLflow 3 na záznamech z vašich produkčních aplikací GenAI a neustále sledovat kvalitu.

Můžete naplánovat skórovací nástroje tak, aby automaticky vyhodnocovaly vzorek vašeho produkčního provozu. Výsledky hodnocení hodnotitele se automaticky připojují jako zpětná vazba ke zhodnoceným stopám.

Monitorování v produkčním prostředí zahrnuje následující:

• Automatizované hodnocení kvality pomocí integrovaných nebo vlastních výkonnostních metrik.
• Konfigurovatelné vzorkovací frekvence, abyste mohli řídit kompromis mezi pokrytím a výpočetními náklady.
• Pomocí stejných výkonnostních metrik ve vývoji a produkčním prostředí zajistíte konzistentní hodnocení.
• Průběžné hodnocení kvality s monitorováním běžícím na pozadí

Poznámka:

Produkční monitorování v MLflow 3 je kompatibilní se stopami zaznamenávanými v MLflow 2.

Informace o monitorování starší verze produkčního prostředí najdete v tématu Referenční informace k rozhraní API pro monitorování produkčního prostředí (starší verze).

Požadavky

Před nastavením monitorování kvality se ujistěte, že máte:

• Experiment MLflow: Experiment MLflow, kde se zaznamenávají stopy. Pokud není zadaný žádný experiment, použije se aktivní experiment.
• Instrumentovaná produkční aplikace: Vaše aplikace GenAI musí protokolovat trasování pomocí MLflow Tracing. Viz průvodce trasováním produkce.
• Definovaní skórovací moduly: Testované skórovací moduly, které pracují s formátem trasování vaší aplikace. Pokud jste produkční aplikaci použili jako predict_fn při mlflow.genai.evaluate() vývoji, vaše vyhodnocovací nástroje jsou pravděpodobně již kompatibilní.

• ID SQL skladu (pro trasování Unity Catalog): Pokud jsou trasování uložená v Unity Catalog, musíte nakonfigurovat ID SQL skladu, aby monitorování fungovalo. Viz Povolení provozního monitorování.

Začínáme s monitorováním v produkčním prostředí

Tato část obsahuje ukázkový kód, který ukazuje, jak vytvořit různé typy výkonnostních metrik.

Další informace o hodnotitelích najdete v následujícím:

• Hodnotitelé a posuzovatelé modelů LLM
• Vlastní hodnoticí systémy
• Vytváření vlastních výkonnostních metrik založených na kódu

Poznámka:

Kdykoli může být k experimentu pro průběžné monitorování kvality přiřazeno maximálně 20 hodnotitelů.

Vytvoření a naplánování porotců LLM pomocí uživatelského rozhraní

Pomocí uživatelského rozhraní MLflow můžete vytvářet a testovat hodnotitele založené na porotcích LLM.

1. V uživatelském rozhraní experimentu MLflow přejděte na kartu Scorers (Přehled výkonnostních metrik ).
2. Klikněte na Nový hodnotitel.
3. V rozevírací nabídce šablony LLM vyberte integrovanou porotu LLM.
4. (Volitelné) Kliknutím na Spustit scorer spusťte podmnožinu trasování.
5. (Volitelné) Upravte nastavení vyhodnocení pro monitorování produkčního prostředí pro budoucí trasování.
6. Klikněte na Vytvořit hodnotitel.

Použití integrovaných hodnotitelů LLM

MLflow poskytuje několik integrovaných hodnotitelů LLM , které můžete použít ke sledování.

from mlflow.genai.scorers import Safety, ScorerSamplingConfig

# Register the scorer with a name and start monitoring
safety_judge = Safety().register(name="my_safety_judge")  # name must be unique to experiment
safety_judge = safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

Ve výchozím nastavení každý soudce používá LLM hostovaný v Databricks, který je navržený k provádění hodnocení kvality GenAI. Model pro posuzování můžete změnit tak, aby místo toho používal koncový bod obsluhující model Databricks pomocí model argumentu v definici skóre. Model musí být zadán ve formátu databricks:/<databricks-serving-endpoint-name>.

safety_judge = Safety(model="databricks:/databricks-gpt-oss-20b").register(name="my_custom_safety_judge")

Použití pokynů pro soudce LLM

Pokyny LLM Porotci vyhodnocují vstupy a výstupy pomocí kritérií přirozeného jazyka pass/fail.

from mlflow.genai.scorers import Guidelines

# Create and register the guidelines scorer
english_judge = Guidelines(
  name="english",
  guidelines=["The response must be in English"]
).register(name="is_english")  # name must be unique to experiment

# Start monitoring with the specified sample rate
english_judge = english_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.7))

Stejně jako u vestavěných soudců můžete změnit model rozhodování tak, aby místo toho používal koncový bod nasazení modelu Databricks.

english_judge = Guidelines(
  name="english",
  guidelines=["The response must be in English"],
  model="databricks:/databricks-gpt-oss-20b",
).register(name="custom_is_english")

Použití porotců LLM s vlastními výzvami

Pokud chcete větší flexibilitu než porotci pokynů, použijte soudce LLM s vlastními výzvami , které umožňují hodnocení kvality na více úrovních s přizpůsobitelnými kategoriemi voleb.

from typing import Literal
from mlflow.genai import make_judge
from mlflow.genai.scorers import ScorerSamplingConfig

# Create a custom judge using make_judge
formality_judge = make_judge(
    name="formality",
    instructions="""You will look at the response and determine the formality of the response.

Request: {{ inputs }}
Response: {{ outputs }}

Evaluate whether the response is formal, somewhat formal, or not formal.
A response is somewhat formal if it mentions friendship, etc.""",
    feedback_value_type=Literal["formal", "semi_formal", "not_formal"],
    model="databricks:/databricks-gpt-oss-20b",  # optional
)

# Register the custom judge and start monitoring
registered_judge = formality_judge.register(name="my_formality_judge")  # name must be unique to experiment
registered_judge = registered_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.1))

Použití vlastních funkcí scoreru

Pro maximální flexibilitu definujte a použijte vlastní hodnotící funkci.

Při definování vlastních hodnotitelů nepoužívejte typové nápovědy, které je potřeba importovat v podpisu funkce. Pokud tělo funkce scorer používá balíčky, které je potřeba importovat, importujte tyto balíčky přímo ve funkci, aby se zajistila správná serializace.

Některé balíčky jsou ve výchozím nastavení k dispozici bez nutnosti vloženého importu. Patří mezi ně databricks-agents, mlflow-skinny, openai a všechny balíčky obsažené v bezserverovém prostředí verze 2.

from mlflow.genai.scorers import scorer, ScorerSamplingConfig


# Custom metric: Check if response mentions Databricks
@scorer
def mentions_databricks(outputs):
    """Check if the response mentions Databricks"""
    return "databricks" in str(outputs.get("response", "")).lower()

# Custom metric: Response length check
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(str(outputs.get("response", "")))

# Custom metric with multiple inputs
@scorer
def response_relevance_score(inputs, outputs):
    """Score relevance based on keyword matching"""
    query = str(inputs.get("query", "")).lower()
    response = str(outputs.get("response", "")).lower()

    # Simple keyword matching (replace with your logic)
    query_words = set(query.split())
    response_words = set(response.split())

    if not query_words:
        return 0.0

    overlap = len(query_words & response_words)
    return overlap / len(query_words)

# Register and start monitoring custom scorers
databricks_scorer = mentions_databricks.register(name="databricks_mentions")
databricks_scorer = databricks_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

length_scorer = response_length.register(name="response_length")
length_scorer = length_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

relevance_scorer = response_relevance_score.register(name="response_relevance_score")  # name must be unique to experiment
relevance_scorer = relevance_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))

Konfigurace více skórovacích systémů

Pro komplexní nastavení monitorování zaregistrujte a spusťte více výkonnostních metrik jednotlivě.

from mlflow.genai.scorers import Safety, Guidelines, ScorerSamplingConfig, list_scorers

# # Register multiple scorers for comprehensive monitoring
safety_judge = Safety().register(name="safety") # name must be unique within an MLflow experiment
safety_judge = safety_judge.start(
    sampling_config=ScorerSamplingConfig(sample_rate=1.0), # Check all traces
)

guidelines_judge = Guidelines(
    name="english",
    guidelines=["Response must be in English"]
).register(name="english_check")
guidelines_judge = guidelines_judge.start(
    sampling_config=ScorerSamplingConfig(sample_rate=0.5), # Sample 50%
)

# List and manage all scorers
all_scorers = list_scorers()
for scorer in all_scorers:
    if scorer.sample_rate > 0:
        print(f"{scorer.name} is active")
    else:
        print(f"{scorer.name} is stopped")

Životní cyklus scoreru

Životní cyklus scorerů se zaměřuje na experimenty v MLflow. Následující tabulka uvádí stavy životního cyklu nástroje pro hodnocení.

Scorery jsou neměnné, takže operace životního cyklu neupravuje původní scorer. Místo toho vrátí novou instanci scoreru.

Stát	Description	API
Neregistrované	Funkce _scorer_ je definována, ale není serverem rozpoznána.
Registrovaný	Scorer se zaregistruje do aktivního experimentu MLflow.	.register()
Aktivní	Scorer běží s vzorkovací frekvencí > 0.	.start()
Zastavený	Scorer je zaregistrovaný, ale neběží (vzorkovací frekvence = 0).	.stop()
Odstraněné	Nástroj scorer byl odebrán ze serveru a už není přidružený k experimentu MLflow.	delete_scorer()

Životní cyklus základního hodnotitele

from mlflow.genai.scorers import Safety, scorer, ScorerSamplingConfig

# Built-in scorer lifecycle
safety_judge = Safety().register(name="safety_check")
safety_judge = safety_judge.start(
    sampling_config=ScorerSamplingConfig(sample_rate=1.0),
)
safety_judge = safety_judge.update(
    sampling_config=ScorerSamplingConfig(sample_rate=0.8),
)
safety_judge = safety_judge.stop()
delete_scorer(name="safety_check")

# Custom scorer lifecycle
@scorer
def response_length(outputs):
    return len(str(outputs.get("response", "")))

length_scorer = response_length.register(name="length_check")
length_scorer = length_scorer.start(
    sampling_config=ScorerSamplingConfig(sample_rate=0.5),
)

Správa skórovacích systémů

Pro správu výkonnostních metrik jsou k dispozici následující rozhraní API.

API	Description	Example
list_scorers()	Vypište seznam všech registrovaných skórujících pro aktuální experiment.	Vypsat současné střelce
get_scorer()	Načte registrovaný hodnotitel podle názvu.	Získejte a aktualizujte skórovací systém
Scorer.update()	Upravte konfiguraci vzorkování aktivního scoreru. Jedná se o neměnnou operaci.	Získejte a aktualizujte skórovací systém
backfill_scorer()	Retroaktivní aplikace nových nebo aktualizovaných metrik na historické záznamy.	Vyhodnocení historických stop (doplnění metrik)
delete_scorer()	Odstraň registrovaný skórovací modul podle názvu.	Zastavení a odstranění výkonnostních metrik

Seznam aktuálních střelců

Zobrazení všech registrovaných výkonnostních metrik pro experiment:

from mlflow.genai.scorers import list_scorers

# List all registered scorers
scorers = list_scorers()
for scorer in scorers:
    print(f"Name: {scorer._server_name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")
    print("---")

Získat a aktualizovat vyhodnocovací modul

K úpravě existujících konfigurací hodnotícího systému:

from mlflow.genai.scorers import get_scorer

# Get existing scorer and update its configuration (immutable operation)
safety_judge = get_scorer(name="safety_monitor")
updated_judge = safety_judge.update(sampling_config=ScorerSamplingConfig(sample_rate=0.8))  # Increased from 0.5

# Note: The original scorer remains unchanged; update() returns a new scorer instance
print(f"Original sample rate: {safety_judge.sample_rate}")  # Original rate
print(f"Updated sample rate: {updated_judge.sample_rate}")   # New rate

Zastavení a odstranění skórovacích nástrojů

Chcete-li zastavit monitorování nebo úplně odstranit skórovač:

from mlflow.genai.scorers import get_scorer, delete_scorer

# Get existing scorer
databricks_scorer = get_scorer(name="databricks_mentions")

# Stop monitoring (sets sample_rate to 0, keeps scorer registered)
stopped_scorer = databricks_scorer.stop()
print(f"Sample rate after stop: {stopped_scorer.sample_rate}")  # 0

# Remove scorer entirely from the server
delete_scorer(name=databricks_scorer.name)

# Or restart monitoring from a stopped scorer
restarted_scorer = stopped_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

Neměnné aktualizace

Skórovače, včetně porotců LLM, jsou neměnné objekty. Při aktualizaci skórovače neupravíte původní skórovač. Místo toho je vytvořena aktualizovaná verze skórovacího systému. Tato neměnnost pomáhá zajistit, aby výkonnostní metriky určené pro produkční prostředí nebyly omylem změněny. Následující fragment kódu ukazuje, jak fungují neměnné aktualizace.

# Demonstrate immutability
original_judge = Safety().register(name="safety")
original_judge = original_judge.start(
   sampling_config=ScorerSamplingConfig(sample_rate=0.3),
)

# Update returns new instance
updated_judge = original_judge.update(
    sampling_config=ScorerSamplingConfig(sample_rate=0.8),
)

# Original remains unchanged
print(f"Original: {original_judge.sample_rate}")  # 0.3
print(f"Updated: {updated_judge.sample_rate}")    # 0.8

Vyhodnotit historické sledování (doplnění metrik)

Na historické stopování můžete zpětně použít nové nebo aktualizované metriky.

Základní doplňování historických dat metrik s využitím aktuálních frekvencí vzorkování

from databricks.agents.scorers import backfill_scorers

safety_judge = Safety()
safety_judge.register(name="safety_check")
safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Use existing sample rates for specified scorers
job_id = backfill_scorers(
    scorers=["safety_check", "response_length"]
)

Zpětné naplnění metrik s využitím vlastních vzorkovacích frekvencí a časového rozsahu

from databricks.agents.scorers import backfill_scorers, BackfillScorerConfig
from datetime import datetime
from mlflow.genai.scorers import Safety, Correctness

safety_judge = Safety()
safety_judge.register(name="safety_check")
safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

#custom scorer
@scorer(aggregations=["mean", "min", "max"])
def response_length(outputs):
    """Measure response length in characters"""
    return len(outputs)

response_length.register(name="response_length")
response_length.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))

# Define custom sample rates for backfill
custom_scorers = [
    BackfillScorerConfig(scorer=safety_judge, sample_rate=0.8),
    BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
]

job_id = backfill_scorers(
    experiment_id=YOUR_EXPERIMENT_ID,
    scorers=custom_scorers,
    start_time=datetime(2024, 6, 1),
    end_time=datetime(2024, 6, 30)
)

Nedávné doplnění dat

from datetime import datetime, timedelta

# Backfill last week's data with higher sample rates
one_week_ago = datetime.now() - timedelta(days=7)

job_id = backfill_scorers(
    scorers=[
        BackfillScorerConfig(scorer=safety_judge, sample_rate=0.8),
        BackfillScorerConfig(scorer=response_length, sample_rate=0.9)
    ],
    start_time=one_week_ago
)

Zobrazení výsledků

Po naplánování hodnotitelů počítejte s počátečním zpracováním 15 až 20 minut. Potom:

1. Přejděte do experimentu MLflow.
2. Otevřete kartu Trasování a zobrazte hodnocení připojená k trasování.
3. Pomocí řídicích panelů monitorování můžete sledovat trendy kvality.

Osvědčené postupy

Správa stavu vyhodnocovacího modulu

• Před zahájením operací pomocí sample_rate zkontrolujte stav skórovacího zařízení.
• Použijte neměnný vzor. Přiřaďte výsledky funkce .start(), .update().stop() proměnným.
• Seznamte se s životním cyklem scoreru. .stop() zachovává registraci, delete_scorer() zcela odstraní skórovací modul.

Doplňování metrik

• Začněte v malém. Začněte s menšími časovými rozsahy, abyste odhadli dobu trvání úlohy a využití prostředků.
• Použijte odpovídající vzorkovací frekvence. Zamyslete se nad náklady a časem při používání vysokých vzorkových sazeb.

Strategie vzorkování

• Pro kritické hodnoty, jako jsou bezpečnostní a ochranné kontroly, použijte sample_rate=1.0.

• U drahých výkonnostních metrik, jako jsou komplexní porotci LLM, použijte nižší vzorkovací sazby (0,05-0,2).

• Pro iterativní zlepšení během vývoje používejte střední míru (0,3-0,5).

• Vyvážit pokrytí a náklady, jak je uvedeno v následujících příkladech:

  # High-priority scorers: higher sampling
  safety_judge = Safety().register(name="safety")
  safety_judge = safety_judge.start(sampling_config=ScorerSamplingConfig(sample_rate=1.0))  # 100% coverage for critical safety
  
  # Expensive scorers: lower sampling
  complex_scorer = ComplexCustomScorer().register(name="complex_analysis")
  complex_scorer = complex_scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.05))  # 5% for expensive operations
  

Návrh vlastního skórovacího systému

Zachovejte vlastní posuzovače samostatně, jak je znázorněno v následujícím příkladu:

@scorer
def well_designed_scorer(inputs, outputs):
    # All imports inside the function
    import re
    import json

    # Handle missing data gracefully
    response = outputs.get("response", "")
    if not response:
        return 0.0

    # Return consistent types
    return float(len(response) > 100)

Řešení problémů

Neběží výkonnostní metriky

Pokud se skórovací nástroje nespouští, zkontrolujte následující:

1. Kontrola experimentu: Ujistěte se, že jsou do experimentu zaevidována trasování, a nikoli k jednotlivým spuštěním.
2. Vzorkovací frekvence: S nízkými vzorkovacími mírami může zobrazení výsledků nějakou dobu trvat.
3. Ověřte filtrační řetězec: Ujistětefilter_string se, že skutečně odpovídá fyzickým záznamům.

Problémy se serializací

Při vytváření vlastního scoreru zahrňte do definice funkce importy.

# ❌ Avoid external dependencies
import external_library  # Outside function

@scorer
def bad_scorer(outputs):
    return external_library.process(outputs)

# ✅ Include imports in the function definition
@scorer
def good_scorer(outputs):
    import json  # Inside function
    return len(json.dumps(outputs))

# ❌ Avoid using type hints in scorer function signature that requires imports
from typing import List

@scorer
def scorer_with_bad_types(outputs: List[str]):
    return False

Problémy s doplňováním metrik

"Naplánovaný hodnotitel 'X' nebyl nalezen v experimentu"

• Ujistěte se, že název skórovacího systému odpovídá registrovanému skórovacímu systému ve vašem experimentu.
• Pomocí metody list_scorers zkontrolujte dostupné scorery.

Archivní stopy

Trasování a jejich přidružená hodnocení můžete uložit do tabulky Unity Catalog Delta pro dlouhodobé ukládání a pokročilou analýzu. To je užitečné při vytváření vlastních řídicích panelů, provádění hloubkové analýzy dat trasování a udržování trvalého záznamu o chování vaší aplikace.

Poznámka:

Musíte mít potřebná oprávnění k zápisu do zadané tabulky Unity Catalog Delta. Cílová tabulka se vytvoří, pokud ještě neexistuje.

Pokud tabulka již existuje, připojí se k ní záznamy.

Povolit archivaci trasování

API MLflow

Pro zahájení archivace stop experimentu použijte funkci enable_databricks_trace_archival. Musíte zadat úplný název cílové tabulky Delta, včetně katalogu a schématu. Pokud nezadáte experiment_id, archivace sledování je povolena pro aktuálně aktivní experiment.

from mlflow.tracing.archival import enable_databricks_trace_archival

# Archive traces from a specific experiment to a Unity Catalog Delta table
enable_databricks_trace_archival(
    delta_table_fullname="my_catalog.my_schema.archived_traces",
    experiment_id="YOUR_EXPERIMENT_ID",
)

Pomocí funkce disable_databricks_trace_archival můžete kdykoli zastavit archivaci stop experimentu.

from mlflow.tracing.archival import disable_databricks_trace_archival

# Stop archiving traces for the specified experiment
disable_databricks_trace_archival(experiment_id="YOUR_EXPERIMENT_ID")

Uživatelské rozhraní Databricks

Ke archivaci trasování experimentu v uživatelském rozhraní:

1. V pracovním prostoru Databrick přejděte na stránku Experimenty.
2. Klikněte na Delta synchronizace: Nepovoleno.
3. Zadejte úplný název cílové tabulky Delta, včetně katalogu a schématu.

Zakázání archivace trasování:

1. V pracovním prostoru Databrick přejděte na stránku Experimenty.
2. Klikněte na Delta synchronizace: Povoleno>Zakázat synchronizaci.

Další kroky

• Vytvářejte vlastní výkonnostní metriky – Vytvářejte výkonnostní metriky přizpůsobené vašim potřebám.
• Vytváření zkušebních datových sad – ke zlepšení kvality použijte výsledky monitorování.

Referenční příručky

• Referenční příručka k rozhraní API pro správu životního cyklu skórovačů – rozhraní API a příklady pro správu skórovačů pro sledování.
• Hodnotitelé - Seznamte se s metrikami, které podporují monitorování.
• Vyhodnocovací nástroj – jak offline vyhodnocení souvisí s produkčním prostředím