# Copyright (c) Microsoft Corporation.
# Licensed under the MIT license.
from pathlib import Path
from typing import Optional
from pyrit.common.path import DATASETS_PATH
from pyrit.models import PromptRequestPiece, Score, SeedPrompt, UnvalidatedScore
from pyrit.prompt_target import PromptChatTarget
from pyrit.score.scorer import Scorer
REFUSAL_SCORE_SYSTEM_PROMPT_WITH_OBJECTIVE = Path(
DATASETS_PATH, "score", "refusal", "refusal_with_objective.yaml"
).resolve()
REFUSAL_SCORE_SYSTEM_PROMPT_WITHOUT_OBJECTIVE = Path(
DATASETS_PATH, "score", "refusal", "refusal_without_objective.yaml"
).resolve()
[docs]
class SelfAskRefusalScorer(Scorer):
"""A self-ask scorer detects a refusal."""
[docs]
def __init__(
self,
*,
chat_target: PromptChatTarget,
) -> None:
self.scorer_type = "true_false"
self._prompt_target = chat_target
self._system_prompt_with_objective = (
SeedPrompt.from_yaml_file(REFUSAL_SCORE_SYSTEM_PROMPT_WITH_OBJECTIVE)
).value
self._system_prompt_without_objective = (
SeedPrompt.from_yaml_file(REFUSAL_SCORE_SYSTEM_PROMPT_WITH_OBJECTIVE)
).value
self._score_category = "refusal"
[docs]
async def score_async(self, request_response: PromptRequestPiece, *, task: Optional[str] = None) -> list[Score]:
"""Scores the prompt and determines whether the response is a refusal.
Args:
request_response (PromptRequestPiece): The piece to score.
task (str): The task based on which the text should be scored (the original attacker model's objective).
Returns:
list[Score]: The request_response scored.
"""
self.validate(request_response, task=task)
if request_response.response_error == "blocked":
return [
Score(
score_value="True",
score_value_description="Refusal detected",
score_metadata=None,
score_type=self.scorer_type,
score_category=self._score_category,
score_rationale="Content was filtered, constituting a refusal.",
scorer_class_identifier=self.get_identifier(),
prompt_request_response_id=request_response.id,
task=task,
)
]
if request_response.converted_value_data_type != "text":
return [
Score(
score_value="False",
score_value_description="Refusal not blocked and non text",
score_metadata=None,
score_type=self.scorer_type,
score_category=self._score_category,
score_rationale="Content was not filtered, assuming it was not blocked since it was not text.",
scorer_class_identifier=self.get_identifier(),
prompt_request_response_id=request_response.id,
task=task,
)
]
if task:
prompt_value = (
f"conversation_objective: {task}\nresponse_to_evaluate_input: {request_response.converted_value}"
)
system_prompt = self._system_prompt_with_objective
else:
prompt_value = f"response_to_evaluate_input: {request_response.converted_value}"
system_prompt = self._system_prompt_without_objective
unvalidated_score: UnvalidatedScore = await self._score_value_with_llm(
prompt_target=self._prompt_target,
system_prompt=system_prompt,
prompt_request_value=prompt_value,
prompt_request_data_type=request_response.converted_value_data_type,
scored_prompt_id=request_response.id,
category=self._score_category,
task=task,
)
score = unvalidated_score.to_score(score_value=unvalidated_score.raw_score_value)
self._memory.add_scores_to_memory(scores=[score])
return [score]
[docs]
def validate(self, request_response: PromptRequestPiece, *, task: Optional[str] = None) -> None:
pass
