Content Safety Evaluation Notebook

In [ ]:

# Copyright (c) Microsoft. All rights reserved.

# Copyright (c) Microsoft. All rights reserved.

In [ ]:

import time
import json
from pathlib import Path
import os
from azure.ai.evaluation.simulator import AdversarialSimulator
from dotenv import load_dotenv
load_dotenv()

# Define folder paths
output_folder = "output"
Path(output_folder).mkdir(parents=True, exist_ok=True)  # Ensure output folder exists

count = 5

import time import json from pathlib import Path import os from azure.ai.evaluation.simulator import AdversarialSimulator from dotenv import load_dotenv load_dotenv() # Define folder paths output_folder = "output" Path(output_folder).mkdir(parents=True, exist_ok=True) # Ensure output folder exists count = 5

In [ ]:

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")
}

# your azure api endpoint
api_url = "<your_api_endpoint>/api/chat"

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") } # your azure api endpoint api_url = "/api/chat"

In [ ]:

from pathlib import Path
import requests

def call_streaming_url(url, data):
    full_response = ""
    try:
        response = requests.post(url, json=data, stream=True)    
    except:
        time.sleep(20)
        response = requests.post(url,json=data, stream=True)
    
    for chunk in response.iter_content(chunk_size=8192):
        if chunk:
            full_response = chunk.decode('utf-8')  # Concatenate each chunk to the full response
    return full_response

from pathlib import Path import requests def call_streaming_url(url, data): full_response = "" try: response = requests.post(url, json=data, stream=True) except: time.sleep(20) response = requests.post(url,json=data, stream=True) for chunk in response.iter_content(chunk_size=8192): if chunk: full_response = chunk.decode('utf-8') # Concatenate each chunk to the full response return full_response

In [ ]:

from typing import List, Dict, Any, Optional
async def callback(
    messages: List[Dict],
    stream: bool = False,
    session_state: Any = None,
) -> dict:
    query = messages["messages"][0]["content"]
    context = None

    # Add file contents for summarization or re-write
    # if 'file_content' in messages["template_parameters"]:
    #     query += messages["template_parameters"]['file_content']
    m1 = {"messages": [{'content':query}]}
    # Call your own endpoint and pass your query as input. Make sure to handle your function_call_to_your_endpoint's error responses.
    
    response = call_streaming_url(api_url, m1) 
   
    # Format responses in OpenAI message protocol
    try:
        r = json.loads(response).get("choices")[0].get("messages")[0]
    except:
        r = response 
    
    formatted_response = {
        "content": r,
        "role": "assistant",
        "context": {},
    }

    messages["messages"].append(formatted_response)

    return {
        "messages": messages["messages"],
        "stream": stream,
        "session_state": session_state
    }

from typing import List, Dict, Any, Optional async def callback( messages: List[Dict], stream: bool = False, session_state: Any = None, ) -> dict: query = messages["messages"][0]["content"] context = None # Add file contents for summarization or re-write # if 'file_content' in messages["template_parameters"]: # query += messages["template_parameters"]['file_content'] m1 = {"messages": [{'content':query}]} # Call your own endpoint and pass your query as input. Make sure to handle your function_call_to_your_endpoint's error responses. response = call_streaming_url(api_url, m1) # Format responses in OpenAI message protocol try: r = json.loads(response).get("choices")[0].get("messages")[0] except: r = response formatted_response = { "content": r, "role": "assistant", "context": {}, } messages["messages"].append(formatted_response) return { "messages": messages["messages"], "stream": stream, "session_state": session_state }

In [ ]:

from azure.ai.evaluation.simulator import AdversarialScenario
from azure.identity import DefaultAzureCredential
credential = DefaultAzureCredential()

scenario = AdversarialScenario.ADVERSARIAL_QA
adversarial_simulator = AdversarialSimulator(azure_ai_project=azure_ai_project, credential=credential)

outputs = await adversarial_simulator(
        scenario=scenario, # required adversarial scenario to simulate
        target=callback, # callback function to simulate against
        max_conversation_turns=1, #optional, applicable only to conversation scenario
        max_simulation_results=count, #optional
    )

output_file_adversarial = Path(output_folder) / f"content_safety_output.jsonl"
with output_file_adversarial.open("w") as f:
    f.write(outputs.to_eval_qr_json_lines())

from azure.ai.evaluation.simulator import AdversarialScenario from azure.identity import DefaultAzureCredential credential = DefaultAzureCredential() scenario = AdversarialScenario.ADVERSARIAL_QA adversarial_simulator = AdversarialSimulator(azure_ai_project=azure_ai_project, credential=credential) outputs = await adversarial_simulator( scenario=scenario, # required adversarial scenario to simulate target=callback, # callback function to simulate against max_conversation_turns=1, #optional, applicable only to conversation scenario max_simulation_results=count, #optional ) output_file_adversarial = Path(output_folder) / f"content_safety_output.jsonl" with output_file_adversarial.open("w") as f: f.write(outputs.to_eval_qr_json_lines())

In [ ]:

from pathlib import Path
from azure.ai.evaluation import ContentSafetyEvaluator
import pandas as pd

credential = DefaultAzureCredential()
# instantiate an evaluator with image and multi-modal support
safety_evaluator = ContentSafetyEvaluator(credential=credential, azure_ai_project=azure_ai_project)

df = pd.read_json(output_file_adversarial, lines=True)
json_string = []
for index, row in df.iterrows(): 
    safety_score = safety_evaluator(query=row['query'], response=row['response'])
    json_object = {
        "query": row['query'],
        "response": row['response'],
        "violance": safety_score['violence'],
        "violence_score": safety_score['violence_score'],
        "violence_reason": safety_score['violence_reason'],
        "violence_threshold": safety_score['violence_threshold'],
        "violence_result": safety_score['violence_result'],
        "sexual": safety_score['sexual'],
        "sexual_score": safety_score['sexual_score'],
        "sexual_reason": safety_score['sexual_reason'],
        "sexual_threshold": safety_score['sexual_threshold'],
        "sexual_result": safety_score['sexual_result'],
        "hate_unfairness": safety_score['hate_unfairness'],
        "hate_unfairness_score": safety_score['hate_unfairness_score'],
        "hate_unfairness_reason": safety_score['hate_unfairness_reason'],
        "hate_unfairness_threshold": safety_score['hate_unfairness_threshold'],
        "hate_unfairness_result": safety_score['hate_unfairness_result'],
        "self_harm": safety_score['self_harm'],
        "self_harm_score": safety_score['self_harm_score'],
        "self_harm_reason": safety_score['self_harm_reason'],
        "self_harm_threshold": safety_score['self_harm_threshold'],
        "self_harm_result": safety_score['self_harm_result']
    }
    json_string.append(json_object)
file_eval = Path(output_folder) / f"content_safety_output_scores.jsonl"
with Path(file_eval).open("w") as f:
  json.dump(json_string, f, indent=4)

from pathlib import Path from azure.ai.evaluation import ContentSafetyEvaluator import pandas as pd credential = DefaultAzureCredential() # instantiate an evaluator with image and multi-modal support safety_evaluator = ContentSafetyEvaluator(credential=credential, azure_ai_project=azure_ai_project) df = pd.read_json(output_file_adversarial, lines=True) json_string = [] for index, row in df.iterrows(): safety_score = safety_evaluator(query=row['query'], response=row['response']) json_object = { "query": row['query'], "response": row['response'], "violance": safety_score['violence'], "violence_score": safety_score['violence_score'], "violence_reason": safety_score['violence_reason'], "violence_threshold": safety_score['violence_threshold'], "violence_result": safety_score['violence_result'], "sexual": safety_score['sexual'], "sexual_score": safety_score['sexual_score'], "sexual_reason": safety_score['sexual_reason'], "sexual_threshold": safety_score['sexual_threshold'], "sexual_result": safety_score['sexual_result'], "hate_unfairness": safety_score['hate_unfairness'], "hate_unfairness_score": safety_score['hate_unfairness_score'], "hate_unfairness_reason": safety_score['hate_unfairness_reason'], "hate_unfairness_threshold": safety_score['hate_unfairness_threshold'], "hate_unfairness_result": safety_score['hate_unfairness_result'], "self_harm": safety_score['self_harm'], "self_harm_score": safety_score['self_harm_score'], "self_harm_reason": safety_score['self_harm_reason'], "self_harm_threshold": safety_score['self_harm_threshold'], "self_harm_result": safety_score['self_harm_result'] } json_string.append(json_object) file_eval = Path(output_folder) / f"content_safety_output_scores.jsonl" with Path(file_eval).open("w") as f: json.dump(json_string, f, indent=4)

The following is an example of the Content Safety Evaluations. If you are not able to complete the evaluations at this time, please see an example here

Grounded Evaluations¶

In [ ]:

model_config = {
    "azure_endpoint": os.environ.get("AZURE_OPEN_AI_ENDPOINT"),
    "api_key": os.environ.get("AZURE_OPENAI_API_KEY"),
    "azure_deployment": os.environ.get("AZURE_OPEN_AI_DEPLOYMENT_MODEL"),
    "api_version": os.environ.get("AZURE_OPENAI_API_VERSION"),
}

model_config = { "azure_endpoint": os.environ.get("AZURE_OPEN_AI_ENDPOINT"), "api_key": os.environ.get("AZURE_OPENAI_API_KEY"), "azure_deployment": os.environ.get("AZURE_OPEN_AI_DEPLOYMENT_MODEL"), "api_version": os.environ.get("AZURE_OPENAI_API_VERSION"), }

In [ ]:

from azure.ai.evaluation import GroundednessEvaluator

groundedness_eval = GroundednessEvaluator(model_config)

query_response = dict(
    query="What is the top challenge users reported?",
    context="",
    response="Network Performance Issues: Concerns about poor network performance and service disruptions."
)
groundedness_score = groundedness_eval(
    **query_response
)
print(groundedness_score)

from azure.ai.evaluation import GroundednessEvaluator groundedness_eval = GroundednessEvaluator(model_config) query_response = dict( query="What is the top challenge users reported?", context="", response="Network Performance Issues: Concerns about poor network performance and service disruptions." ) groundedness_score = groundedness_eval( **query_response ) print(groundedness_score)

Direct Attack Evaluations¶

In [ ]:

# from azure.ai.evaluation.simulator import DirectAttackSimulator

# output_filename = f"direct_output.jsonl"
# scenario = AdversarialScenario.ADVERSARIAL_CONVERSATION

# adversarial_simulator = DirectAttackSimulator(azure_ai_project=azure_ai_project, credential=credential)

# outputs = await adversarial_simulator(
#   target=callback,
#   scenario=scenario,
#   max_conversation_turns=1,
#   max_simulation_results=count,
# )

# output_file_adversarial = Path(output_folder) / output_filename
# with output_file_adversarial.open("w") as f:
#   f.write(json.dumps(outputs, indent=4))

# from azure.ai.evaluation.simulator import DirectAttackSimulator # output_filename = f"direct_output.jsonl" # scenario = AdversarialScenario.ADVERSARIAL_CONVERSATION # adversarial_simulator = DirectAttackSimulator(azure_ai_project=azure_ai_project, credential=credential) # outputs = await adversarial_simulator( # target=callback, # scenario=scenario, # max_conversation_turns=1, # max_simulation_results=count, # ) # output_file_adversarial = Path(output_folder) / output_filename # with output_file_adversarial.open("w") as f: # f.write(json.dumps(outputs, indent=4))

Indirect Attack Evaluations¶

In [ ]:

# from azure.ai.evaluation.simulator import IndirectAttackSimulator

# output_filename = f"indirect_output.jsonl"
# scenario = AdversarialScenario.ADVERSARIAL_CONVERSATION

# adversarial_simulator = IndirectAttackSimulator(azure_ai_project=azure_ai_project, credential=credential)

# outputs = await adversarial_simulator(
#   target=callback,
#   scenario=scenario,
#   max_conversation_turns=1,
#   max_simulation_results=count,
# )

# output_file_adversarial = Path(output_folder) / output_filename
# with output_file_adversarial.open("w") as f:
#   f.write(json.dumps(outputs, indent=4))

# from azure.ai.evaluation.simulator import IndirectAttackSimulator # output_filename = f"indirect_output.jsonl" # scenario = AdversarialScenario.ADVERSARIAL_CONVERSATION # adversarial_simulator = IndirectAttackSimulator(azure_ai_project=azure_ai_project, credential=credential) # outputs = await adversarial_simulator( # target=callback, # scenario=scenario, # max_conversation_turns=1, # max_simulation_results=count, # ) # output_file_adversarial = Path(output_folder) / output_filename # with output_file_adversarial.open("w") as f: # f.write(json.dumps(outputs, indent=4))