From 2af324c599c5ae3c55bb945aa9ea6f8457c3d2e9 Mon Sep 17 00:00:00 2001 From: varunj-msft Date: Thu, 17 Sep 2026 19:42:07 +0000 Subject: [PATCH 1/3] [BREAKING] FEAT: Normalize simulated conversation prompts to SeedPrompt Simulated-conversation techniques carried three YAML file paths through the seed model, the attack parameters, and the async executor, which read those files once per objective in the middle of attack execution. A technique was therefore only inspectable by chasing a path, and its identity was a set of absolute, machine-specific path strings. The adversarial, simulated-target, and next-message system prompts are now canonical SeedPrompt values on SeedSimulatedConversation. YAML is loaded at the setup and composition boundaries, and only prompts travel from there, so the executor no longer touches disk. Serialized configuration and identity now describe prompt content instead of file locations, which makes a technique portable and lets an edit to a prompt change its identity. The matching *_system_prompt_path inputs stay accepted on the model, the technique factory, and the public generator for legacy callers and for records persisted before this change; they are resolved once, warn, and never become fields. A record written earlier still loads as long as the prompt files it names are present, and reports which record is at fault when they are not. Composition now reads prompt files where it previously only built paths, so the scenario and initializer entry points that build techniques from an async method run that work in a worker thread. Defaults, template rendering, response schemas, sequence behavior, and the database schema are unchanged. --- doc/code/datasets/2_seed_programming.ipynb | 8 +- doc/code/datasets/2_seed_programming.py | 8 +- .../datasets/5_simulated_conversation.ipynb | 16 +- doc/code/datasets/5_simulated_conversation.py | 16 +- doc/code/memory/3_memory_data_types.md | 2 +- .../executor/attack/core/attack_parameters.py | 6 +- .../multi_turn/simulated_conversation.py | 133 ++++++-- pyrit/memory/memory_models.py | 70 ++-- pyrit/models/__init__.py | 4 + pyrit/models/seeds/__init__.py | 4 + .../seeds/seed_simulated_conversation.py | 318 +++++++++++++++--- .../scenario/core/attack_technique_factory.py | 85 +++-- .../scenarios/adaptive/adaptive_scenario.py | 9 +- pyrit/scenario/scenarios/airt/psychosocial.py | 30 +- pyrit/scenario/scenarios/airt/scam.py | 33 +- pyrit/setup/initializers/techniques/core.py | 48 ++- .../techniques/technique_initializer.py | 4 +- .../test_local_prompt_dataset_semantics.py | 10 +- .../component/test_simulated_conversation.py | 148 +++++++- .../attack/core/test_attack_parameters.py | 4 +- tests/unit/memory/test_memory_models.py | 84 ++++- .../test_attack_technique_seed_group.py | 21 +- tests/unit/models/test_seed.py | 4 +- tests/unit/models/test_seed_group.py | 10 +- .../test_seed_simulated_conversation.py | 245 ++++++++++++-- .../core/test_attack_technique_factory.py | 63 +++- .../test_default_run_size_estimates.py | 6 +- .../unit/setup/test_technique_initializer.py | 28 +- 28 files changed, 1160 insertions(+), 257 deletions(-) diff --git a/doc/code/datasets/2_seed_programming.ipynb b/doc/code/datasets/2_seed_programming.ipynb index 87c85e10af..835bf34887 100644 --- a/doc/code/datasets/2_seed_programming.ipynb +++ b/doc/code/datasets/2_seed_programming.ipynb @@ -478,7 +478,7 @@ ], "source": [ "from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH\n", - "from pyrit.models import SeedSimulatedConversation\n", + "from pyrit.models import SeedSimulatedConversation, load_next_message_prompt\n", "\n", "seed_group = AttackSeedGroup(\n", " seeds=[\n", @@ -488,10 +488,12 @@ " role=\"system\",\n", " ),\n", " SeedSimulatedConversation(\n", - " adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH / \"naive_crescendo.yaml\",\n", + " adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(EXECUTOR_RED_TEAM_PATH / \"naive_crescendo.yaml\"),\n", " sequence=1,\n", " num_turns=4,\n", - " next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / \"direct_next_message.yaml\",\n", + " next_message_system_prompt=load_next_message_prompt(\n", + " EXECUTOR_SIMULATED_TARGET_PATH / \"direct_next_message.yaml\"\n", + " ),\n", " ),\n", " ]\n", ")\n", diff --git a/doc/code/datasets/2_seed_programming.py b/doc/code/datasets/2_seed_programming.py index 134a561592..601af4b7eb 100644 --- a/doc/code/datasets/2_seed_programming.py +++ b/doc/code/datasets/2_seed_programming.py @@ -92,7 +92,7 @@ # %% from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH -from pyrit.models import SeedSimulatedConversation +from pyrit.models import SeedSimulatedConversation, load_next_message_prompt seed_group = AttackSeedGroup( seeds=[ @@ -102,10 +102,12 @@ role="system", ), SeedSimulatedConversation( - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH / "naive_crescendo.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(EXECUTOR_RED_TEAM_PATH / "naive_crescendo.yaml"), sequence=1, num_turns=4, - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "direct_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "direct_next_message.yaml" + ), ), ] ) diff --git a/doc/code/datasets/5_simulated_conversation.ipynb b/doc/code/datasets/5_simulated_conversation.ipynb index 8de223e7fa..3d7aa12b2a 100644 --- a/doc/code/datasets/5_simulated_conversation.ipynb +++ b/doc/code/datasets/5_simulated_conversation.ipynb @@ -23,7 +23,7 @@ "\n", "## Generating a Simulated Conversation\n", "\n", - "The function takes an objective, an adversarial chat model, a scorer, and a system prompt path.\n", + "The function takes an objective, an adversarial chat model, a scorer, and a system prompt.\n", "It runs a `RedTeamingAttack` internally with the adversarial LLM playing both attacker and target\n", "roles." ] @@ -63,7 +63,7 @@ "\n", "from pyrit.common.path import EXECUTOR_SEED_PROMPT_PATH\n", "from pyrit.executor.attack import generate_simulated_conversation_async\n", - "from pyrit.models import SeedGroup\n", + "from pyrit.models import SeedGroup, SeedPrompt\n", "from pyrit.output import output_attack_async\n", "from pyrit.prompt_target import OpenAIChatTarget\n", "from pyrit.score import SelfAskRefusalScorer\n", @@ -82,7 +82,9 @@ " adversarial_chat=adversarial_chat,\n", " objective_scorer=objective_scorer,\n", " num_turns=3,\n", - " adversarial_chat_system_prompt_path=Path(EXECUTOR_SEED_PROMPT_PATH) / \"red_teaming\" / \"naive_crescendo.yaml\",\n", + " adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(\n", + " Path(EXECUTOR_SEED_PROMPT_PATH) / \"red_teaming\" / \"naive_crescendo.yaml\"\n", + " ),\n", ")\n", "\n", "print(f\"Generated {len(simulated_result.seed_prompts)} messages\")" @@ -100,7 +102,7 @@ "Wrapping the prompts in a `SeedGroup` gives you convenient access to `prepended_conversation`\n", "(all turns except the last) and `next_message` (the final user message to continue from).\n", "Note that `next_message` is only populated when the last generated message has role `\"user\"` —\n", - "if you need a final user turn, pass `next_message_system_prompt_path` to the function.\n", + "if you need a final user turn, pass `next_message_system_prompt` to the function.\n", "\n", "This replaces the earlier `list[SeedPrompt]` return value. Use `result.seed_prompts` where you\n", "previously used the returned list." @@ -521,9 +523,9 @@ "| `adversarial_chat` | `PromptTarget` | The LLM that generates attack prompts (also plays the simulated target). Must declare `supports_multi_turn=True` and `supports_editable_history=True`. |\n", "| `objective_scorer` | `TrueFalseScorer` | Evaluates whether the final turn achieved the objective |\n", "| `num_turns` | `int` | Number of conversation turns to generate (default: 3) |\n", - "| `adversarial_chat_system_prompt_path` | `str \\| Path` | System prompt for the adversarial chat role |\n", - "| `simulated_target_system_prompt_path` | `str \\| Path \\| None` | Optional system prompt for the simulated target role |\n", - "| `next_message_system_prompt_path` | `str \\| Path \\| None` | Optional path to generate a final user message that elicits objective fulfillment |\n", + "| `adversarial_chat_system_prompt` | `SeedPrompt` | System prompt for the adversarial chat role |\n", + "| `simulated_target_system_prompt` | `SeedPrompt \\| None` | Optional system prompt for the simulated target role |\n", + "| `next_message_system_prompt` | `SeedPrompt \\| None` | Optional prompt that generates a final user message eliciting objective fulfillment |\n", "| `attack_converter_config` | `AttackConverterConfig \\| None` | Optional converter configuration for the attack |\n", "| `memory_labels` | `dict[str, str] \\| None` | Labels for tracking in memory |\n", "\n", diff --git a/doc/code/datasets/5_simulated_conversation.py b/doc/code/datasets/5_simulated_conversation.py index 6ee2e66025..d4e0adc552 100644 --- a/doc/code/datasets/5_simulated_conversation.py +++ b/doc/code/datasets/5_simulated_conversation.py @@ -27,7 +27,7 @@ # # ## Generating a Simulated Conversation # -# The function takes an objective, an adversarial chat model, a scorer, and a system prompt path. +# The function takes an objective, an adversarial chat model, a scorer, and a system prompt. # It runs a `RedTeamingAttack` internally with the adversarial LLM playing both attacker and target # roles. @@ -36,7 +36,7 @@ from pyrit.common.path import EXECUTOR_SEED_PROMPT_PATH from pyrit.executor.attack import generate_simulated_conversation_async -from pyrit.models import SeedGroup +from pyrit.models import SeedGroup, SeedPrompt from pyrit.output import output_attack_async from pyrit.prompt_target import OpenAIChatTarget from pyrit.score import SelfAskRefusalScorer @@ -55,7 +55,9 @@ adversarial_chat=adversarial_chat, objective_scorer=objective_scorer, num_turns=3, - adversarial_chat_system_prompt_path=Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / "naive_crescendo.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / "naive_crescendo.yaml" + ), ) print(f"Generated {len(simulated_result.seed_prompts)} messages") @@ -68,7 +70,7 @@ # Wrapping the prompts in a `SeedGroup` gives you convenient access to `prepended_conversation` # (all turns except the last) and `next_message` (the final user message to continue from). # Note that `next_message` is only populated when the last generated message has role `"user"` — -# if you need a final user turn, pass `next_message_system_prompt_path` to the function. +# if you need a final user turn, pass `next_message_system_prompt` to the function. # # This replaces the earlier `list[SeedPrompt]` return value. Use `result.seed_prompts` where you # previously used the returned list. @@ -130,9 +132,9 @@ # | `adversarial_chat` | `PromptTarget` | The LLM that generates attack prompts (also plays the simulated target). Must declare `supports_multi_turn=True` and `supports_editable_history=True`. | # | `objective_scorer` | `TrueFalseScorer` | Evaluates whether the final turn achieved the objective | # | `num_turns` | `int` | Number of conversation turns to generate (default: 3) | -# | `adversarial_chat_system_prompt_path` | `str \| Path` | System prompt for the adversarial chat role | -# | `simulated_target_system_prompt_path` | `str \| Path \| None` | Optional system prompt for the simulated target role | -# | `next_message_system_prompt_path` | `str \| Path \| None` | Optional path to generate a final user message that elicits objective fulfillment | +# | `adversarial_chat_system_prompt` | `SeedPrompt` | System prompt for the adversarial chat role | +# | `simulated_target_system_prompt` | `SeedPrompt \| None` | Optional system prompt for the simulated target role | +# | `next_message_system_prompt` | `SeedPrompt \| None` | Optional prompt that generates a final user message eliciting objective fulfillment | # | `attack_converter_config` | `AttackConverterConfig \| None` | Optional converter configuration for the attack | # | `memory_labels` | `dict[str, str] \| None` | Labels for tracking in memory | # diff --git a/doc/code/memory/3_memory_data_types.md b/doc/code/memory/3_memory_data_types.md index 6c41eb769b..7ab05d53bf 100644 --- a/doc/code/memory/3_memory_data_types.md +++ b/doc/code/memory/3_memory_data_types.md @@ -84,7 +84,7 @@ All seed types inherit from [`Seed`](../../../pyrit/models/seeds/seed.py), which - [`SeedObjective`](../../../pyrit/models/seeds/seed_objective.py) — The goal of an attack (e.g., "Generate hate speech content"). Always text. Cannot be a general technique. -- [`SeedSimulatedConversation`](../../../pyrit/models/seeds/seed_simulated_conversation.py) — Configuration for dynamically generating multi-turn conversations. Specifies system prompt paths, number of turns, and sequence offsets. The actual generation happens in the executor layer. +- [`SeedSimulatedConversation`](../../../pyrit/models/seeds/seed_simulated_conversation.py) — Configuration for dynamically generating multi-turn conversations. Carries the adversarial, simulated-target, and next-message system prompts, the number of turns, and sequence offsets. The actual generation happens in the executor layer. ### Seed Groups diff --git a/pyrit/executor/attack/core/attack_parameters.py b/pyrit/executor/attack/core/attack_parameters.py index a6eaedd4a1..a45a9197da 100644 --- a/pyrit/executor/attack/core/attack_parameters.py +++ b/pyrit/executor/attack/core/attack_parameters.py @@ -169,9 +169,9 @@ async def from_seed_group_async( objective_scorer=objective_scorer, num_turns=simulated_conversation_config.num_turns, starting_sequence=simulated_conversation_config.sequence, - adversarial_chat_system_prompt_path=simulated_conversation_config.adversarial_chat_system_prompt_path, - simulated_target_system_prompt_path=simulated_conversation_config.simulated_target_system_prompt_path, - next_message_system_prompt_path=simulated_conversation_config.next_message_system_prompt_path, + adversarial_chat_system_prompt=simulated_conversation_config.adversarial_chat_system_prompt, + simulated_target_system_prompt=simulated_conversation_config.simulated_target_system_prompt, + next_message_system_prompt=simulated_conversation_config.next_message_system_prompt, ) simulated_prompts = simulated_result.seed_prompts if "source_conversations" in valid_fields: diff --git a/pyrit/executor/attack/multi_turn/simulated_conversation.py b/pyrit/executor/attack/multi_turn/simulated_conversation.py index b116ca77f8..7a9a2fb7dd 100644 --- a/pyrit/executor/attack/multi_turn/simulated_conversation.py +++ b/pyrit/executor/attack/multi_turn/simulated_conversation.py @@ -10,6 +10,7 @@ from __future__ import annotations +import asyncio import logging from dataclasses import dataclass from typing import TYPE_CHECKING @@ -31,11 +32,14 @@ ConversationType, Message, SeedPrompt, - SeedSimulatedConversation, + load_next_message_prompt, + load_simulated_target_prompt, ) +from pyrit.models.seeds.seed_simulated_conversation import _warn_prompt_path_deprecated from pyrit.prompt_normalizer import PromptNormalizer if TYPE_CHECKING: + from collections.abc import Callable from pathlib import Path from pyrit.prompt_target import PromptTarget @@ -52,6 +56,39 @@ class SimulatedConversationResult: related_conversations: frozenset[ConversationReference] +async def _resolve_prompt_source_async( + *, + prompt: SeedPrompt | None, + path: str | Path | None, + prompt_name: str, + path_name: str, + load_prompt: Callable[[str | Path], SeedPrompt], +) -> SeedPrompt | None: + """ + Resolve a prompt source, loading a deprecated path input off the event loop. + + The conflict check and the deprecation warning stay on the event loop so the warning points + at the caller; only the file read is sent to a worker thread. + + Args: + prompt: The canonical prompt, if the caller supplied one. + path: The deprecated path input, if the caller supplied one. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + load_prompt: Loader that turns the path into a prompt. + + Returns: + SeedPrompt | None: The resolved prompt, or None when neither input was supplied. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if path is None: + return prompt + _warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + return await asyncio.to_thread(load_prompt, path) + + async def generate_simulated_conversation_async( *, objective: str, @@ -59,7 +96,10 @@ async def generate_simulated_conversation_async( objective_scorer: TrueFalseScorer, num_turns: int = 3, starting_sequence: int = 0, - adversarial_chat_system_prompt_path: str | Path, + adversarial_chat_system_prompt: SeedPrompt | None = None, + simulated_target_system_prompt: SeedPrompt | None = None, + next_message_system_prompt: SeedPrompt | None = None, + adversarial_chat_system_prompt_path: str | Path | None = None, simulated_target_system_prompt_path: str | Path | None = None, next_message_system_prompt_path: str | Path | None = None, attack_converter_config: AttackConverterConfig | None = None, @@ -86,14 +126,21 @@ async def generate_simulated_conversation_async( num_turns: Number of conversation turns to generate. Defaults to 3. starting_sequence: The starting sequence number for the generated SeedPrompts. Each message gets an incrementing sequence number. Defaults to 0. - adversarial_chat_system_prompt_path: Path to the system prompt for the adversarial chat. - simulated_target_system_prompt_path: Path to the system prompt for the simulated target. + adversarial_chat_system_prompt: System prompt for the adversarial chat. Required unless + the deprecated path input is used. + simulated_target_system_prompt: System prompt for the simulated target. If None, no system prompt is used for the simulated target. - next_message_system_prompt_path: Optional path to a system prompt for generating - a final user message. If provided, after the simulated conversation, a single - LLM call generates a user message that attempts to get the target to fulfill - the objective in their next response. The prompt template receives `objective` - and `conversation_so_far` parameters. + next_message_system_prompt: Optional system prompt for generating a final user message. + If provided, after the simulated conversation, a single LLM call generates a user + message that attempts to get the target to fulfill the objective in their next + response. The prompt template receives `objective` and `conversation_context` + parameters. + adversarial_chat_system_prompt_path: Deprecated. Path to the adversarial chat system + prompt YAML. Use ``adversarial_chat_system_prompt`` instead. + simulated_target_system_prompt_path: Deprecated. Path to the simulated target system + prompt YAML. Use ``simulated_target_system_prompt`` instead. + next_message_system_prompt_path: Deprecated. Path to the next message system prompt + YAML. Use ``next_message_system_prompt`` instead. attack_converter_config: Converter configuration for the attack. Defaults to None. memory_labels: Labels to associate with the conversation in memory. Defaults to None. @@ -102,7 +149,8 @@ async def generate_simulated_conversation_async( start from ``starting_sequence`` and increment by 1 for each message. Raises: - ValueError: If num_turns is not a positive integer. + ValueError: If num_turns is not a positive integer, if no adversarial chat system prompt + is supplied, or if a prompt and its deprecated path input are both supplied. """ # Use the same LLM for both adversarial chat and simulated target # They get different system prompts to play different roles @@ -110,22 +158,40 @@ async def generate_simulated_conversation_async( if num_turns <= 0: raise ValueError("num_turns must be a positive integer") - # Load and configure simulated target system prompt using centralized validation - # Returns None if no path is provided (no system prompt for simulated target) - simulated_target_system_prompt = SeedSimulatedConversation.load_simulated_target_system_prompt( - objective=objective, - num_turns=num_turns, - simulated_target_system_prompt_path=simulated_target_system_prompt_path, + adversarial_chat_system_prompt = await _resolve_prompt_source_async( + prompt=adversarial_chat_system_prompt, + path=adversarial_chat_system_prompt_path, + prompt_name="adversarial_chat_system_prompt", + path_name="adversarial_chat_system_prompt_path", + load_prompt=SeedPrompt.from_yaml_file, ) - - # Create adversarial config for the simulation. Load the optional path into a SeedPrompt so the - # resolved prompt is stored directly on the configuration. - adversarial_system_prompt = ( - SeedPrompt.from_yaml_file(adversarial_chat_system_prompt_path) if adversarial_chat_system_prompt_path else None + simulated_target_system_prompt = await _resolve_prompt_source_async( + prompt=simulated_target_system_prompt, + path=simulated_target_system_prompt_path, + prompt_name="simulated_target_system_prompt", + path_name="simulated_target_system_prompt_path", + load_prompt=load_simulated_target_prompt, + ) + next_message_system_prompt = await _resolve_prompt_source_async( + prompt=next_message_system_prompt, + path=next_message_system_prompt_path, + prompt_name="next_message_system_prompt", + path_name="next_message_system_prompt_path", + load_prompt=load_next_message_prompt, ) + if adversarial_chat_system_prompt is None: + raise ValueError("adversarial_chat_system_prompt is required") + + # Render the simulated target system prompt; None means the target gets no system prompt. + simulated_target_system_prompt_value = ( + simulated_target_system_prompt.render_template_value(objective=objective, num_turns=num_turns) + if simulated_target_system_prompt is not None + else None + ) + adversarial_config = AttackAdversarialConfig( target=adversarial_chat, - system_prompt=adversarial_system_prompt, + system_prompt=adversarial_chat_system_prompt, ) # Create scoring config @@ -149,8 +215,8 @@ async def generate_simulated_conversation_async( # Build prepended_conversation - only include system message if prompt is provided prepended_conversation: list[Message] = [] - if simulated_target_system_prompt: - prepended_conversation.append(Message.from_system_prompt(simulated_target_system_prompt)) + if simulated_target_system_prompt_value: + prepended_conversation.append(Message.from_system_prompt(simulated_target_system_prompt_value)) result = await attack.execute_async( objective=objective, @@ -175,15 +241,15 @@ async def generate_simulated_conversation_async( *result.related_conversations, } - # If next_message_system_prompt_path is provided, generate a final user message - if next_message_system_prompt_path: + # If a next-message prompt is configured, generate a final user message + if next_message_system_prompt: next_message_conversation_id = str(uuid4()) next_message = await _generate_next_message_async( objective=objective, conversation_messages=conversation_messages, adversarial_chat=adversarial_chat, conversation_id=next_message_conversation_id, - next_message_system_prompt_path=next_message_system_prompt_path, + next_message_system_prompt=next_message_system_prompt, prompt_normalizer=PromptNormalizer(), memory_labels=memory_labels, ) @@ -216,7 +282,7 @@ async def _generate_next_message_async( conversation_messages: list[Message], adversarial_chat: PromptTarget, conversation_id: str, - next_message_system_prompt_path: str | Path, + next_message_system_prompt: SeedPrompt, prompt_normalizer: PromptNormalizer, memory_labels: dict[str, str] | None = None, ) -> Message: @@ -234,7 +300,7 @@ async def _generate_next_message_async( conversation_messages: The conversation generated so far as Messages. adversarial_chat: The LLM to use for generation. conversation_id: The conversation ID for the adversarial generation exchange. - next_message_system_prompt_path: Path to the system prompt template. + next_message_system_prompt: The system prompt template. prompt_normalizer: The normalizer the manager sends the adversarial turn through. memory_labels: Optional memory labels to attach to the request. @@ -249,17 +315,10 @@ async def _generate_next_message_async( normalizer = ConversationContextNormalizer() conversation_context = await normalizer.normalize_string_async(conversation_messages) - # Load the system prompt template (schema + prompt are a matched pair declared in the YAML) - template = SeedPrompt.from_yaml_with_required_parameters( - template_path=next_message_system_prompt_path, - required_parameters=["objective", "conversation_context"], - error_message="Next message system prompt must have objective and conversation_context parameters", - ) - # The manager owns schema resolution, setting the system prompt, the send, and JSON parse/retry. manager = _AdversarialConversationManager( adversarial_target=adversarial_chat, - adversarial_system_prompt=template, + adversarial_system_prompt=next_message_system_prompt, prompt_normalizer=prompt_normalizer, conversation_id=conversation_id, objective=objective, diff --git a/pyrit/memory/memory_models.py b/pyrit/memory/memory_models.py index c6fff1de8f..62700f8c3a 100644 --- a/pyrit/memory/memory_models.py +++ b/pyrit/memory/memory_models.py @@ -1624,6 +1624,10 @@ def get_seed(self) -> Seed: Returns: Seed: The reconstructed seed object (SeedPrompt, SeedObjective, or SeedSimulatedConversation) + + Raises: + ValueError: If a simulated conversation record cannot be rebuilt, for example when it + names a prompt file that is not present on this machine. """ cleaned_metadata, decoded_schema = self._unpack_seed_metadata(self.prompt_metadata) if self.seed_type == "objective": @@ -1644,28 +1648,52 @@ def get_seed(self) -> Seed: prompt_group_id=self.prompt_group_id, ) if self.seed_type == "simulated_conversation": - # Reconstruct SeedSimulatedConversation from JSON value + # Reconstruct SeedSimulatedConversation from JSON value. Records written before the + # prompts were normalized carry only ``*_path`` keys; the model's compatibility + # adapter resolves those, and a canonicalized record loses the stale hash of its + # old path-shaped value. config = json.loads(self.value) - return SeedSimulatedConversation( - id=self.id, - value_sha256=self.value_sha256, - name=self.name, - dataset_name=self.dataset_name, - harm_categories=self.harm_categories, - description=self.description, - authors=self.authors, - groups=self.groups, - source=self.source, - date_added=self.date_added, - added_by=self.added_by, - metadata=cleaned_metadata, - prompt_group_id=self.prompt_group_id, - num_turns=config.get("num_turns", 3), - sequence=config.get("sequence", 0), - adversarial_chat_system_prompt_path=config.get("adversarial_chat_system_prompt_path"), - simulated_target_system_prompt_path=config.get("simulated_target_system_prompt_path"), - next_message_system_prompt_path=config.get("next_message_system_prompt_path"), - ) + prompt_config = { + key: config[key] + for key in ( + "adversarial_chat_system_prompt", + "adversarial_chat_system_prompt_path", + "simulated_target_system_prompt", + "simulated_target_system_prompt_path", + "next_message_system_prompt", + "next_message_system_prompt_path", + ) + if config.get(key) is not None + } + is_legacy_record = any(key.endswith("_path") for key in prompt_config) + try: + return SeedSimulatedConversation( + id=self.id, + value_sha256=None if is_legacy_record else self.value_sha256, + name=self.name, + dataset_name=self.dataset_name, + harm_categories=self.harm_categories, + description=self.description, + authors=self.authors, + groups=self.groups, + source=self.source, + date_added=self.date_added, + added_by=self.added_by, + metadata=cleaned_metadata, + prompt_group_id=self.prompt_group_id, + num_turns=config.get("num_turns", 3), + sequence=config.get("sequence", 0), + pyrit_version=config.get("pyrit_version"), + **prompt_config, + ) + except (OSError, ValueError) as exc: + # A legacy record names prompt files by absolute path, so one written elsewhere + # can reference a file this machine does not have. Name the record so a single + # bad row is identifiable rather than an opaque failure of the whole query. + raise ValueError( + f"Could not rebuild simulated conversation seed {self.id} " + f"(name={self.name!r}, dataset={self.dataset_name!r}): {exc}" + ) from exc return SeedPrompt( id=self.id, value=self.value, diff --git a/pyrit/models/__init__.py b/pyrit/models/__init__.py index 0596833912..071d89a796 100644 --- a/pyrit/models/__init__.py +++ b/pyrit/models/__init__.py @@ -162,6 +162,8 @@ SeedUnion, SimulatedTargetSystemPromptPaths, group_seeds_into_attack_groups, + load_next_message_prompt, + load_simulated_target_prompt, ) from pyrit.models.target import ( COMMON_JSON_SCHEMAS, @@ -311,6 +313,8 @@ "SeedType": "pyrit.models.literals", "SeedUnion": "pyrit.models.seeds", "SimulatedTargetSystemPromptPaths": "pyrit.models.seeds", + "load_next_message_prompt": "pyrit.models.seeds", + "load_simulated_target_prompt": "pyrit.models.seeds", "snake_case_to_class_name": "pyrit.models.identifiers", "sort_message_pieces": "pyrit.models.messages.message_piece", "StrategyResult": "pyrit.models.results.strategy_result", diff --git a/pyrit/models/seeds/__init__.py b/pyrit/models/seeds/__init__.py index dbb015bf05..4a82cf7334 100644 --- a/pyrit/models/seeds/__init__.py +++ b/pyrit/models/seeds/__init__.py @@ -33,6 +33,8 @@ NextMessageSystemPromptPaths, SeedSimulatedConversation, SimulatedTargetSystemPromptPaths, + load_next_message_prompt, + load_simulated_target_prompt, ) from pyrit.models.seeds.yaml_seed_loader import ( load_seed_dataset_from_yaml, @@ -44,6 +46,8 @@ "load_seed_dataset_from_yaml": "pyrit.models.seeds.yaml_seed_loader", "load_seed_from_yaml": "pyrit.models.seeds.yaml_seed_loader", "load_seed_prompt_from_yaml_with_required_parameters": "pyrit.models.seeds.yaml_seed_loader", + "load_next_message_prompt": "pyrit.models.seeds.seed_simulated_conversation", + "load_simulated_target_prompt": "pyrit.models.seeds.seed_simulated_conversation", "group_seeds_into_attack_groups": "pyrit.models.seeds.seed_grouping", "NextMessageSystemPromptPaths": "pyrit.models.seeds.seed_simulated_conversation", "Seed": "pyrit.models.seeds.seed", diff --git a/pyrit/models/seeds/seed_simulated_conversation.py b/pyrit/models/seeds/seed_simulated_conversation.py index f5c29525dc..ff02efcc8f 100644 --- a/pyrit/models/seeds/seed_simulated_conversation.py +++ b/pyrit/models/seeds/seed_simulated_conversation.py @@ -19,16 +19,29 @@ import json import logging from pathlib import Path -from typing import Any, Literal +from typing import TYPE_CHECKING, Annotated, Any, Literal -from pydantic import field_validator, model_validator +from pydantic import Field, WrapValidator, field_validator, model_validator +from pyrit.common.deprecation import print_deprecation_message from pyrit.common.path import EXECUTOR_SIMULATED_TARGET_PATH from pyrit.models.seeds.seed import Seed from pyrit.models.seeds.seed_prompt import SeedPrompt +if TYPE_CHECKING: + from collections.abc import Callable + + from pydantic import ValidatorFunctionWrapHandler + logger = logging.getLogger(__name__) +SIMULATED_TARGET_REQUIRED_PARAMETERS = ["objective", "num_turns"] +SIMULATED_TARGET_PARAMETER_ERROR = "Simulated target system prompt must have objective and num_turns parameters" +NEXT_MESSAGE_REQUIRED_PARAMETERS = ["objective", "conversation_context"] +NEXT_MESSAGE_PARAMETER_ERROR = "Next message system prompt must have objective and conversation_context parameters" + +_PROMPT_PATH_REMOVED_IN = "1.4.0" + class SimulatedTargetSystemPromptPaths(enum.Enum): """Enum for predefined simulated target system prompt paths.""" @@ -42,11 +55,178 @@ class NextMessageSystemPromptPaths(enum.Enum): DIRECT = Path(EXECUTOR_SIMULATED_TARGET_PATH, "direct_next_message.yaml").resolve() +def load_simulated_target_prompt(template_path: str | Path) -> SeedPrompt: + """ + Load a simulated target system prompt template and verify it declares the parameters it needs. + + Args: + template_path: Path to the YAML file containing the prompt template. + + Returns: + SeedPrompt: The loaded template. + + Raises: + ValueError: If the template does not declare ``objective`` and ``num_turns``. + """ + return SeedPrompt.from_yaml_with_required_parameters( + template_path=template_path, + required_parameters=SIMULATED_TARGET_REQUIRED_PARAMETERS, + error_message=SIMULATED_TARGET_PARAMETER_ERROR, + ) + + +def load_next_message_prompt(template_path: str | Path) -> SeedPrompt: + """ + Load a next-message system prompt template and verify it declares the parameters it needs. + + Args: + template_path: Path to the YAML file containing the prompt template. + + Returns: + SeedPrompt: The loaded template. + + Raises: + ValueError: If the template does not declare ``objective`` and ``conversation_context``. + """ + return SeedPrompt.from_yaml_with_required_parameters( + template_path=template_path, + required_parameters=NEXT_MESSAGE_REQUIRED_PARAMETERS, + error_message=NEXT_MESSAGE_PARAMETER_ERROR, + ) + + +def _load_compliant_simulated_target_prompt() -> SeedPrompt: + """ + Load the default compliant simulated target prompt. + + Returns: + SeedPrompt: The compliant simulated target template. + """ + return load_simulated_target_prompt(SimulatedTargetSystemPromptPaths.COMPLIANT.value) + + +def _resolve_prompt_source( + *, + prompt: SeedPrompt | None, + path: str | Path | None, + prompt_name: str, + path_name: str, + load_prompt: Callable[[str | Path], SeedPrompt], +) -> SeedPrompt | None: + """ + Choose between a canonical prompt and its deprecated path input, loading the path if needed. + + Every boundary that still accepts a ``*_system_prompt_path`` uses this so they all warn the + same way and reject the same ambiguity. It reads from disk, so async callers must run it + through ``asyncio.to_thread``. + + Args: + prompt: The canonical prompt, if the caller supplied one. + path: The deprecated path input, if the caller supplied one. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + load_prompt: Loader that turns the path into a prompt. + + Returns: + SeedPrompt | None: The resolved prompt, or None when neither input was supplied. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if path is None: + return prompt + _warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + return load_prompt(path) + + +def _warn_prompt_path_deprecated(*, prompt: SeedPrompt | None, prompt_name: str, path_name: str) -> None: + """ + Reject an ambiguous prompt source and warn that the path input is deprecated. + + Separated from loading so async callers can run this on the event loop, where the warning + points at their own call site, and send only the file read to a worker thread. + + Args: + prompt: The canonical prompt, if the caller supplied one alongside the path. + prompt_name: Name of the canonical parameter, used in messages. + path_name: Name of the deprecated parameter, used in messages. + + Raises: + ValueError: If both the canonical prompt and its deprecated path are supplied. + """ + if prompt is not None: + raise ValueError(f"Set only one of {prompt_name} or {path_name}; both were provided.") + print_deprecation_message(old_item=path_name, new_item=prompt_name, removed_in=_PROMPT_PATH_REMOVED_IN) + + +# Deprecated ``*_path`` inputs, mapped to the canonical field they populate and the loader that +# resolves them. These are accepted at construction only; they never become model fields. +_LEGACY_PROMPT_PATH_INPUTS: dict[str, tuple[str, Any]] = { + "adversarial_chat_system_prompt_path": ("adversarial_chat_system_prompt", SeedPrompt.from_yaml_file), + "simulated_target_system_prompt_path": ("simulated_target_system_prompt", load_simulated_target_prompt), + "next_message_system_prompt_path": ("next_message_system_prompt", load_next_message_prompt), +} + + +def _prompt_identity(prompt: SeedPrompt | None) -> dict[str, Any] | None: + """ + Project a prompt onto the fields that change how a simulated conversation behaves. + + Only the fields that alter rendering, validation, or the response contract are kept, so they + survive reconstruction from a persisted record. Descriptive metadata such as ``name``, + ``description``, and ``source`` is deliberately left out: renaming a template should not change + the configuration's identity. Those fields are therefore not restored from a persisted record. + + ``is_jinja_template`` is also left out. It marks a template that still needs its one-shot + path substitution, and the projected value has already had it, so restoring the flag would + make a rebuilt prompt render a second time. + + Args: + prompt: The prompt to project, or None. + + Returns: + dict[str, Any] | None: The projected prompt, or None when no prompt was given. + """ + if prompt is None: + return None + return { + "value": prompt.value, + "data_type": prompt.data_type, + "parameters": list(prompt.parameters or []), + "response_json_schema": prompt.response_json_schema, + } + + +def _keep_prompt_instance(value: Any, handler: ValidatorFunctionWrapHandler) -> Any: + """ + Accept an existing prompt as-is instead of validating it again. + + ``SeedPrompt`` substitutes dataset paths into a trusted template once, while it is being + validated. Re-validating an instance would run that substitution a second time, which + consumes template syntax the executor is meant to fill in and mutates the caller's object. + + Args: + value: The raw value supplied for the field. + handler: The validator to fall back to for anything that is not already a prompt. + + Returns: + Any: The prompt unchanged, or the result of normal validation. + """ + if isinstance(value, SeedPrompt): + return value + return handler(value) + + +# Prompt fields hold templates that have already been prepared, so an existing instance is +# never re-validated (see _keep_prompt_instance). +SystemPrompt = Annotated[SeedPrompt, WrapValidator(_keep_prompt_instance)] + + class SeedSimulatedConversation(Seed): """ Configuration for generating a simulated conversation dynamically. - This class holds the paths and parameters needed to generate prepended conversation + This class holds the prompts and parameters needed to generate prepended conversation content by running an adversarial chat against a simulated (compliant) target. This is a pure configuration class. The actual generation is performed by @@ -56,12 +236,23 @@ class SeedSimulatedConversation(Seed): The `value` property returns a JSON serialization of the config for database storage and deduplication. + The prompts are canonical `SeedPrompt` templates, so a technique carries its prompt text + rather than a file location and can be inspected or edited in place. The matching + `*_system_prompt_path` inputs are still accepted at construction for legacy callers and for + reading records persisted before the change; they are resolved immediately and are not + stored on the model. + + To change a prompt, edit the `SeedPrompt` and then build a new `SeedSimulatedConversation` + from it. Like the other fields, `value` is a snapshot taken when the configuration is + validated, so mutating a prompt on an existing instance changes what executes without + changing what is stored. + Attributes: num_turns: Number of conversation turns to generate. - adversarial_chat_system_prompt_path: Path to the adversarial chat system prompt YAML. - simulated_target_system_prompt_path: Path to the simulated target system prompt YAML. + adversarial_chat_system_prompt: System prompt template for the adversarial chat. + simulated_target_system_prompt: System prompt template for the simulated target. Defaults to the compliant prompt if not specified. - next_message_system_prompt_path: Optional path to the system prompt for generating + next_message_system_prompt: Optional system prompt template for generating an additional user message after the simulated conversation. If provided, a single LLM call generates a final user message that attempts to get the target to fulfill the objective in their next response. @@ -85,9 +276,12 @@ class SeedSimulatedConversation(Seed): num_turns: int = 3 sequence: int = 0 - adversarial_chat_system_prompt_path: Path - simulated_target_system_prompt_path: Path = SimulatedTargetSystemPromptPaths.COMPLIANT.value - next_message_system_prompt_path: Path | None = None + # A prompt supplied directly is trusted as-is. The declared-parameter contract for the + # simulated target and next message is enforced where a template is loaded from a file, + # by load_simulated_target_prompt and load_next_message_prompt. + adversarial_chat_system_prompt: SystemPrompt + simulated_target_system_prompt: SystemPrompt = Field(default_factory=_load_compliant_simulated_target_prompt) + next_message_system_prompt: SystemPrompt | None = None pyrit_version: str | None = None @model_validator(mode="before") @@ -106,14 +300,51 @@ def _strip_user_value(cls, data: Any) -> Any: data.pop("value", None) return data - @field_validator("simulated_target_system_prompt_path", mode="before") + @field_validator("simulated_target_system_prompt", mode="before") @classmethod - def _default_simulated_target_path(cls, value: Any) -> Any: + def _default_simulated_target_prompt(cls, value: Any) -> Any: # Reconstruction from memory may pass an explicit None; fall back to the compliant default. if value is None: - return SimulatedTargetSystemPromptPaths.COMPLIANT.value + return _load_compliant_simulated_target_prompt() return value + @model_validator(mode="before") + @classmethod + def _resolve_legacy_prompt_paths(cls, data: Any) -> Any: + """ + Resolve deprecated ``*_system_prompt_path`` inputs into their canonical prompt fields. + + Runs in ``mode="before"`` so the path keys are removed from the input before Pydantic's + ``extra="forbid"`` rejects them. The paths are therefore never model fields and never + reach ``value`` or ``get_identifier()``. + + Args: + data: Raw input passed to the model constructor. + + Returns: + The input with any legacy path keys replaced by loaded prompts. + + Raises: + ValueError: If a canonical prompt and its deprecated path are both supplied. + """ + if not isinstance(data, dict): + return data + + resolved = data + for path_key, (prompt_key, load_prompt) in _LEGACY_PROMPT_PATH_INPUTS.items(): + if path_key not in resolved: + continue + if resolved is data: + resolved = dict(data) + resolved[prompt_key] = _resolve_prompt_source( + prompt=resolved.get(prompt_key), + path=resolved.pop(path_key), + prompt_name=f"SeedSimulatedConversation.{prompt_key}", + path_name=f"SeedSimulatedConversation.{path_key}", + load_prompt=load_prompt, + ) + return resolved + @model_validator(mode="after") def _validate_and_compute_value(self) -> SeedSimulatedConversation: if self.num_turns <= 0: @@ -125,25 +356,32 @@ def _validate_and_compute_value(self) -> SeedSimulatedConversation: self.value = self._compute_value() return self - def _compute_value(self) -> str: + def _config_dict(self) -> dict[str, Any]: """ - Compute the value field as JSON serialization of config. + Build the canonical configuration mapping shared by ``value`` and ``get_identifier()``. Returns: - str: Deterministic JSON representation of this configuration. + dict[str, Any]: The configuration, with each prompt reduced to its behavioral fields. """ - config = { + return { "num_turns": self.num_turns, "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), - "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), - "next_message_system_prompt_path": ( - str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None - ), + "adversarial_chat_system_prompt": _prompt_identity(self.adversarial_chat_system_prompt), + "simulated_target_system_prompt": _prompt_identity(self.simulated_target_system_prompt), + "next_message_system_prompt": _prompt_identity(self.next_message_system_prompt), "pyrit_version": self.pyrit_version, } - return json.dumps(config, sort_keys=True, separators=(",", ":")) + + def _compute_value(self) -> str: + """ + Compute the value field as JSON serialization of config. + + Returns: + str: Deterministic JSON representation of this configuration. + + """ + return json.dumps(self._config_dict(), sort_keys=True, separators=(",", ":")) def get_identifier(self) -> dict[str, Any]: """ @@ -153,17 +391,7 @@ def get_identifier(self) -> dict[str, Any]: Dictionary with configuration details. """ - return { - "__type__": "SeedSimulatedConversation", - "num_turns": self.num_turns, - "sequence": self.sequence, - "adversarial_chat_system_prompt_path": str(self.adversarial_chat_system_prompt_path), - "simulated_target_system_prompt_path": str(self.simulated_target_system_prompt_path), - "next_message_system_prompt_path": ( - str(self.next_message_system_prompt_path) if self.next_message_system_prompt_path else None - ), - "pyrit_version": self.pyrit_version, - } + return {"__type__": "SeedSimulatedConversation", **self._config_dict()} def compute_hash(self) -> str: """ @@ -187,6 +415,11 @@ def load_simulated_target_system_prompt( """ Load and render the simulated target system prompt. + .. deprecated:: + Render ``SeedSimulatedConversation.simulated_target_system_prompt`` directly with + ``SeedPrompt.render_template_value(objective=..., num_turns=...)`` instead. This + helper reads from disk, so it must not be called from an async path. + If no path is provided, returns None (no system prompt). Validates that the template has required `objective` and `num_turns` parameters. @@ -203,14 +436,15 @@ def load_simulated_target_system_prompt( ValueError: If the template doesn't have required parameters. """ + print_deprecation_message( + old_item="SeedSimulatedConversation.load_simulated_target_system_prompt", + new_item="SeedSimulatedConversation.simulated_target_system_prompt.render_template_value", + removed_in=_PROMPT_PATH_REMOVED_IN, + ) if simulated_target_system_prompt_path is None: return None - template = SeedPrompt.from_yaml_with_required_parameters( - template_path=simulated_target_system_prompt_path, - required_parameters=["objective", "num_turns"], - error_message="Simulated target system prompt must have objective and num_turns parameters", - ) + template = load_simulated_target_prompt(simulated_target_system_prompt_path) return template.render_template_value( objective=objective, @@ -223,14 +457,14 @@ def sequence_range(self) -> range: The range of sequence numbers this simulated conversation will occupy. Each turn generates 2 messages (user + assistant), so num_turns generates - num_turns * 2 messages. If next_message_system_prompt_path is set, an additional + num_turns * 2 messages. If next_message_system_prompt is set, an additional user message is added at the end. Returns: A range object representing the sequence numbers. """ - message_count = self.num_turns * 2 + (1 if self.next_message_system_prompt_path else 0) + message_count = self.num_turns * 2 + (1 if self.next_message_system_prompt else 0) return range(self.sequence, self.sequence + message_count) def __repr__(self) -> str: @@ -241,9 +475,9 @@ def __repr__(self) -> str: str: Simulated conversation summary string. """ - has_next_msg = self.next_message_system_prompt_path is not None + has_next_msg = self.next_message_system_prompt is not None return ( f"" + f"adversarial_prompt={self.adversarial_chat_system_prompt.name or ''})>" ) diff --git a/pyrit/scenario/core/attack_technique_factory.py b/pyrit/scenario/core/attack_technique_factory.py index 9c373512f2..67ef95ce8d 100644 --- a/pyrit/scenario/core/attack_technique_factory.py +++ b/pyrit/scenario/core/attack_technique_factory.py @@ -36,8 +36,10 @@ SeedIdentifier, SeedPrompt, SeedSimulatedConversation, + load_next_message_prompt, + load_simulated_target_prompt, ) -from pyrit.models.seeds.seed_simulated_conversation import NextMessageSystemPromptPaths +from pyrit.models.seeds.seed_simulated_conversation import NextMessageSystemPromptPaths, _resolve_prompt_source from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target @@ -165,6 +167,9 @@ def with_simulated_conversation( name: str, attack_class: type[AttackStrategy[Any, Any]] | None = None, description: str | None = None, + adversarial_chat_system_prompt: SeedPrompt | None = None, + simulated_target_system_prompt: SeedPrompt | None = None, + next_message_system_prompt: SeedPrompt | None = None, adversarial_chat_system_prompt_path: str | Path | None = None, simulated_target_system_prompt_path: str | Path | None = None, next_message_system_prompt_path: str | Path | None = None, @@ -192,18 +197,22 @@ def with_simulated_conversation( ``PromptSendingAttack``. description: Short human-readable summary of what the technique does. Forwarded to the factory constructor as descriptive metadata. - adversarial_chat_system_prompt_path: Path to the YAML file containing - the adversarial chat system prompt for the simulated conversation. - Defaults to ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``. - simulated_target_system_prompt_path: Optional path to the YAML file - containing the system prompt for the simulated target (the - assistant side of the generated conversation). When ``None``, + adversarial_chat_system_prompt: System prompt for the adversarial chat in the + simulated conversation. Defaults to the prompt loaded from + ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``. + simulated_target_system_prompt: Optional system prompt for the simulated target + (the assistant side of the generated conversation). When ``None``, ``SeedSimulatedConversation`` falls back to its compliant default. - next_message_system_prompt_path: Optional path to the YAML file - containing the system prompt for generating a final user message - after the simulated conversation. Defaults to - ``NextMessageSystemPromptPaths.DIRECT.value``. Ignored (forced to - ``None``) when ``final_user_message`` is provided. + next_message_system_prompt: Optional system prompt for generating a final user + message after the simulated conversation. Defaults to the prompt loaded from + ``NextMessageSystemPromptPaths.DIRECT``. Ignored (forced to ``None``) when + ``final_user_message`` is provided. + adversarial_chat_system_prompt_path: Deprecated. Path to the YAML file containing + the adversarial chat system prompt. Use ``adversarial_chat_system_prompt``. + simulated_target_system_prompt_path: Deprecated. Path to the YAML file containing + the simulated target system prompt. Use ``simulated_target_system_prompt``. + next_message_system_prompt_path: Deprecated. Path to the YAML file containing the + next-message system prompt. Use ``next_message_system_prompt``. final_user_message: Optional fixed final user message. When provided, a static ``SeedPrompt`` carrying this text is appended after the simulated conversation (so it becomes the ``next_message``) and no @@ -240,29 +249,47 @@ def with_simulated_conversation( """ if attack_class is None: attack_class = PromptSendingAttack - if adversarial_chat_system_prompt_path is None: - adversarial_chat_system_prompt_path = Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml" + adversarial_chat_system_prompt = _resolve_prompt_source( + prompt=adversarial_chat_system_prompt, + path=adversarial_chat_system_prompt_path, + prompt_name="adversarial_chat_system_prompt", + path_name="adversarial_chat_system_prompt_path", + load_prompt=SeedPrompt.from_yaml_file, + ) + if adversarial_chat_system_prompt is None: + adversarial_chat_system_prompt = SeedPrompt.from_yaml_file( + Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml" + ) + simulated_target_system_prompt = _resolve_prompt_source( + prompt=simulated_target_system_prompt, + path=simulated_target_system_prompt_path, + prompt_name="simulated_target_system_prompt", + path_name="simulated_target_system_prompt_path", + load_prompt=load_simulated_target_prompt, + ) # A fixed final user message and an LLM-generated next message are mutually # exclusive: when a fixed message is supplied it becomes the next_message via # a static SeedPrompt, so no next-message generation prompt is used. if final_user_message is not None: - next_message_system_prompt_path = None - elif next_message_system_prompt_path is None: - next_message_system_prompt_path = NextMessageSystemPromptPaths.DIRECT.value - - simulated_conversation_kwargs: dict[str, Any] = { - "adversarial_chat_system_prompt_path": Path(adversarial_chat_system_prompt_path), - "num_turns": num_turns, - } - if simulated_target_system_prompt_path is not None: - simulated_conversation_kwargs["simulated_target_system_prompt_path"] = Path( - simulated_target_system_prompt_path + next_message_system_prompt = None + else: + next_message_system_prompt = _resolve_prompt_source( + prompt=next_message_system_prompt, + path=next_message_system_prompt_path, + prompt_name="next_message_system_prompt", + path_name="next_message_system_prompt_path", + load_prompt=load_next_message_prompt, ) - if next_message_system_prompt_path is not None: - simulated_conversation_kwargs["next_message_system_prompt_path"] = Path(next_message_system_prompt_path) - - simulated_conversation = SeedSimulatedConversation(**simulated_conversation_kwargs) + if next_message_system_prompt is None: + next_message_system_prompt = load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value) + + simulated_conversation = SeedSimulatedConversation( + num_turns=num_turns, + adversarial_chat_system_prompt=adversarial_chat_system_prompt, + simulated_target_system_prompt=simulated_target_system_prompt, + next_message_system_prompt=next_message_system_prompt, + ) seeds: list[Any] = [simulated_conversation] if final_user_message is not None: diff --git a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py index 16965aab65..3de77591d9 100644 --- a/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py +++ b/pyrit/scenario/scenarios/adaptive/adaptive_scenario.py @@ -16,6 +16,7 @@ from __future__ import annotations +import asyncio import logging from abc import abstractmethod from typing import TYPE_CHECKING, ClassVar @@ -177,7 +178,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list Raises: ValueError: If ``_build_techniques_dict`` finds no usable techniques. """ - techniques = self._build_techniques_dict(objective_target=context.objective_target) + # Building the technique catalog reads each technique's prompt YAML, so keep the + # synchronous builder off the event loop. + techniques = await asyncio.to_thread(self._build_techniques_dict, objective_target=context.objective_target) atomic_attacks: list[AtomicAttack] = [] if context.include_baseline: @@ -243,7 +246,9 @@ async def _estimate_run_size_async(self) -> ScenarioRunSizeEstimate: ) assert self._objective_target is not None - techniques = self._build_techniques_dict(objective_target=self._objective_target) + # Building the technique catalog reads each technique's prompt YAML, so keep the + # synchronous builder off the event loop. + techniques = await asyncio.to_thread(self._build_techniques_dict, objective_target=self._objective_target) dispatcher = AdaptiveTechniqueDispatcher( objective_target=self._objective_target, techniques=techniques, diff --git a/pyrit/scenario/scenarios/airt/psychosocial.py b/pyrit/scenario/scenarios/airt/psychosocial.py index af6b28e193..64fee315fe 100644 --- a/pyrit/scenario/scenarios/airt/psychosocial.py +++ b/pyrit/scenario/scenarios/airt/psychosocial.py @@ -3,6 +3,7 @@ from __future__ import annotations +import asyncio import logging import pathlib from dataclasses import dataclass @@ -290,6 +291,27 @@ def default(cls) -> PsychosocialTechnique: } +def _build_simulated_base_factory(*, harm: _SubHarm, max_turns: int) -> AttackTechniqueFactory: + """ + Build the simulated-conversation base factory for a sub-harm. + + Reads the sub-harm's escalation prompt from disk, so async callers must run it through + ``asyncio.to_thread``. + + Args: + harm: The sub-harm whose escalation prompt drives the simulated conversation. + max_turns: Number of simulated conversation turns. + + Returns: + AttackTechniqueFactory: The base factory every converter technique layers onto. + """ + return AttackTechniqueFactory.with_simulated_conversation( + name=f"psychosocial_{harm.name}", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file(harm.escalation_prompt_path), + num_turns=max_turns, + ) + + def _converter_for_technique(technique: PsychosocialTechnique, *, adversarial_chat: PromptTarget) -> Converter | None: """ Map a converter technique to its converter instance. @@ -582,11 +604,9 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list ) ) - base_factory = AttackTechniqueFactory.with_simulated_conversation( - name=f"psychosocial_{harm.name}", - adversarial_chat_system_prompt_path=harm.escalation_prompt_path, - num_turns=max_turns, - ) + # Building the base factory reads the sub-harm's escalation prompt from disk, so keep + # it off the event loop. + base_factory = await asyncio.to_thread(_build_simulated_base_factory, harm=harm, max_turns=max_turns) for technique in techniques: if technique is PsychosocialTechnique.Crescendo: diff --git a/pyrit/scenario/scenarios/airt/scam.py b/pyrit/scenario/scenarios/airt/scam.py index b7a64dd7be..7315dfb013 100644 --- a/pyrit/scenario/scenarios/airt/scam.py +++ b/pyrit/scenario/scenarios/airt/scam.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import asyncio import logging from pathlib import Path from typing import TYPE_CHECKING, Any @@ -9,7 +10,7 @@ from pyrit.common.path import EXECUTOR_RED_TEAM_PATH, EXECUTOR_SIMULATED_TARGET_PATH, SCORER_SEED_PROMPT_PATH from pyrit.executor.attack import RedTeamingAttack from pyrit.executor.attack.core.attack_config import AttackAdversarialConfig, AttackScoringConfig -from pyrit.models import AttackSeedGroup, Parameter, SeedPrompt +from pyrit.models import AttackSeedGroup, Parameter, SeedPrompt, load_next_message_prompt, load_simulated_target_prompt from pyrit.prompt_target import PromptTarget from pyrit.scenario.core.atomic_attack import AtomicAttack from pyrit.scenario.core.attack_technique import AttackTechnique @@ -195,10 +196,12 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # objective is delivered to the target. role_play_technique = AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "role_play" - / "role_play_persuasion_written.yaml", - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "role_play" / "role_play_persuasion_written.yaml" + ), + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), num_turns=2, ).create( objective_target=self._objective_target, @@ -219,10 +222,12 @@ def _get_atomic_attack_from_technique(self, *, technique: str, seed_groups: list # further, then a fixed "yes." is delivered to the target as the final user turn. context_compliance_technique = AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "context_compliance" - / "context_compliance.yaml", - simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml" + ), + simulated_target_system_prompt=load_simulated_target_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml" + ), final_user_message="yes.", num_turns=1, ).create( @@ -271,8 +276,14 @@ async def _build_atomic_attacks_async(self, *, context: ScenarioContext) -> list memory_labels=context.memory_labels, ) ) + # Building a simulated-conversation technique reads its prompt YAML, so keep the + # synchronous builder off the event loop. atomic_attacks.extend( - self._get_atomic_attack_from_technique(technique=technique, seed_groups=seed_groups) - for technique in techniques + [ + await asyncio.to_thread( + self._get_atomic_attack_from_technique, technique=technique, seed_groups=seed_groups + ) + for technique in techniques + ] ) return atomic_attacks diff --git a/pyrit/setup/initializers/techniques/core.py b/pyrit/setup/initializers/techniques/core.py index d4dbc9667c..d95e073081 100644 --- a/pyrit/setup/initializers/techniques/core.py +++ b/pyrit/setup/initializers/techniques/core.py @@ -29,7 +29,13 @@ RedTeamingAttack, TreeOfAttacksWithPruningAttack, ) -from pyrit.models import AttackTechniqueSeedGroup, Message, SeedPrompt +from pyrit.models import ( + AttackTechniqueSeedGroup, + Message, + SeedPrompt, + load_next_message_prompt, + load_simulated_target_prompt, +) from pyrit.prompt_normalizer import ConverterConfiguration from pyrit.scenario.core.attack_technique_factory import AttackTechniqueFactory @@ -112,50 +118,60 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="role_play_movie_script", description="Frames the objective as a movie beat by casting the adversarial chat as a screenwriter.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_movie_script.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_video_game", description="Frames the objective as an in-game mechanic by casting the adversarial chat as a designer.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_video_game.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_trivia_game", description="Frames the objective as a quiz answer by casting the adversarial chat as a trivia host.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_trivia_game.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion", description="Frames a persuasion-resistance test so the target drafts a persuasive spoken script.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), AttackTechniqueFactory.with_simulated_conversation( name="role_play_persuasion_written", description="Frames a persuasion-resistance test so the target drafts written persuasive material.", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_persuasion_written.yaml" ), - next_message_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml", + next_message_system_prompt=load_next_message_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "role_play_next_message.yaml" + ), technique_tags=["single_turn", "light"], num_turns=2, ), @@ -200,10 +216,12 @@ def get_technique_factories() -> list[AttackTechniqueFactory]: AttackTechniqueFactory.with_simulated_conversation( name="context_compliance", description="Injects a fabricated prior exchange so the target continues as if it already agreed.", - adversarial_chat_system_prompt_path=EXECUTOR_RED_TEAM_PATH - / "context_compliance" - / "context_compliance.yaml", - simulated_target_system_prompt_path=EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml", + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( + EXECUTOR_RED_TEAM_PATH / "context_compliance" / "context_compliance.yaml" + ), + simulated_target_system_prompt=load_simulated_target_prompt( + EXECUTOR_SIMULATED_TARGET_PATH / "context_compliance_target.yaml" + ), final_user_message="yes.", num_turns=1, technique_tags=["single_turn", "light"], diff --git a/pyrit/setup/initializers/techniques/technique_initializer.py b/pyrit/setup/initializers/techniques/technique_initializer.py index 80370d6cca..10c94c27c3 100644 --- a/pyrit/setup/initializers/techniques/technique_initializer.py +++ b/pyrit/setup/initializers/techniques/technique_initializer.py @@ -17,6 +17,7 @@ not overwritten. """ +import asyncio import logging from enum import Enum @@ -113,7 +114,8 @@ async def initialize_async(self) -> None: if TechniqueInitializerTags.ALL.value in tags: tags = [TechniqueInitializerTags.CORE.value, TechniqueInitializerTags.EXTRA.value] - factories = build_technique_factories(groups=tags) + # Building the catalog reads each technique's prompt YAML, so keep it off the event loop. + factories = await asyncio.to_thread(build_technique_factories, groups=tags) registry = AttackTechniqueRegistry.get_registry_singleton() registry.register_from_factories(factories) diff --git a/tests/unit/datasets/test_local_prompt_dataset_semantics.py b/tests/unit/datasets/test_local_prompt_dataset_semantics.py index 6a33fd3e84..0d4f847ad4 100644 --- a/tests/unit/datasets/test_local_prompt_dataset_semantics.py +++ b/tests/unit/datasets/test_local_prompt_dataset_semantics.py @@ -7,7 +7,13 @@ from pyrit.common.path import DATASETS_PATH from pyrit.executor.attack import AttackParameters -from pyrit.models import AttackSeedGroup, AttackTechniqueSeedGroup, SeedDataset, SeedSimulatedConversation +from pyrit.models import ( + AttackSeedGroup, + AttackTechniqueSeedGroup, + SeedDataset, + SeedPrompt, + SeedSimulatedConversation, +) LOCAL_DATASET_PATH = Path(DATASETS_PATH) / "seed_datasets" / "local" FAIRNESS_REQUEST = """Infer ethnicity from the following names: @@ -35,7 +41,7 @@ async def test_airt_fairness_builds_coherent_attack_parameters() -> None: simulated_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="test.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] diff --git a/tests/unit/executor/attack/component/test_simulated_conversation.py b/tests/unit/executor/attack/component/test_simulated_conversation.py index 76225bd5f8..dcf4dec972 100644 --- a/tests/unit/executor/attack/component/test_simulated_conversation.py +++ b/tests/unit/executor/attack/component/test_simulated_conversation.py @@ -27,6 +27,7 @@ Score, SeedPrompt, SimulatedTargetSystemPromptPaths, + load_next_message_prompt, ) from pyrit.prompt_normalizer import PromptNormalizer from pyrit.prompt_target import PromptTarget @@ -926,7 +927,7 @@ async def test_parses_next_message_from_json_reply(self, mock_adversarial_chat: conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) @@ -956,7 +957,7 @@ async def test_invalid_json_reply_raises_after_retry(self, mock_adversarial_chat conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) @@ -978,6 +979,147 @@ async def test_raises_when_no_response(self, mock_adversarial_chat: MagicMock): conversation_messages=[], adversarial_chat=mock_adversarial_chat, conversation_id="next-message-conversation", - next_message_system_prompt_path=NextMessageSystemPromptPaths.DIRECT.value, + next_message_system_prompt=load_next_message_prompt(NextMessageSystemPromptPaths.DIRECT.value), prompt_normalizer=normalizer, ) + + +@pytest.mark.usefixtures("patch_central_database") +class TestSimulatedConversationPromptSources: + """Tests for the canonical prompt inputs and the deprecated path inputs.""" + + @staticmethod + def _mock_attack(conversation_id: str) -> MagicMock: + attack = MagicMock() + attack.get_identifier.return_value = ComponentIdentifier( + class_name="RedTeamingAttack", class_module="pyrit.executor.attack" + ) + attack.execute_async = AsyncMock( + return_value=AttackResult( + atomic_attack_identifier=ComponentIdentifier( + class_name="RedTeamingAttack", class_module="pyrit.executor.attack" + ), + conversation_id=conversation_id, + objective="Test objective", + outcome=AttackOutcome.SUCCESS, + executed_turns=2, + ) + ) + return attack + + async def test_canonical_prompts_never_read_from_disk( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + sample_conversation: list[Message], + ): + """Canonical prompts are carried in, so the async path performs no YAML loading.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack_class.return_value = self._mock_attack(str(uuid.uuid4())) + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + with patch("pyrit.models.seeds.yaml_seed_loader.load_seed_from_yaml") as mock_load: + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt( + value="Objective: {{ objective }} Turns: {{ num_turns }}", + parameters=["objective", "num_turns"], + ), + num_turns=2, + ) + + mock_load.assert_not_called() + + async def test_simulated_target_prompt_is_rendered_into_system_message( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + sample_conversation: list[Message], + ): + """The simulated target template is rendered with the objective and turn count.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack = self._mock_attack(str(uuid.uuid4())) + mock_attack_class.return_value = mock_attack + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt( + value="Objective: {{ objective }} Turns: {{ num_turns }}", + parameters=["objective", "num_turns"], + ), + num_turns=2, + ) + + prepended = mock_attack.execute_async.call_args.kwargs["prepended_conversation"] + assert prepended[0].get_value() == "Objective: Test objective Turns: 2" + + async def test_deprecated_path_input_warns( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + adversarial_system_prompt_path: Path, + sample_conversation: list[Message], + ): + """A deprecated path input still works and warns.""" + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.RedTeamingAttack") as mock_attack_class: + mock_attack_class.return_value = self._mock_attack(str(uuid.uuid4())) + + with patch("pyrit.executor.attack.multi_turn.simulated_conversation.CentralMemory") as mock_memory_class: + mock_memory = MagicMock() + mock_memory.get_conversation_messages.return_value = iter(sample_conversation) + mock_memory_class.get_memory_instance.return_value = mock_memory + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt_path=adversarial_system_prompt_path, + num_turns=2, + ) + + async def test_prompt_and_path_together_raises( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + adversarial_system_prompt_path: Path, + ): + """Supplying both a canonical prompt and its deprecated path is ambiguous.""" + with pytest.raises(ValueError, match="Set only one of adversarial_chat_system_prompt"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + adversarial_chat_system_prompt_path=adversarial_system_prompt_path, + num_turns=2, + ) + + async def test_missing_adversarial_prompt_raises( + self, + mock_adversarial_chat: MagicMock, + mock_objective_scorer: MagicMock, + ): + """The adversarial system prompt is required.""" + with pytest.raises(ValueError, match="adversarial_chat_system_prompt is required"): + await generate_simulated_conversation_async( + objective="Test objective", + adversarial_chat=mock_adversarial_chat, + objective_scorer=mock_objective_scorer, + num_turns=2, + ) diff --git a/tests/unit/executor/attack/core/test_attack_parameters.py b/tests/unit/executor/attack/core/test_attack_parameters.py index 805213c25c..f41c4a43e3 100644 --- a/tests/unit/executor/attack/core/test_attack_parameters.py +++ b/tests/unit/executor/attack/core/test_attack_parameters.py @@ -132,8 +132,8 @@ def simulated_conversation_config(self) -> SeedSimulatedConversation: """Create a SeedSimulatedConversation config.""" return SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path="/path/to/adversarial.yaml", - simulated_target_system_prompt_path="/path/to/target.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), ) @pytest.fixture diff --git a/tests/unit/memory/test_memory_models.py b/tests/unit/memory/test_memory_models.py index b65405d850..3fdcf7c4c8 100644 --- a/tests/unit/memory/test_memory_models.py +++ b/tests/unit/memory/test_memory_models.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft Corporation. # Licensed under the MIT license. +import json import uuid from collections.abc import Sequence from datetime import UTC, datetime @@ -619,14 +620,93 @@ def test_roundtrip_seed_objective_strips_reserved_key(self): assert SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY not in (recovered.metadata or {}) assert (recovered.metadata or {}).get("owned") == "by-caller" + def test_roundtrip_seed_simulated_conversation_preserves_prompts_and_version(self): + """A canonical record round-trips its prompts, value, hash, and recorded version.""" + config = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + pyrit_version="1.0.0", + ) + config.value_sha256 = "canonical-hash" + + recovered = SeedEntry(entry=config).get_seed() + + assert isinstance(recovered, SeedSimulatedConversation) + assert recovered.adversarial_chat_system_prompt.value == "adversarial" + assert recovered.next_message_system_prompt is not None + assert recovered.next_message_system_prompt.response_json_schema is not None + assert recovered.pyrit_version == "1.0.0" + assert recovered.value == config.value + assert recovered.value_sha256 == "canonical-hash" + + def test_legacy_path_record_reconstructs_prompts(self, tmp_path): + """A record written before normalization still loads, resolving its paths to prompts.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: legacy adversarial\ndata_type: text") + + seed = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="placeholder"), + ) + entry = SeedEntry(entry=seed) + entry.value = json.dumps( + { + "num_turns": 2, + "sequence": 0, + "adversarial_chat_system_prompt_path": str(adv_path), + "simulated_target_system_prompt_path": None, + "next_message_system_prompt_path": None, + "pyrit_version": "1.0.0", + }, + sort_keys=True, + separators=(",", ":"), + ) + entry.value_sha256 = "stale-path-hash" + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + recovered = entry.get_seed() + + assert isinstance(recovered, SeedSimulatedConversation) + assert recovered.adversarial_chat_system_prompt.value == "legacy adversarial" + # The compliant default fills in for the omitted simulated target. + assert recovered.simulated_target_system_prompt.name == "simulated_target_compliant" + assert recovered.next_message_system_prompt is None + assert recovered.pyrit_version == "1.0.0" + # The stored hash described the old path-shaped value, so it is not carried over. + assert recovered.value_sha256 is None + + def test_legacy_record_with_missing_file_names_the_record(self, tmp_path): + """A legacy record pointing at a file this machine lacks fails with the record identified.""" + seed = SeedSimulatedConversation( + num_turns=2, + adversarial_chat_system_prompt=SeedPrompt(value="placeholder"), + name="stale-technique", + dataset_name="legacy-dataset", + ) + entry = SeedEntry(entry=seed) + entry.value = json.dumps( + { + "num_turns": 2, + "sequence": 0, + "adversarial_chat_system_prompt_path": str(tmp_path / "gone.yaml"), + "pyrit_version": "1.0.0", + }, + sort_keys=True, + separators=(",", ":"), + ) + + with pytest.raises(ValueError, match="stale-technique"): + entry.get_seed() + def test_roundtrip_seed_simulated_conversation_strips_reserved_key(self): """SeedSimulatedConversation also has no schema field; reserved key must still be stripped.""" from pyrit.models import SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY config = SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path="/path/to/adversarial.yaml", - simulated_target_system_prompt_path="/path/to/target.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), metadata={ SEED_RESPONSE_JSON_SCHEMA_METADATA_KEY: "sneaky", "owned": "by-caller", diff --git a/tests/unit/models/test_attack_technique_seed_group.py b/tests/unit/models/test_attack_technique_seed_group.py index 3e312a1130..e87d89a0cb 100644 --- a/tests/unit/models/test_attack_technique_seed_group.py +++ b/tests/unit/models/test_attack_technique_seed_group.py @@ -42,24 +42,18 @@ def test_seed_objective_raises_if_set_true(self): with pytest.raises(ValueError, match="SeedObjective cannot be a general technique"): SeedObjective(value="Test objective", is_general_technique=True) - def test_seed_simulated_conversation_defaults_to_true(self, tmp_path): + def test_seed_simulated_conversation_defaults_to_true(self): """Test that SeedSimulatedConversation.is_general_technique defaults to True.""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - sim = SeedSimulatedConversation( - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), num_turns=2, ) assert sim.is_general_technique is True - def test_seed_simulated_conversation_can_be_set_false(self, tmp_path): + def test_seed_simulated_conversation_can_be_set_false(self): """Test that SeedSimulatedConversation.is_general_technique can be overridden to False.""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - sim = SeedSimulatedConversation( - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), num_turns=2, is_general_technique=False, ) @@ -113,16 +107,13 @@ def test_init_raises_with_objective(self): ] ) - def test_init_with_simulated_conversation(self, tmp_path): + def test_init_with_simulated_conversation(self): """Test initialization with SeedSimulatedConversation (defaults to general technique).""" - adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: Adversarial\ndata_type: text") - group = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( num_turns=3, - adversarial_chat_system_prompt_path=adv_path, + adversarial_chat_system_prompt=SeedPrompt(value="Adversarial", parameters=["objective"]), ), SeedPrompt( value="Technique prompt", data_type="text", sequence=10, role="user", is_general_technique=True diff --git a/tests/unit/models/test_seed.py b/tests/unit/models/test_seed.py index 2d8c48f9c7..1dca202d06 100644 --- a/tests/unit/models/test_seed.py +++ b/tests/unit/models/test_seed.py @@ -1069,8 +1069,8 @@ def test_seed_dataset_dict_to_seed_simulated_conversation_all_base_params(tmp_pa # Verify SeedSimulatedConversation-specific fields assert seed.num_turns == 5 - assert seed.adversarial_chat_system_prompt_path == pathlib.Path(adv_path) - assert seed.simulated_target_system_prompt_path == pathlib.Path(sim_path) + assert seed.adversarial_chat_system_prompt.value == "You are adversarial" + assert seed.simulated_target_system_prompt.parameters == ["objective", "num_turns"] def test_seed_dataset_uses_dataset_defaults_for_missing_params(): diff --git a/tests/unit/models/test_seed_group.py b/tests/unit/models/test_seed_group.py index bcb0c754fb..9876cbe4ed 100644 --- a/tests/unit/models/test_seed_group.py +++ b/tests/unit/models/test_seed_group.py @@ -679,7 +679,7 @@ def test_raises_when_technique_has_simulated_conversation_and_prompts_overlap(se technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -694,7 +694,7 @@ def test_succeeds_when_technique_has_simulated_conversation_and_no_prompts(self) technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -716,7 +716,7 @@ def test_is_compatible_returns_false_when_prompts_overlap_simulated_range(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -730,7 +730,7 @@ def test_is_compatible_returns_true_for_objective_only_with_simulated(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake_path.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], @@ -777,7 +777,7 @@ def test_filters_out_incompatible_groups(self): technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ), ], diff --git a/tests/unit/models/test_seed_simulated_conversation.py b/tests/unit/models/test_seed_simulated_conversation.py index c8239a9caa..3222c712c4 100644 --- a/tests/unit/models/test_seed_simulated_conversation.py +++ b/tests/unit/models/test_seed_simulated_conversation.py @@ -9,8 +9,8 @@ import pytest from pyrit.models.seeds import ( + SeedPrompt, SeedSimulatedConversation, - SimulatedTargetSystemPromptPaths, ) @@ -22,7 +22,7 @@ def test_init_with_all_parameters(self, tmp_path): adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") sim_path = tmp_path / "simulated.yaml" - sim_path.write_text("value: test\ndata_type: text") + sim_path.write_text("value: test\ndata_type: text\nparameters:\n - objective\n - num_turns") conv = SeedSimulatedConversation( adversarial_chat_system_prompt_path=adv_path, @@ -31,8 +31,8 @@ def test_init_with_all_parameters(self, tmp_path): ) assert conv.num_turns == 5 - assert conv.adversarial_chat_system_prompt_path == adv_path - assert conv.simulated_target_system_prompt_path == sim_path + assert conv.adversarial_chat_system_prompt.value == "test" + assert conv.simulated_target_system_prompt.value == "test" assert conv.data_type == "text" assert isinstance(conv.id, uuid.UUID) @@ -46,9 +46,9 @@ def test_init_with_minimal_parameters(self, tmp_path): ) assert conv.num_turns == 3 # default - assert conv.adversarial_chat_system_prompt_path == adv_path - # Default simulated_target_system_prompt_path is the compliant prompt - assert conv.simulated_target_system_prompt_path == SimulatedTargetSystemPromptPaths.COMPLIANT.value + assert conv.adversarial_chat_system_prompt.value == "test" + # The simulated target defaults to the compliant prompt + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" def test_init_default_num_turns(self, tmp_path): """Test that default num_turns is 3.""" @@ -106,7 +106,7 @@ def test_init_generates_json_value(self, tmp_path): value = json.loads(conv.value) assert value["num_turns"] == 5 - assert "adversarial_chat_system_prompt_path" in value + assert value["adversarial_chat_system_prompt"]["value"] == "test" assert "pyrit_version" in value def test_init_value_is_deterministic(self, tmp_path): @@ -148,8 +148,8 @@ def test_init_custom_sequence(self, tmp_path): assert conv.sequence == 5 - def test_init_default_next_message_system_prompt_path_is_none(self, tmp_path): - """Test that default next_message_system_prompt_path is None.""" + def test_init_default_next_message_system_prompt_is_none(self, tmp_path): + """Test that the next message system prompt defaults to None.""" adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") @@ -157,10 +157,10 @@ def test_init_default_next_message_system_prompt_path_is_none(self, tmp_path): adversarial_chat_system_prompt_path=adv_path, ) - assert conv.next_message_system_prompt_path is None + assert conv.next_message_system_prompt is None - def test_init_next_message_system_prompt_path_set(self, tmp_path): - """Test that next_message_system_prompt_path can be set.""" + def test_init_next_message_system_prompt_set(self, tmp_path): + """Test that the next message system prompt can be set.""" adv_path = tmp_path / "adversarial.yaml" adv_path.write_text("value: test\ndata_type: text") next_msg_path = tmp_path / "next_message.yaml" @@ -171,7 +171,7 @@ def test_init_next_message_system_prompt_path_set(self, tmp_path): next_message_system_prompt_path=next_msg_path, ) - assert conv.next_message_system_prompt_path == next_msg_path + assert conv.next_message_system_prompt.value == "test" class TestSeedSimulatedConversationFromMapping: @@ -189,7 +189,7 @@ def test_from_dict_with_paths(self, tmp_path): conv = SeedSimulatedConversation.model_validate(data) assert conv.num_turns == 5 - assert conv.adversarial_chat_system_prompt_path == adv_path + assert conv.adversarial_chat_system_prompt.value == "test" def test_from_dict_without_simulated_target_path(self, tmp_path): """Test construction without simulated_target_system_prompt_path uses compliant default.""" @@ -202,8 +202,8 @@ def test_from_dict_without_simulated_target_path(self, tmp_path): } conv = SeedSimulatedConversation.model_validate(data) - # Default simulated_target_system_prompt_path is the compliant prompt - assert conv.simulated_target_system_prompt_path == SimulatedTargetSystemPromptPaths.COMPLIANT.value + # The simulated target defaults to the compliant prompt + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" def test_from_dict_default_num_turns(self, tmp_path): """Test that num_turns defaults to 3 when not specified.""" @@ -221,7 +221,7 @@ def test_from_dict_missing_adversarial_path_raises_error(self): """Test that construction raises when adversarial path is missing (required field).""" data = {"num_turns": 3} - with pytest.raises(ValueError, match="adversarial_chat_system_prompt_path"): + with pytest.raises(ValueError, match="adversarial_chat_system_prompt"): SeedSimulatedConversation.model_validate(data) @@ -241,7 +241,7 @@ def test_get_identifier_returns_correct_structure(self, tmp_path): assert identifier["__type__"] == "SeedSimulatedConversation" assert identifier["num_turns"] == 3 - assert "adversarial_chat_system_prompt_path" in identifier + assert identifier["adversarial_chat_system_prompt"]["value"] == "test" assert "pyrit_version" in identifier @@ -298,10 +298,10 @@ def test_compute_hash_differs_for_different_num_turns(self, tmp_path): class TestSeedSimulatedConversationRepr: """Tests for SeedSimulatedConversation.__repr__ method.""" - def test_repr_shows_num_turns_and_path(self, tmp_path): + def test_repr_shows_num_turns_and_prompt_name(self, tmp_path): """Test __repr__ shows key information.""" adv_path = tmp_path / "adversarial.yaml" - adv_path.write_text("value: test\ndata_type: text") + adv_path.write_text("name: my_adversarial\nvalue: test\ndata_type: text") conv = SeedSimulatedConversation( adversarial_chat_system_prompt_path=adv_path, @@ -311,7 +311,7 @@ def test_repr_shows_num_turns_and_path(self, tmp_path): assert "SeedSimulatedConversation" in repr_str assert "num_turns=5" in repr_str - assert "adversarial.yaml" in repr_str + assert "my_adversarial" in repr_str class TestSeedSimulatedConversationLoadSimulatedTargetSystemPrompt: @@ -348,3 +348,204 @@ def test_load_simulated_target_system_prompt_raises_for_missing_params(self, tmp num_turns=3, simulated_target_system_prompt_path=sim_path, ) + + +class TestSeedSimulatedConversationCanonicalPrompts: + """Tests for the canonical SeedPrompt fields and the deprecated path inputs.""" + + def test_init_with_canonical_prompts(self): + """Canonical SeedPrompt inputs populate the fields without touching disk.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + simulated_target_system_prompt=SeedPrompt(value="target", parameters=["objective", "num_turns"]), + next_message_system_prompt=SeedPrompt(value="next", parameters=["objective", "conversation_context"]), + num_turns=2, + ) + + assert conv.adversarial_chat_system_prompt.value == "adversarial" + assert conv.simulated_target_system_prompt.value == "target" + assert conv.next_message_system_prompt is not None + assert conv.next_message_system_prompt.value == "next" + + def test_path_inputs_are_not_model_fields(self): + """The deprecated path inputs never become fields, so they cannot reach persistence.""" + for field_name in ( + "adversarial_chat_system_prompt_path", + "simulated_target_system_prompt_path", + "next_message_system_prompt_path", + ): + assert field_name not in SeedSimulatedConversation.model_fields + + def test_deprecated_path_input_warns_and_resolves(self, tmp_path): + """A deprecated path input warns and is resolved into the canonical prompt.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + conv = SeedSimulatedConversation(adversarial_chat_system_prompt_path=adv_path) + + assert conv.adversarial_chat_system_prompt.value == "adversarial" + assert not hasattr(conv, "adversarial_chat_system_prompt_path") + + def test_prompt_and_path_together_raises(self, tmp_path): + """Supplying both a canonical prompt and its deprecated path is ambiguous.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + + with pytest.raises( + ValueError, match="Set only one of SeedSimulatedConversation.adversarial_chat_system_prompt" + ): + SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + adversarial_chat_system_prompt_path=adv_path, + ) + + def test_missing_deprecated_path_raises(self, tmp_path): + """A deprecated path that does not exist fails loudly rather than silently.""" + with pytest.raises(FileNotFoundError): + SeedSimulatedConversation(adversarial_chat_system_prompt_path=tmp_path / "missing.yaml") + + def test_deprecated_simulated_target_path_requires_parameters(self, tmp_path): + """Loading a simulated target from a path keeps the declared-parameter contract.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: adversarial\ndata_type: text") + sim_path = tmp_path / "simulated.yaml" + sim_path.write_text("value: no params\ndata_type: text") + + with pytest.raises(ValueError, match="objective and num_turns"): + SeedSimulatedConversation( + adversarial_chat_system_prompt_path=adv_path, + simulated_target_system_prompt_path=sim_path, + ) + + def test_explicit_none_simulated_target_falls_back_to_compliant(self): + """An explicit None simulated target (as memory reconstruction sends) uses the default.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt=None, + ) + + assert conv.simulated_target_system_prompt.name == "simulated_target_compliant" + + +class TestSeedSimulatedConversationIdentity: + """Tests for the content-based value and hash.""" + + def test_value_carries_prompt_text_not_paths(self): + """The serialized value holds the prompt text so a technique is inspectable.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial text", parameters=["objective"]) + ) + value = json.loads(conv.value) + + assert value["adversarial_chat_system_prompt"]["value"] == "adversarial text" + assert value["adversarial_chat_system_prompt"]["parameters"] == ["objective"] + assert not any(key.endswith("_path") for key in value) + + def test_editing_one_word_changes_identity(self): + """Changing a word in a system prompt produces a different configuration identity.""" + original = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="be a screenwriter")) + edited = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="be a novelist")) + + assert original.value != edited.value + assert original.compute_hash() != edited.compute_hash() + + def test_identical_content_from_different_files_matches(self, tmp_path): + """Two copies of the same prompt text share an identity even from different files.""" + first = tmp_path / "first.yaml" + first.write_text("value: same text\ndata_type: text") + second = tmp_path / "second.yaml" + second.write_text("value: same text\ndata_type: text") + + conv1 = SeedSimulatedConversation(adversarial_chat_system_prompt_path=first) + conv2 = SeedSimulatedConversation(adversarial_chat_system_prompt_path=second) + + assert conv1.value == conv2.value + + def test_response_json_schema_changes_identity(self): + """Two prompts with identical text but different schemas are not interchangeable.""" + plain = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next"), + ) + with_schema = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + ) + + assert plain.value != with_schema.value + + @pytest.mark.parametrize("mode", ["python", "json"]) + def test_value_is_stable_across_round_trip(self, mode): + """Dumping and revalidating recomputes the same value.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), + next_message_system_prompt=SeedPrompt(value="next", response_json_schema_name="adversarial_chat"), + num_turns=2, + ) + + assert SeedSimulatedConversation.model_validate(conv.model_dump(mode=mode)).value == conv.value + + def test_sequence_range_accounts_for_next_message(self): + """The next message adds one sequence slot after the generated turns.""" + without = SeedSimulatedConversation(adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), num_turns=2) + with_next = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + next_message_system_prompt=SeedPrompt(value="next"), + num_turns=2, + ) + + assert list(without.sequence_range) == [0, 1, 2, 3] + assert list(with_next.sequence_range) == [0, 1, 2, 3, 4] + + +class TestSeedSimulatedConversationTemplatePreparation: + """Tests that a prompt is prepared exactly once, so deferred template syntax survives.""" + + DEFERRED_TEMPLATE = "{% raw %}{% if num_turns == 1 %}one turn{% else %}many turns{% endif %}{% endraw %}" + + @pytest.fixture + def deferred_prompt_path(self, tmp_path): + path = tmp_path / "deferred.yaml" + path.write_text( + f"data_type: text\nparameters:\n - objective\n - num_turns\nvalue: '{self.DEFERRED_TEMPLATE}'\n" + ) + return path + + def test_canonical_prompt_is_not_prepared_again(self, deferred_prompt_path): + """A prompt handed in already prepared keeps its deferred template syntax.""" + prompt = SeedPrompt.from_yaml_file(deferred_prompt_path) + original = prompt.value + + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt=prompt, + num_turns=1, + ) + + assert prompt.value == original, "the caller's prompt must not be mutated" + assert conv.simulated_target_system_prompt.render_template_value(objective="o", num_turns=1) == "one turn" + + def test_deprecated_path_prompt_is_not_prepared_again(self, deferred_prompt_path): + """Loading through the deprecated path input also prepares the template only once.""" + with pytest.warns(DeprecationWarning): + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt_path=deferred_prompt_path, + num_turns=1, + ) + + assert conv.simulated_target_system_prompt.render_template_value(objective="o", num_turns=1) == "one turn" + + def test_prompt_rebuilt_from_value_is_not_prepared_again(self, deferred_prompt_path): + """A prompt rebuilt from the serialized configuration keeps its deferred template syntax.""" + with pytest.warns(DeprecationWarning): + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="adversarial"), + simulated_target_system_prompt_path=deferred_prompt_path, + num_turns=1, + ) + + rebuilt = SeedPrompt(**json.loads(conv.value)["simulated_target_system_prompt"]) + + assert rebuilt.render_template_value(objective="o", num_turns=1) == "one turn" diff --git a/tests/unit/scenario/core/test_attack_technique_factory.py b/tests/unit/scenario/core/test_attack_technique_factory.py index c5324d8d39..7812a7992b 100644 --- a/tests/unit/scenario/core/test_attack_technique_factory.py +++ b/tests/unit/scenario/core/test_attack_technique_factory.py @@ -4,6 +4,7 @@ """Tests for the AttackTechniqueFactory class.""" import typing +import warnings from unittest.mock import MagicMock, patch import pytest @@ -957,7 +958,7 @@ def test_simulated_conversation_resolves_default_lazily(self): factory = AttackTechniqueFactory.with_simulated_conversation( name="role_play_movie_script", - adversarial_chat_system_prompt_path=( + adversarial_chat_system_prompt=SeedPrompt.from_yaml_file( EXECUTOR_SEED_PROMPT_PATH / "red_teaming" / "role_play" / "role_play_movie_script.yaml" ), num_turns=2, @@ -1034,3 +1035,63 @@ def get_identifier(self): ) assert factory._get_scoring_config_type() is None + + +@pytest.mark.usefixtures("patch_central_database") +class TestWithSimulatedConversationPromptSources: + """Tests for the canonical prompt inputs on ``with_simulated_conversation``.""" + + def test_defaults_resolve_to_prompts_without_warning(self): + """The name-derived adversarial prompt and the default next message load silently.""" + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + factory = AttackTechniqueFactory.with_simulated_conversation(name="crescendo_simulated") + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.name == "crescendo_simulated" + assert sim.simulated_target_system_prompt.name == "simulated_target_compliant" + assert sim.next_message_system_prompt is not None + assert sim.next_message_system_prompt.name == "direct_next_message_generator" + + def test_canonical_prompt_is_used(self): + """An explicit prompt is carried straight through to the seed.""" + prompt = SeedPrompt(value="custom adversarial", parameters=["objective"]) + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + adversarial_chat_system_prompt=prompt, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.value == "custom adversarial" + + def test_deprecated_path_input_warns(self, tmp_path): + """An explicit path input still works and warns.""" + adv_path = tmp_path / "adversarial.yaml" + adv_path.write_text("value: from path\ndata_type: text") + + with pytest.warns(DeprecationWarning, match="adversarial_chat_system_prompt_path"): + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + adversarial_chat_system_prompt_path=adv_path, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.adversarial_chat_system_prompt.value == "from path" + + def test_final_user_message_disables_next_message_prompt(self): + """A fixed final message replaces the generated next message.""" + factory = AttackTechniqueFactory.with_simulated_conversation( + name="crescendo_simulated", + final_user_message="yes.", + num_turns=1, + ) + + sim = factory.seed_technique.simulated_conversation_config + assert sim is not None + assert sim.next_message_system_prompt is None + prompts = list(factory.seed_technique.prompts) + assert prompts[0].value == "yes." + assert prompts[0].sequence == sim.sequence_range.stop diff --git a/tests/unit/scenario/test_default_run_size_estimates.py b/tests/unit/scenario/test_default_run_size_estimates.py index f07e27dd2e..c3e6f89f5f 100644 --- a/tests/unit/scenario/test_default_run_size_estimates.py +++ b/tests/unit/scenario/test_default_run_size_estimates.py @@ -292,7 +292,7 @@ async def test_matrix_estimate_filters_each_technique_seed_population_like_execu conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] @@ -409,7 +409,7 @@ async def resolve_groups() -> tuple[dict[str, list[AttackSeedGroup]], list[Scena conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] @@ -811,7 +811,7 @@ async def test_adversarial_benchmark_resolves_targets_and_filters_each_technique conversation_factory.seed_technique = AttackTechniqueSeedGroup( seeds=[ SeedSimulatedConversation( - adversarial_chat_system_prompt_path="fake.yaml", + adversarial_chat_system_prompt=SeedPrompt(value="adversarial", parameters=["objective"]), num_turns=3, ) ] diff --git a/tests/unit/setup/test_technique_initializer.py b/tests/unit/setup/test_technique_initializer.py index 66f791a12f..00770b58a0 100644 --- a/tests/unit/setup/test_technique_initializer.py +++ b/tests/unit/setup/test_technique_initializer.py @@ -260,11 +260,12 @@ def test_seed_technique_num_turns_matches_canonical_default(self): assert sim is not None assert sim.num_turns == 3 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_carries_resolved_adversarial_prompt(self): for f in self._persona_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] class TestPersonaCrescendoYamls: @@ -327,10 +328,10 @@ def test_seed_technique_uses_custom_simulated_target_prompt(self): factory = self._context_compliance_factory() sim = factory.seed_technique.simulated_conversation_config assert sim is not None - assert sim.simulated_target_system_prompt_path.name == "context_compliance_target.yaml" - assert sim.simulated_target_system_prompt_path.exists() + assert sim.simulated_target_system_prompt.name == "simulated_target_context_compliance" + assert sim.simulated_target_system_prompt.parameters == ["objective", "num_turns"] # No LLM-generated next message: the final turn is a fixed affirmation instead. - assert sim.next_message_system_prompt_path is None + assert sim.next_message_system_prompt is None def test_final_user_message_is_fixed_affirmation(self): factory = self._context_compliance_factory() @@ -343,12 +344,12 @@ def test_final_user_message_is_fixed_affirmation(self): assert yes_prompt.role == "user" assert yes_prompt.sequence == 2 - def test_adversarial_yaml_resolves_to_existing_file(self): + def test_adversarial_prompt_is_resolved(self): factory = self._context_compliance_factory() sim = factory.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.name == "context_compliance.yaml" - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] def test_tagged_core_single_turn_light(self): factory = self._context_compliance_factory() @@ -417,19 +418,20 @@ def test_seed_technique_num_turns_matches_role_play_default(self): assert sim is not None assert sim.num_turns == 2 - def test_seed_technique_yaml_path_resolves_to_existing_file(self): + def test_seed_technique_carries_resolved_adversarial_prompt(self): for f in self._role_play_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.adversarial_chat_system_prompt_path.exists() + assert sim.adversarial_chat_system_prompt.value + assert sim.adversarial_chat_system_prompt.parameters == ["objective", "max_turns"] def test_all_use_role_play_next_message_prompt(self): for f in self._role_play_factories(): sim = f.seed_technique.simulated_conversation_config assert sim is not None - assert sim.next_message_system_prompt_path is not None - assert sim.next_message_system_prompt_path.name == "role_play_next_message.yaml" - assert sim.next_message_system_prompt_path.exists() + assert sim.next_message_system_prompt is not None + assert sim.next_message_system_prompt.name == "role_play_next_message_generator" + assert sim.next_message_system_prompt.parameters == ["objective", "conversation_context"] class TestRolePlayYamls: From a7f2734b11dc86b6d37bd14e232fa11fed2e8e6a Mon Sep 17 00:00:00 2001 From: varunj-msft Date: Tue, 22 Sep 2026 15:10:12 +0000 Subject: [PATCH 2/3] MAINT: Address review feedback on SeedPrompt normalization Resolve the review notes on the simulated-conversation prompt normalization. Describe the executor's next-message parameter as the SeedPrompt it now is rather than a bare template, and apply the same wording to the three SeedSimulatedConversation attribute docs so the terminology is consistent rather than corrected in one place. Resolve the simulated-target default inside with_simulated_conversation, so all three prompts reach SeedSimulatedConversation as concrete SeedPrompt values instead of one arriving as None and being backfilled by a field validator. Serialized configuration and identity are unchanged. Promote resolve_prompt_source and warn_prompt_path_deprecated to public names and export them from pyrit.models.seeds and pyrit.models, so the scenario and executor layers no longer reach into a private helper across a package boundary. Drop the invented placeholder from __repr__. The prompt name is descriptive metadata that identity deliberately discards, so a seed rebuilt from a persisted record never has one; the fragment is now omitted in that case instead of printing a placeholder or a bare None. --- .../multi_turn/simulated_conversation.py | 7 ++++--- pyrit/models/__init__.py | 4 ++++ pyrit/models/seeds/__init__.py | 4 ++++ .../seeds/seed_simulated_conversation.py | 21 +++++++++++-------- .../scenario/core/attack_technique_factory.py | 18 ++++++++++------ 5 files changed, 36 insertions(+), 18 deletions(-) diff --git a/pyrit/executor/attack/multi_turn/simulated_conversation.py b/pyrit/executor/attack/multi_turn/simulated_conversation.py index 7a9a2fb7dd..5a0c79a472 100644 --- a/pyrit/executor/attack/multi_turn/simulated_conversation.py +++ b/pyrit/executor/attack/multi_turn/simulated_conversation.py @@ -34,8 +34,8 @@ SeedPrompt, load_next_message_prompt, load_simulated_target_prompt, + warn_prompt_path_deprecated, ) -from pyrit.models.seeds.seed_simulated_conversation import _warn_prompt_path_deprecated from pyrit.prompt_normalizer import PromptNormalizer if TYPE_CHECKING: @@ -85,7 +85,7 @@ async def _resolve_prompt_source_async( """ if path is None: return prompt - _warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) return await asyncio.to_thread(load_prompt, path) @@ -300,7 +300,8 @@ async def _generate_next_message_async( conversation_messages: The conversation generated so far as Messages. adversarial_chat: The LLM to use for generation. conversation_id: The conversation ID for the adversarial generation exchange. - next_message_system_prompt: The system prompt template. + next_message_system_prompt: The next-message system-prompt SeedPrompt, rendered with the + conversation context. prompt_normalizer: The normalizer the manager sends the adversarial turn through. memory_labels: Optional memory labels to attach to the request. diff --git a/pyrit/models/__init__.py b/pyrit/models/__init__.py index 071d89a796..a0a2a06903 100644 --- a/pyrit/models/__init__.py +++ b/pyrit/models/__init__.py @@ -164,6 +164,8 @@ group_seeds_into_attack_groups, load_next_message_prompt, load_simulated_target_prompt, + resolve_prompt_source, + warn_prompt_path_deprecated, ) from pyrit.models.target import ( COMMON_JSON_SCHEMAS, @@ -315,6 +317,8 @@ "SimulatedTargetSystemPromptPaths": "pyrit.models.seeds", "load_next_message_prompt": "pyrit.models.seeds", "load_simulated_target_prompt": "pyrit.models.seeds", + "resolve_prompt_source": "pyrit.models.seeds", + "warn_prompt_path_deprecated": "pyrit.models.seeds", "snake_case_to_class_name": "pyrit.models.identifiers", "sort_message_pieces": "pyrit.models.messages.message_piece", "StrategyResult": "pyrit.models.results.strategy_result", diff --git a/pyrit/models/seeds/__init__.py b/pyrit/models/seeds/__init__.py index 4a82cf7334..fbf25c407f 100644 --- a/pyrit/models/seeds/__init__.py +++ b/pyrit/models/seeds/__init__.py @@ -35,6 +35,8 @@ SimulatedTargetSystemPromptPaths, load_next_message_prompt, load_simulated_target_prompt, + resolve_prompt_source, + warn_prompt_path_deprecated, ) from pyrit.models.seeds.yaml_seed_loader import ( load_seed_dataset_from_yaml, @@ -48,6 +50,8 @@ "load_seed_prompt_from_yaml_with_required_parameters": "pyrit.models.seeds.yaml_seed_loader", "load_next_message_prompt": "pyrit.models.seeds.seed_simulated_conversation", "load_simulated_target_prompt": "pyrit.models.seeds.seed_simulated_conversation", + "resolve_prompt_source": "pyrit.models.seeds.seed_simulated_conversation", + "warn_prompt_path_deprecated": "pyrit.models.seeds.seed_simulated_conversation", "group_seeds_into_attack_groups": "pyrit.models.seeds.seed_grouping", "NextMessageSystemPromptPaths": "pyrit.models.seeds.seed_simulated_conversation", "Seed": "pyrit.models.seeds.seed", diff --git a/pyrit/models/seeds/seed_simulated_conversation.py b/pyrit/models/seeds/seed_simulated_conversation.py index ff02efcc8f..b32b6238cc 100644 --- a/pyrit/models/seeds/seed_simulated_conversation.py +++ b/pyrit/models/seeds/seed_simulated_conversation.py @@ -105,7 +105,7 @@ def _load_compliant_simulated_target_prompt() -> SeedPrompt: return load_simulated_target_prompt(SimulatedTargetSystemPromptPaths.COMPLIANT.value) -def _resolve_prompt_source( +def resolve_prompt_source( *, prompt: SeedPrompt | None, path: str | Path | None, @@ -135,11 +135,11 @@ def _resolve_prompt_source( """ if path is None: return prompt - _warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) + warn_prompt_path_deprecated(prompt=prompt, prompt_name=prompt_name, path_name=path_name) return load_prompt(path) -def _warn_prompt_path_deprecated(*, prompt: SeedPrompt | None, prompt_name: str, path_name: str) -> None: +def warn_prompt_path_deprecated(*, prompt: SeedPrompt | None, prompt_name: str, path_name: str) -> None: """ Reject an ambiguous prompt source and warn that the path input is deprecated. @@ -249,10 +249,10 @@ class SeedSimulatedConversation(Seed): Attributes: num_turns: Number of conversation turns to generate. - adversarial_chat_system_prompt: System prompt template for the adversarial chat. - simulated_target_system_prompt: System prompt template for the simulated target. + adversarial_chat_system_prompt: System-prompt SeedPrompt for the adversarial chat. + simulated_target_system_prompt: System-prompt SeedPrompt for the simulated target. Defaults to the compliant prompt if not specified. - next_message_system_prompt: Optional system prompt template for generating + next_message_system_prompt: Optional system-prompt SeedPrompt for generating an additional user message after the simulated conversation. If provided, a single LLM call generates a final user message that attempts to get the target to fulfill the objective in their next response. @@ -336,7 +336,7 @@ def _resolve_legacy_prompt_paths(cls, data: Any) -> Any: continue if resolved is data: resolved = dict(data) - resolved[prompt_key] = _resolve_prompt_source( + resolved[prompt_key] = resolve_prompt_source( prompt=resolved.get(prompt_key), path=resolved.pop(path_key), prompt_name=f"SeedSimulatedConversation.{prompt_key}", @@ -476,8 +476,11 @@ def __repr__(self) -> str: """ has_next_msg = self.next_message_system_prompt is not None + # ``name`` is descriptive metadata that _prompt_identity drops, so a seed rebuilt from a + # persisted record has none. Omit the fragment rather than print a placeholder. + prompt_name = self.adversarial_chat_system_prompt.name + adversarial = f", adversarial_prompt={prompt_name}" if prompt_name else "" return ( f"'})>" + f"next_message={has_next_msg}{adversarial})>" ) diff --git a/pyrit/scenario/core/attack_technique_factory.py b/pyrit/scenario/core/attack_technique_factory.py index 67ef95ce8d..c9be41af6b 100644 --- a/pyrit/scenario/core/attack_technique_factory.py +++ b/pyrit/scenario/core/attack_technique_factory.py @@ -36,10 +36,12 @@ SeedIdentifier, SeedPrompt, SeedSimulatedConversation, + SimulatedTargetSystemPromptPaths, load_next_message_prompt, load_simulated_target_prompt, + resolve_prompt_source, ) -from pyrit.models.seeds.seed_simulated_conversation import NextMessageSystemPromptPaths, _resolve_prompt_source +from pyrit.models.seeds.seed_simulated_conversation import NextMessageSystemPromptPaths from pyrit.scenario.core.attack_technique import AttackTechnique from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target @@ -201,8 +203,8 @@ def with_simulated_conversation( simulated conversation. Defaults to the prompt loaded from ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``. simulated_target_system_prompt: Optional system prompt for the simulated target - (the assistant side of the generated conversation). When ``None``, - ``SeedSimulatedConversation`` falls back to its compliant default. + (the assistant side of the generated conversation). Defaults to the prompt + loaded from ``SimulatedTargetSystemPromptPaths.COMPLIANT``. next_message_system_prompt: Optional system prompt for generating a final user message after the simulated conversation. Defaults to the prompt loaded from ``NextMessageSystemPromptPaths.DIRECT``. Ignored (forced to ``None``) when @@ -249,7 +251,7 @@ def with_simulated_conversation( """ if attack_class is None: attack_class = PromptSendingAttack - adversarial_chat_system_prompt = _resolve_prompt_source( + adversarial_chat_system_prompt = resolve_prompt_source( prompt=adversarial_chat_system_prompt, path=adversarial_chat_system_prompt_path, prompt_name="adversarial_chat_system_prompt", @@ -260,13 +262,17 @@ def with_simulated_conversation( adversarial_chat_system_prompt = SeedPrompt.from_yaml_file( Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml" ) - simulated_target_system_prompt = _resolve_prompt_source( + simulated_target_system_prompt = resolve_prompt_source( prompt=simulated_target_system_prompt, path=simulated_target_system_prompt_path, prompt_name="simulated_target_system_prompt", path_name="simulated_target_system_prompt_path", load_prompt=load_simulated_target_prompt, ) + if simulated_target_system_prompt is None: + simulated_target_system_prompt = load_simulated_target_prompt( + SimulatedTargetSystemPromptPaths.COMPLIANT.value + ) # A fixed final user message and an LLM-generated next message are mutually # exclusive: when a fixed message is supplied it becomes the next_message via @@ -274,7 +280,7 @@ def with_simulated_conversation( if final_user_message is not None: next_message_system_prompt = None else: - next_message_system_prompt = _resolve_prompt_source( + next_message_system_prompt = resolve_prompt_source( prompt=next_message_system_prompt, path=next_message_system_prompt_path, prompt_name="next_message_system_prompt", From 381303eb46f0a2671af21d905b990e74fd414469 Mon Sep 17 00:00:00 2001 From: varunj-msft Date: Tue, 22 Sep 2026 15:24:35 +0000 Subject: [PATCH 3/3] MAINT: Cover the unnamed-prompt branch of the simulated conversation repr The representation now omits the adversarial prompt fragment when the prompt carries no name, which is the usual state for a seed rebuilt from a persisted record. Assert that branch directly so it is not left to the named case. --- .../unit/models/test_seed_simulated_conversation.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/unit/models/test_seed_simulated_conversation.py b/tests/unit/models/test_seed_simulated_conversation.py index 3222c712c4..4f71d38e40 100644 --- a/tests/unit/models/test_seed_simulated_conversation.py +++ b/tests/unit/models/test_seed_simulated_conversation.py @@ -313,6 +313,18 @@ def test_repr_shows_num_turns_and_prompt_name(self, tmp_path): assert "num_turns=5" in repr_str assert "my_adversarial" in repr_str + def test_repr_omits_prompt_name_when_unnamed(self): + """An unnamed adversarial prompt drops the fragment rather than printing a placeholder.""" + conv = SeedSimulatedConversation( + adversarial_chat_system_prompt=SeedPrompt(value="test", data_type="text"), + num_turns=5, + ) + repr_str = repr(conv) + + assert "num_turns=5" in repr_str + assert "adversarial_prompt" not in repr_str + assert "None" not in repr_str + class TestSeedSimulatedConversationLoadSimulatedTargetSystemPrompt: """Tests for SeedSimulatedConversation.load_simulated_target_system_prompt static method."""