pyrit.executor.attack.AttackExecutor

class AttackExecutor(*, max_concurrency: int = 1)

Bases: object

Manages the execution of attack strategies with support for different execution patterns.

The AttackExecutor provides controlled execution of attack strategies with features like concurrency limiting and parallel execution. It can handle multiple objectives against the same target or execute different strategies concurrently.

__init__(*, max_concurrency: int = 1)

Initialize the attack executor with configurable concurrency control.

Parameters:

max_concurrency (int) – Maximum number of concurrent attack executions allowed. Must be a positive integer (defaults to 1).

Raises:

ValueError – If max_concurrency is not a positive integer.

Methods

__init__(*[, max_concurrency])	Initialize the attack executor with configurable concurrency control.
execute_multi_objective_attack_async(*, ...)	Execute the same attack strategy with multiple objectives against the same target in parallel.
execute_multi_objective_attack_with_context_async(*, ...)	Execute the same attack strategy with multiple objectives using context objects.
execute_multi_turn_attacks_async(*, attack, ...)	Execute a batch of multi-turn attacks with multiple objectives.
execute_single_turn_attacks_async(*, attack, ...)	Execute a batch of single-turn attacks with multiple objectives.

async execute_multi_objective_attack_async(*, attack: AttackStrategy[AttackStrategyContextT, AttackStrategyResultT], objectives: List[str], prepended_conversation: List[PromptRequestResponse] | None = None, memory_labels: Dict[str, str] | None = None, **attack_params) → List[AttackStrategyResultT]

Execute the same attack strategy with multiple objectives against the same target in parallel.

This method provides a simplified interface for executing multiple objectives without requiring users to create context objects. It uses the attack’s execute_async method which accepts parameters directly.

Parameters:

• attack (AttackStrategy[ContextT, AttackStrategyResultT]) – The attack strategy to use for all objectives.

• objectives (List[str]) – List of attack objectives to test.

• prepended_conversation (Optional[List[PromptRequestResponse]]) – Conversation to prepend to the target model.

• memory_labels (Optional[Dict[str, str]]) – Additional labels that can be applied to the prompts.

• **attack_params – Additional parameters specific to the attack strategy.

Returns:

List of attack results in the same order as the objectives list.

Return type:

List[AttackStrategyResultT]

Example

>>> executor = AttackExecutor(max_concurrency=3)
>>> results = await executor.execute_multi_objective_attack_async(
...     attack=red_teaming_attack,
...     objectives=["how to make a Molotov cocktail", "how to escalate privileges"],
... )

async execute_multi_objective_attack_with_context_async(*, attack: AttackStrategy[AttackStrategyContextT, AttackStrategyResultT], context_template: AttackStrategyContextT, objectives: List[str]) → List[AttackStrategyResultT]

Execute the same attack strategy with multiple objectives using context objects.

This method works with context objects directly, duplicating the template context for each objective. Use this when you need fine-grained control over the context or have an existing context template to reuse.

Parameters:

• attack (AttackStrategy[AttackStrategyContextT, AttackStrategyResultT]) – The attack strategy to use for all objectives.

• context_template (AttackStrategyContextT) – Template context that will be duplicated for each objective. Must have a ‘duplicate()’ method and an ‘objective’ attribute.

• objectives (List[str]) – List of attack objectives to test. Each objective will be executed as a separate attack using a copy of the context template.

Returns:

List of attack results in the same order as the objectives list.

Each result corresponds to the execution of the strategy with the objective at the same index.

Return type:

List[AttackStrategyResultT]

Raises:

• AttributeError – If the context_template doesn’t have required ‘duplicate()’ method or ‘objective’ attribute.

• Any exceptions raised during strategy execution will be propagated. –

Example

>>> executor = AttackExecutor(max_concurrency=3)
>>> context = MultiTurnAttackContext(max_turns=5, ...)
>>> results = await executor.execute_multi_objective_attack_with_context_async(
...     attack=prompt_injection_attack,
...     context_template=context,
...     objectives=["how to make a Molotov cocktail", "how to escalate privileges"]
... )

async execute_multi_turn_attacks_async(*, attack: AttackStrategy[_MultiTurnContextT, AttackStrategyResultT], objectives: List[str], custom_prompts: List[str] | None = None, prepended_conversations: List[List[PromptRequestResponse]] | None = None, memory_labels: Dict[str, str] | None = None) → List[AttackStrategyResultT]

Execute a batch of multi-turn attacks with multiple objectives.

This method is specifically designed for multi-turn attacks, allowing you to execute multiple objectives in parallel while managing the contexts and custom prompts.

Parameters:

• attack (AttackStrategy[_MultiTurnContextT, AttackStrategyResultT]) – The multi-turn attack strategy to use, the context must be a MultiTurnAttackContext or a subclass of it.

• objectives (List[str]) – List of attack objectives to test.

• custom_prompts (Optional[List[str]]) – List of custom prompts to use for this execution. If provided, must match the length of objectives. custom prompts will be sent along the objective with the same list index.

• prepended_conversations (Optional[List[List[PromptRequestResponse]]]) – Conversations to prepend to each objective. If provided, must match the length of objectives. Conversation will be sent along the objective with the same list index.

• memory_labels (Optional[Dict[str, str]]) – Additional labels that can be applied to the prompts. The labels will be the same across all executions.

Returns:

List of attack results in the same order as the objectives list.

Return type:

List[AttackStrategyResultT]

Example

>>> executor = AttackExecutor(max_concurrency=3)
>>> results = await executor.execute_multi_turn_attacks_async(
...     attack=multi_turn_attack,
...     objectives=["how to make a Molotov cocktail", "how to escalate privileges"],
...     custom_prompts=["Tell me about chemistry", "Explain system administration"]
... )

async execute_single_turn_attacks_async(*, attack: AttackStrategy[_SingleTurnContextT, AttackStrategyResultT], objectives: List[str], seed_prompt_groups: List[SeedPromptGroup] | None = None, prepended_conversations: List[List[PromptRequestResponse]] | None = None, memory_labels: Dict[str, str] | None = None) → List[AttackStrategyResultT]

Execute a batch of single-turn attacks with multiple objectives.

This method is specifically designed for single-turn attacks, allowing you to execute multiple objectives in parallel while managing the contexts and prompts.

Parameters:

• attack (AttackStrategy[_SingleTurnContextT, AttackStrategyResultT]) – The single-turn attack strategy to use, the context must be a SingleTurnAttackContext or a subclass of it.

• objectives (List[str]) – List of attack objectives to test.

• seed_prompt_groups (Optional[List[SeedPromptGroup]]) – List of seed prompt groups to use for this execution. If provided, must match the length of objectives. Seed prompt group will be sent along the objective with the same list index.

• prepended_conversations (Optional[List[List[PromptRequestResponse]]]) – Conversations to prepend to each objective. If provided, must match the length of objectives. Conversation will be sent along the objective with the same list index.

• memory_labels (Optional[Dict[str, str]]) – Additional labels that can be applied to the prompts. The labels will be the same across all executions.

Returns:

List of attack results in the same order as the objectives list.

Return type:

List[AttackStrategyResultT]

Example

>>> executor = AttackExecutor(max_concurrency=3)
>>> results = await executor.execute_single_turn_batch_async(
...     attack=single_turn_attack,
...     objectives=["how to make a Molotov cocktail", "how to escalate privileges"],
...     seed_prompts=[SeedPromptGroup(...), SeedPromptGroup(...)]
... )