指標値の計算¶
カスタム指標値を計算するには、MetricEvaluatorを使用して時間の経過に伴う指標値を計算、BatchMetricEvaluatorを使用してバッチごとの指標値を計算、またはIndividualMetricEvaluatorを使用してデータ集計なしで指標を評価できます。
指標の評価¶
MetricEvaluatorクラスは、選択したソースを使用して、時間の経過に伴う指標値を計算します。 このクラスは、指標オブジェクトを介してデータを「ストリーム」し、指標値を生成するために使用されます。 次の必須パラメーターでMetricEvaluatorを初期化します。
from dmm import MetricEvaluator, TimeBucket
from dmm import DataRobotSource
from dmm.metric import MedianAbsoluteError
source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(weeks=1),
end=datetime.utcnow(),
)
metric = MedianAbsoluteError()
metric_evaluator = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.MINUTE)
MetricEvaluatorを使用するには、MetricBaseインターフェイスを実装する指標クラスとDataSourceBaseを実装するソースを作成します。 次に、集計のきめ細かさのレベルを指定します。 すべてのパラメーターでMetricEvaluatorを初期化します。
from dmm import ColumnName, MetricEvaluator, TimeBucket
metric_evaluator = MetricEvaluator(
metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
prediction_col=ColumnName.PREDICTIONS,
actuals_col=ColumnName.ACTUALS,
timestamp_col=ColumnName.TIMESTAMP,
filter_actuals=False,
filter_predictions=False,
filter_scoring_data=False,
segment_attribute=None,
segment_value=None,
)
| パラメーター | 説明 |
|---|---|
metric: Union[str, MetricBase, List[str], List[MetricBase]] |
文字列または文字列のリストが渡された場合、MetricEvaluatorは一致するSklearn指標を検索します。 指標またはオブジェクトのリストが渡された場合、MetricBaseインターフェイスを実装する必要があります。 |
source: DataSourceBase |
データの取得元のソースであるDataRobotSourceまたはDataFrameSource、またはDataSourceBaseインターフェイスを実装するその他のソース。 |
time_bucket: TimeBucket |
指標の評価、集計のきめ細かさの決定に使用するタイムバケットのサイズ。 |
prediction_col: Optional[str] |
予測を含む列名。 |
actuals_col: Optional[str] |
実測値を含む列名。 |
timestamp_col: Optional[str] |
タイムスタンプを含む列名。 |
filter_actuals: Optional[bool] |
スコアリングの前に、MetricEvaluatorが欠損実測値を削除するかどうか。 Trueは欠損実測値を削除します。デフォルト値はFalseです。 |
filter_predictions: Optional[bool] |
スコアリングの前に、MetricEvaluatorが欠損予測値を削除するかどうか。 Trueは欠損予測値を削除します。デフォルト値はFalseです。 |
filter_scoring_data: Optional[bool] |
スコアリングの前に、MetricEvaluatorが欠損スコアリング値を削除するかどうか。 Trueは欠損しているスコアリング値を削除します。デフォルト値はFalseです。 |
segment_attribute: Optional[str] |
セグメント値を含む列名。 |
segment_value: Optional[Union[str or List[str]]] |
セグメント化するセグメント属性の単一値または値リスト。 |
scoreメソッドは、TimeBucketで定義された集計指標を返します。 pandas DataFrameとして返される出力には、ソースからのすべてのデータのタイムバケットごとの結果が含まれます。
source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(hours=3),
end=datetime.utcnow(),
)
metric = LogLossFromSklearn()
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)
aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss
0 2023-09-14 13:29:48.065000+00:00 499 0.539315
1 2023-09-14 14:01:51.484000+00:00 499 0.539397
# we can see the evaluator's statistics
stats = me.stats()
print(stats)
total rows: 998, score calls: 2, reduce calls: 2
一度に複数の指標に渡すには、以下を実行します。
metrics = [LogLossFromSklearn(), AsymmetricError(), RocAuc()]
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)
aggregated_metric_per_time_bucket = me.score()
stats = me.stats()
print(aggregated_metric_per_time_bucket.to_string())
print(stats)
timestamp samples log_loss Asymmetric Error roc_auc_score
0 2023-09-14 13:29:48.065000+00:00 499 0.539315 0.365571 0.787030
1 2023-09-14 14:01:51.484000+00:00 499 0.539397 0.365636 0.786837
total rows: 998, score calls: 6, reduce calls: 6
データに対して、評価する列の名前を指定します。
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=5,
rows_per_time_bucket=1,
prediction_value=1,
time_bucket=TimeBucket.DAY,
prediction_col="my_pred_col",
actuals_col="my_actuals_col",
timestamp_col="my_timestamp_col"
)
print(test_df)
my_timestamp_col my_pred_col my_actuals_col
0 01/06/2005 13:00:00.000000 1 0.999
1 02/06/2005 13:00:00.000000 1 0.999
2 03/06/2005 13:00:00.000000 1 0.999
3 04/06/2005 13:00:00.000000 1 0.999
4 05/06/2005 13:00:00.000000 1 0.999
source = DataFrameSource(
df=test_df,
max_rows=10000,
timestamp_col="timestamp",
)
metric = LogLossFromSklearn()
me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.DAY,
prediction_col="my_pred_col",
actuals_col="my_actuals_col",
timestamp_col="my_timestamp_col"
)
aggregated_metric_per_time_bucket = me.score()
データフィルターの設定¶
データが欠損している場合は、フィルターフラグを使用します。 次の例では、データの実測値が欠損しています。 フラグのないこのシナリオでは、例外が発生します。
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=10,
rows_per_time_bucket=5,
prediction_value=1,
time_bucket=TimeBucket.HOUR,
)
test_df["actuals"].loc[2] = None
test_df["actuals"].loc[5] = None
print(test_df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 0.999
1 01/06/2005 13:00:00.000000 1 0.999
2 01/06/2005 13:00:00.000000 1 NaN
3 01/06/2005 13:00:00.000000 1 0.999
4 01/06/2005 13:00:00.000000 1 0.999
5 01/06/2005 14:00:00.000000 1 NaN
6 01/06/2005 14:00:00.000000 1 0.999
7 01/06/2005 14:00:00.000000 1 0.999
8 01/06/2005 14:00:00.000000 1 0.999
9 01/06/2005 14:00:00.000000 1 0.999
source = DataFrameSource(df=test_df)
metric = MedianAbsoluteError()
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR)
aggregated_metric_per_time_bucket = me.score()
"ValueError: Could not apply metric median_absolute_error, make sure you are passing the right data (see the sklearn docs).
The error message was: Input contains NaN."
前の結果を、filter_actualsフラグを有効化したときの結果と比較します。
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR, filter_actuals=True)
aggregated_metric_per_time_bucket = me.score()
"removed 1 rows out of 5 in the data chunk before scoring, due to missing values in ['actuals'] data"
"removed 1 rows out of 5 in the data chunk before scoring, due to missing values in ['actuals'] data"
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples median_absolute_error
0 01/06/2005 13:00:00.000000 4 0.001
1 01/06/2005 14:00:00.000000 4 0.001
指標を計算する前に、filter_actuals、filter_predictions、およびfilter_scoring_dataフラグを使用して、データから欠損値を除外できます。 デフォルトでは、これらのフラグはFalseに設定されています。 指標の計算に必要なすべてのデータがデータチャンクから欠損している場合、データチャンクは該当のログでスキップされます。
test_df = gen_dataframe_for_accuracy_metric(
nr_rows=4,
rows_per_time_bucket=2,
prediction_value=1,
time_bucket=TimeBucket.HOUR,
)
test_df["actuals"].loc[0] = None
test_df["actuals"].loc[1] = None
print(test_df)
timestamp predictions actuals
0 01/06/2005 13:00:00.000000 1 NaN
1 01/06/2005 13:00:00.000000 1 NaN
2 01/06/2005 14:00:00.000000 1 0.999
3 01/06/2005 14:00:00.000000 1 0.999
source = DataFrameSource(df=test_df)
metric = MedianAbsoluteError()
me = MetricEvaluator(metric=metric, source=source, time_bucket=TimeBucket.HOUR, filter_actuals=True)
aggregated_metric_per_time_bucket = me.score()
"removed 2 rows out of 2 in the data chunk before scoring, due to missing values in ['actuals'] data"
"data chunk is empty, skipping scoring..."
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples median_absolute_error
1 01/06/2005 14:00:00.000000 2 0.001
セグメント化された分析の実行¶
segment_attributeおよび各segment_valueを定義して、セグメント化された分析を実行します。
metrics = LogLossFromSklearn()
me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value="Down",
)
aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down]
0 2023-09-14 13:29:49.737000+00:00 49 0.594483
1 2023-09-14 14:01:52.437000+00:00 49 0.594483
# passing more than one segment value
me = MetricEvaluator(metric=metric,
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value=["Down", "Steady"],
)
aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down] log_loss [Steady]
0 2023-09-14 13:29:48.502000+00:00 199 0.594483 0.515811
1 2023-09-14 14:01:51.758000+00:00 199 0.594483 0.515811
# passing more than one segment value and more than one metric
me = MetricEvaluator(metric=[LogLossFromSklearn(), RocAuc()],
source=source,
time_bucket=TimeBucket.HOUR,
segment_attribute="insulin",
segment_value=["Down", "Steady"],
)
aggregated_metric_per_time_bucket = me.score()
print(aggregated_metric_per_time_bucket.to_string())
timestamp samples log_loss [Down] log_loss [Steady] roc_auc_score [Down] roc_auc_score [Steady]
0 2023-09-14 13:29:48.502000+00:00 199 0.594483 0.515811 0.783333 0.826632
1 2023-09-14 14:01:51.758000+00:00 199 0.594483 0.515811 0.783333 0.826632
バッチごとの集計による指標の評価¶
BatchMetricEvaluatorクラスは、時間経過に伴う集計ではなく、バッチごとの集計を使用します。 バッチの場合、TimeBucketを定義しないでください。
from dmm.batch_metric_evaluator import BatchMetricEvaluator
from dmm.data_source.datarobot_source import BatchDataRobotSource
from dmm.metric import MissingValuesFraction
source = BatchDataRobotSource(
deployment_id=DEPLOYMENT_ID,
batch_ids=BATCH_IDS,
model_id=MODEL_ID,
)
feature_name = 'RAD'
metric = MissingValuesFraction(feature_name=feature_name)
missing_values_fraction_evaluator = BatchMetricEvaluator(metric=metric, source=source)
aggregated_metric_per_batch = missing_values_fraction_evaluator.score()
print(aggregated_metric_per_batch.to_string())
batch_id samples Missing Values Fraction
0 <batch_id> 506 0.0
1 <batch_id> 506 0.0
2 <batch_id> 506 0.0
!!! note "備考"
バッチの場合、実測値および複数のセグメントはサポートされていません。
データ集計なしの指標評価¶
IndividualMetricEvaluatorクラスは、データ集計なしで指標を評価するために使用されます。 エクスポートされたすべてのデータに対して指標計算を実行し、各結果のリストを返します。 MetricEvaluatorを使用すると、対応する関連付けIDと一緒に、個々のデータポイントを送信できます。これは、予測値と実測値とともに指標結果を視覚化する必要がある場合に役立ちます。 カスタム指標でMetricEvaluatorを使用するには、特にパラメーターtimestampsおよびassociation_idsを含むscore()メソッドを指定します。
from itertools import zip_longest
from typing import List
from datetime import datetime
from datetime import timedelta
from dmm import CustomMetric
from dmm import DataRobotSource
from dmm import SingleMetricResult
from dmm.individual_metric_evaluator import IndividualMetricEvaluator
from dmm.metric import LLMMetricBase
from nltk import sent_tokenize
import numpy as np
import pandas as pd
source = DataRobotSource(
deployment_id=DEPLOYMENT_ID,
start=datetime.utcnow() - timedelta(weeks=1),
end=datetime.utcnow(),
)
custom_metric = CustomMetric.from_id()
class SentenceCount(LLMMetricBase):
"""
Calculates the total number of sentences created while working with the LLM model.
Returns the sum of the number of sentences from prompts and completions.
"""
def __init__(self):
super().__init__(
name=custom_metric.name,
description="Calculates the total number of sentences created while working with the LLM model.",
need_training_data=False,
)
self.prompt_column = "promptColumn"
def score(
self,
scoring_data: pd.DataFrame,
predictions: np.ndarray,
timestamps: np.ndarray,
association_ids: np.ndarray,
**kwargs,
) -> List[SingleMetricResult]:
if self.prompt_column not in scoring_data.columns:
raise ValueError(
f"Prompt column {self.prompt_column} not found in the exported data, "
f"modify 'PROMPT_COLUMN' runtime parameter"
)
prompts = scoring_data[self.prompt_column].to_numpy()
sentence_count = []
for prompt, completion, ts, a_id in zip_longest(
prompts, predictions, timestamps, association_ids
):
if not isinstance(prompt, str) or not isinstance(completion, str):
continue
value = len(sent_tokenize(prompt)) + len(sent_tokenize(completion))
sentence_count.append(
SingleMetricResult(value=value, timestamp=ts, association_id=a_id)
)
return sentence_count
sentence_count_evaluator = IndividualMetricEvaluator(
metric=SentenceCount(),
source=source,
)
metric_results = sentence_count_evaluator.score()
更新しました March 12, 2025
このページは役に立ちましたか？
ありがとうございます。どのような点が役に立ちましたか？
より良いコンテンツを提供するには、どうすればよいでしょうか？
アンケートにご協力いただき、ありがとうございました。