FEAT text adaptive scenario

doc/code/scenarios/3_adaptive_scenarios.ipynb

@@ -0,0 +1,282 @@
+{
+ "cells": [
+  {
+   "cell_type": "markdown",
+   "id": "0",
+   "metadata": {},
+   "source": [
+    "# Adaptive Scenarios\n",
+    "\n",
+    "An **adaptive scenario** doesn't run every attack technique against every objective.\n",
+    "Instead, it picks which technique to try next per-objective, learns from what worked,\n",
+    "and stops as soon as one technique succeeds. This concentrates spend on techniques\n",
+    "that actually work on your target.\n",
+    "\n",
+    "## How it works (high level)\n",
+    "\n",
+    "For each objective, the scenario tries up to `max_attempts_per_objective` techniques:\n",
+    "\n",
+    "- With probability `epsilon`, it **explores** \u2014 picks a random technique.\n",
+    "- Otherwise it **exploits** \u2014 picks the technique with the highest observed success\n",
+    "  rate so far.\n",
+    "- It records the outcome and stops early on success.\n",
+    "\n",
+    "Unseen techniques are tried first, so the first few objectives effectively round-robin\n",
+    "through every technique before the scenario settles on the best performers.\n",
+    "\n",
+    "## Adaptive vs. static scenarios\n",
+    "\n",
+    "| Feature             | Static scenarios                  | Adaptive scenarios                 |\n",
+    "|---------------------|-----------------------------------|------------------------------------|\n",
+    "| Technique selection | Run every selected technique      | Pick per-objective from outcomes   |\n",
+    "| Early stopping      | No                                | Yes \u2014 stops on first success       |\n",
+    "| Cost                | O(techniques \u00d7 objectives)        | O(max_attempts \u00d7 objectives)       |\n",
+    "\n",
+    "`AdaptiveScenario` is the modality-agnostic base class.\n",
+    "[`TextAdaptive`](../../../pyrit/scenario/scenarios/adaptive/text_adaptive.py) is the\n",
+    "text subclass used in the examples below."
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "1",
+   "metadata": {},
+   "source": [
+    "## Setup"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "2",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "from pathlib import Path\n",
+    "\n",
+    "from pyrit.registry import TargetRegistry\n",
+    "from pyrit.scenario import DatasetConfiguration\n",
+    "from pyrit.scenario.printer.console_printer import ConsoleScenarioResultPrinter\n",
+    "from pyrit.scenario.scenarios.adaptive import TextAdaptive, harm_category_context\n",
+    "from pyrit.setup import initialize_from_config_async\n",
+    "\n",
+    "await initialize_from_config_async(config_path=Path(\"../../scanner/pyrit_conf.yaml\"))  # type: ignore\n",
+    "\n",
+    "objective_target = TargetRegistry.get_registry_singleton().get_instance_by_name(\"openai_chat\")\n",
+    "printer = ConsoleScenarioResultPrinter()"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "3",
+   "metadata": {},
+   "source": [
+    "## Basic usage\n",
+    "\n",
+    "Defaults: `epsilon=0.2`, `max_attempts_per_objective=3`, the subclass's default datasets."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "4",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "scenario = TextAdaptive()\n",
+    "\n",
+    "await scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    ")\n",
+    "result = await scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "5",
+   "metadata": {},
+   "source": [
+    "## Tuning exploration (`epsilon`)\n",
+    "\n",
+    "- `epsilon=0.0` \u2014 pure exploitation (always pick the best-known technique).\n",
+    "- `epsilon=1.0` \u2014 pure exploration (random every time).\n",
+    "- `epsilon=0.2` (default) \u2014 20% exploration."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "6",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "explorative_scenario = TextAdaptive(epsilon=0.5)\n",
+    "\n",
+    "await explorative_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    dataset_config=DatasetConfiguration(dataset_names=[\"airt_hate\"], max_dataset_size=4),\n",
+    ")\n",
+    "explorative_result = await explorative_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(explorative_result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "7",
+   "metadata": {},
+   "source": [
+    "## Attempts per objective\n",
+    "\n",
+    "`max_attempts_per_objective` caps how many techniques are tried per objective before\n",
+    "moving on. Higher = more chances to succeed, more API calls."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "8",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "persistent_scenario = TextAdaptive(max_attempts_per_objective=5)\n",
+    "\n",
+    "await persistent_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    dataset_config=DatasetConfiguration(dataset_names=[\"airt_violence\"], max_dataset_size=4),\n",
+    ")\n",
+    "persistent_result = await persistent_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(persistent_result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "9",
+   "metadata": {},
+   "source": [
+    "## Learning per harm category\n",
+    "\n",
+    "By default, the scenario keeps one global success-rate table \u2014 what works on hate\n",
+    "objectives boosts the same technique on violence objectives. Pass `harm_category_context`\n",
+    "to learn each category independently:"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "10",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "contextual_scenario = TextAdaptive(context_extractor=harm_category_context)\n",
+    "\n",
+    "await contextual_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    dataset_config=DatasetConfiguration(\n",
+    "        dataset_names=[\"airt_hate\", \"airt_violence\"],\n",
+    "        max_dataset_size=4,\n",
+    "    ),\n",
+    ")\n",
+    "contextual_result = await contextual_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(contextual_result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "11",
+   "metadata": {},
+   "source": [
+    "## Restricting which techniques participate\n",
+    "\n",
+    "Use `scenario_strategies` to limit which techniques the scenario can pick from."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "12",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "strategy_class = TextAdaptive.get_strategy_class()\n",
+    "\n",
+    "single_turn_scenario = TextAdaptive()\n",
+    "\n",
+    "await single_turn_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    scenario_strategies=[strategy_class(\"single_turn\")],\n",
+    "    dataset_config=DatasetConfiguration(dataset_names=[\"airt_hate\"], max_dataset_size=4),\n",
+    ")\n",
+    "single_turn_result = await single_turn_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(single_turn_result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "13",
+   "metadata": {},
+   "source": [
+    "## Reproducible runs\n",
+    "\n",
+    "Pass `seed` to make every selection decision deterministic."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "14",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "deterministic_scenario = TextAdaptive(seed=42, epsilon=0.3)\n",
+    "\n",
+    "await deterministic_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    dataset_config=DatasetConfiguration(dataset_names=[\"airt_hate\"], max_dataset_size=2),\n",
+    ")\n",
+    "deterministic_result = await deterministic_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(deterministic_result)  # type: ignore"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "id": "15",
+   "metadata": {},
+   "source": [
+    "## Resuming a run\n",
+    "\n",
+    "Adaptive scenarios are resumable \u2014 pass `scenario_result_id=...` to the `TextAdaptive`\n",
+    "constructor and the run picks up where it left off, with prior outcomes replayed into\n",
+    "the selector. Here we resume the deterministic run from the previous cell."
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "id": "14",
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "resumed_scenario = TextAdaptive(\n",
+    "    seed=42,\n",
+    "    epsilon=0.3,\n",
+    "    scenario_result_id=str(deterministic_result.id),\n",
+    ")\n",
+    "\n",
+    "await resumed_scenario.initialize_async(  # type: ignore\n",
+    "    objective_target=objective_target,\n",
+    "    dataset_config=DatasetConfiguration(dataset_names=[\"airt_hate\"], max_dataset_size=2),\n",
+    ")\n",
+    "resumed_result = await resumed_scenario.run_async()  # type: ignore\n",
+    "await printer.write_async(resumed_result)  # type: ignore"
+   ]
+  }
+ ],
+ "metadata": {
+  "jupytext": {
+   "main_language": "python"
+  }
+ },
+ "nbformat": 4,
+ "nbformat_minor": 5
+}