diff --git a/src/strands_evals/experimental/redteam/experiment.py b/src/strands_evals/experimental/redteam/experiment.py index adc01f0c..f1ea20da 100644 --- a/src/strands_evals/experimental/redteam/experiment.py +++ b/src/strands_evals/experimental/redteam/experiment.py @@ -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 @@ -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]): @@ -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] = {} @@ -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, + ) diff --git a/src/strands_evals/experimental/redteam/strategies/base.py b/src/strands_evals/experimental/redteam/strategies/base.py index 0e980947..7ba16b9a 100644 --- a/src/strands_evals/experimental/redteam/strategies/base.py +++ b/src/strands_evals/experimental/redteam/strategies/base.py @@ -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) diff --git a/src/strands_evals/experimental/redteam/strategies/crescendo/__init__.py b/src/strands_evals/experimental/redteam/strategies/crescendo/__init__.py index 472913c0..4cf562ef 100644 --- a/src/strands_evals/experimental/redteam/strategies/crescendo/__init__.py +++ b/src/strands_evals/experimental/redteam/strategies/crescendo/__init__.py @@ -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 @@ -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 diff --git a/src/strands_evals/experimental/redteam/strategies/prompt_strategy/__init__.py b/src/strands_evals/experimental/redteam/strategies/prompt_strategy/__init__.py index 3be34d84..96c658d2 100644 --- a/src/strands_evals/experimental/redteam/strategies/prompt_strategy/__init__.py +++ b/src/strands_evals/experimental/redteam/strategies/prompt_strategy/__init__.py @@ -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"] diff --git a/src/strands_evals/experimental/redteam/utils.py b/src/strands_evals/experimental/redteam/utils.py new file mode 100644 index 00000000..b0173b6f --- /dev/null +++ b/src/strands_evals/experimental/redteam/utils.py @@ -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: + """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 diff --git a/tests/strands_evals/experimental/redteam/test_experiment.py b/tests/strands_evals/experimental/redteam/test_experiment.py index 008d2e0a..f3cfdafe 100644 --- a/tests/strands_evals/experimental/redteam/test_experiment.py +++ b/tests/strands_evals/experimental/redteam/test_experiment.py @@ -1,15 +1,37 @@ """Tests for RedTeamExperiment.""" import pytest +from strands.models.model import Model from strands_evals.experimental.redteam.case import RedTeamCase from strands_evals.experimental.redteam.evaluators import AttackSuccessEvaluator from strands_evals.experimental.redteam.experiment import RedTeamExperiment from strands_evals.experimental.redteam.report import RedTeamReport +from strands_evals.experimental.redteam.strategies import CrescendoStrategy, PromptStrategy from strands_evals.experimental.redteam.strategies.base import AttackRunResult, AttackStrategy from strands_evals.experimental.redteam.types import AttackGoal, RedTeamConfig +class _StubModel(Model): + """Minimal Model subclass used to exercise the runtime-object branch of `_serialize_model`.""" + + def __init__(self, config: dict | None) -> None: + # `config` may be a non-dict (e.g. None) on purpose for the non-coercible branch. + self.config = config # type: ignore[assignment] + + def get_config(self): + return self.config + + def update_config(self, **kwargs): + pass + + def structured_output(self, *args, **kwargs): + raise NotImplementedError + + async def stream(self, *args, **kwargs): + raise NotImplementedError + + class _StubStrategy(AttackStrategy): def __init__(self, name="stub", *, label=None): super().__init__(label=label) @@ -142,6 +164,114 @@ async def _async_task(case): exp.run_evaluations(task=_async_task) +def test_agent_setter_round_trip(): + """`exp.agent = ...` is the canonical way to attach a target after `from_file`.""" + exp = RedTeamExperiment(cases=[_case()], attack_strategies=[_StubStrategy()]) + assert exp.agent is None + sess = _FakeSession() + exp.agent = sess + assert exp.agent is sess + + +def test_to_dict_persists_strategies_and_model(): + exp = RedTeamExperiment( + cases=[_case()], + attack_strategies=[ + CrescendoStrategy(max_turns=3, max_backtracks=2, success_threshold=0.6, label="cre-fast"), + PromptStrategy("gradual_escalation", "TPL", max_turns=4), + ], + model="claude-3-5", + ) + out = exp.to_dict() + assert out["model"] == "claude-3-5" + assert "agent" not in out # live target is never persisted + strategies = out["attack_strategies"] + assert strategies[0] == { + "strategy_type": "CrescendoStrategy", + "label": "cre-fast", + "max_turns": 3, + "max_backtracks": 2, + "success_threshold": 0.6, + } + assert strategies[1] == { + "strategy_type": "PromptStrategy", + "strategy_name": "gradual_escalation", + "system_prompt_template": "TPL", + "max_turns": 4, + } + + +def test_from_dict_round_trip_runs_after_setting_agent(tmp_path): + """Reload via to_file/from_file, then attach an agent and run.""" + exp = RedTeamExperiment( + cases=[_case("c0"), _case("c1")], + attack_strategies=[CrescendoStrategy(max_turns=3, label="cre")], + model="claude-3-5", + ) + p = tmp_path / "rt.json" + exp.to_file(str(p)) + loaded = RedTeamExperiment.from_file(str(p)) + + assert loaded.agent is None + assert loaded._model == "claude-3-5" + assert [type(c).__name__ for c in loaded.cases] == ["RedTeamCase", "RedTeamCase"] + assert loaded.cases[0].config.attack_goal.actor_goal == "goal" + assert [s.label for s in loaded.attack_strategies] == ["cre"] + assert isinstance(loaded.evaluators[0], AttackSuccessEvaluator) + # Symmetric: any field-level drift (added/dropped/reordered key, lossy coercion) + # would be caught here in one assertion, complementing the per-field checks above. + assert RedTeamExperiment.from_dict(exp.to_dict()).to_dict() == exp.to_dict() + + # Without agent, run_evaluations raises the existing message. + with pytest.raises(ValueError, match="agent.*task"): + loaded.run_evaluations() + + # Attach an agent and supply a stub task to skip real LLM calls. + loaded.agent = _FakeSession() + report = loaded.run_evaluations(task=lambda case: {"output": []}) + assert isinstance(report, RedTeamReport) + + +def test_from_dict_accepts_custom_strategies(tmp_path): + exp = RedTeamExperiment(cases=[_case()], attack_strategies=[_StubStrategy(label="stub-a")]) + p = tmp_path / "rt.json" + exp.to_file(str(p)) + loaded = RedTeamExperiment.from_file(str(p), custom_strategies=[_StubStrategy]) + assert [type(s).__name__ for s in loaded.attack_strategies] == ["_StubStrategy"] + assert loaded.attack_strategies[0].label == "stub-a" + + +def test_to_dict_serializes_model_instance(): + """A Model instance with `config['model_id']` round-trips into out['model'].""" + model = _StubModel(config={"model_id": "claude-stub"}) + exp = RedTeamExperiment( + cases=[_case()], + attack_strategies=[CrescendoStrategy(model=model)], + model=model, + ) + out = exp.to_dict() + assert out["model"] == "claude-stub" + assert out["attack_strategies"][0]["model"] == "claude-stub" + + +def test_to_dict_drops_non_coercible_model_with_warning(caplog): + """A Model that doesn't expose dict config + model_id logs a warning and is dropped.""" + model = _StubModel(config=None) + exp = RedTeamExperiment(cases=[_case()], model=model) + with caplog.at_level("WARNING"): + out = exp.to_dict() + assert "model" not in out + assert any("non-coercible Model" in record.message for record in caplog.records) + + +def test_from_dict_unknown_strategy_raises(tmp_path): + exp = RedTeamExperiment(cases=[_case()], attack_strategies=[_StubStrategy()]) + p = tmp_path / "rt.json" + exp.to_file(str(p)) + with pytest.raises(ValueError, match="_StubStrategy"): + RedTeamExperiment.from_file(str(p)) # no custom_strategies + + def test_run_evaluations_twice_is_idempotent(): """Re-running must not re-expand an already-expanded case list (no c0__cre__cre).""" runs: list[list[str]] = []