pyrit.executor.attack.AttackExecutor

class AttackExecutor(*, max_concurrency: int = 1)

Bases: object

Manages the execution of attack strategies with support for parallel execution.

The AttackExecutor provides controlled execution of attack strategies with concurrency limiting. It uses the attack’s params_type to create parameters from seed groups.

__init__(*, max_concurrency: int = 1)

Initialize the attack executor with configurable concurrency control.

Parameters:

max_concurrency – Maximum number of concurrent attack executions (default: 1).

Raises:

ValueError – If max_concurrency is not a positive integer.

Methods

__init__(*[, max_concurrency])	Initialize the attack executor with configurable concurrency control.
execute_attack_async(*, attack, objectives)	Execute attacks in parallel for each objective.
execute_attack_from_seed_groups_async(*, ...)	Execute attacks in parallel, extracting parameters from SeedGroups.
execute_multi_objective_attack_async(*, ...)	Execute the same attack strategy with multiple objectives against the same target in parallel.
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_attack_async(*, attack: AttackStrategy[AttackStrategyContextT, AttackStrategyResultT], objectives: Sequence[str], field_overrides: Sequence[Dict[str, Any]] | None = None, return_partial_on_failure: bool = False, **broadcast_fields) → AttackExecutorResult[AttackStrategyResultT]

Execute attacks in parallel for each objective.

Creates AttackParameters directly from objectives and field values.

Parameters:

• attack – The attack strategy to execute.

• objectives – List of attack objectives.

• field_overrides – Optional per-objective field overrides. If provided, must match the length of objectives.

• return_partial_on_failure – If True, returns partial results when some objectives fail. If False (default), raises the first exception.

• **broadcast_fields – Fields applied to all objectives (e.g., memory_labels). Per-objective field_overrides take precedence.

Returns:

AttackExecutorResult with completed results and any incomplete objectives.

Raises:

• ValueError – If objectives is empty or field_overrides length doesn’t match.

• BaseException – If return_partial_on_failure=False and any objective fails.

async execute_attack_from_seed_groups_async(*, attack: AttackStrategy[AttackStrategyContextT, AttackStrategyResultT], seed_groups: Sequence[SeedGroup], field_overrides: Sequence[Dict[str, Any]] | None = None, return_partial_on_failure: bool = False, **broadcast_fields) → AttackExecutorResult[AttackStrategyResultT]

Execute attacks in parallel, extracting parameters from SeedGroups.

Uses the attack’s params_type.from_seed_group() to extract parameters, automatically handling which fields the attack accepts.

Parameters:

• attack – The attack strategy to execute.

• seed_groups – SeedGroups containing objectives and optional prompts.

• field_overrides – Optional per-seed-group field overrides. If provided, must match the length of seed_groups. Each dict is passed to from_seed_group() as overrides.

• return_partial_on_failure – If True, returns partial results when some objectives fail. If False (default), raises the first exception.

• **broadcast_fields – Fields applied to all seed groups (e.g., memory_labels). Per-seed-group field_overrides take precedence.

Returns:

AttackExecutorResult with completed results and any incomplete objectives.

Raises:

• ValueError – If seed_groups is empty or field_overrides length doesn’t match.

• BaseException – If return_partial_on_failure=False and any objective fails.

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

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

Deprecated since version Use: execute_attack_async() instead. This method will be removed in a future version.

Parameters:

• attack – The attack strategy to use for all objectives.

• objectives – List of attack objectives to test.

• prepended_conversation – Conversation to prepend to the target model.

• memory_labels – Additional labels that can be applied to the prompts.

• return_partial_on_failure – If True, returns partial results on failure.

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

Returns:

AttackExecutorResult with completed results and any incomplete objectives.

async execute_multi_turn_attacks_async(*, attack: AttackStrategy[_MultiTurnContextT, AttackStrategyResultT], objectives: List[str], messages: List[Message] | None = None, prepended_conversations: List[List[Message]] | None = None, memory_labels: Dict[str, str] | None = None, return_partial_on_failure: bool = False, **attack_params) → AttackExecutorResult[AttackStrategyResultT]

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

Deprecated since version Use: execute_attack_async() instead. This method will be removed in a future version.

Parameters:

• attack – The multi-turn attack strategy to use.

• objectives – List of attack objectives to test.

• messages – List of messages to use for this execution (per-objective).

• prepended_conversations – Conversations to prepend to each objective (per-objective).

• memory_labels – Additional labels that can be applied to the prompts.

• return_partial_on_failure – If True, returns partial results on failure.

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

Returns:

AttackExecutorResult with completed results and any incomplete objectives.

Raises:

TypeError – If the attack does not use MultiTurnAttackContext.

async execute_single_turn_attacks_async(*, attack: AttackStrategy[_SingleTurnContextT, AttackStrategyResultT], objectives: List[str], messages: List[Message] | None = None, prepended_conversations: List[List[Message]] | None = None, memory_labels: Dict[str, str] | None = None, return_partial_on_failure: bool = False, **attack_params) → AttackExecutorResult[AttackStrategyResultT]

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

Deprecated since version Use: execute_attack_async() instead. This method will be removed in a future version.

Parameters:

• attack – The single-turn attack strategy to use.

• objectives – List of attack objectives to test.

• messages – List of messages to use for this execution (per-objective).

• prepended_conversations – Conversations to prepend to each objective (per-objective).

• memory_labels – Additional labels that can be applied to the prompts.

• return_partial_on_failure – If True, returns partial results on failure.

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

Returns:

AttackExecutorResult with completed results and any incomplete objectives.

Raises:

TypeError – If the attack does not use SingleTurnAttackContext.