AdversarialSimulator Class
Note
This is an experimental class, and may change at any time. Please see https://aka.ms/azuremlexperimental for more information.
Initializes the adversarial simulator with a project scope.
Constructor.
Constructor
AdversarialSimulator(*, azure_ai_project: str | AzureAIProject, credential: TokenCredential)
Parameters
Name | Description |
---|---|
azure_ai_project
Required
|
The Azure AI project, which can either be a string representing the project endpoint or an instance of AzureAIProject. It contains subscription id, resource group, and project name. |
credential
Required
|
The credential for connecting to Azure AI project. |
Keyword-Only Parameters
Name | Description |
---|---|
azure_ai_project
Required
|
|
credential
Required
|
|
Examples
Run the AdversarialSimulator with an AdversarialConversation scenario to produce 2 results with 2 conversation turns each (4 messages per result).
import os
import asyncio
from typing import List, Dict, Any, Optional
from azure.ai.evaluation.simulator import AdversarialScenario, AdversarialSimulator
from azure.identity import DefaultAzureCredential
azure_ai_project = {
"subscription_id": os.environ.get("AZURE_SUBSCRIPTION_ID"),
"resource_group_name": os.environ.get("AZURE_RESOURCE_GROUP_NAME"),
"project_name": os.environ.get("AZURE_PROJECT_NAME"),
}
async def callback(
messages: List[Dict],
stream: bool = False,
session_state: Any = None,
context: Optional[Dict[str, Any]] = None,
) -> dict:
query = messages["messages"][0]["content"]
formatted_response = {"content": query, "role": "assistant"}
messages["messages"].append(formatted_response)
return {
"messages": messages["messages"],
"stream": stream,
"session_state": session_state,
"context": context,
}
simulator = AdversarialSimulator(azure_ai_project=azure_ai_project, credential=DefaultAzureCredential())
outputs = asyncio.run(
simulator(
scenario=AdversarialScenario.ADVERSARIAL_CONVERSATION,
max_conversation_turns=2,
max_simulation_results=2,
target=callback,
api_call_retry_limit=3,
api_call_retry_sleep_sec=1,
api_call_delay_sec=30,
concurrent_async_task=1,
randomization_seed=1,
)
)
Methods
call_sync |
Call the adversarial simulator synchronously. :keyword scenario: Enum value specifying the adversarial scenario used for generating inputs. .. admonition:: Example
|
call_sync
Call the adversarial simulator synchronously. :keyword scenario: Enum value specifying the adversarial scenario used for generating inputs. .. admonition:: Example
<xref:azure.ai.evaluation.simulator.adversarial_scenario.AdversarialScenario.ADVERSARIAL_QA>
<xref:azure.ai.evaluation.simulator.adversarial_scenario.AdversarialScenario.ADVERSARIAL_CONVERSATION>
call_sync(*, scenario: AdversarialScenario, max_conversation_turns: int, max_simulation_results: int, target: Callable, api_call_retry_limit: int, api_call_retry_sleep_sec: int, api_call_delay_sec: int, concurrent_async_task: int) -> List[Dict[str, Any]]
Keyword-Only Parameters
Name | Description |
---|---|
max_conversation_turns
|
The maximum number of conversation turns to simulate. |
max_simulation_results
|
The maximum number of simulation results to return. |
target
|
The target function to simulate adversarial inputs against. |
api_call_retry_limit
|
The maximum number of retries for each API call within the simulation. |
api_call_retry_sleep_sec
|
The sleep duration (in seconds) between retries for API calls. |
api_call_delay_sec
|
The delay (in seconds) before making an API call. |
concurrent_async_task
|
The number of asynchronous tasks to run concurrently during the simulation. |
scenario
Required
|
|
Returns
Type | Description |
---|---|
A list of dictionaries, each representing a simulated conversation. |