Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 10 additions & 14 deletions pyrit/score/scorer_evaluation/scorer_evaluator.py
Original file line number Diff line number Diff line change
Expand Up @@ -481,23 +481,19 @@ async def _score_responses_grouped_async(
@staticmethod
def _score_matches_harm_category(*, score: Score, harm_category: str) -> bool:
"""Return whether a score category matches a canonical or aliased harm category."""
labeled_categories = set(HarmCategory.parse_many(harm_category))
if labeled_categories == {HarmCategory.OTHER} and harm_category.casefold() not in {
HarmCategory.OTHER.name.casefold(),
HarmCategory.OTHER.value.casefold(),
}:
labeled_categories = set()
target_casefold = harm_category.casefold()
if any(c.casefold() == target_casefold for c in score.score_category or []):
return True

labeled_category = HarmCategory.parse(harm_category)
if labeled_category == HarmCategory.OTHER and target_casefold != "other":
return False

for score_category in score.score_category or []:
if score_category == harm_category:
return True
score_categories = set(HarmCategory.parse_many(score_category))
if score_categories == {HarmCategory.OTHER} and score_category.casefold() not in {
HarmCategory.OTHER.name.casefold(),
HarmCategory.OTHER.value.casefold(),
}:
score_category_parsed = HarmCategory.parse(score_category)
if score_category_parsed == HarmCategory.OTHER and score_category.casefold() != "other":
continue
if score_categories & labeled_categories:
if score_category_parsed == labeled_category:
return True
return False

Expand Down
66 changes: 54 additions & 12 deletions tests/unit/score/test_scorer_evaluator.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.

import csv
from unittest.mock import AsyncMock, MagicMock, patch

import numpy as np
import pytest
from azure.ai.contentsafety.models import TextCategory

from pyrit.common.path import SCORER_EVALS_PATH
from pyrit.memory import MemoryInterface
from pyrit.models import Message, MessagePiece, Score, ScoreStatus
from pyrit.score import (
Expand All @@ -16,6 +17,7 @@
HarmScorerEvaluator,
HarmScorerMetrics,
HumanLabeledDataset,
LikertScalePaths,
MetricsType,
ObjectiveHumanLabeledEntry,
ObjectiveScorerEvaluator,
Expand Down Expand Up @@ -958,20 +960,60 @@ class TestSelectEvaluationScore:
def _score(*, category: list[str] | None) -> Score:
return Score(score_type="float_scale", score_value="0.5", score_category=category)

@pytest.mark.parametrize("category", list(AzureContentFilterScorer._CATEGORY_EVAL_FILES))
@pytest.mark.parametrize(
("emitted_category", "csv_relative_path", "registered_category"),
[
*(
(cat.value, files[0][0], files[2])
for cat, files in AzureContentFilterScorer._CATEGORY_EVAL_FILES.items()
),
*(
(
preset.load().category,
preset.evaluation_files.human_labeled_datasets_files[0],
preset.evaluation_files.harm_category,
)
for preset in LikertScalePaths
if preset.evaluation_files is not None
),
],
)
@pytest.mark.parametrize("multiple_scores", [False, True])
def test_azure_categories_match_registered_evaluation(self, category: TextCategory, multiple_scores: bool) -> None:
config = AzureContentFilterScorer._get_eval_files_for_category(category)
assert config is not None
selected = self._score(category=[category.value])
def test_shipped_pairings_match_csv_evaluation(
self,
emitted_category: str,
csv_relative_path: str,
registered_category: str,
multiple_scores: bool,
) -> None:
csv_path = SCORER_EVALS_PATH / csv_relative_path
with open(csv_path, encoding="utf-8") as f:
reader = csv.DictReader(line for line in f if not line.startswith("#"))
csv_categories = {row["harm_category"] for row in reader if "harm_category" in row}
assert len(csv_categories) == 1, f"Expected exactly one harm_category in {csv_path}, got {csv_categories}"
csv_harm_category = csv_categories.pop()

assert registered_category == csv_harm_category

selected = self._score(category=[emitted_category])
scores = [selected]
if multiple_scores:
scores = [
self._score(category=[other.value])
for other in AzureContentFilterScorer._CATEGORY_EVAL_FILES
if other != category
] + scores
assert ScorerEvaluator._select_evaluation_score(scores=scores, harm_category=config.harm_category) is selected
scores = [self._score(category=["unrelated_harm_category"]), selected]
assert ScorerEvaluator._select_evaluation_score(scores=scores, harm_category=csv_harm_category) is selected

def test_rejects_hate_score_for_representational_dataset(self):
score = self._score(category=["Hate"])
with pytest.raises(ValueError, match="requires a score for harm category 'REPRESENTATIONAL'"):
ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="REPRESENTATIONAL")

def test_rejects_bias_score_for_hate_speech_dataset(self):
score = self._score(category=["bias"])
with pytest.raises(ValueError, match="requires a score for harm category 'hate_speech'"):
ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="hate_speech")

def test_accepts_case_insensitive_unrecognized_category(self):
score = self._score(category=["Jailbreak"])
assert ScorerEvaluator._select_evaluation_score(scores=[score], harm_category="jailbreak") is score

def test_returns_none_when_the_scorer_returned_nothing(self):
assert ScorerEvaluator._select_evaluation_score(scores=[], harm_category="hate_speech") is None
Expand Down
Loading