Share via


DirectAttackSimulator Class

Note

This is an experimental class, and may change at any time. Please see https://aka.ms/azuremlexperimental for more information.

Initialize a UPIA (user prompt injected attack) jailbreak adversarial simulator with a project scope. This simulator converses with your AI system using prompts designed to interrupt normal functionality.

Constructor.

Constructor

DirectAttackSimulator(*, 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 DirectAttackSimulator to produce 2 results with 3 conversation turns each (6 messages in each result).


   import os
   import asyncio
   from azure.ai.evaluation.simulator import AdversarialScenario, DirectAttackSimulator
   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 = DirectAttackSimulator(azure_ai_project=azure_ai_project, credential=DefaultAzureCredential())

   outputs = asyncio.run(
       simulator(
           scenario=AdversarialScenario.ADVERSARIAL_REWRITE,
           max_conversation_turns=3,
           max_simulation_results=2,
           target=callback,
       )
   )