Create app.py

app.py

@@ -0,0 +1,390 @@
+import os
+import openai
+import datetime
+import gradio as gr
+import json
+from jinja2 import Template
+import requests
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# Initialize OpenAI
+openai.api_key = os.environ.get('OPENAI_API_KEY')
+
+# Configuration variables
+airtable_api_key = os.environ.get('AIRTABLE_API_KEY')
+
+# Airtable table names
+policies_table_name = 'tbla6PC65qZfqdJhE'
+prompts_table_name = 'tblYIZEB8m6JkGDEP'
+qalog_table_name = 'tbl4oNgFPWM5xH1XO'
+examples_table_name = 'tblu7sraOEmRgEGkp'
+users_table_name = 'tblLNe5ZL47SvrAEk'
+user_log_table_name = 'tblrlTsRrkl6BqMAJ'
+
+# Define the style and content for the response field
+label_text = "NILI Response"
+color = "#6562F4"
+background_color = "white"
+border_radius = "10px"
+response_label = f'<h3 style="color: {color}; background-color: {background_color}; border-radius: {border_radius}; padding: 10px;display: inline-block;">{label_text}</h3>'
+
+#Airtable Base ID
+base_id = 'appcUK3hUWC7GM2Kb'
+
+#Name of the prompt temlate record
+prompt_name = "NILI_v1"
+
+#Header for the Airtable requests
+headers = {
+    "Authorization": f"Bearer {airtable_api_key}",
+    "Content-Type": "application/json",
+    "Accept": "application/json",
+}
+
+#Function to trim prompts....not used
+def prompt_trim(prompt: str) -> str:
+    lines = prompt.split('\n')
+    trimmed = '\n'.join([l.strip() for l in lines])
+    return trimmed
+
+#Get the policies for the selected schools and concatenate them.
+def get_policies(school_selection):
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{policies_table_name}'    
+
+    # Parameters for the API request to filter by 'school' field and retrieve 'policy_text'
+    params = {
+        'filterByFormula': "OR({})".format(','.join(["school='{}'".format(school) for school in school_selection])),
+        'fields[]': 'policy_text',  # Replace with the name of your field
+    }
+
+    # Initialize an empty string to store concatenated policies
+    concatenated_policies = ''
+
+    #print(params)
+
+    try:
+        # Send a GET request to the Airtable API
+        response = requests.get(airtable_endpoint, headers=headers, params=params)
+
+        # Check if the request was successful (status code 200)
+        if response.status_code == 200:
+            # Parse the JSON response
+            data = response.json()
+
+            # Check if there are records in the response
+            if data.get('records'):
+                
+                # Extract the 'policy_text' values from each record and concatenate them
+                for record in data['records']:
+                    policy_text = record['fields']['policy_text']
+                    if concatenated_policies:
+                        concatenated_policies += "\n----------\n"
+                    concatenated_policies += policy_text
+
+            else:
+                print("No records found in the 'policies' table for the selected schools.")
+        else:
+            print(f"Failed to retrieve data. Status code: {response.status_code}")
+    except Exception as e:
+        print(f"An error occurred: {str(e)}")
+
+    #print(concatenated_policies)   
+
+    return concatenated_policies
+
+#Get a list of School Name from the policies for the UI dropdown
+def get_schools():
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{policies_table_name}'    
+
+    # Parameters for the API request to select only the 'school' field
+    params = {
+        'fields[]': 'school',  # Replace with the name of your field
+        'sort[0][field]': 'school',  # Sort by the 'school' field
+        'sort[0][direction]': 'asc',  # Sort in ascending order
+    }
+
+    schools = ''
+
+    try:
+        # Send a GET request to the Airtable API
+        response = requests.get(airtable_endpoint, headers=headers, params=params)
+
+        # Check if the request was successful (status code 200)
+        if response.status_code == 200:
+            # Parse the JSON response
+            data = response.json()
+
+            # Check if there are records in the response
+            if data.get('records'):
+                # Extract the 'school' values from each record
+                schools = [record['fields']['school'] for record in data['records']]
+
+            else:
+                print("No records found in the 'policies' table.")
+        else:
+            print(f"Failed to retrieve data. Status code: {response.status_code}")
+    except Exception as e:
+        print(f"An error occurred: {str(e)}")
+
+    return schools
+
+#Get the designated prompt template record
+def get_prompt(header, template_content):
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{prompts_table_name}'    
+
+    params = {
+        'filterByFormula': "prompt_name='NILI_v1'"
+    }
+
+    response = requests.get(airtable_endpoint, headers=headers, params=params)
+
+    # Check for errors
+    response.raise_for_status()
+
+    data = response.json()
+
+    # Check if there is at least one record matching the condition
+    if data.get('records'):
+        # Get the first record (there should be only one)
+        record = data['records'][0]['fields']
+            
+        # Assign system_prompt and user_prompt to variables
+        header = record.get('system_prompt', '')
+        template_content = record.get('user_prompt', '')
+
+    return header, template_content
+
+
+def get_examples():
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{examples_table_name}'    
+
+    # Send your request and parse the response
+    response = requests.get(airtable_endpoint, headers=headers)
+    data = json.loads(response.text)
+
+    # Check for errors
+    response.raise_for_status()
+
+    for record in data['records']:
+        nil_question = record['fields']['nil_question']
+        ui_examples.append([None, None, None, nil_question])
+
+    #print(ui_examples)
+           
+
+def append_to_at_qalog(your_role, school_selection, output_format, input_text, gpt_response,response_time,question_cost,prompt_tokens,completion_tokens):
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{qalog_table_name}'
+    
+    # Organize data for Airtable
+    new_fields = {
+            'your_role': str(your_role),
+            'school_selection': str(school_selection),
+            'output_format': str(output_format),
+            'input_text': str(input_text),
+            'gpt_response': str(gpt_response),
+            'response_time': str(response_time),
+            'question_cost': question_cost,
+            'user_name': str(logged_in_user),
+            'prompt_tokens': prompt_tokens,
+            'completion_tokens': completion_tokens
+        }
+
+    data = {
+        'fields': new_fields
+        }
+    try:
+        # Post data to Airtable
+        response = requests.post(airtable_endpoint, headers=headers, json=data)
+
+        # Check for errors
+        response.raise_for_status()
+
+    except requests.exceptions.HTTPError as http_error:
+        # Handle the HTTP error (e.g., log it or display an error message)
+        print(f"HTTP error occurred: {http_error}")
+
+    except Exception as e:
+        # Handle exceptions, log errors, or raise them as needed
+        print(f"An error occurred: {str(e)}")
+
+
+#Chatbot Function
+def chatbot(your_role,school_selection,output_format,input_text):
+
+    start_time = datetime.datetime.now()
+
+    # school_selection holds an array of one or more schools
+    #print(school_selection)
+
+    # Read the Hydrated policies
+  
+    policies = get_policies(school_selection)
+
+    template_content = ''
+    header = ''
+
+    header, template_content = get_prompt(header, template_content)
+
+    #print(header)
+    #print(template_content)
+
+    header_template = Template(header)
+    merged_header = header_template.render(your_role=your_role)
+
+    # Create a Jinja2 template from the content
+    template = Template(template_content)
+
+    # Render the template with the policy JSON
+    analysis_input = template.render(policies=policies, question=input_text,format=output_format,your_role=your_role)
+
+    trimmed_input = prompt_trim(analysis_input)
+
+    with open('analysis_input.txt', 'w', encoding='utf-8') as out_file:
+        out_file.write(trimmed_input)
+  
+    response = openai.ChatCompletion.create(
+        model="gpt-4",
+        #model="gpt-3.5-turbo",
+        temperature=0,
+        messages=[
+            {
+                "role": "system",
+                "content": merged_header
+            },
+            {
+                "role": "user",
+                "content": analysis_input
+            }
+          ]
+        )
+
+    gpt_response = response.choices[0].message["content"]
+
+    tokens_used = response.usage
+
+    question_cost = (tokens_used.get('total_tokens', 0) / 1000) * .03
+    prompt_tokens = tokens_used.get('prompt_tokens',)
+    completion_tokens = tokens_used.get('completion_tokens', 0)
+
+    """
+    with open('response.txt', 'w', encoding='utf-8') as out_file:
+        out_file.write(gpt_response)
+    """
+    end_time = datetime.datetime.now()
+    response_time = end_time - start_time
+
+    append_to_at_qalog(your_role, school_selection, output_format, input_text, gpt_response,response_time,question_cost,prompt_tokens,completion_tokens)
+
+    return response_label,gpt_response
+
+def log_login(username):
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{user_log_table_name}'
+
+    # Organize data for Airtable
+    new_fields = {
+            'user_name': str(username),
+        }
+
+    data = {
+        'fields': new_fields
+        }
+
+    try:
+    # Post data to Airtable
+        response = requests.post(airtable_endpoint, headers=headers, json=data)
+
+        # Check for errors
+        response.raise_for_status()
+
+    except requests.exceptions.HTTPError as http_error:
+        # Handle the HTTP error (e.g., log it or display an error message)
+        print(f"HTTP error occurred: {http_error}")
+
+    except Exception as e:
+        # Handle exceptions, log errors, or raise them as needed
+        print(f"An error occurred: {str(e)}")
+
+
+def login_auth(username, password):
+
+    airtable_endpoint = f'https://api.airtable.com/v0/{base_id}/{users_table_name}'
+    
+    # Query the 'users' table to check for a match with the provided username and password
+    params = {
+        'filterByFormula': f'AND(user_name = "{username}", password = "{password}")'
+    }
+
+    response = requests.get(airtable_endpoint, headers=headers, params=params)
+
+    if response.status_code == 200:
+        data = response.json()
+        #If the matching user/password record is found:
+        if data.get('records'):
+
+            #Log that the user logged in
+            log_login(username)
+
+            #Set the global logged_in_user variable. This used in the append_to_at_qalog function to track what user asked the question
+            global logged_in_user
+            logged_in_user = username
+
+            return True
+
+        print(f"Invalid user/password combination")
+
+    return False
+
+#Gradio UI
+CIMStheme = gr.themes.Soft().set(button_primary_background_fill='#6562F4')
+
+# Initialize an empty list to store the examples
+ui_examples = []
+school_selection = []
+
+schools = get_schools()
+
+get_examples()
+
+logged_in_user = 'admin'
+
+with gr.Blocks(CIMStheme) as iface:
+    with gr.Row():
+        with gr.Column(scale=2):
+            gr.Image(label="Logo",value="CIMS Logo Purple.png",width=10,show_download_button=False,interactive=False,show_label=False,elem_id="logo",container=False)
+        with gr.Column(scale=2):
+            #gr.Textbox(value="# NILI - Powered by CIMS.AI",show_label=False,interactive=False,text_align="center",elem_id="CIMSTitle")
+            gr.Markdown(value="# NILI - Powered by CIMS.AI")
+        with gr.Column(scale=2):
+            gr.Markdown("")
+    with gr.Row():
+        with gr.Column():
+            gr.Interface(fn=chatbot,
+                     inputs=[
+                         gr.components.Dropdown(["Student Athlete","Parent","Athletic Director"],multiselect=False,info="Select a role.",label="User Role", ),
+                         gr.components.Dropdown(schools,multiselect=True,info="Select one or more schools. This will help set the context of your question.",label="School Context"),
+                         gr.components.Dropdown(["Summary","Detailed Analysis","Table"],multiselect=False,info="Select the desired output format.",label="Output Format"),
+                         gr.components.Textbox(lines=5, placeholder="Enter your question here", label="NIL Question")],
+                     outputs=[
+                         gr.components.Markdown(response_label),
+                         gr.components.HTML(label="NILI Response")
+                         ],
+                     description="Ask any question about Name, Image, Likeness (NIL)",
+                     allow_flagging="manual",
+                     examples=ui_examples,
+                     cache_examples=False,
+                     flagging_options=["The response is incorrect","The response is inappropriate","The response doesn't make sense"]
+                      )
+    with gr.Row():
+        with gr.Column():
+            gr.HTML('<center><i>CIMS.AI Confidential 2023</i></center>')
+
+iface.launch(auth=login_auth, auth_message= "Enter your username and password that you received from CIMS.AI. To request a login, please email 'info@cims.ai'")
+#iface.launch(share=True)