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
17 changes: 10 additions & 7 deletions pyrit/executor/attack/multi_turn/tree_of_attacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -1339,18 +1339,21 @@ async def _get_response_score_async(self, response_id: str) -> str:
response_id (str): The unique identifier of the response to retrieve the score for.

Returns:
str: The score value as a string representation. Returns "unavailable" if no score
exists for the given response ID. For numeric scores, this will be the string
str: The score value as a string representation. Returns "unavailable" if no identified
objective score exists for the given response ID. For numeric scores, this will be the string
representation of the float value (e.g., "0.75").

Note:
The method assumes that if scores exist, at least one score will be present in the
list. It takes the first score if multiple scores are associated with the response,
which is typically the objective score in the TAP algorithm context.
Auxiliary scorers also score the response, so only the objective scorer's score is used.
"""
scores = await self._memory.get_prompt_scores_async(prompt_ids=[str(response_id)])
if scores:
return str(normalize_score_to_float(scores[0]))
objective_scorer_hash = self._objective_scorer.get_identifier().hash
for score in scores:
if (
score.scorer_class_identifier is not None
and score.scorer_class_identifier.hash == objective_scorer_hash
):
return str(normalize_score_to_float(score))
return "unavailable"

async def _send_to_adversarial_chat_async(
Expand Down
41 changes: 41 additions & 0 deletions tests/unit/executor/attack/multi_turn/test_tree_of_attacks.py
Original file line number Diff line number Diff line change
Expand Up @@ -1781,6 +1781,47 @@ def test_node_initialization(self, node_components):
assert node.auxiliary_scores == {}
assert node.error_message is None

@pytest.mark.parametrize("include_legacy", [False, True])
@pytest.mark.parametrize("include_auxiliary", [False, True])
@pytest.mark.parametrize("include_objective", [False, True])
async def test_score_feedback_uses_only_the_identified_objective_score_async(
self,
*,
node_components: dict[str, Any],
include_legacy: bool,
include_auxiliary: bool,
include_objective: bool,
) -> None:
node = _TreeOfAttacksNode(**node_components)

def score(*, value: str, identifier: ComponentIdentifier | None) -> Score:
return Score(
score_value=value,
score_type="float_scale",
score_category=["test"],
score_rationale="r",
score_value_description="d",
score_metadata=None,
message_piece_id=str(uuid.uuid4()),
scorer_class_identifier=identifier,
)

scores = []
if include_legacy:
scores.append(score(value="0.8", identifier=None))
if include_auxiliary:
scores.append(
score(value="0.95", identifier=ComponentIdentifier(class_name="AuxScorer", class_module="test"))
)
if include_objective:
scores.append(score(value="0.2", identifier=node._objective_scorer.get_identifier()))

with patch.object(node._memory, "get_prompt_scores_async", new=AsyncMock(return_value=scores)) as get_scores:
assert await node._get_response_score_async("response-id") == (
"0.2" if include_objective else "unavailable"
)
assert get_scores.await_args.kwargs["prompt_ids"] == ["response-id"]

async def test_subsequent_prompt_omits_score_when_feedback_disabled(self, node_components):
"""A disabled score-feedback setting preserves response context without exposing the score."""
node = _TreeOfAttacksNode(**node_components, use_score_as_feedback=False)
Expand Down
Loading