app.py

import os
import requests
import gradio as gr

from pydantic.v1 import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser
from langchain.agents import AgentExecutor, create_openai_functions_agent
from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.messages import AIMessage, HumanMessage
from langchain.tools import StructuredTool
from datetime import datetime, timedelta

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

def repo_get_all_employees_from_database():
    url = "https://api.airtable.com/v0/appopGmlHujYnd6Vw/Interviewers?maxRecords=100&view=Grid%20view"
    headers = {
        "Authorization": os.getenv("DB_AUTH_TOKEN")
    }
    response = requests.get(url, headers=headers)
    records = response.json()
    records_list = records['records']
    employees_list = []
    for record in records_list:
        employee = record["fields"]
        employees_list.append(employee)
    
    return employees_list

def get_all_interviewers() -> str:
    """
    A function to get a list of all interviewers from database. 
    Returns: 
        str: A list of all employees in json.
    """
    return repo_get_all_employees_from_database()

def get_interviewer_teams(
        team_size: int,
        division: str,
        interview_level: str) -> str:
    """
    A function to get all possible teams of interviewers.
    Args:
        team_size: int - The number of interviewers in each team.
        division: str - A division interviewers belong to.
        interview_level: str - A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.

    Returns: 
        str: A list of interviewers teams in json.
    """
    return get_interviewer_teams_by_skills(
        team_size=team_size, 
        division=division, 
        interview_level=interview_level,
        skills=[])

def get_interviewer_teams_by_skills(
        team_size: int,
        division: str,
        interview_level: str,
        skills: list) -> str:
    """
    A function to get all possible teams of interviewers by their skills.
    Args:
        team_size: int - The number of interviewers in each team.
        division: str - A division interviewers belong to.
        interview_level: str - A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.
        skills: list - Skills that an interviewers must have to be able to conduct an interview.

    Returns: 
        str: A list of interviewers teams in json.
    """
    return get_interviewer_teams_by_skills_in_time_slot(
        team_size=team_size, 
        division=division, 
        interview_level=interview_level, 
        start_date_time=None,
        duration_hours=0, 
        skills=skills)

def get_interviewer_teams_in_time_slot(
        team_size: int,
        division: str,
        interview_level: str,
        start_date_time: str, 
        duration_hours: int) -> str:
    """
    A function to get all possible teams of interviewers in requested time slot.
    Args:
        team_size: int - The number of interviewers in each team.
        division: str - A division interviewers belong to.
        interview_level: str - A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.
        start_date_time: str - Date and time of start of interview.
        duration_hours: int - Required duration of the availability in hours.

    Returns: 
        str: A list of interviewers teams in json.
    """
    return get_interviewer_teams_by_skills_in_time_slot(
        team_size=team_size, 
        division=division, 
        interview_level=interview_level, 
        start_date_time=start_date_time,
        duration_hours=duration_hours, 
        skills=[])

def get_interviewer_teams_by_skills_in_time_slot(
        team_size: int,
        division: str,
        interview_level: str,
        start_date_time: str, 
        duration_hours: int, 
        skills: list) -> str:
    """
    A function to get all possible teams of interviewers by their skills in requested time slot.
    Args:
        team_size: int - The number of interviewers in each team.
        division: str - A division interviewers belong to.
        interview_level: str - A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.
        start_date_time: str - Date and time of start of interview.
        duration_hours: int - Required duration of the availability in hours.
        skills: list - Skills that an interviewers must have to be able to conduct an interview.

    Returns: 
        str: A list of interviewers teams in json.
    """
    
    params = "?teamSize={team_size}&division={division}&interviewLevel={interview_level}".format(
        team_size=team_size, 
        division=division, 
        interview_level=interview_level)

    if start_date_time is not None:
        end_date_time = start_date_time
        start_date_time_obj = datetime.fromisoformat(start_date_time)
        end_date_time_obj = start_date_time_obj + timedelta(hours=duration_hours)
        end_date_time = end_date_time_obj.isoformat()
        optional_params = "&startDateTime={start}&endDateTime={end}".format(start=start_date_time, end=end_date_time)
        params += optional_params

    params += "".join(f'&skills={skill}' for skill in skills)
    print(params)    
    
    basic_url = "http://18.133.247.78/search/interviewerTeams"    
    url = basic_url + params
    return requests.get(url).content

class GetAllInterviewers(BaseModel):
    """
    Pydantic arguments schema for get_all_interviewers function
    """

class GetInterviewerTeams(BaseModel):
    """
    Pydantic arguments schema for get_interviewer_teams function
    """
    team_size: int = Field(..., description="The number of interviewers in each team.")
    division: str = Field(..., description="A division interviewers belong to.")
    interview_level: str = Field(..., description="A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.")  

class GetInterviewerTeamsBySkills(BaseModel):
    """
    Pydantic arguments schema for get_interviewer_teams_by_skills function
    """
    team_size: int = Field(..., description="The number of interviewers in each team.")
    division: str = Field(..., description="A division interviewers belong to.")
    interview_level: str = Field(..., description="A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.")
    skills: list = Field(..., description="Skills that an interviewers must have to be able to conduct an interview.")

class GetInterviewerTeamsInTimeSlot(BaseModel):
    """
    Pydantic arguments schema for get_interviewer_teams_in_time_slot function
    """
    team_size: int = Field(..., description="The number of interviewers in each team.")
    division: str = Field(..., description="A division interviewers belong to.")
    interview_level: str = Field(..., description="A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.")
    start_date_time: str = Field(..., description="Date and time of start of interview.")
    duration_hours: int = Field(..., description="Required duration of the availability in hours.")   

class GetInterviewerTeamsBySkillsInTimeSlot(BaseModel):
    """
    Pydantic arguments schema for get_interviewer_teams_by_skills_in_time_slot function
    """
    team_size: int = Field(..., description="The number of interviewers in each team.")
    division: str = Field(..., description="A division interviewers belong to.")
    interview_level: str = Field(..., description="A level of interview. E.g. for Junior interview level we need interviewers level Junior or higher. For Middle - Middle and higher. And so on.")
    start_date_time: str = Field(..., description="Date and time of start of interview.")
    duration_hours: int = Field(..., description="Required duration of the availability in hours.")
    skills: list = Field(..., description="Skills that an interviewers must have to be able to conduct an interview.")

llm = ChatOpenAI(temperature=0.0, model_name="gpt-3.5-turbo", openai_api_key=OPENAI_API_KEY)

tools = [
    StructuredTool.from_function(
        func=get_all_interviewers,
        args_schema=GetAllInterviewers,
        description="A function to get a list of all interviewers from database."
    ),
    StructuredTool.from_function(
        func=get_interviewer_teams,
        args_schema=GetInterviewerTeams,
        description="A function to get all possible teams of interviewers."
    ),
    StructuredTool.from_function(
        func=get_interviewer_teams_by_skills,
        args_schema=GetInterviewerTeamsBySkills,
        description="A function to get all possible teams of interviewers by their skills."
    ),
    StructuredTool.from_function(
        func=get_interviewer_teams_in_time_slot,
        args_schema=GetInterviewerTeamsInTimeSlot,
        description="A function to get all possible teams of interviewers in requested time slot."
    ),
    StructuredTool.from_function(
        func=get_interviewer_teams_by_skills_in_time_slot,
        args_schema=GetInterviewerTeamsBySkillsInTimeSlot,
        description="A function to get all possible teams of interviewers by their skills in requested time slot."
    )
]

system_content = """
    You are an AI Interview Team Assistant that is developed by "Godel Technologies Europe" corporation.
    You help to choose employees who can interview newcomers.
    For this you select employees that are correspond to request parameters.
    You select employees from the data that is stored in json format.
    You always strictly and directly follow all instructions from the user.
    """

def predict(message, history):
    chat_history = []

    for human, assistant in history:
        chat_history.extend([HumanMessage(content=human), AIMessage(content=assistant)])

    prompt = ChatPromptTemplate.from_messages(
    [
        ("system", system_content),
        MessagesPlaceholder(variable_name="chat_history"),
        ("user", message),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
    )

    agent = create_openai_functions_agent(llm, tools, prompt)
    agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
    gpt_response = agent_executor.invoke({"input": message, "chat_history": chat_history})
    gpt_output = gpt_response["output"]
    chat_history.extend([HumanMessage(content=message), AIMessage(content=gpt_output)])
    return gpt_output

examples = [
    "Hello! Tell me please about yourself: who are you, what is your purpose, what tools do you have?",
    "I need a team of 2 interviewers from Java division who can interview Senior level newcomers",
    "I need a team of 2 interviewers from Java division with Java and SQL skills who can interview Senior level newcomers",
    "I need a team of 2 interviewers from Java division who can interview Senior level newcomers in given time slot: start time is 2024-02-12T11:00:00, duration 1 hour",
    "I need a team of 2 interviewers from Java division with Java and SQL skills who can interview Senior level newcomers in given time slot: start time is 2024-02-12T11:00:00, duration 1 hour",
    "List all interviewers"
]

description = '''
# AI Interview Team Assistant | Empowered by Godel Technologies AI \n
\n
This is an AI Interview Team Assistant. You can ask any questions about recruiting a team for an interview.\n
'''

gr.ChatInterface(predict, examples=examples, description=description).launch()