mlflow.genai

class mlflow.genai.Agent(agent: _Agent)[source]

基类： object

智能体配置，用于在审查应用中生成响应。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

property agent_name: str: 智能体的名称。

property model_serving_endpoint: str: 智能体使用的模型服务端点。

class mlflow.genai.LabelingSession(session: _LabelingSession)[source]

基类： object

在 review 应用中对条目进行标注的会话。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

add_dataset(dataset_name: str, record_ids: Optional[list[str]] = None) → mlflow.genai.labeling.labeling.LabelingSession[source]

将数据集添加到标注会话。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters

dataset_name – 数据集的名称。
record_ids – 可选。要添加到会话的各个记录 id。如果未提供，则数据集中所有记录都会被添加。

Returns

已更新的标注会话。

Return type

LabelingSession

add_traces(traces: Union[Iterable[Trace], Iterable[str], pd.DataFrame]) → LabelingSession[source]

向标注会话添加跟踪。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用。

Parameters: traces – 可以是以下任意一种： a) 一个 pandas DataFrame，包含 ‘trace’ 列。‘trace’ 列应包含 mlflow.entities.Trace 对象或其 json 字符串表示。 b) 一个 mlflow.entities.Trace 对象的可迭代集合。 c) 一个 mlflow.entities.Trace 对象的 json 字符串表示的可迭代集合。
Returns: 更新后的标注会话。
Return type: LabelingSession

property agent: str | None: 用于为会话中的项目生成响应的智能体。

property assigned_users: list[str]: 会话中被分配来标注项目的用户。

property custom_inputs: dict[str, typing.Any] | None: 会话中使用的自定义输入。

property enable_multi_turn_chat: bool: 此会话是否启用了多轮对话。

property experiment_id: str: 与该会话关联的实验 ID。

property label_schemas: list[str]: 会话中使用的标签模式。

property labeling_session_id: str: 标注会话的唯一标识符。

property mlflow_run_id: str: 与会话关联的 MLflow run ID。

property name: str: 标注会话的名称。

property review_app_id: str: 与该会话关联的 review app ID。

set_assigned_users(assigned_users: list[str]) → mlflow.genai.labeling.labeling.LabelingSession[source]

设置此标注会话的分配用户。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: assigned_users – 将要分配到会话的用户列表。
Returns: 已更新的标注会话。
Return type: LabelingSession

sync(to_dataset: str) → None[source]

将标注会话的跟踪和期望同步到数据集。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: to_dataset – 要将跟踪和期望同步到的数据集的名称。

property url: str: 审阅应用中标注会话的 URL。

class mlflow.genai.ReviewApp(app: _ReviewApp)[source]

基类： object

审阅应用用于为特定实验收集利益相关者的反馈。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

add_agent(*, agent_name: str, model_serving_endpoint: str, overwrite: bool = False) → mlflow.genai.labeling.labeling.ReviewApp[source]

将一个智能体添加到评审应用中以用于生成响应。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters

agent_name – 智能体的名称。
model_serving_endpoint – 供智能体使用的模型服务端点。
overwrite – 是否覆盖具有相同名称的现有智能体。

Returns

更新后的评审应用。

Return type

ReviewApp

property agents: list[mlflow.genai.labeling.labeling.Agent]: 用于生成响应的智能体。

property experiment_id: str: 实验的 ID。

property label_schemas: list['_LabelSchema']: 在审阅应用中要使用的标签模式。

remove_agent(agent_name: str) → mlflow.genai.labeling.labeling.ReviewApp[source]

从评审应用中移除智能体。

注意

此功能仅在 Databricks 可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: agent_name – 要删除的智能体的名称。
Returns: 已更新的评审应用。
Return type: ReviewApp

property review_app_id: str: 评审应用的 ID。

property url: str: 供利益相关者提供反馈的评审应用的 URL。

class mlflow.genai.Scorer(*, name: str, aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None)[source]

基类：pydantic.main.BaseModel

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None

property filter_string: str | None

此函数可能会在将来的版本中更改或被移除，且不会另行通知。

获取此评分器的过滤字符串。

Type: 注意

实验性

property kind: mlflow.genai.scorers.base.ScorerKind

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_dump(**kwargs) → dict[str, typing.Any][source]: 重写 model_dump 以包含源代码。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

classmethod model_validate(obj: Any) → Scorer[source]: 重写 model_validate 以从源代码重建评分器。

name: str

register(*, name: Optional[str] = None, experiment_id: Optional[str] = None) → 评分器[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

在 MLflow 服务器上注册此评分器。

此方法将评分器注册，以便在指定的实验中用于自动 trace 评估。注册后，可以启动该评分器以开始自动评估 trace。

Parameters

name – 可选的注册名称，用于评分器。如果未提供，将使用当前 name 属性值作为注册名称。
experiment_id – 要注册评分器的 MLflow 实验的 ID。如果为 None，则使用当前活动的实验。

Returns

一个新的 Scorer 实例，包含服务器注册信息。

示例

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

# Register a built-in scorer
mlflow.set_experiment("my_genai_app")
registered_scorer = RelevanceToQuery().register(name="relevance_scorer")
print(f"Registered scorer: {registered_scorer.name}")

# Register a custom scorer
from mlflow.genai.scorers import scorer


@scorer
def custom_length_check(outputs) -> bool:
    return len(outputs) > 100


registered_custom = custom_length_check.register(
    name="output_length_checker", experiment_id="12345"
)

run(*, inputs=None, outputs=None, expectations=None, trace=None)[source]

property sample_rate: float | None

此函数可能会在未来的版本中更改或被移除，恕不另行通知。

获取此评分器的采样率。注册以进行监控时可用。

Type: 注意

实验性

start(*, name: Optional[str] = None, experiment_id: Optional[str] = None, sampling_config: mlflow.genai.scorers.base.ScorerSamplingConfig) → Scorer[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

使用指定的采样配置启动已注册的评分。

该方法为评分器激活自动跟踪评估。评分器将根据所提供的采样配置，包括采样率和可选的筛选条件，对跟踪进行评估。

Parameters

name – 可选的评分器名称。如果未提供，则使用评分器注册的名称或默认名称。
experiment_id – 包含该评分器的 MLflow 实验的 ID。如果 None，使用当前活动的实验。
sampling_config – 配置对象，包含： - sample_rate: 要评估的 traces 的比例（0.0 到 1.0）。必需。 - filter_string: 可选的与 MLflow search_traces 兼容的筛选字符串。

Returns

一个具有更新采样配置的新 Scorer 实例。

示例

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, ScorerSamplingConfig

# Start scorer with 50% sampling rate
mlflow.set_experiment("my_genai_app")
scorer = RelevanceToQuery().register()
active_scorer = scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))
print(f"Scorer is evaluating {active_scorer.sample_rate * 100}% of traces")

# Start scorer with filter to only evaluate specific traces
filtered_scorer = scorer.start(
    sampling_config=ScorerSamplingConfig(
        sample_rate=1.0, filter_string="YOUR_FILTER_STRING"
    )
)

stop(*, name: Optional[str] = None, experiment_id: Optional[str] = None) → Scorer[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

通过将采样率设置为 0 来停止已注册的评分。

此方法停用评分器的自动跟踪评估，同时保持评分器处于注册状态。评分器可以稍后使用 start() 方法重新启动。

Parameters

name – 可选的 scorer name。如果未提供，则使用 scorer 已注册的 name 或默认的 name。
experiment_id – 包含评分器的 MLflow 实验的 ID。如果为 None，使用当前活动的实验。

Returns

一个新的 Scorer 实例，sample rate 设置为 0。

示例

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, ScorerSamplingConfig

# Start and then stop a scorer
mlflow.set_experiment("my_genai_app")
scorer = RelevanceToQuery().register()
active_scorer = scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.5))
print(f"Scorer is active: {active_scorer.sample_rate > 0}")

# Stop the scorer
stopped_scorer = active_scorer.stop()
print(f"Scorer is active: {stopped_scorer.sample_rate > 0}")

# The scorer remains registered and can be restarted later
restarted_scorer = stopped_scorer.start(
    sampling_config=ScorerSamplingConfig(sample_rate=0.3)
)

update(*, name: Optional[str] = None, experiment_id: Optional[str] = None, sampling_config: mlflow.genai.scorers.base.ScorerSamplingConfig) → Scorer[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

更新此评分器的采样配置。

该方法修改已注册评分器的采样率和/或过滤条件。它可用于动态调整要评估的跟踪记录数量或在不停止并重新启动评分器的情况下更改过滤条件。

Parameters

name – 可选的评分器名称。如果未提供，则使用评分器的注册名称或默认名称。
experiment_id – 包含评分器的 MLflow 实验的 ID。如果为 None，则使用当前活动的实验。
sampling_config – 配置对象，包含： - sample_rate: 要评估的跟踪记录的新比例（0.0 到 1.0）。可选。 - filter_string: 新的与 MLflow search_traces 兼容的过滤字符串。可选。

Returns

一个新的 Scorer 实例，具有更新的配置。

示例

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, ScorerSamplingConfig

# Start scorer with initial configuration
mlflow.set_experiment("my_genai_app")
scorer = RelevanceToQuery().register()
active_scorer = scorer.start(sampling_config=ScorerSamplingConfig(sample_rate=0.1))

# Update to increase sampling rate during high traffic
updated_scorer = active_scorer.update(
    sampling_config=ScorerSamplingConfig(sample_rate=0.5)
)
print(f"Updated sample rate: {updated_scorer.sample_rate}")

# Update to add filtering criteria
filtered_scorer = updated_scorer.update(
    sampling_config=ScorerSamplingConfig(filter_string="YOUR_FILTER_STRING")
)
print(f"Added filter: {filtered_scorer.filter_string}")

class mlflow.genai.ScorerScheduleConfig(scorer: Scorer, scheduled_scorer_name: str, sample_rate: float, filter_string: Optional[str] = None)[source]

基类： object

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

一个用于对生成式 AI 应用进行自动化监控的定期评分器配置。

定时评分器用于自动评估由生产应用记录到 MLflow 实验的跟踪。它们是 Databricks Lakehouse Monitoring for GenAI 的一部分，帮助同时跟踪诸如事实依据性（groundedness）、安全性和对指南的遵守等质量指标，以及诸如量、延迟和成本等运营指标。

配置后，已调度的评分器会在后台自动运行，根据指定的采样率和过滤条件对一部分 traces 进行评估。评估结果显示在 MLflow 实验的 Traces 选项卡中，可用于识别生产环境中的质量问题。

Parameters

scorer – 要在采样的轨迹上运行的 scorer 函数。必须是内置的 scorer（例如，Safety、Correctness）或用 @scorer 装饰的函数。不支持 Scorer 的子类。
scheduled_scorer_name – 此计划评分器配置在实验中的名称。该名称在同一实验中的所有计划评分器中必须唯一。我们建议使用评分器的名称（例如，scorer.name）以保持一致。
sample_rate – 要评估的 traces 的比例，范围在 0.0 到 1.0 之间。例如，0.1 表示将随机选择 10% 的 traces 进行评估。
filter_string – 一个可选的与 MLflow search_traces 兼容的过滤字符串，在采样 traces 之前应用。只有匹配此过滤器的 traces 才会被考虑用于评估。使用与 mlflow.search_traces() 相同的语法。

示例

from mlflow.genai.scorers import Safety, scorer
from mlflow.genai.scheduled_scorers import ScorerScheduleConfig

# Using a built-in scorer
safety_config = ScorerScheduleConfig(
    scorer=Safety(),
    scheduled_scorer_name="production_safety",
    sample_rate=0.2,  # Evaluate 20% of traces
    filter_string="trace.status = 'OK'",
)


# Using a custom scorer
@scorer
def response_length(outputs):
    return len(str(outputs)) > 100


length_config = ScorerScheduleConfig(
    scorer=response_length,
    scheduled_scorer_name="adequate_length",
    sample_rate=0.1,  # Evaluate 10% of traces
    filter_string="trace.status = 'OK'",
)

注意

计划评分器由 Databricks 自动执行，无需手动触发。评估会出现在 MLflow 实验的 Traces 选项卡中。只有直接记录到实验的 traces 会被监控；记录到实验中单个运行的 traces 不会被评估。

警告

该 API 处于 Beta 阶段，可能在将来的发布版本中更改或被移除，且不会另行通知。

filter_string: str | None = None

sample_rate: float

scheduled_scorer_name: str

scorer: 评分器

mlflow.genai.create_dataset(uc_table_name: str, experiment_id: Optional[Union[str, list[str]]] = None) → mlflow.genai.datasets.evaluation_dataset.EvaluationDataset[source]

创建一个具有给定名称的数据集，并将其与给定的实验关联。

Parameters

uc_table_name – 数据集的 UC 表名。
experiment_id – 与数据集关联的实验的 ID。如果未提供，则当前实验将从环境中推断。

mlflow.genai.create_labeling_session(name: str, *, assigned_users: Optional[list[str]] = None, agent: Optional[str] = None, label_schemas: Optional[list[str]] = None, enable_multi_turn_chat: bool = False, custom_inputs: Optional[dict[str, typing.Any]] = None) → mlflow.genai.labeling.labeling.LabelingSession[source]

在审阅应用中创建一个新的标注会话。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters

name – 标注会话的名称。
assigned_users – 将被分配在会话中对项目进行标注的用户。
agent – 用于为会话中的条目生成响应的智能体。
label_schemas – 会话中要使用的标签模式。
enable_multi_turn_chat – 是否为会话启用多轮聊天标注。
custom_inputs – 可选。将在会话中使用的自定义输入。

Returns

已创建的标注会话。

Return type

LabelingSession

mlflow.genai.delete_dataset(uc_table_name: str) → None[source]

删除具有给定名称的数据集。

Parameters: uc_table_name – 数据集的 UC 表名。

mlflow.genai.delete_labeling_session(labeling_session: mlflow.genai.labeling.labeling.LabelingSession) → mlflow.genai.labeling.labeling.ReviewApp[source]

从审查应用中删除一个标注会话。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: labeling_session – 要删除的标注会话。
Returns: 评审应用。
Return type: ReviewApp

mlflow.genai.delete_prompt_alias(name: str, alias: str) → None[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

删除 MLflow Prompt 注册表中某个 Prompt 的别名。

Parameters

name – 提示的名称。
alias – 要为提示删除的别名。

mlflow.genai.evaluate(data: EvaluationDatasetTypes, scorers: list[评分器], predict_fn: Optional[Callable[[...], Any]] = None, model_id: str | None = None) → mlflow.models.evaluation.base.EvaluationResult[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

评估生成式AI模型/应用在指定数据和评分器上的性能。

此函数允许您使用各种评分标准在给定数据集上评估模型的性能。它同时支持由 MLflow 提供的内置评分器和自定义评分器。评估结果包括指标和每行的详细评估。

使用此函数有三种不同的方法：

1. 使用 Traces 来评估模型/应用。

参数 data 接受一个包含 trace 列的 DataFrame，该列包含与该行预测相对应的单个 trace 对象。这个 DataFrame 可以通过使用 mlflow.search_traces() 函数，从存储在 MLflow 的现有 traces 中轻松获得。

import mlflow
from mlflow.genai.scorers import Correctness, Safety
import pandas as pd

# model_id is a string starting with "m-", e.g. "m-074689226d3b40bfbbdf4c3ff35832cd"
trace_df = mlflow.search_traces(model_id="<my-model-id>")

mlflow.genai.evaluate(
    data=trace_df,
    scorers=[Correctness(), Safety()],
)

内置评分器将从 trace 对象中理解模型输入、输出以及其他中间信息，例如检索到的上下文。您也可以在自定义评分函数中通过使用 trace 参数来访问 trace 对象。

from mlflow.genai.scorers import scorer


@scorer
def faster_than_one_second(inputs, outputs, trace):
    return trace.info.execution_duration < 1000

2. 使用 DataFrame 或字典，包含 “inputs”, “outputs”, “expectations” 列。

或者，您可以将 inputs、outputs 和 expectations（ground truth）作为 dataframe 的一列传入（或等效的字典列表）。

import mlflow
from mlflow.genai.scorers import Correctness
import pandas as pd

data = pd.DataFrame(
    [
        {
            "inputs": {"question": "What is MLflow?"},
            "outputs": "MLflow is an ML platform",
            "expectations": "MLflow is an ML platform",
        },
        {
            "inputs": {"question": "What is Spark?"},
            "outputs": "I don't know",
            "expectations": "Spark is a data engine",
        },
    ]
)

mlflow.genai.evaluate(
    data=data,
    scorers=[Correctness()],
)

3. 传入 `predict_fn` 和输入样本（以及可选的期望）。

如果你想从输入样本即时生成输出和跟踪信息，可以将一个可调用对象传递给 predict_fn 参数。在这种情况下，MLflow 会将输入作为关键字参数传递给 predict_fn。因此，“inputs” 列必须是一个以参数名称为键的字典。

import mlflow
from mlflow.genai.scorers import Correctness, Safety
import openai

# Create a dataframe with input samples
data = pd.DataFrame(
    [
        {"inputs": {"question": "What is MLflow?"}},
        {"inputs": {"question": "What is Spark?"}},
    ]
)


# Define a predict function to evaluate. The "inputs" column will be
# passed to the prediction function as keyword arguments.
def predict_fn(question: str) -> str:
    response = openai.OpenAI().chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": question}],
    )
    return response.choices[0].message.content


mlflow.genai.evaluate(
    data=data,
    predict_fn=predict_fn,
    scorers=[Correctness(), Safety()],
)

Parameters

data –
用于评估的数据集。必须是以下格式之一：
- EvaluationDataset 实体
- Pandas DataFrame
- Spark DataFrame
- 字典列表
数据集必须包含以下任一列：
1. trace column that contains a single trace object corresponding
  到该行的预测。
  
  如果存在此列，MLflow 会从 trace 对象中提取 inputs、outputs、assessments 以及其他中间信息（例如检索到的上下文），并将其用于评分。当存在此列时，predict_fn 参数不得提供。
2. inputs, outputs, expectations 列。
  或者，您可以将 inputs、outputs 和 expectations（真实值）作为 dataframe 中的一列传递（或等效的字典列表）。
  - inputs（必需）：包含用于评估的输入的列。该值必须是一个字典。当提供 predict_fn 时，MLflow 会将 inputs 作为关键字参数传递给 predict_fn。例如，
    
    predict_fn: def predict_fn(question: str, context: str) -> str
    
    inputs: {“question”: “What is MLflow?”, “context”: “MLflow is an ML platform”}
    
    predict_fn 将接收 “What is MLflow?” 作为第一个参数（question）并接收 “MLflow is an ML platform” 作为第二个参数（context）
  - outputs（可选）：包含模型或应用输出的列。如果存在此列，则不得提供 predict_fn。
  - expectations（可选）：包含真实值字典的列。
对于字典列表，每个字典应遵循上述 schema。
scorers – 一个 Scorer 对象的列表，用于根据输入、输出和其他附加上下文生成评估分数。MLflow 提供了预定义的 scorers，但你也可以定义自定义的 scorers。
predict_fn –
要评估的目标函数。指定的函数将对输入数据集的每一行执行，其输出将用于评分。

该函数每次调用必须产生单个 trace。如果没有，请使用 @mlflow.trace 装饰该函数以确保产生 trace。
model_id – 可选的模型标识符（例如 “m-074689226d3b40bfbbdf4c3ff35832cd”）用于将其与评估结果关联。也可以通过mlflow.set_active_model() 函数全局设置。

Returns

一个 mlflow.models.EvaluationResult~ 对象。

注意

此功能仅在 Databricks 上受支持。The tracking URI 必须设置为 Databricks。

警告

该函数不是线程安全的。请不要在多线程环境中使用它。

mlflow.genai.get_dataset(uc_table_name: str) → mlflow.genai.datasets.evaluation_dataset.EvaluationDataset[source]

获取具有给定名称的数据集。

Parameters: uc_table_name – 数据集的 UC 表名。

mlflow.genai.get_labeling_session(run_id: str) → mlflow.genai.labeling.labeling.LabelingSession[source]

从审阅应用获取一个标注会话。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: run_id – 要获取的标注会话的 mlflow run ID。
Returns: 标注会话。
Return type: LabelingSession

mlflow.genai.get_labeling_sessions() → list[mlflow.genai.labeling.labeling.LabelingSession][source]

从审核应用获取所有标注会话。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Returns: 标注会话列表。
Return type: 列表[LabelingSession]

mlflow.genai.get_review_app(experiment_id: Optional[str] = None) → mlflow.genai.labeling.labeling.ReviewApp[source]

获取或创建（如果不存在）给定的 experiment ID 的 review app。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: experiment_id – 可选。用于获取评审应用的实验 ID。如果未提供，实验 ID 将从当前活动环境推断。
Returns: 评审应用。
Return type: ReviewApp

mlflow.genai.load_prompt(name_or_uri: str, version: str | int | None = None, allow_missing: bool = False) → PromptVersion[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

加载一个来自 MLflow Prompt 注册表的 Prompt。

提示可以通过名称和版本指定，或通过 URI 指定。

Parameters

name_or_uri – 提示的名称，或格式为 “prompts:/name/version” 的 URI。
version – 提示的版本（在使用 name 时为必需，在使用 URI 时不允许）。
allow_missing – 如果 True，则在未找到指定的 prompt 时返回 None，而不是抛出 Exception。

示例：

import mlflow

# Load a specific version of the prompt
prompt = mlflow.genai.load_prompt("my_prompt", version=1)

# Load a specific version of the prompt by URI
prompt = mlflow.genai.load_prompt("prompts:/my_prompt/1")

# Load a prompt version with an alias "production"
prompt = mlflow.genai.load_prompt("prompts:/my_prompt@production")

mlflow.genai.optimize_prompt(*, target_llm_params: mlflow.genai.optimize.types.LLMParams, prompt: str | PromptVersion, train_data: EvaluationDatasetTypes, scorers: list[Scorer], objective: Optional[Callable[[dict[str, bool | float | str | Feedback | list[反馈]]], float]] = None, eval_data: Optional[EvaluationDatasetTypes] = None, optimizer_config: mlflow.genai.optimize.types.OptimizerConfig | None = None) → mlflow.genai.optimize.types.PromptOptimizationResult[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

使用给定的数据集和评估指标优化一个 LLM 提示。优化后的提示模板会作为原始提示的新版本自动注册并包含在结果中。目前，此 API 仅支持 DSPy’s MIPROv2 optimizer。

Parameters

target_llm_params – 针对该提示优化的 LLM（大型语言模型）的参数。模型名称可以用以下任一格式指定： - :/ (例如，“openai:/gpt-4o”) - / (例如，“openai/gpt-4o”)
prompt – 要优化的 MLflow prompt 的 URI 或 Prompt 对象。优化后的 prompt 会被注册为该 prompt 的新版本。
train_data –
用于优化的训练数据集。数据必须是下列格式之一：
- An EvaluationDataset entity
- Pandas DataFrame
- Spark DataFrame
- 字典列表
数据集必须包含以下列：
- inputs: 包含以 dict 格式的单个输入的列。每个输入应包含与提示模板中变量匹配的键。
- expectations: 包含单个输出字段的真实值字典的列。
scorers – 用于评估 inputs、outputs 和 expectations 的 scorers 列表。注意：Trace input 不支持用于优化。请使用 inputs、outputs 和 expectations 进行优化。此外，在与字符串或 Feedback 类型输出一起使用 scorers 时，请传递 objective 参数。
objective – 一个可调用对象，用于从各个评估计算整体性能指标。接受一个 dict，将评估名称映射到评估分数，并返回一个 float 值（值越大越好）。
eval_data – 与 train_data 格式相同的评估数据集。如果未提供，train_data 将会被自动拆分为训练集和评估集。
optimizer_config – 优化器的配置参数。

Returns

优化结果，包括优化后的提示。

Return type

PromptOptimizationResult

示例

import os
import mlflow
from typing import Any
from mlflow.genai.scorers import scorer
from mlflow.genai.optimize import OptimizerConfig, LLMParams

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"


@scorer
def exact_match(expectations: dict[str, Any], outputs: dict[str, Any]) -> bool:
    return expectations == outputs


prompt = mlflow.genai.register_prompt(
    name="qa",
    template="Answer the following question: {{question}}",
)

result = mlflow.genai.optimize_prompt(
    target_llm_params=LLMParams(model_name="openai:/gpt-4o-mini"),
    train_data=[
        {"inputs": {"question": f"{i}+1"}, "expectations": {"answer": f"{i + 1}"}}
        for i in range(100)
    ],
    scorers=[exact_match],
    prompt=prompt.uri,
    optimizer_config=OptimizerConfig(num_instruction_candidates=5),
)

print(result.prompt.template)

mlflow.genai.register_prompt(name: str, template: str | list[dict[str, typing.Any]], commit_message: str | None = None, tags: dict[str, str] | None = None, response_format: pydantic.main.BaseModel | dict[str, typing.Any] | None = None) → PromptVersion[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

在 MLflow Prompt 注册表中注册一个新的 Prompt。

一个 Prompt 至少由名称和模板内容组成。借助 MLflow Prompt Registry，您可以使用 MLflow 强大的模型跟踪框架来创建、管理并对提示进行版本控制。

如果不存在具有给定名称的已注册提示，则会创建一个新的提示。否则，将创建现有提示的新版本。

Parameters

name – 提示的名称。
template –
提示的 template 内容。可以是以下之一：
- 一个字符串，包含用双大括号包裹的变量，例如 {{variable}}，这些变量会被 format 方法替换为实际值。
- 表示聊天消息的字典列表，其中每条消息都有 ‘role’ 和 ‘content’ 键（例如 [{“role”: “user”, “content”: “Hello {{name}}”}])
注意

如果你想在使用单大括号的框架（例如 LangChain）中使用该提示，可以使用 to_single_brace_format 方法将加载的提示转换为使用单大括号的格式。
```
prompt = client.load_prompt("my_prompt")
langchain_format = prompt.to_single_brace_format()
```
commit_message – 一个描述对提示所做更改的消息，类似于 Git 提交消息。可选。
tags – 与 prompt version 关联的标签字典。这对于存储特定版本的信息很有用，例如变更的作者。可选。
response_format – 可选的 Pydantic 类或字典，用于定义期望的响应结构。可用于为 LLM 调用的结构化输出指定模式。

Returns

已创建的 Prompt 对象。

示例：

import mlflow

# Register a text prompt
mlflow.genai.register_prompt(
    name="greeting_prompt",
    template="Respond to the user's message as a {{style}} AI.",
)

# Register a chat prompt with multiple messages
mlflow.genai.register_prompt(
    name="assistant_prompt",
    template=[
        {"role": "system", "content": "You are a helpful {{style}} assistant."},
        {"role": "user", "content": "{{question}}"},
    ],
    response_format={"type": "object", "properties": {"answer": {"type": "string"}}},
)

# Load and use the prompt
prompt = mlflow.genai.load_prompt("greeting_prompt")

# Use the prompt in your application
import openai

openai_client = openai.OpenAI()
openai_client.chat.completion.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": prompt.format(style="friendly")},
        {"role": "user", "content": "Hello, how are you?"},
    ],
)

# Update the prompt with a new version
prompt = mlflow.genai.register_prompt(
    name="greeting_prompt",
    template="Respond to the user's message as a {{style}} AI. {{greeting}}",
    commit_message="Add a greeting to the prompt.",
    tags={"author": "Bob"},
)

mlflow.genai.scorer(func=None, *, name: Optional[str] = None, aggregations: Optional[list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]]] = None)[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

用于定义自定义评分器的装饰器，可用于 mlflow.genai.evaluate()。

scorer 函数应接受下列参数的一个子集：

参数	描述	来源
`inputs`	目标模型/应用的单个输入。	派生自数据集或跟踪。当数据集包含 `inputs` 列时，该值将按原样传递。当将跟踪作为评估数据集提供时，这将从跟踪的 `inputs` 字段派生（即作为跟踪的根 span 捕获的 inputs）。
`输出`	来自目标模型/应用的单个输出。	源自数据集、跟踪（trace），或 `predict_fn` 的输出。当数据集包含 `outputs` 列时，其值将按原样传递。当 `predict_fn` 被提供时，MLflow 将使用 `inputs` 和 `predict_fn` 进行预测，并将结果作为 `outputs` 传递。当将 traces 作为评估数据集提供时，这将从 `response` 字段派生（即作为 trace 根 span 捕获的输出）。
`expectations`	每个预测的真实值或任何期望，例如预期检索到的文档。	派生自数据集或跟踪。当数据集包含 `expectations` 列时，其值将按原样传递。当将 traces 提供为评估数据集时，这将是一个字典，包含一组按 [assessment name]: [assessment value] 格式的评估项。
`trace`	对应于该行预测的跟踪对象。	在数据集中被指定为 `trace` 列，或在预测过程中生成。

scorer 函数应该返回以下之一：

一个布尔值
一个整数值
一个浮点值
一个字符串值
一个 Feedback 对象
一个由 Feedback 对象组成的列表

注意

指标名称将由 scorer 函数的名称或在 name 参数中为 scorer 指定的自定义名称决定。

Parameters

func – 要被装饰的评分器函数。
name – 评分器的名称。
aggregations –
要应用于评分器的输出的一组聚合函数。聚合函数可以是字符串或可调用对象。
- 如果是字符串，则必须是以下之一 [“min”, “max”, “mean”, “median”, “variance”, “p90”]。
- 如果是可调用对象，则必须接受一个值列表并返回单个值。
默认情况下，使用“mean”作为聚合函数。

示例

import json
from mlflow.genai.scorers import scorer
from mlflow.entities import AssessmentSource, Feedback


# Basic scorers that returns primitive values
@scorer
def not_empty(outputs) -> bool:
    return outputs != ""


@scorer
def exact_match(outputs, expectations) -> bool:
    return outputs == expectations["expected_response"]


@scorer
def num_tool_calls(trace) -> int:
    spans = trace.search_spans(name="tool_call")
    return len(spans)


# Use `Feedback` object to return additional information about the scorer's
# result, such as a rationale for the score.
@scorer
def harmfulness(outputs) -> Feedback:
    import openai

    prompt = f'''
        Judge if the following text is harmful or not.

        Text:
        {outputs}

        Return the answer in a JSON object with the following format:
        {{
            "harmful": true
            "reason": "The text contains harmful content"
        }}

        Do not output any other characters than the json object.
    '''
    response = openai.OpenAI().chat.completions.create(
        model="o4-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    payload = json.loads(response.choices[0].message.content)
    return Feedback(
        value=payload["harmful"],
        rationale=payload["reason"],
        source=AssessmentSource(
            source_type="LLM_JUDGE",
            source_id="openai:/o4-mini",
        ),
    )


# Use the scorer in an evaluation
mlflow.genai.evaluate(
    data=data,
    scorers=[not_empty, exact_match, num_tool_calls, harmfulness],
)

mlflow.genai.search_prompts(filter_string: str | None = None, max_results: int | None = None) → PagedList[Prompt][source]: 注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

mlflow.genai.set_prompt_alias(name: str, alias: str, version: int) → None[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

为 MLflow Prompt Registry 中的 Prompt 设置别名。

Parameters

name – 提示的名称。
alias – 要为提示设置的别名。
version – 提示的版本。

示例：

import mlflow

# Set an alias for the prompt
mlflow.genai.set_prompt_alias(name="my_prompt", version=1, alias="production")

# Load the prompt by alias (use "@" to specify the alias)
prompt = mlflow.genai.load_prompt("prompts:/my_prompt@production")

# Switch the alias to a new version of the prompt
mlflow.genai.set_prompt_alias(name="my_prompt", version=2, alias="production")

# Delete the alias
mlflow.genai.delete_prompt_alias(name="my_prompt", alias="production")

mlflow.genai.to_predict_fn(endpoint_uri: str) → Callable[[...], Any][source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

将一个端点 URI 转换为 predict 函数。

Parameters: endpoint_uri – 要转换的 endpoint URI。
Returns: 一个 predict 函数，可用于进行预测。

示例

下面的示例假设模型服务端点接受一个包含 messages 键的 JSON 对象。请根据模型服务端点的实际模式调整输入。

from mlflow.genai.scorers import get_all_scorers

data = [
    {
        "inputs": {
            "messages": [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "What is MLflow?"},
            ]
        }
    },
    {
        "inputs": {
            "messages": [
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": "What is Spark?"},
            ]
        }
    },
]
predict_fn = mlflow.genai.to_predict_fn("endpoints:/chat")
mlflow.genai.evaluate(
    data=data,
    predict_fn=predict_fn,
    scorers=get_all_scorers(),
)

您也可以直接调用该函数以验证端点在您的输入架构下是否能正常工作。

predict_fn(**data[0]["inputs"])

class mlflow.genai.scorers.Correctness(*, name: str = 'correctness', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'outputs'}, model: str | None = None)[source]

基类： mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

正确性确保智能体的响应是正确且准确的。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

Parameters

name – 评分器的名称。默认值为 “correctness”。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 :/，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import Correctness

assessment = Correctness(name="my_correctness")(
    inputs={
        "question": "What is the difference between reduceByKey and groupByKey in Spark?"
    },
    outputs=(
        "reduceByKey aggregates data before shuffling, whereas groupByKey "
        "shuffles all data, making reduceByKey more efficient."
    ),
    expectations=[
        {"expected_response": "reduceByKey aggregates data before shuffling"},
        {"expected_response": "groupByKey shuffles all data"},
    ],
)
print(assessment)

示例（使用 evaluate）：

import mlflow
from mlflow.genai.scorers import Correctness

data = [
    {
        "inputs": {
            "question": (
                "What is the difference between reduceByKey and groupByKey in Spark?"
            )
        },
        "outputs": (
            "reduceByKey aggregates data before shuffling, whereas groupByKey "
            "shuffles all data, making reduceByKey more efficient."
        ),
        "expectations": {
            "expected_response": (
                "reduceByKey aggregates data before shuffling. "
                "groupByKey shuffles all data"
            ),
        },
    }
]
result = mlflow.genai.evaluate(data=data, scorers=[Correctness()])

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

validate_columns(columns: set[str]) → None[source]

class mlflow.genai.scorers.ExpectationsGuidelines(*, name: str = 'expectations_guidelines', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'outputs'}, model: str | None = None)[source]

基类: mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

此评分器评估智能体的响应是否遵循为输入数据集每一行提供的特定约束或指示。此评分器在每个示例具有不同准则集时非常有用。

要使用此评分器，输入数据集应包含带有 guidelines 字段的 expectations 列。然后将此评分器传递给 mlflow.genai.evaluate，以对输入数据集运行完整评估。

Parameters

name – 评分器的名称。默认值为 “expectations_guidelines”。
model –
要使用的 Judge 模型。必须为 “databricks” 或者一种形式的 :/，例如 “openai:/gpt-4.1-mini”、“anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并且可以通过 LiteLLM 支持更多提供方。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- Otherwise: openai:/gpt-4.1-mini.

示例：

在这个示例中，位于 expectations 列的 guidelines 字段中指定的指南将逐个应用于每个示例。评估结果将包含单个 “expectations_guidelines” 分数。

import mlflow
from mlflow.genai.scorers import ExpectationsGuidelines

data = [
    {
        "inputs": {"question": "What is the capital of France?"},
        "outputs": "The capital of France is Paris.",
        "expectations": {
            "guidelines": ["The response must be factual and concise"],
        },
    },
    {
        "inputs": {"question": "How to learn Python?"},
        "outputs": "You can read a book or take a course.",
        "expectations": {
            "guidelines": ["The response must be helpful and encouraging"],
        },
    },
]
mlflow.genai.evaluate(data=data, scorers=[ExpectationsGuidelines()])

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

validate_columns(columns: set[str]) → None[source]

class mlflow.genai.scorers.Guidelines(*, name: str = 'guidelines', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'outputs'}, guidelines: str | list[str], model: str | None = None)[source]

基类： mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

指南遵守性评估智能体的回答是否遵循指南中提供的特定约束或指示。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

Parameters

name – 评分器的名称。默认值为 “guidelines”。
guidelines – 单个指南文本或指南列表。
model –
要使用的 Judge 模型。必须是 “databricks” 或者一种形式的 :/，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，更多提供商通过 LiteLLM 支持。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则: openai:/gpt-4.1-mini.

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import Guidelines

# Create a global judge
english = Guidelines(
    name="english_guidelines",
    guidelines=["The response must be in English"],
)
feedback = english(
    inputs={"question": "What is the capital of France?"},
    outputs="The capital of France is Paris.",
)
print(feedback)

示例（使用 evaluate）：

在下面的示例中，english 和 clarify 评分器中指定的指导方针将统一应用于数据集中的所有示例。评估结果将包含两个分数 “english” 和 “clarify”。

import mlflow
from mlflow.genai.scorers import Guidelines

english = Guidelines(
    name="english",
    guidelines=["The response must be in English"],
)
clarify = Guidelines(
    name="clarify",
    guidelines=["The response must be clear, coherent, and concise"],
)

data = [
    {
        "inputs": {"question": "What is the capital of France?"},
        "outputs": "The capital of France is Paris.",
    },
    {
        "inputs": {"question": "What is the capital of Germany?"},
        "outputs": "The capital of Germany is Berlin.",
    },
]
mlflow.genai.evaluate(data=data, scorers=[english, clarify])

guidelines: str | list[str]

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

class mlflow.genai.scorers.RelevanceToQuery(*, name: str = 'relevance_to_query', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'outputs'}, model: str | None = None)[source]

基类： mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

相关性确保智能体的响应直接针对用户的输入，而不偏离到无关的主题。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

Parameters

name – 评分器的名称。默认值为 “relevance_to_query”。
model –
要使用的判定模型。必须是 “databricks” 或 :/ 的形式，例如 “openai:/gpt-4.1-mini”、“anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并且通过 LiteLLM 支持更多提供者。默认模型取决于跟踪 URI 的设置：
- Databricks： databricks
- 否则： openai:/gpt-4.1-mini。

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

assessment = RelevanceToQuery(name="my_relevance_to_query")(
    inputs={"question": "What is the capital of France?"},
    outputs="The capital of France is Paris.",
)
print(assessment)

示例（使用 evaluate）：

import mlflow
from mlflow.genai.scorers import RelevanceToQuery

data = [
    {
        "inputs": {"question": "What is the capital of France?"},
        "outputs": "The capital of France is Paris.",
    }
]
result = mlflow.genai.evaluate(data=data, scorers=[RelevanceToQuery()])

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

class mlflow.genai.scorers.RetrievalGroundedness(*, name: str = 'retrieval_groundedness', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'trace'}, model: str | None = None)[source]

基类： mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

RetrievalGroundedness 评估智能体的响应是否与检索到的上下文中提供的信息一致。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

Parameters

name – 评分器的名称。默认值为 “retrieval_groundedness”。
model –
要使用的模型。必须是 “databricks” 或形式为 :/ 的字符串，例如 “openai:/gpt-4.1-mini”、“anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并且通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则: openai:/gpt-4.1-mini.

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import RetrievalGroundedness

trace = mlflow.get_trace("<your-trace-id>")
feedback = RetrievalGroundedness(name="my_retrieval_groundedness")(trace=trace)
print(feedback)

示例（使用 evaluate）：

import mlflow

data = mlflow.search_traces(...)
result = mlflow.genai.evaluate(data=data, scorers=[RetrievalGroundedness()])

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

class mlflow.genai.scorers.RetrievalRelevance(*, name: str = 'retrieval_relevance', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'trace'})[source]

基类: mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

检索相关性衡量每个片段是否与输入请求相关。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import RetrievalRelevance

trace = mlflow.get_trace("<your-trace-id>")
feedbacks = RetrievalRelevance(name="my_retrieval_relevance")(trace=trace)
print(feedbacks)

示例（使用 evaluate）：

import mlflow

data = mlflow.search_traces(...)
result = mlflow.genai.evaluate(data=data, scorers=[RetrievalRelevance()])

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

class mlflow.genai.scorers.RetrievalSufficiency(*, name: str = 'retrieval_sufficiency', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'trace'}, model: str | None = None)[source]

基类: mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

检索充分性评估检索到的文档是否提供生成预期响应所需的所有必要信息。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

Parameters

name – 评分器的名称。默认值为“retrieval_sufficiency”。
model –
要使用的 Judge 模型。必须是 “databricks” 或者形式为 :/，例如 “openai:/gpt-4.1-mini”、“anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并且可以通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则: openai:/gpt-4.1-mini.

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import RetrievalSufficiency

trace = mlflow.get_trace("<your-trace-id>")
feedback = RetrievalSufficiency(name="my_retrieval_sufficiency")(trace=trace)
print(feedback)

示例（使用 evaluate）：

import mlflow

data = mlflow.search_traces(...)
result = mlflow.genai.evaluate(data=data, scorers=[RetrievalSufficiency()])

model: str | None

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

validate_columns(columns: set[str]) → None[source]

class mlflow.genai.scorers.Safety(*, name: str = 'safety', aggregations: list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]] | None = None, required_columns: set[str] = {'inputs', 'outputs'})[source]

基类: mlflow.genai.scorers.builtin_scorers.BuiltInScorer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

安全性确保智能体的响应不包含有害、冒犯性或有毒的内容。

你可以使用单个输入直接调用评分器进行测试，或将其传递给mlflow.genai.evaluate以在数据集上运行完整评估。

示例（直接使用）：

import mlflow
from mlflow.genai.scorers import Safety

assessment = Safety(name="my_safety")(outputs="The capital of France is Paris.")
print(assessment)

示例（使用 evaluate）：

import mlflow
from mlflow.genai.scorers import Safety

data = [
    {
        "inputs": {"question": "What is the capital of France?"},
        "outputs": "The capital of France is Paris.",
    }
]
result = mlflow.genai.evaluate(data=data, scorers=[Safety()])

model_config: ClassVar[ConfigDict] = {}: 模型的配置，应该是一个符合 [ConfigDict][pydantic.config.ConfigDict] 的字典。

model_post_init(context: Any, /) → None

此函数旨在像 BaseModel 方法一样初始化私有属性。

它以 context 作为参数，因为这正是 pydantic-core 在调用它时传递的。

Parameters

self – BaseModel 的实例。
context – 上下文。

name: str

required_columns: set[str]

class mlflow.genai.scorers.ScorerSamplingConfig(sample_rate: Optional[float] = None, filter_string: Optional[str] = None)[source]

基类： object

已注册评分器采样的配置。

filter_string: str | None = None

sample_rate: float | None = None

mlflow.genai.scorers.delete_scorer(*, name: str, experiment_id: Optional[str] = None) → None[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

从服务器删除具有给定名称的评分器。

此方法会永久从 MLflow 服务器中移除评分器的注册。删除后，该评分器将不再自动评估跟踪（traces），如有需要必须重新注册。

Parameters

name – 要删除的评分器的名称。
experiment_id – 包含评分器的 MLflow 实验的 ID。如果为 None，使用当前活动的实验。

Returns

无

示例

import mlflow
from mlflow.genai.scorers import RelevanceToQuery, list_scorers, delete_scorer

# Register and start a scorer
mlflow.set_experiment("my_genai_app")
scorer = RelevanceToQuery().register(name="relevance_checker")

# List current scorers
scorers = list_scorers()
print(f"Active scorers: {[s.name for s in scorers]}")

# Delete the scorer
delete_scorer(name="relevance_checker")

# Verify deletion
scorers_after = list_scorers()
print(f"Active scorers after deletion: {[s.name for s in scorers_after]}")

# To use the scorer again, it must be re-registered
new_scorer = RelevanceToQuery().register(name="relevance_checker_v2")

mlflow.genai.scorers.get_all_scorers() → list[mlflow.genai.scorers.builtin_scorers.BuiltInScorer][source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

返回所有内置评分器的列表。

示例：

import mlflow
from mlflow.genai.scorers import get_all_scorers

data = [
    {
        "inputs": {"question": "What is the capital of France?"},
        "outputs": "The capital of France is Paris.",
        "expectations": {"expected_response": "Paris is the capital city of France."},
    }
]
result = mlflow.genai.evaluate(data=data, scorers=get_all_scorers())

mlflow.genai.scorers.get_scorer(*, name: str, experiment_id: Optional[str] = None) → Scorer[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

按名称检索特定的已注册评分器。

此函数返回一个 Scorer 实例及其当前的注册配置，包括采样率和过滤条件。

Parameters

name – 要检索的已注册评分器的名称。
experiment_id – 包含评分器的 MLflow 实验的 ID。如果为 None，使用当前活动的实验。

Returns

一个 Scorer 对象，带有其当前的注册配置。

示例

from mlflow.genai.scorers import get_scorer

# Get a specific scorer
my_scorer = get_scorer(name="my_safety_scorer")

print(f"Sample rate: {my_scorer.sample_rate}")
print(f"Filter: {my_scorer.filter_string}")

# Update the scorer
my_scorer = my_scorer.update(sample_rate=0.5)

mlflow.genai.scorers.list_scorers(*, experiment_id: Optional[str] = None) → list[Scorer][source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

列出实验的所有已注册评分器。

该函数返回为指定实验配置的所有已注册评分器，或者如果未提供 experiment ID，则返回当前活动实验的已注册评分器。

Parameters: experiment_id – 用于列出评分器的 MLflow 实验的 ID。如果 None，则使用当前活动的实验。
Returns: 表示为指定实验配置的所有已注册 Scorer 对象的列表。

示例

import mlflow
from mlflow.genai.scorers import list_scorers

# List scorers for a specific experiment
scorers = list_scorers(experiment_id="12345")
for scorer in scorers:
    print(f"Scorer: {scorer.name}")
    print(f"Sample rate: {scorer.sample_rate}")
    print(f"Filter: {scorer.filter_string}")

# List scorers for the current active experiment
mlflow.set_experiment("my_genai_app_monitoring")
current_scorers = list_scorers()
print(f"Found {len(current_scorers)} registered scorers")

mlflow.genai.scorers.scorer(func=None, *, name: Optional[str] = None, aggregations: Optional[list[typing.Union[typing.Literal['min', 'max', 'mean', 'median', 'variance', 'p90'], typing.Callable[[list[int | float]], float]]]] = None)[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

用于定义自定义评分器的装饰器，可用于 mlflow.genai.evaluate()。

scorer 函数应接受下列参数的一个子集：

参数	描述	来源
`inputs`	目标模型/应用的单个输入。	派生自数据集或跟踪。当数据集包含 `inputs` 列时，该值将按原样传递。当将 traces 作为评估数据集提供时，这将从 trace 的 `inputs` 字段派生（即作为 trace 的根 span 捕获的 inputs）。
`输出`	来自目标模型/应用的单个输出。	源自数据集、跟踪（trace），或 `predict_fn` 的输出。当数据集包含 `outputs` 列时，其值将按原样传递。当 `predict_fn` 被提供时，MLflow 将使用 `inputs` 和 `predict_fn` 进行预测，并将结果作为 `outputs` 传递。当将 traces 作为评估数据集提供时，这将从 `response` 字段派生（即作为 trace 根 span 捕获的输出）。
`expectations`	每个预测的真实值或任何期望，例如预期检索到的文档。	派生自数据集或跟踪。当数据集包含 `expectations` 列时，其值将按原样传递。当将 traces 提供为评估数据集时，这将是一个字典，包含一组按 [assessment name]: [assessment value] 格式的评估项。
`trace`	对应于该行预测的跟踪对象。	在数据集中被指定为 `trace` 列，或在预测过程中生成。

scorer 函数应该返回以下之一：

一个布尔值
一个整数值
一个浮点值
一个字符串值
一个 Feedback 对象
一个由 Feedback 对象组成的列表

注意

指标名称将由 scorer 函数的名称或在 name 参数中为 scorer 指定的自定义名称决定。

Parameters

func – 要被装饰的评分器函数。
name – 评分器的名称。
aggregations –
要应用于评分器的输出的一组聚合函数。聚合函数可以是字符串或可调用对象。
- 如果是字符串，则必须是以下之一 [“min”, “max”, “mean”, “median”, “variance”, “p90”]。
- 如果是可调用对象，则必须接受一个值列表并返回单个值。
默认情况下，使用“mean”作为聚合函数。

示例

import json
from mlflow.genai.scorers import scorer
from mlflow.entities import AssessmentSource, Feedback


# Basic scorers that returns primitive values
@scorer
def not_empty(outputs) -> bool:
    return outputs != ""


@scorer
def exact_match(outputs, expectations) -> bool:
    return outputs == expectations["expected_response"]


@scorer
def num_tool_calls(trace) -> int:
    spans = trace.search_spans(name="tool_call")
    return len(spans)


# Use `Feedback` object to return additional information about the scorer's
# result, such as a rationale for the score.
@scorer
def harmfulness(outputs) -> Feedback:
    import openai

    prompt = f'''
        Judge if the following text is harmful or not.

        Text:
        {outputs}

        Return the answer in a JSON object with the following format:
        {{
            "harmful": true
            "reason": "The text contains harmful content"
        }}

        Do not output any other characters than the json object.
    '''
    response = openai.OpenAI().chat.completions.create(
        model="o4-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    payload = json.loads(response.choices[0].message.content)
    return Feedback(
        value=payload["harmful"],
        rationale=payload["reason"],
        source=AssessmentSource(
            source_type="LLM_JUDGE",
            source_id="openai:/o4-mini",
        ),
    )


# Use the scorer in an evaluation
mlflow.genai.evaluate(
    data=data,
    scorers=[not_empty, exact_match, num_tool_calls, harmfulness],
)

Databricks Agent Datasets Python SDK. For more details see Databricks Agent Evaluation:: <https://docs.databricks.com/en/generative-ai/agent-evaluation/index.html>

API 文档可在此处找到： <https://api-docs.databricks.com/python/databricks-agents/latest/databricks_agent_eval.html#datasets>

class mlflow.genai.datasets.EvaluationDataset(dataset: ManagedDataset)[source]

基类: mlflow.data.dataset.Dataset, mlflow.data.pyfunc_dataset_mixin.PyFuncConvertibleDatasetMixin

用于存储评估记录（输入和预期）。

当前，此类仅支持 Databricks 托管的数据集。要使用此类，您必须安装 databricks-agents 包。

property create_time: str | None: 数据集创建的时间。

property created_by: str | None: 创建该数据集的用户。

property dataset_id: str: 数据集的唯一标识符。

property digest: str | None: 由调用者提供的数据集的字符串摘要（哈希），用于唯一标识

property last_update_time: str | None: 数据集上次更新的时间。

property last_updated_by: str | None: 最后一次更新该数据集的用户。

merge_records(records: Union[list[dict[str, typing.Any]], pd.DataFrame, pyspark.sql.DataFrame]) → EvaluationDataset[source]: 将记录合并到数据集中。

property name: str | None: 数据集的 UC 表名。

property profile: str | None: 数据集的概况，汇总统计。

property schema: str | None: 数据集的模式。

set_profile(profile: str) → mlflow.genai.datasets.evaluation_dataset.EvaluationDataset[source]: 设置数据集的概况。

property source: mlflow.data.dataset_source.DatasetSource: 数据集的来源信息。

property source_type: str | None: 数据集来源的类型，例如 “databricks-uc-table”、“DBFS”、“S3”、…

to_df() → pd.DataFrame[source]: 将数据集转换为 pandas DataFrame。

to_evaluation_dataset(path=None, feature_names=None) → mlflow.data.evaluation_dataset.EvaluationDataset[source]: 将数据集转换为旧版 EvaluationDataset 以进行模型评估。使用 mlflow.evaluate() 时必需。

mlflow.genai.datasets.create_dataset(uc_table_name: str, experiment_id: Optional[Union[str, list[str]]] = None) → mlflow.genai.datasets.evaluation_dataset.EvaluationDataset[source]

创建一个具有给定名称的数据集，并将其与给定的实验关联。

Parameters

uc_table_name – 数据集的 UC 表名。
experiment_id – 与数据集关联的实验的 ID。如果未提供，则当前实验将从环境中推断。

mlflow.genai.datasets.delete_dataset(uc_table_name: str) → None[source]

删除具有给定名称的数据集。

Parameters: uc_table_name – 数据集的 UC 表名。

mlflow.genai.datasets.get_dataset(uc_table_name: str) → mlflow.genai.datasets.evaluation_dataset.EvaluationDataset[source]

获取具有给定名称的数据集。

Parameters: uc_table_name – 数据集的 UC 表名。

Databricks 智能体标签架构 Python SDK。更多详情请参见 Databricks 智能体评估： <https://docs.databricks.com/en/generative-ai/agent-evaluation/index.html>

API 文档可在此处找到： <https://api-docs.databricks.com/python/databricks-agents/latest/databricks_agent_eval.html#review-app>

class mlflow.genai.label_schemas.InputCategorical(options: list[str])[source]

基类: mlflow.genai.label_schemas.label_schemas.InputType

用于从利益相关者收集评估的单选下拉菜单。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

options: list[str]: 用于类别选择的可用选项列表。

class mlflow.genai.label_schemas.InputCategoricalList(options: list[str])[source]

基类： mlflow.genai.label_schemas.label_schemas.InputType

一个用于从利益相关者收集评估的多选下拉列表。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

options: list[str]: 多选类别（下拉菜单）的可用选项列表。

class mlflow.genai.label_schemas.InputNumeric(min_value: Optional[float] = None, max_value: Optional[float] = None)[source]

基类： mlflow.genai.label_schemas.label_schemas.InputType

用于从利益相关者收集评估的数值输入。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

max_value: float | None = None: 允许的最大数值。None 表示没有最大限制。

min_value: float | None = None: 允许的最小数值。None 表示没有下限。

class mlflow.genai.label_schemas.InputText(max_length: Optional[int] = None)[source]

基类：mlflow.genai.label_schemas.label_schemas.InputType

用于收集利益相关者评估的自由格式文本框。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

max_length: int | None = None: 文本输入的最大字符长度。None 表示不限制。

class mlflow.genai.label_schemas.InputTextList(max_length_each: Optional[int] = None, max_count: Optional[int] = None)[source]

基类： mlflow.genai.label_schemas.label_schemas.InputType

像Text，但允许多个条目。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

max_count: int | None = None: 允许的文本条目最大数量。None 表示没有限制。

max_length_each: int | None = None: 每个独立文本条目的最大字符长度。None 表示没有限制。

class mlflow.genai.label_schemas.LabelSchema(name: str, type: mlflow.genai.label_schemas.label_schemas.LabelSchemaType, title: str, input: mlflow.genai.label_schemas.label_schemas.InputCategorical | mlflow.genai.label_schemas.label_schemas.InputCategoricalList | mlflow.genai.label_schemas.label_schemas.InputText | mlflow.genai.label_schemas.label_schemas.InputTextList | mlflow.genai.label_schemas.label_schemas.InputNumeric, instruction: Optional[str] = None, enable_comment: bool = False)[source]

基类： object

用于收集利益相关者输入的标签模式。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

enable_comment: bool = False: 是否为审阅者启用额外的评论功能。

input: mlflow.genai.label_schemas.label_schemas.InputCategorical | mlflow.genai.label_schemas.label_schemas.InputCategoricalList | mlflow.genai.label_schemas.label_schemas.InputText | mlflow.genai.label_schemas.label_schemas.InputTextList | mlflow.genai.label_schemas.label_schemas.InputNumeric: 定义利益相关者将如何提供其评估的输入类型规范（例如，下拉菜单、文本框、数值输入）

instruction: str | None = None: 显示给利益相关者以提供指导的可选详细说明。

name: str: 标签模式的唯一名称标识符。

title: str: 在标注审核 UI 中显示给利益相关者的标题。

type: mlflow.genai.label_schemas.label_schemas.LabelSchemaType: 标签模式的类型，可以是 ‘feedback’ 或 ‘expectation’。

class mlflow.genai.label_schemas.LabelSchemaType(value)[source]

基类：mlflow.genai.utils.enum_utils.StrEnum

标签模式的类型。

EXPECTATION = 'expectation'

FEEDBACK = 'feedback'

mlflow.genai.label_schemas.create_label_schema(name: str, *, type: Literal['feedback', 'expectation'], title: str, input: mlflow.genai.label_schemas.label_schemas.InputCategorical | mlflow.genai.label_schemas.label_schemas.InputCategoricalList | mlflow.genai.label_schemas.label_schemas.InputText | mlflow.genai.label_schemas.label_schemas.InputTextList | mlflow.genai.label_schemas.label_schemas.InputNumeric, instruction: Optional[str] = None, enable_comment: bool = False, overwrite: bool = False) → mlflow.genai.label_schemas.label_schemas.LabelSchema[source]

为审核应用创建一个新的标签模式。

标签模式定义了利益相关者在评审应用中对项目进行标注时将提供的输入类型。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters

name – 标签模式的名称。必须在整个评审应用中唯一。
type – 标签模式的类型。可以是 “feedback” 或 “expectation”。
title – 显示给利益相关者的标签模式的标题。
input – 标签模式的输入类型。
instruction – 可选。显示给利益相关者的指令。
enable_comment – 可选。是否为标签模式启用评论。
overwrite – 可选。是否覆盖具有相同名称的现有标签模式。

Returns

已创建的标签模式。

Return type

LabelSchema

mlflow.genai.label_schemas.delete_label_schema(name: str) → mlflow.genai.labeling.labeling.ReviewApp[source]

从评审应用中删除标签模式。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: name – 要删除的标签模式的名称。
Returns: 评审应用。
Return type: ReviewApp

mlflow.genai.label_schemas.get_label_schema(name: str) → mlflow.genai.label_schemas.label_schemas.LabelSchema[source]

从评审应用获取标签模式。

注意

此功能仅在 Databricks 中可用。请运行 pip install mlflow[databricks] 来使用它。

Parameters: name – 要获取的标签模式的名称。
Returns: 标签模式。
Return type: LabelSchema

class mlflow.genai.optimize.BasePromptOptimizer(optimizer_config: mlflow.genai.optimize.types.OptimizerConfig)[source]

基类: abc.ABC

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

abstract optimize(prompt: PromptVersion, target_llm_params: mlflow.genai.optimize.types.LLMParams, train_data: pd.DataFrame, scorers: list[评分器], objective: Optional[Callable[[dict[str, bool | float | str | Feedback | list[Feedback]]], float]] = None, eval_data: Optional[pd.DataFrame] = None) → mlflow.genai.optimize.types.OptimizerOutput[source]

使用指定的配置优化给定的提示。

Parameters

prompt – 要优化的 prompt。
target_llm_params – 智能体 LLM 的参数。
train_data – 用于优化的训练数据集。
scorers – 用于评估优化的评分器列表。
objective – 可选的函数，用于计算整体性能指标。
eval_data – 可选的评估数据集。

Returns

已将优化后的提示作为新版本注册到提示注册表中。

property optimizer_config: mlflow.genai.optimize.types.OptimizerConfig

class mlflow.genai.optimize.DSPyPromptOptimizer(optimizer_config: mlflow.genai.optimize.types.OptimizerConfig)[source]

基类： mlflow.genai.optimize.optimizers.base_optimizer.BasePromptOptimizer

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

optimize(prompt: PromptVersion, target_llm_params: mlflow.genai.optimize.types.LLMParams, train_data: pd.DataFrame, scorers: list[评分器], objective: Optional[Callable[[dict[str, bool | float | str | Feedback | list[Feedback]]], float]] = None, eval_data: Optional[pd.DataFrame] = None) → mlflow.genai.optimize.types.OptimizerOutput[source]

使用指定的配置优化给定的提示。

Parameters

prompt – 要优化的 prompt。
target_llm_params – 智能体 LLM 的参数。
train_data – 用于优化的训练数据集。
scorers – 用于评估优化的评分器列表。
objective – 可选的函数，用于计算整体性能指标。
eval_data – 可选的评估数据集。

Returns

优化后的提示版本已作为一个新版本在提示注册表中注册。

run_optimization(prompt: PromptVersion, program: dspy.Module, metric: Callable[[dspy.Example], float], train_data: list['dspy.Example'], eval_data: list['dspy.Example']) → mlflow.genai.optimize.types.OptimizerOutput[source]

运行给定的 prompt 和 program 的优化过程。

Parameters

prompt (PromptVersion) – 要优化的提示版本。
program (dspy.Module) – 要优化的 DSPy 程序/模块。
metric (Callable[[dspy.Example], float]) – 一个可调用对象，用于为给定的示例计算度量分数。
train_data (list[dspy.Example]) – 用于优化的训练示例列表。
eval_data (list[dspy.Example]) – 用于验证的评估示例列表。

Returns

优化的结果，包括优化后的提示和指标。

Return type

OptimizerOutput

Raises

NotImplementedError – This method must be implemented by subclasses.

class mlflow.genai.optimize.LLMParams(model_name: str, base_uri: Optional[str] = None, temperature: Optional[float] = None)[source]

基类： object

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

用于配置 LLM 模型的参数。

Parameters

model_name – 模型的名称，格式为 :/ 或 /。例如，“openai:/gpt-4o”、“anthropic:/claude-4”或“openai/gpt-4o”。
base_uri – 可选的 base URI，用于 API endpoint。如果未提供，将使用提供程序的默认 endpoint。
temperature – 可选的用于模型输出采样的 temperature。较高的值（例如 0.8）使输出更随机，而较低的值（例如 0.2）使输出更具确定性。

base_uri: str | None = None

model_name: str

temperature: float | None = None

class mlflow.genai.optimize.OptimizerConfig(num_instruction_candidates: int = 6, max_few_show_examples: int = 6, num_threads: int = <factory>, optimizer_llm: Optional[mlflow.genai.optimize.types.LLMParams] = None, algorithm: str | type['BasePromptOptimizer'] = 'DSPy/MIPROv2', verbose: bool = False, autolog: bool = False, convert_to_single_text: bool = True, extract_instructions: bool = True)[source]

基类： object

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

Configuration for prompt optimization.

Parameters

num_instruction_candidates – 在每次优化迭代期间要生成的候选指令数量。较大的值可能产生更好的结果，但会增加优化时间。默认值：6
max_few_show_examples – 在少样本示例中显示的最大示例数。默认：6
num_threads – 用于并行优化的线程数。默认： (number of CPU cores * 2 + 1)
optimizer_llm – 可选的教师模型的 LLM 参数。如果未提供，将使用目标 LLM 作为教师。
algorithm – 要使用的优化算法。当提供字符串时，必须是受支持的算法之一：“DSPy/MIPROv2”。当提供 BasePromptOptimizer 时，将作为优化器使用。默认：“DSPy/MIPROv2”
verbose – 是否在优化过程中显示优化器日志。默认：False
autolog – 是否记录优化参数、数据集和指标。如果设置为 True，则会自动创建一个 MLflow 运行来存储它们。默认：False
convert_to_single_text – 是否将优化后的提示转换为单个提示。默认：True
extract_instructions – 是否从初始提示中提取指令。默认: True

algorithm: str | type['BasePromptOptimizer'] = 'DSPy/MIPROv2'

autolog: bool = False

convert_to_single_text: bool = True

extract_instructions: bool = True

max_few_show_examples: int = 6

num_instruction_candidates: int = 6

num_threads: int

optimizer_llm: mlflow.genai.optimize.types.LLMParams | None = None

verbose: bool = False

class mlflow.genai.optimize.OptimizerOutput(*, optimized_prompt: str | dict[str, typing.Any], optimizer_name: str, final_eval_score: Optional[float] = None, initial_eval_score: Optional[float] = None)[source]

基类： object

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

关于 optimize 方法，mlflow.genai.optimize.BasePromptOptimizer 的输出。

Parameters

optimized_prompt – 优化后的 prompt 版本实体。
optimizer_name – 优化器的名称。
final_eval_score – 优化后的提示词的最终评估分数。
initial_eval_score – 优化后提示的初始评估分数。

final_eval_score: float | None = None

initial_eval_score: float | None = None

optimized_prompt: str | dict[str, typing.Any]

optimizer_name: str

class mlflow.genai.optimize.PromptOptimizationResult(prompt: PromptVersion, initial_prompt: PromptVersion, optimizer_name: str, final_eval_score: float | None, initial_eval_score: float | None)[source]

基类： object

注意

实验性：此类可能在将来的版本中更改或在未发出警告的情况下被移除。

mlflow.genai.optimize_prompt() API 的结果。

Parameters

prompt – 一个包含优化模板的 prompt 版本实体。
initial_prompt – 一个包含初始模板的提示版本实体。
optimizer_name – 优化器的名称。
final_eval_score – 优化后提示的最终评估得分。
initial_eval_score – 优化提示的初始评估得分。

final_eval_score: float | None

initial_eval_score: float | None

initial_prompt: PromptVersion

optimizer_name: str

prompt: PromptVersion

mlflow.genai.optimize.format_dspy_prompt(program: dspy.Predict, convert_to_single_text: bool) → dict[str, typing.Any] | str[source]: 注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

mlflow.genai.optimize.optimize_prompt(*, target_llm_params: mlflow.genai.optimize.types.LLMParams, prompt: str | PromptVersion, train_data: EvaluationDatasetTypes, scorers: list[评分器], objective: Optional[Callable[[dict[str, bool | float | str | Feedback | list[Feedback]]], float]] = None, eval_data: Optional[EvaluationDatasetTypes] = None, optimizer_config: mlflow.genai.optimize.types.OptimizerConfig | None = None) → mlflow.genai.optimize.types.PromptOptimizationResult[source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

使用给定的数据集和评估指标优化一个 LLM 提示。优化后的提示模板会作为原始提示的新版本自动注册并包含在结果中。目前，此 API 仅支持 DSPy’s MIPROv2 optimizer。

Parameters

target_llm_params – 针对该提示优化的 LLM（大型语言模型）的参数。模型名称可以用以下任一格式指定： - <provider>:/<model> (例如，“openai:/gpt-4o”) - <provider>/<model> (例如，“openai/gpt-4o”)
prompt – 要优化的 MLflow prompt 的 URI 或 Prompt 对象。优化后的 prompt 会被注册为该 prompt 的新版本。
train_data –
用于优化的训练数据集。数据必须是下列格式之一：
- An EvaluationDataset entity
- Pandas DataFrame
- Spark DataFrame
- 字典列表
数据集必须包含以下列：
- inputs: 包含以 dict 格式的单个输入的列。每个输入应包含与提示模板中变量匹配的键。
- expectations: 包含单个输出字段的真实值字典的列。
scorers – 用于评估 inputs、outputs 和 expectations 的 scorers 列表。注意：Trace input 不支持用于优化。请使用 inputs、outputs 和 expectations 进行优化。此外，在与字符串或 Feedback 类型输出一起使用 scorers 时，请传递 objective 参数。
objective – 一个可调用对象，用于从各个评估计算整体性能指标。接受一个 dict，将评估名称映射到评估分数，并返回一个 float 值（值越大越好）。
eval_data – 与 train_data 格式相同的评估数据集。如果未提供，train_data 将会被自动拆分为训练集和评估集。
optimizer_config – 优化器的配置参数。

Returns

优化结果，包括优化后的提示。

Return type

PromptOptimizationResult

示例

import os
import mlflow
from typing import Any
from mlflow.genai.scorers import scorer
from mlflow.genai.optimize import OptimizerConfig, LLMParams

os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"


@scorer
def exact_match(expectations: dict[str, Any], outputs: dict[str, Any]) -> bool:
    return expectations == outputs


prompt = mlflow.genai.register_prompt(
    name="qa",
    template="Answer the following question: {{question}}",
)

result = mlflow.genai.optimize_prompt(
    target_llm_params=LLMParams(model_name="openai:/gpt-4o-mini"),
    train_data=[
        {"inputs": {"question": f"{i}+1"}, "expectations": {"answer": f"{i + 1}"}}
        for i in range(100)
    ],
    scorers=[exact_match],
    prompt=prompt.uri,
    optimizer_config=OptimizerConfig(num_instruction_candidates=5),
)

print(result.prompt.template)

class mlflow.genai.judges.CategoricalRating(value)[source]

基类：mlflow.genai.utils.enum_utils.StrEnum

评估的类别评级。

示例

from mlflow.genai.judges import CategoricalRating
from mlflow.entities import Feedback

# Create feedback with categorical rating
feedback = Feedback(
    name="my_metric", value=CategoricalRating.YES, rationale="The metric is passing."
)

NO = 'no'

UNKNOWN = 'unknown'

YES = 'yes'

mlflow.genai.judges.custom_prompt_judge(*, name: str, prompt_template: str, numeric_values: Optional[dict[str, float]] = None, model: Optional[str] = None) → Callable[[...], Feedback][source]

注意

实验性：此功能可能在将来的发布中更改或在不另行通知的情况下被移除。

创建一个自定义提示评判器，使用模板评估输入。

Parameters

name – 评判器的 name，用作返回的 mlflow.entities.Feedback 对象的 name。
prompt_template – 带有 {{var_name}} 占位符用于变量替换的模板字符串。应以带有选项的输出进行提示。
numeric_values – 可选的将类别值映射为数值分数的映射。若你想创建一个返回连续值输出的自定义评判器，这会很有用。默认值为 None。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个可调用对象，接受关键字参数，这些参数映射到模板变量，并返回一个 mlflow mlflow.entities.Feedback。

示例提示模板：

You will look at the response and determine the formality of the response.

<request>{{request}}</request>
<response>{{response}}</response>

You must choose one of the following categories.

[[formal]]: The response is very formal.
[[semi_formal]]: The response is somewhat formal. The response is somewhat formal if the
response mentions friendship, etc.
[[not_formal]]: The response is not formal.

模板中的变量名应使用双大括号括起，例如 {{request}}、{{response}}。它们应为字母数字字符，可以包含下划线，但不应包含空格或特殊字符。

提示模板必须将选项作为输出请求，每个选项用方括号括起。选项名称应为字母数字，并且可以包含下划线和空格。

mlflow.genai.judges.is_context_relevant(*, request: str, context: Any, name: Optional[str] = None, model: Optional[str] = None) → Feedback[source]

LLM 判定器判断给定的上下文是否与输入请求相关。

Parameters

request – 传入应用以供评估的输入，即用户的问题或查询。
context – 用于评估与请求相关性的上下文。支持任何可 JSON 序列化的对象。
name – 可选名称，用于覆盖返回反馈的默认名称。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，值为“yes”或“no”，表示上下文是否与请求相关。

示例

下面的示例展示了如何评估检索器检索到的文档是否与用户的问题相关。

from mlflow.genai.judges import is_context_relevant

feedback = is_context_relevant(
    request="What is the capital of France?",
    context="Paris is the capital of France.",
)
print(feedback.value)  # "yes"

feedback = is_context_relevant(
    request="What is the capital of France?",
    context="Paris is known for its Eiffel Tower.",
)
print(feedback.value)  # "no"

mlflow.genai.judges.is_context_sufficient(*, request: str, context: Any, expected_facts: list[str], expected_response: Optional[str] = None, name: Optional[str] = None, model: Optional[str] = None) → Feedback[source]

LLM judge 判断给定的上下文是否足以回答输入的请求。

Parameters

request – 要评估的应用程序输入，用户的问题或查询。
context – 用于评估充分性的上下文。支持任何可 JSON 序列化的对象。
expected_facts – 一组应存在于上下文中的预期事实。可选。
expected_response – 应用程序的预期响应。可选。
name – 可选名称，用于覆盖返回的反馈的默认名称。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，值为 “yes” 或 “no”，表示上下文是否足以回答该请求。

示例

下面的示例演示如何评估检索器返回的文档是否提供足够的上下文来回答用户的问题。

from mlflow.genai.judges import is_context_sufficient

feedback = is_context_sufficient(
    request="What is the capital of France?",
    context=[
        {"content": "Paris is the capital of France."},
        {"content": "Paris is known for its Eiffel Tower."},
    ],
    expected_facts=["Paris is the capital of France."],
)
print(feedback.value)  # "yes"

feedback = is_context_sufficient(
    request="What is the capital of France?",
    context={"content": "France is a country in Europe."},
    expected_response="Paris is the capital of France.",
)
print(feedback.value)  # "no"

mlflow.genai.judges.is_correct(*, request: str, response: str, expected_facts: Optional[list[str]] = None, expected_response: Optional[str] = None, name: Optional[str] = None, model: Optional[str] = None) → Feedback[source]

LLM judge 判定给定的响应是否正确地满足输入请求。

Parameters

request – 提供给应用以评估的输入，用户的问题或查询。
response – 来自应用程序的响应，用于评估。
expected_facts – 一个应该出现在响应中的预期事实列表。可选。
expected_response – 应用程序的预期响应。可选。
name – 可选名称，用于覆盖返回反馈的默认名称。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，值为“yes”或“no”，表示响应是否正确地满足请求。

示例

下面的示例展示了如何评估响应是否正确。

from mlflow.genai.judges import is_correct

feedback = is_correct(
    request="What is the capital of France?",
    response="Paris is the capital of France.",
    expected_response="Paris",
)
print(feedback.value)  # "yes"

feedback = is_correct(
    request="What is the capital of France?",
    response="London is the capital of France.",
    expected_facts=["Paris is the capital of France"],
)
print(feedback.value)  # "no"

mlflow.genai.judges.is_grounded(*, request: str, response: str, context: Any, name: Optional[str] = None, model: Optional[str] = None) → Feedback[source]

LLM 判定器确定给定的响应是否基于给定的上下文。

Parameters

request – 要由应用评估的输入，用户的问题或查询。
response – 来自应用程序的响应以供评估。
context – 用于评估响应的上下文。支持任何可序列化为 JSON 的对象。
name – 可选名称，用于覆盖返回反馈的默认名称。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，值为 “yes” 或 “no”，指示响应是否基于上下文。

示例

以下示例展示如何评估响应是否基于上下文。

from mlflow.genai.judges import is_grounded

feedback = is_grounded(
    request="What is the capital of France?",
    response="Paris",
    context=[
        {"content": "Paris is the capital of France."},
        {"content": "Paris is known for its Eiffel Tower."},
    ],
)
print(feedback.value)  # "yes"

feedback = is_grounded(
    request="What is the capital of France?",
    response="London is the capital of France.",
    context=[
        {"content": "Paris is the capital of France."},
        {"content": "Paris is known for its Eiffel Tower."},
    ],
)
print(feedback.value)  # "no"

mlflow.genai.judges.is_safe(*, content: str, name: str | None = None) → Feedback[source]

LLM 判定器判断给定的回复是否安全。

Parameters

content – 用于评估安全性的文本内容。
name – 可选名称，用于覆盖返回反馈的默认名称。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，值为 “yes” 或 “no”，用于指示响应是否安全。

示例

from mlflow.genai.judges import is_safe

feedback = is_safe(content="I am a happy person.")
print(feedback.value)  # "yes"

mlflow.genai.judges.meets_guidelines(*, guidelines: str | list[str], context: dict[str, typing.Any], name: Optional[str] = None, model: Optional[str] = None) → Feedback[source]

LLM judge 用于判断给定的响应是否符合给定的指南。

Parameters

guidelines – 单个指南或指南列表。
context – 用于根据指南评估的上下文映射。例如，传入 {“response”: “”} 来评估响应是否符合给定的指南。
name – 可选名称，用于覆盖返回反馈的默认名称。
model –
要使用的评估模型。必须是 “databricks” 或者形式为 <provider>:/<model-name>，例如 “openai:/gpt-4.1-mini”、 “anthropic:/claude-3.5-sonnet-20240620”。MLflow 原生支持 [“openai”, “anthropic”, “bedrock”, “mistral”]，并可通过 LiteLLM 支持更多提供者。默认模型取决于 tracking URI 的设置：
- Databricks: databricks
- 否则：openai:/gpt-4.1-mini。

Returns

一个 mlflow.entities.assessment.Feedback~ 对象，其值为“yes”或“no”，用于指示响应是否符合指南。

示例

以下示例展示如何评估响应是否符合给定的指导原则。

from mlflow.genai.judges import meets_guidelines

feedback = meets_guidelines(
    guidelines="Be polite and respectful.",
    context={"response": "Hello, how are you?"},
)
print(feedback.value)  # "yes"

feedback = meets_guidelines(
    guidelines=["Be polite and respectful.", "Must be in English."],
    context={"response": "Hola, ¿cómo estás?"},
)
print(feedback.value)  # "no"