FEAT: Adding TargetIdentifier

doc/api.rst

@@ -279,6 +279,7 @@ API Reference
     IdentifierType
     ScorerIdentifier
     snake_case_to_class_name
+    TargetIdentifier
 
 :py:mod:`pyrit.memory`
 ======================

pyrit/executor/attack/multi_turn/crescendo.py

@@ -586,7 +586,7 @@ async def _send_prompt_to_objective_target_async(
         Raises:
             ValueError: If no response is received from the objective target.
         """
-        objective_target_type = self._objective_target.get_identifier()["__type__"]
+        objective_target_type = self._objective_target.get_identifier().class_name
 
         # Send the generated prompt to the objective target
         prompt_preview = attack_message.get_value()[:100] if attack_message.get_value() else ""

pyrit/identifiers/__init__.py

@@ -14,6 +14,7 @@
     IdentifierType,
 )
 from pyrit.identifiers.scorer_identifier import ScorerIdentifier
+from pyrit.identifiers.target_identifier import TargetIdentifier
 
 __all__ = [
     "class_name_to_snake_case",
@@ -25,4 +26,5 @@
     "LegacyIdentifiable",
     "ScorerIdentifier",
     "snake_case_to_class_name",
+    "TargetIdentifier",
 ]

pyrit/identifiers/target_identifier.py

@@ -0,0 +1,55 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT license.
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any, Dict, Optional, Type, cast
+
+from pyrit.identifiers.identifier import Identifier
+
+
+@dataclass(frozen=True)
+class TargetIdentifier(Identifier):
+    """
+    Identifier for PromptTarget instances.
+
+    This frozen dataclass extends Identifier with target-specific fields.
+    It provides a stable, hashable identifier for prompt targets that can be
+    used for scorer evaluation, registry tracking, and memory storage.
+    """
+
+    endpoint: str = ""
+    """The target endpoint URL."""
+
+    model_name: str = ""
+    """The model or deployment name."""
+
+    temperature: Optional[float] = None
+    """The temperature parameter for generation."""
+
+    top_p: Optional[float] = None
+    """The top_p parameter for generation."""
+
+    max_requests_per_minute: Optional[int] = None
+    """Maximum number of requests per minute."""
+
+    target_specific_params: Optional[Dict[str, Any]] = None
+    """Additional target-specific parameters."""
+
+    @classmethod
+    def from_dict(cls: Type["TargetIdentifier"], data: dict[str, Any]) -> "TargetIdentifier":
+        """
+        Create a TargetIdentifier from a dictionary (e.g., retrieved from database).
+
+        Extends the base Identifier.from_dict() to handle legacy key mappings.
+
+        Args:
+            data: The dictionary representation.
+
+        Returns:
+            TargetIdentifier: A new TargetIdentifier instance.
+        """
+        # Delegate to parent class for standard processing
+        result = Identifier.from_dict.__func__(cls, data)  # type: ignore[attr-defined]
+        return cast(TargetIdentifier, result)

pyrit/memory/memory_models.py

@@ -32,7 +32,7 @@
 
 import pyrit
 from pyrit.common.utils import to_sha256
-from pyrit.identifiers import ConverterIdentifier, ScorerIdentifier
+from pyrit.identifiers import ConverterIdentifier, ScorerIdentifier, TargetIdentifier
 from pyrit.models import (
     AttackOutcome,
     AttackResult,
@@ -216,7 +216,10 @@ def __init__(self, *, entry: MessagePiece):
         self.prompt_metadata = entry.prompt_metadata
         self.targeted_harm_categories = entry.targeted_harm_categories
         self.converter_identifiers = [conv.to_dict() for conv in entry.converter_identifiers]
-        self.prompt_target_identifier = entry.prompt_target_identifier
+        # Normalize prompt_target_identifier and convert to dict for JSON serialization
+        self.prompt_target_identifier = (
+            entry.prompt_target_identifier.to_dict() if entry.prompt_target_identifier else {}
+        )
         self.attack_identifier = entry.attack_identifier
 
         self.original_value = entry.original_value
@@ -247,6 +250,12 @@ def get_message_piece(self) -> MessagePiece:
                 ConverterIdentifier.from_dict({**c, "pyrit_version": stored_version})
                 for c in self.converter_identifiers
             ]
+
+        # Reconstruct TargetIdentifier with the stored pyrit_version
+        target_id: Optional[TargetIdentifier] = None
+        if self.prompt_target_identifier:
+            target_id = TargetIdentifier.from_dict({**self.prompt_target_identifier, "pyrit_version": stored_version})
+
         message_piece = MessagePiece(
             role=self.role,
             original_value=self.original_value,
@@ -260,7 +269,7 @@ def get_message_piece(self) -> MessagePiece:
             prompt_metadata=self.prompt_metadata,
             targeted_harm_categories=self.targeted_harm_categories,
             converter_identifiers=converter_ids,
-            prompt_target_identifier=self.prompt_target_identifier,
+            prompt_target_identifier=target_id,
             attack_identifier=self.attack_identifier,
             original_value_data_type=self.original_value_data_type,
             converted_value_data_type=self.converted_value_data_type,
@@ -279,7 +288,11 @@ def __str__(self) -> str:
             str: Formatted string representation of the memory entry.
         """
         if self.prompt_target_identifier:
-            return f"{self.prompt_target_identifier['__type__']}: {self.role}: {self.converted_value}"
+            # prompt_target_identifier is stored as dict in the database
+            class_name = self.prompt_target_identifier.get("class_name") or self.prompt_target_identifier.get(
+                "__type__", "Unknown"
+            )
+            return f"{class_name}: {self.role}: {self.converted_value}"
         return f": {self.role}: {self.converted_value}"
 
 
@@ -902,7 +915,8 @@ def __init__(self, *, entry: ScenarioResult):
         self.scenario_version = entry.scenario_identifier.version
         self.pyrit_version = entry.scenario_identifier.pyrit_version
         self.scenario_init_data = entry.scenario_identifier.init_data
-        self.objective_target_identifier = entry.objective_target_identifier
+        # Convert TargetIdentifier to dict for JSON storage
+        self.objective_target_identifier = entry.objective_target_identifier.to_dict()
         # Convert ScorerIdentifier to dict for JSON storage
         self.objective_scorer_identifier = (
             entry.objective_scorer_identifier.to_dict() if entry.objective_scorer_identifier else None
@@ -952,10 +966,13 @@ def get_scenario_result(self) -> ScenarioResult:
                 {**self.objective_scorer_identifier, "pyrit_version": stored_version}
             )
 
+        # Convert dict back to TargetIdentifier for reconstruction
+        target_identifier = TargetIdentifier.from_dict(self.objective_target_identifier)
+
         return ScenarioResult(
             id=self.id,
             scenario_identifier=scenario_identifier,
-            objective_target_identifier=self.objective_target_identifier,
+            objective_target_identifier=target_identifier,
             attack_results=attack_results,
             objective_scorer_identifier=scorer_identifier,
             scenario_run_state=self.scenario_run_state,

pyrit/models/message_piece.py

@@ -5,11 +5,10 @@
 
 import uuid
 from datetime import datetime
-from typing import Dict, List, Literal, Optional, Union, get_args
+from typing import Any, Dict, List, Literal, Optional, Union, get_args
 from uuid import uuid4
 
-from pyrit.common.deprecation import print_deprecation_message
-from pyrit.identifiers import ConverterIdentifier, ScorerIdentifier
+from pyrit.identifiers import ConverterIdentifier, ScorerIdentifier, TargetIdentifier
 from pyrit.models.literals import ChatMessageRole, PromptDataType, PromptResponseError
 from pyrit.models.score import Score
 
@@ -39,7 +38,7 @@ def __init__(
         labels: Optional[Dict[str, str]] = None,
         prompt_metadata: Optional[Dict[str, Union[str, int]]] = None,
         converter_identifiers: Optional[List[Union[ConverterIdentifier, Dict[str, str]]]] = None,
-        prompt_target_identifier: Optional[Dict[str, str]] = None,
+        prompt_target_identifier: Optional[Union[TargetIdentifier, Dict[str, Any]]] = None,
         attack_identifier: Optional[Dict[str, str]] = None,
         scorer_identifier: Optional[Union[ScorerIdentifier, Dict[str, str]]] = None,
         original_value_data_type: PromptDataType = "text",
@@ -107,35 +106,24 @@ def __init__(
         self.labels = labels or {}
         self.prompt_metadata = prompt_metadata or {}
 
-        # Handle converter_identifiers: convert dicts to ConverterIdentifier with deprecation warning
-        self.converter_identifiers: List[ConverterIdentifier] = []
-        if converter_identifiers:
-            for conv_id in converter_identifiers:
-                if isinstance(conv_id, dict):
-                    print_deprecation_message(
-                        old_item="dict for converter_identifiers",
-                        new_item="ConverterIdentifier",
-                        removed_in="0.14.0",
-                    )
-                    self.converter_identifiers.append(ConverterIdentifier.from_dict(conv_id))
-                else:
-                    self.converter_identifiers.append(conv_id)
-
-        self.prompt_target_identifier = prompt_target_identifier or {}
+        # Handle converter_identifiers: normalize to ConverterIdentifier (handles dict with deprecation warning)
+        self.converter_identifiers: List[ConverterIdentifier] = (
+            [ConverterIdentifier.normalize(conv_id) for conv_id in converter_identifiers]
+            if converter_identifiers
+            else []
+        )
+
+        # Handle prompt_target_identifier: normalize to TargetIdentifier (handles dict with deprecation warning)
+        self.prompt_target_identifier: Optional[TargetIdentifier] = (
+            TargetIdentifier.normalize(prompt_target_identifier) if prompt_target_identifier else None
+        )
+
         self.attack_identifier = attack_identifier or {}
 
-        # Handle scorer_identifier: convert dict to ScorerIdentifier with deprecation warning
-        if scorer_identifier is None:
-            self.scorer_identifier: Optional[ScorerIdentifier] = None
-        elif isinstance(scorer_identifier, dict):
-            print_deprecation_message(
-                old_item="dict for scorer_identifier",
-                new_item="ScorerIdentifier",
-                removed_in="0.13.0",
-            )
-            self.scorer_identifier = ScorerIdentifier.from_dict(scorer_identifier)
-        else:
-            self.scorer_identifier = scorer_identifier
+        # Handle scorer_identifier: normalize to ScorerIdentifier (handles dict with deprecation warning)
+        self.scorer_identifier: Optional[ScorerIdentifier] = (
+            ScorerIdentifier.normalize(scorer_identifier) if scorer_identifier else None
+        )
 
         self.original_value = original_value
 
@@ -292,7 +280,9 @@ def to_dict(self) -> dict[str, object]:
             "targeted_harm_categories": self.targeted_harm_categories if self.targeted_harm_categories else None,
             "prompt_metadata": self.prompt_metadata,
             "converter_identifiers": [conv.to_dict() for conv in self.converter_identifiers],
-            "prompt_target_identifier": self.prompt_target_identifier,
+            "prompt_target_identifier": (
+                self.prompt_target_identifier.to_dict() if self.prompt_target_identifier else None
+            ),
             "attack_identifier": self.attack_identifier,
             "scorer_identifier": self.scorer_identifier.to_dict() if self.scorer_identifier else None,
             "original_value_data_type": self.original_value_data_type,
@@ -308,7 +298,8 @@ def to_dict(self) -> dict[str, object]:
         }
 
     def __str__(self) -> str:
-        return f"{self.prompt_target_identifier}: {self._role}: {self.converted_value}"
+        target_str = self.prompt_target_identifier.class_name if self.prompt_target_identifier else "Unknown"
+        return f"{target_str}: {self._role}: {self.converted_value}"
 
     __repr__ = __str__
 

pyrit/models/scenario_result.py

@@ -10,7 +10,7 @@
 from pyrit.models import AttackOutcome, AttackResult
 
 if TYPE_CHECKING:
-    from pyrit.identifiers import ScorerIdentifier
+    from pyrit.identifiers import ScorerIdentifier, TargetIdentifier
     from pyrit.score import Scorer
     from pyrit.score.scorer_evaluation.scorer_metrics import ScorerMetrics
 
@@ -59,7 +59,7 @@ def __init__(
         self,
         *,
         scenario_identifier: ScenarioIdentifier,
-        objective_target_identifier: dict[str, str],
+        objective_target_identifier: Union[Dict[str, Any], "TargetIdentifier"],
         attack_results: dict[str, List[AttackResult]],
         objective_scorer_identifier: Union[Dict[str, Any], "ScorerIdentifier"],
         scenario_run_state: ScenarioRunState = "CREATED",
@@ -71,11 +71,13 @@ def __init__(
         objective_scorer: Optional["Scorer"] = None,
     ) -> None:
         from pyrit.common import print_deprecation_message
-        from pyrit.identifiers import ScorerIdentifier
+        from pyrit.identifiers import ScorerIdentifier, TargetIdentifier
 
         self.id = id if id is not None else uuid.uuid4()
         self.scenario_identifier = scenario_identifier
-        self.objective_target_identifier = objective_target_identifier
+
+        # Normalize objective_target_identifier to TargetIdentifier
+        self.objective_target_identifier = TargetIdentifier.normalize(objective_target_identifier)
 
         # Handle deprecated objective_scorer parameter
         if objective_scorer is not None:

pyrit/prompt_converter/prompt_converter.py

@@ -209,11 +209,13 @@ def _create_identifier(
         target_info: Optional[Dict[str, Any]] = None
         if converter_target:
             target_id = converter_target.get_identifier()
-            # Extract standard fields for converter identification
-            target_info = {}
-            for key in ["__type__", "model_name", "temperature", "top_p"]:
-                if key in target_id:
-                    target_info[key] = target_id[key]
+            # Extract standard fields for converter identification using attribute access
+            target_info = {
+                "class_name": target_id.class_name,
+                "model_name": target_id.model_name,
+                "temperature": target_id.temperature,
+                "top_p": target_id.top_p,
+            }
 
         return ConverterIdentifier(
             class_name=self.__class__.__name__,

pyrit/prompt_target/azure_blob_storage_target.py

@@ -12,6 +12,7 @@
 
 from pyrit.auth import AzureStorageAuth
 from pyrit.common import default_values
+from pyrit.identifiers import TargetIdentifier
 from pyrit.models import Message, construct_response_from_request
 from pyrit.prompt_target.common.prompt_target import PromptTarget
 from pyrit.prompt_target.common.utils import limit_requests_per_minute
@@ -79,6 +80,20 @@ def __init__(
 
         super().__init__(endpoint=self._container_url, max_requests_per_minute=max_requests_per_minute)
 
+    def _build_identifier(self) -> TargetIdentifier:
+        """
+        Build the identifier with Azure Blob Storage-specific parameters.
+
+        Returns:
+            TargetIdentifier: The identifier for this target instance.
+        """
+        return self._create_identifier(
+            target_specific_params={
+                "container_url": self._container_url,
+                "blob_content_type": self._blob_content_type,
+            },
+        )
+
     async def _create_container_client_async(self) -> None:
         """
         Create an asynchronous ContainerClient for Azure Storage. If a SAS token is provided via the

pyrit/prompt_target/azure_ml_chat_target.py

@@ -13,6 +13,7 @@
     handle_bad_request_exception,
     pyrit_target_retry,
 )
+from pyrit.identifiers import TargetIdentifier
 from pyrit.message_normalizer import ChatMessageNormalizer, MessageListNormalizer
 from pyrit.models import (
     Message,
@@ -103,6 +104,23 @@ def __init__(
         self._repetition_penalty = repetition_penalty
         self._extra_parameters = param_kwargs
 
+    def _build_identifier(self) -> TargetIdentifier:
+        """
+        Build the identifier with Azure ML-specific parameters.
+
+        Returns:
+            TargetIdentifier: The identifier for this target instance.
+        """
+        return self._create_identifier(
+            temperature=self._temperature,
+            top_p=self._top_p,
+            target_specific_params={
+                "max_new_tokens": self._max_new_tokens,
+                "repetition_penalty": self._repetition_penalty,
+                "message_normalizer": self.message_normalizer.__class__.__name__,
+            },
+        )
+
     def _initialize_vars(self, endpoint: Optional[str] = None, api_key: Optional[str] = None) -> None:
         """
         Set the endpoint and key for accessing the Azure ML model. Use this function to manually

pyrit/prompt_target/common/prompt_chat_target.py

@@ -121,7 +121,7 @@ def _get_json_response_config(self, *, message_piece: MessagePiece) -> _JsonResp
         config = _JsonResponseConfig.from_metadata(metadata=message_piece.prompt_metadata)
 
         if config.enabled and not self.is_json_response_supported():
-            target_name = self.get_identifier()["__type__"]
+            target_name = self.get_identifier().class_name
             raise ValueError(f"This target {target_name} does not support JSON response format.")
 
         return config