Skip to content
Merged
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
112 changes: 112 additions & 0 deletions src/strands_evals/experimental/redteam/experiment.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,9 @@

import asyncio
import inspect
import json
from collections.abc import Callable
from pathlib import Path
from typing import Any, cast

from strands import Agent
Expand All @@ -15,11 +17,13 @@
from ...evaluators.evaluator import Evaluator
from ...experiment import Experiment
from ...types import InputT, OutputT
from .case import RedTeamCase
from .evaluators import AttackSuccessEvaluator
from .report import RedTeamReport
from .strategies import AttackStrategy
from .strategies.target_session import TargetSession
from .task import _build_attacker_task
from .utils import _serialize_model


class RedTeamExperiment(Experiment[InputT, OutputT]):
Expand Down Expand Up @@ -67,6 +71,32 @@ def __init__(
# base Experiment doesn't carry task-returned metadata into EvaluationData.
self._run_meta: dict[str, dict[str, Any]] = {}

@property
def agent(self) -> Agent | TargetSession | None:
"""The target the default attacker task talks to.

Persisted experiments do not include the agent (live `Agent` /
`MultiAgentBase` / `TargetSession` instances are not JSON-serializable),
so the canonical load path is `from_file` followed by `exp.agent = ...`
before `run_evaluations`. Setting to `None` is allowed and restores the
"must supply explicit task" error path.
"""
return self._agent

@agent.setter
def agent(self, value: Agent | TargetSession | None) -> None:
self._agent = value

@property
def attack_strategies(self) -> list[AttackStrategy]:
"""The configured attack strategies (read-only view).

The experiment's `_by_label` index is built from this list at
construction; mutating the returned list will not re-index. Reload
with `from_dict` if you need to change strategies after construction.
"""
return list(self._attack_strategies)

@staticmethod
def _build_by_label(strategies: list[AttackStrategy]) -> dict[str, AttackStrategy]:
by_label: dict[str, AttackStrategy] = {}
Expand Down Expand Up @@ -164,3 +194,85 @@ def _default_task(self) -> Callable[[Case[InputT, OutputT]], Any]:
run_meta=self._run_meta,
),
)

def to_dict(self) -> dict: # type: ignore[override]
"""Serialize the experiment without the live target agent.

Adds `attack_strategies` and `model` on top of the base shape. The
target `agent` is intentionally omitted — `Agent` / `MultiAgentBase` /
`TargetSession` instances are not JSON-serializable, so callers must
re-attach via the `agent` setter (or pass an explicit `task`) after
loading. `_run_meta` is per-run state and is not persisted.
"""
out = super().to_dict()
out["attack_strategies"] = [strategy.to_dict() for strategy in self._attack_strategies]
# Coerce via the strategy helper for consistency with how strategies serialize their own model field.
model_id = _serialize_model(self._model)
if model_id is not None:
out["model"] = model_id
return out

@classmethod
def from_file( # type: ignore[override]
cls,
path: str,
custom_evaluators: list[type[Evaluator]] | None = None,
custom_strategies: list[type[AttackStrategy]] | None = None,
):
"""Load a RedTeamExperiment from a JSON file.

Same contract as :meth:`from_dict`: the loaded experiment has no
`agent` attached, so set `exp.agent = ...` before `run_evaluations`.
"""
file_path = Path(path)
if file_path.suffix != ".json":
raise ValueError(
f"Only .json format is supported. Got file: {path}. Please provide a path with .json extension."
)
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
return cls.from_dict(
data,
custom_evaluators=custom_evaluators,
custom_strategies=custom_strategies,
)

@classmethod
def from_dict( # type: ignore[override]
cls,
data: dict,
custom_evaluators: list[type[Evaluator]] | None = None,
custom_strategies: list[type[AttackStrategy]] | None = None,
):
"""Reconstruct a RedTeamExperiment from its serialized form.

The loaded experiment has no `agent` attached: set `exp.agent = ...`
before `run_evaluations`, or pass an explicit `task=`. Cases are
validated as `RedTeamCase` so the typed `config` survives reload.
`AttackSuccessEvaluator` is registered automatically; pass
`custom_evaluators` for user-defined evaluator subclasses and
`custom_strategies` for user-defined strategy subclasses.
"""
merged_evaluators: list[type[Evaluator]] = [AttackSuccessEvaluator, *(custom_evaluators or [])]
# Reuse the base evaluator-resolving path (registry + custom merge), but skip
# its case parser: we want RedTeamCase, not Case.
payload = dict(data)
case_dicts = payload.pop("cases", [])
strategy_dicts = payload.pop("attack_strategies", [])
model = payload.pop("model", None)
# Drive the base only for evaluator resolution by giving it an empty case list.
base_for_evaluators = super().from_dict(
{"cases": [], "evaluators": payload.get("evaluators", [])},
custom_evaluators=merged_evaluators,
)
cases: list[Case[InputT, OutputT]] = [RedTeamCase.model_validate(case_data) for case_data in case_dicts]
strategies = [
AttackStrategy.from_dict(strategy_data, custom_strategies=custom_strategies)
for strategy_data in strategy_dicts
]
return cls(
cases=cases,
attack_strategies=strategies,
evaluators=base_for_evaluators.evaluators,
model=model,
)
46 changes: 46 additions & 0 deletions src/strands_evals/experimental/redteam/strategies/base.py
Original file line number Diff line number Diff line change
Expand Up @@ -120,3 +120,49 @@ def reset(self) -> None: # noqa: B027
overrides must clear any per-case mutable state here. Called by
the task runner before each case.
"""

def to_dict(self) -> dict[str, Any]:
"""Serialize the strategy's static config.

The base only persists `strategy_type` (the class name, used as the
registry key in `from_dict`) and `label` when explicitly set.
Subclasses override to add their own ctor fields. Per-case runtime
state (cached attacker/judge agents, etc.) is intentionally dropped --
it is rebuilt on the first `run_attack` call after load.
"""
out: dict[str, Any] = {"strategy_type": type(self).__name__}
if self._label is not None:
out["label"] = self._label
return out

@classmethod
def from_dict(
cls,
data: dict[str, Any],
custom_strategies: list[type[AttackStrategy]] | None = None,
) -> AttackStrategy:
"""Reconstruct a strategy from its `to_dict` payload.

Resolves `strategy_type` against the built-in subclasses plus any
`custom_strategies` the caller passes through. Mirrors the
`Experiment.from_dict` evaluator-registry pattern.
"""
# Lazy import: subclasses inherit from AttackStrategy, so importing them at module top
# would cycle (base -> strategies/__init__ -> crescendo -> base).
from . import CrescendoStrategy, PromptStrategy

registry: dict[str, type[AttackStrategy]] = {
CrescendoStrategy.__name__: CrescendoStrategy,
PromptStrategy.__name__: PromptStrategy,
}
for custom in custom_strategies or []:
registry[custom.__name__] = custom

payload = dict(data)
type_name = payload.pop("strategy_type", None)
if type_name is None:
raise ValueError("Strategy dict is missing required key 'strategy_type'.")
if type_name not in registry:
raise ValueError(f"Cannot find strategy {type_name!r}. Pass `custom_strategies=[...]` to register it.")
target_cls = registry[type_name]
return target_cls(**payload)
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
from strands import Agent
from strands.models.model import Model

from ...utils import _serialize_model
from ..base import AttackRunResult, AttackStrategy
from . import crescendo_v0

Expand Down Expand Up @@ -303,6 +304,18 @@ def reset(self) -> None:
self._judge = None
self._refusal_judge = None

def to_dict(self) -> dict[str, Any]:
out = super().to_dict()
out.update(
max_turns=self._max_turns,
max_backtracks=self._max_backtracks,
success_threshold=self._success_threshold,
)
model_id = _serialize_model(self._model)
if model_id is not None:
out["model"] = model_id
return out

def _attacker_agent(self, goal: AttackGoal, model: Model | str | None) -> Agent:
# The attacker is lazily built once per case and its system prompt bakes in
# goal.actor_goal. Correctness relies on the task runner calling reset() before
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -95,5 +95,14 @@ def run_attack(

return AttackRunResult(conversation=conversation, metadata={"turns_used": len(conversation) // 2})

def to_dict(self) -> dict[str, Any]:
out = super().to_dict()
out.update(
strategy_name=self._name,
system_prompt_template=self._system_prompt_template,
max_turns=self._max_turns,
)
return out


__all__ = ["PromptStrategy"]
39 changes: 39 additions & 0 deletions src/strands_evals/experimental/redteam/utils.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,39 @@
"""Shared helpers for the red team module."""

from __future__ import annotations

import logging

from strands.models.model import Model

logger = logging.getLogger(__name__)


def _serialize_model(model: Model | str | None) -> str | None:
Comment thread
poshinchen marked this conversation as resolved.
"""Coerce a `model` value into a JSON-safe string id, or `None`.

Used by `RedTeamExperiment.to_dict` and individual strategy `to_dict`
methods so they all serialize their `model` field the same way. Unlike
`Evaluator._get_model_id`, `None` stays `None`: a strategy with
`model=None` defers to the experiment-level model, and that semantic
must survive a round-trip.

When a non-`None` `Model` instance does not expose a dict config with
a `model_id`, a warning is logged and `None` is returned. Returning `None`
means the caller's `to_dict` will drop the field, so reload will fall
back to the experiment-level model -- a visible behavior change, but the
warning makes it diagnosable rather than silent.
"""
if model is None or isinstance(model, str):
return model
if isinstance(model, Model):
config = model.get_config()
if isinstance(config, dict):
model_id = config.get("model_id")
if model_id is not None:
return str(model_id)
logger.warning(
"type=<%s> | non-coercible Model: missing dict config with 'model_id'; dropping field",
type(model).__name__,
)
return None
Loading
Loading