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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 7 additions & 1 deletion doc/code/scenarios/0_attack_techniques.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,13 @@
"\n",
"The objective is *not* part of the technique — it stays separate and is supplied by the dataset at\n",
"run time. You rarely build a technique by hand; instead you register a **factory** and let scenarios\n",
"construct techniques on demand with the scenario's own objective target and scorer."
"construct techniques on demand with the scenario's own objective target and scorer.\n",
"\n",
"`adversarial_chat_system_prompt` accepts either an inline `SeedPrompt` or a YAML `Path`. The factory\n",
"resolves either representation to a `SeedPrompt` immediately, so created techniques carry portable\n",
"prompt content rather than a runtime file dependency. The legacy\n",
"`adversarial_chat_system_prompt_path` name remains a compatibility alias, and if neither parameter\n",
"is supplied, the existing `red_teaming/{technique_name}.yaml` convention remains the default."
]
},
{
Expand Down
6 changes: 6 additions & 0 deletions doc/code/scenarios/0_attack_techniques.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,12 @@
# The objective is *not* part of the technique — it stays separate and is supplied by the dataset at
# run time. You rarely build a technique by hand; instead you register a **factory** and let scenarios
# construct techniques on demand with the scenario's own objective target and scorer.
#
# `adversarial_chat_system_prompt` accepts either an inline `SeedPrompt` or a YAML `Path`. The factory
# resolves either representation to a `SeedPrompt` immediately, so created techniques carry portable
# prompt content rather than a runtime file dependency. The legacy
# `adversarial_chat_system_prompt_path` name remains a compatibility alias, and if neither parameter
# is supplied, the existing `red_teaming/{technique_name}.yaml` convention remains the default.

# %% [markdown]
# ## Where techniques come from: initializers
Expand Down
1 change: 1 addition & 0 deletions pyrit/executor/attack/core/attack_parameters.py
Original file line number Diff line number Diff line change
Expand Up @@ -167,6 +167,7 @@ async def from_seed_group_async(
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,
adversarial_chat_system_prompt=simulated_conversation_config.adversarial_chat_system_prompt,
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,
)
Expand Down
45 changes: 38 additions & 7 deletions pyrit/executor/attack/multi_turn/simulated_conversation.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@

from __future__ import annotations

import asyncio
import logging
from typing import TYPE_CHECKING

Expand Down Expand Up @@ -43,7 +44,8 @@ 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_path: str | Path | None = None,
adversarial_chat_system_prompt: SeedPrompt | 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,
Expand All @@ -70,7 +72,8 @@ 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.
adversarial_chat_system_prompt_path: Legacy path to the system prompt for the adversarial chat.
adversarial_chat_system_prompt: Canonical inline system prompt for the adversarial chat.
simulated_target_system_prompt_path: Path to the 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
Expand All @@ -89,7 +92,7 @@ async def generate_simulated_conversation_async(
generated to elicit the objective fulfillment.

Raises:
ValueError: If num_turns is not a positive integer.
ValueError: If num_turns is not positive or the adversarial prompt source is ambiguous or missing.
"""
# Use the same LLM for both adversarial chat and simulated target
# They get different system prompts to play different roles
Expand All @@ -105,10 +108,9 @@ async def generate_simulated_conversation_async(
simulated_target_system_prompt_path=simulated_target_system_prompt_path,
)

# 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
adversarial_system_prompt = await _resolve_adversarial_chat_system_prompt_async(
adversarial_chat_system_prompt_path=adversarial_chat_system_prompt_path,
adversarial_chat_system_prompt=adversarial_chat_system_prompt,
)
adversarial_config = AttackAdversarialConfig(
target=adversarial_chat,
Expand Down Expand Up @@ -176,6 +178,35 @@ async def generate_simulated_conversation_async(
return seed_prompts


async def _resolve_adversarial_chat_system_prompt_async(
*,
adversarial_chat_system_prompt_path: str | Path | None,
adversarial_chat_system_prompt: SeedPrompt | None,
) -> SeedPrompt:
"""
Adapt a legacy path-backed prompt or canonical inline prompt for execution.

Args:
adversarial_chat_system_prompt_path: Legacy YAML prompt path.
adversarial_chat_system_prompt: Canonical inline prompt.

Returns:
The resolved adversarial chat system prompt.

Raises:
ValueError: If both or neither prompt sources are provided.
"""
has_prompt_path = adversarial_chat_system_prompt_path is not None
has_inline_prompt = adversarial_chat_system_prompt is not None
if has_prompt_path == has_inline_prompt:
raise ValueError("Set exactly one of adversarial_chat_system_prompt_path or adversarial_chat_system_prompt.")
if adversarial_chat_system_prompt is not None:
return adversarial_chat_system_prompt

assert adversarial_chat_system_prompt_path is not None
return await asyncio.to_thread(SeedPrompt.from_yaml_file, adversarial_chat_system_prompt_path)


async def _generate_next_message_async(
*,
objective: str,
Expand Down
2 changes: 2 additions & 0 deletions pyrit/memory/memory_models.py
Original file line number Diff line number Diff line change
Expand Up @@ -1491,8 +1491,10 @@ def get_seed(self) -> Seed:
num_turns=config.get("num_turns", 3),
sequence=config.get("sequence", 0),
adversarial_chat_system_prompt_path=config.get("adversarial_chat_system_prompt_path"),
adversarial_chat_system_prompt=config.get("adversarial_chat_system_prompt"),
simulated_target_system_prompt_path=config.get("simulated_target_system_prompt_path"),
next_message_system_prompt_path=config.get("next_message_system_prompt_path"),
pyrit_version=config.get("pyrit_version"),
)
return SeedPrompt(
id=self.id,
Expand Down
80 changes: 73 additions & 7 deletions pyrit/models/seeds/seed_simulated_conversation.py
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,7 @@ 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, paths, 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
Expand All @@ -58,7 +58,8 @@ class SeedSimulatedConversation(Seed):

Attributes:
num_turns: Number of conversation turns to generate.
adversarial_chat_system_prompt_path: Path to the adversarial chat system prompt YAML.
adversarial_chat_system_prompt_path: Legacy path to the adversarial chat system prompt YAML.
adversarial_chat_system_prompt: Canonical inline adversarial chat system prompt.
simulated_target_system_prompt_path: Path to the simulated target system prompt YAML.
Defaults to the compliant prompt if not specified.
next_message_system_prompt_path: Optional path to the system prompt for generating
Expand All @@ -85,7 +86,8 @@ class SeedSimulatedConversation(Seed):

num_turns: int = 3
sequence: int = 0
adversarial_chat_system_prompt_path: Path
adversarial_chat_system_prompt_path: Path | None = None
adversarial_chat_system_prompt: SeedPrompt | None = None
simulated_target_system_prompt_path: Path = SimulatedTargetSystemPromptPaths.COMPLIANT.value
next_message_system_prompt_path: Path | None = None
pyrit_version: str | None = None
Expand Down Expand Up @@ -116,6 +118,12 @@ def _default_simulated_target_path(cls, value: Any) -> Any:

@model_validator(mode="after")
def _validate_and_compute_value(self) -> SeedSimulatedConversation:
has_prompt_path = self.adversarial_chat_system_prompt_path is not None
has_inline_prompt = self.adversarial_chat_system_prompt is not None
if has_prompt_path == has_inline_prompt:
raise ValueError(
"Set exactly one of adversarial_chat_system_prompt_path or adversarial_chat_system_prompt."
)
if self.num_turns <= 0:
raise ValueError("num_turns must be a positive integer")
if self.sequence < 0:
Expand All @@ -133,18 +141,63 @@ def _compute_value(self) -> str:
str: Deterministic JSON representation of this configuration.

"""
config = {
config: dict[str, Any] = {
"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,
}
if self.adversarial_chat_system_prompt is not None:
config["adversarial_chat_system_prompt"] = self._serialize_adversarial_chat_system_prompt()
else:
config["adversarial_chat_system_prompt_path"] = str(self.adversarial_chat_system_prompt_path)
return json.dumps(config, sort_keys=True, separators=(",", ":"))

def _serialize_adversarial_chat_system_prompt(self) -> dict[str, Any]:
"""
Serialize the inline prompt without generated identity or timestamp fields.

Returns:
A deterministic JSON-compatible prompt representation.
"""
assert self.adversarial_chat_system_prompt is not None
prompt_data = self.adversarial_chat_system_prompt.model_dump(
mode="python",
exclude={"id", "date_added", "value_sha256", "prompt_group_id"},
)
self._reject_unordered_collections(prompt_data)
return self.adversarial_chat_system_prompt.model_dump(
mode="json",
exclude={"id", "date_added", "value_sha256", "prompt_group_id"},
)

@classmethod
def _reject_unordered_collections(cls, value: Any) -> None:
"""
Reject values whose JSON list order can vary across processes.

Args:
value: Prompt data to inspect recursively.

Raises:
ValueError: If the prompt contains a set or frozenset.
"""
if isinstance(value, (set, frozenset)):
raise ValueError(
"Inline adversarial chat system prompts must use ordered JSON-compatible values; "
"set and frozenset values are not supported."
)
if isinstance(value, dict):
for nested_key, nested_value in value.items():
cls._reject_unordered_collections(nested_key)
cls._reject_unordered_collections(nested_value)
elif isinstance(value, (list, tuple)):
for nested_value in value:
cls._reject_unordered_collections(nested_value)

def get_identifier(self) -> dict[str, Any]:
"""
Get an identifier dict capturing this configuration for comparison/storage.
Expand All @@ -153,17 +206,21 @@ def get_identifier(self) -> dict[str, Any]:
Dictionary with configuration details.

"""
return {
identifier: dict[str, Any] = {
"__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,
}
if self.adversarial_chat_system_prompt is not None:
identifier["adversarial_chat_system_prompt"] = self._serialize_adversarial_chat_system_prompt()
else:
identifier["adversarial_chat_system_prompt_path"] = str(self.adversarial_chat_system_prompt_path)
return identifier

def compute_hash(self) -> str:
"""
Expand Down Expand Up @@ -242,6 +299,15 @@ def __repr__(self) -> str:

"""
has_next_msg = self.next_message_system_prompt_path is not None
if self.adversarial_chat_system_prompt is not None:
adversarial_source = self.adversarial_chat_system_prompt.name or "<inline>"
return (
f"<SeedSimulatedConversation(num_turns={self.num_turns}, sequence={self.sequence}, "
f"next_message={has_next_msg}, "
f"adversarial_prompt={adversarial_source})>"
)

assert self.adversarial_chat_system_prompt_path is not None
return (
f"<SeedSimulatedConversation(num_turns={self.num_turns}, sequence={self.sequence}, "
f"next_message={has_next_msg}, "
Expand Down
4 changes: 4 additions & 0 deletions pyrit/scenario/core/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,9 @@
from pyrit.scenario.core.scenario import BaselineAttackPolicy, Scenario
from pyrit.scenario.core.scenario_target_defaults import get_default_adversarial_target, get_default_scorer_target
from pyrit.scenario.core.scenario_technique import ScenarioTechnique
from pyrit.scenario.core.simulated_conversation_prompt import (
resolve_simulated_conversation_adversarial_prompt,
)

_LAZY_EXPORTS: dict[str, str | tuple[str, str | None]] = {
"AtomicAttack": "pyrit.scenario.core.atomic_attack",
Expand All @@ -46,6 +49,7 @@
"ScorerOverridePolicy": "pyrit.scenario.core.attack_technique_factory",
"get_default_scorer_target": "pyrit.scenario.core.scenario_target_defaults",
"get_default_adversarial_target": "pyrit.scenario.core.scenario_target_defaults",
"resolve_simulated_conversation_adversarial_prompt": "pyrit.scenario.core.simulated_conversation_prompt",
}

__all__ = list(_LAZY_EXPORTS)
Expand Down
20 changes: 15 additions & 5 deletions pyrit/scenario/core/attack_technique_factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,9 @@
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
from pyrit.scenario.core.simulated_conversation_prompt import (
resolve_simulated_conversation_adversarial_prompt,
)

if TYPE_CHECKING:
from pyrit.converter import Converter
Expand Down Expand Up @@ -159,6 +162,7 @@ def with_simulated_conversation(
name: str,
attack_class: type[AttackStrategy[Any, Any]] | None = None,
description: str | None = None,
adversarial_chat_system_prompt: SeedPrompt | Path | 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,
Expand All @@ -185,9 +189,11 @@ 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.
adversarial_chat_system_prompt: Inline adversarial chat system prompt
or ``Path`` to its YAML file.
Defaults to ``EXECUTOR_SEED_PROMPT_PATH/red_teaming/{name}.yaml``.
adversarial_chat_system_prompt_path: Legacy YAML path alias for
``adversarial_chat_system_prompt``. The two parameters are mutually exclusive.
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``,
Expand Down Expand Up @@ -230,8 +236,12 @@ 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"

resolved_adversarial_prompt = resolve_simulated_conversation_adversarial_prompt(
adversarial_chat_system_prompt_path=adversarial_chat_system_prompt_path,
adversarial_chat_system_prompt=adversarial_chat_system_prompt,
default_system_prompt_path=Path(EXECUTOR_SEED_PROMPT_PATH) / "red_teaming" / f"{name}.yaml",
)

# 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
Expand All @@ -242,7 +252,7 @@ def with_simulated_conversation(
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),
"adversarial_chat_system_prompt": resolved_adversarial_prompt,
"num_turns": num_turns,
}
if simulated_target_system_prompt_path is not None:
Expand Down
Loading