orc/plugins/ResponsibleAI/wrapper.py (18 lines of code) (raw):
# imports
import json
import logging
from shared.util import call_semantic_function, get_usage_tokens
async def fairness(kernel, rai_plugin, arguments):
"""
This function is used to evaluate the fairness of a given context.
It calls a semantic function that returns a response indicating whether the context is fair or not.
The response is then parsed and returned in a dictionary format.
Returns:
dict: A dictionary containing the fairness evaluation response. The response includes the fairness status, answer, prompt tokens, and completion tokens.
'fair' (bool): A flag indicating whether the context is fair or not. Defaults to True if not found.
'answer' (str): The answer to the request. Defaults to an empty string if not found.
'prompt_tokens' (str): The prompt tokens for the request. Retrieved using the get_usage_tokens function.
'completion_tokens' (str): The completion tokens for the request. Retrieved using the get_usage_tokens function.
'bypass' (bool): A flag indicating whether to bypass the the reminder flow steps (in case of an error has occurred).
"""
fairness_dict= {"fair": True, "answer": "", "bypass": False}
function_result = await call_semantic_function(kernel, rai_plugin["Fairness"], arguments)
message_content = str(function_result)
try:
response = message_content.strip("`json\n`")
response_json = json.loads(response)
except json.JSONDecodeError:
logging.error(f"[code_orchest] error when executing RAG flow (Fairness). Invalid json: {function_result}")
raise Exception(f"Triage was not successful due to a JSON error. Invalid json: {function_result}")
fairness_dict["fair"] = response_json.get('fair', True)
fairness_dict["answer"] = response_json.get('new_answer', '')
fairness_dict["prompt_tokens"] = get_usage_tokens(function_result, 'prompt')
fairness_dict["completion_tokens"] = get_usage_tokens(function_result, 'completion')
return fairness_dict