Genta Academic Assistant, an AI powered assistant to help you on your academic research
"""
FILE is for UI
"""
import streamlit as st
from PIL import Image

from modules.events import trigger_event
from modules.utilities import EnvironmentVariables, validate_openai_api_key

env = EnvironmentVariables()

IMAGE_DIRECTORY = "./Genta_Logo.png"
image = Image.open(IMAGE_DIRECTORY)

PAGE_CONFIG = {"page_title": "Academic Assistant", "page_icon": image}

st.set_page_config(**PAGE_CONFIG)

TITLE = (
    '<link rel="stylesheet" type="text/css"' +
    'href="http://fonts.googleapis.com/css?family=Ubuntu:regular,bold&subset=Latin">' + 
    '<H1 style="font-family:Ubuntu; font-size: 30px; font-weight:bold;">' +
    '<span style="font-size = 40px;">GENTA</span> - Academic Assistant</H1>')

st.markdown(TITLE, unsafe_allow_html=True)
st.markdown("""Genta Academic Assistant is a tool to help you""" +
            """ find the right research paper for your research.""")

with st.sidebar:
    col1, col2, col3 = st.columns(3)
    img = Image.open("Extension.png")
    with col2:
        st.image(img, use_column_width="always")
    BOX = "<style>#input {box-shadow: 10px 5px 5px black;}</style>"
    st.markdown(BOX, unsafe_allow_html=True)
    st.text_input(
        "Please enter token",
        placeholder="Optional",
        key="token",
    )

if "messages" not in st.session_state:
    st.session_state.messages = [{
        "role": "system",
        "content": "You will be provided with multiple documents " +
                   "delimited by triple quotes and a question. " +
                   "Your task is to answer the question using only " +
                   "the provided documents and to cite " +
                   "the passage(s) of the documents used to answer " +
                   "the question. If the documents does " +
                   "not contain the information needed to " +
                   "answer this question then simply write: " +
                   '"Insufficient information." If an answer ' +
                   "to the question is provided, it must " +
                   "be annotated with a citations. Use the following " +
                   "format for to cite relevant passages " +
                   '({"citations": ..., ..., ...}).',
    }]

for message in st.session_state.messages[1:]:
    with st.chat_message(message["role"]):
        st.markdown(message["content"])

if (len(st.session_state.messages) <= 8
    and ("token" in st.session_state or st.session_state.token == "")) \
   or (validate_openai_api_key(st.session_state.token) and "token" in st.session_state):
    if prompt := st.chat_input("Wassup"):
        with st.chat_message("user"):
            st.markdown(prompt)

        st.session_state.messages.append({"role": "user", "content": prompt})

        response, docs = trigger_event(
            st.session_state.token
            if "token" in st.session_state \
                and validate_openai_api_key(st.session_state.token)
            else env["OPENAI_API_KEY"]
        )

        with st.chat_message("assistant"):
            st.markdown(response)
            with st.expander("Search Results"):
                for doc in docs:
                    st.markdown(
                        f"<a href='https://doi.org/{doc['dOI']}'>" +
                        f"{doc['title']}, {doc['authors']}, {doc['date']}</a>",
                        unsafe_allow_html=True,
                    )

        st.session_state.messages.append({
            "role": "assistant",
            "content": response
        })
else:
    TEXT = (
        '<p style="font-size: 18px; font-weight:bold; color: #FF1D2E; margin-top: 15px;">' +
        "Limit exceeded, refresh the page to start a new conversation or insert your own" + \
        " <a href='https://platform.openai.com/account/api-keys'>OpenAI API key</a> " + \
        "in the sidebar.</p>"
    )
    st.markdown(TEXT, unsafe_allow_html=True)

app.py
Genta-Technology/Academic-Assistant
"""
Populate the database with the data from the json file.
"""

import json
import weaviate

from tqdm import tqdm

from modules.utilities import EnvironmentVariables

# Load environment variables
env = EnvironmentVariables()

# Create a client
client = weaviate.Client(
    url=env.get('WEAVIATE_CLIENT_ENDPOINT'),
    additional_headers={
        'X-OpenAI-Api-Key': env('OPENAI_API_KEY')
    }
)

# Check if schema exists
if not client.schema.contains():
    # Create schema
    client.schema.create(
        {
            'classes': [
                {
                    'class': 'Paper',
                    'description': 'A research paper',
                    'properties': [
                        {
                            'name': 'DOI',
                            'description': 'ID of the paper',
                            'dataType': ['string'],
                        },
                        {
                            'name': 'title',
                            'description': 'Title of the paper',
                            'dataType': ['string']
                        },
                        {
                            'name': 'authors',
                            'description': 'Authors of the paper',
                            'dataType': ['string']
                        },
                        {
                            'name': 'abstract',
                            'description': 'Abstract of the paper',
                            'dataType': ['string']
                        },
                        {
                            'name': 'date',
                            'description': 'Date of the paper',
                            'dataType': ['string']
                        },
                    ]
                }
            ]
        }
    )

def main(data_path: str = './data/ml-arxiv-embeddings.json'):
    """
    Populate the database with data.

    Args:
        data_path (str): Path to the data file.
    """

    # read data json file from ./data/ml-arxiv-embeddings.json
    with open(data_path, encoding='utf-8') as file:
        json_data = []
        # read file as text
        data = file.read()
        # split text into lines
        data = data.split('\n')
        for item in enumerate(data):
            if item != '':
                # parse json
                try:
                    json_data.append(json.loads(item[1]))
                except json.decoder.JSONDecodeError:
                    continue

    client.batch.configure(100)
    with client.batch as batch:
        for sample in tqdm(data):
            batch.add_data_object(
                data_object={
                    'DOI': sample['root']['id'],
                    'title': sample['root']['title'],
                    'authors': sample['root']['authors'],
                    'abstract': sample['root']['abstract'],
                    'date': sample['root']['update_date'],
                },
                class_name='Paper',
                vector=sample["root"]['embedding']
            )

database/populate_database.py
Genta-Technology/Academic-Assistant
"""
Populate the database with data.

Usage:
    python database_initialization.py --data ./data/ml-arxiv-embeddings.json
"""

import argparse

from database import populate_database

# setup command line arguments
parser = argparse.ArgumentParser(description='Populate the database with data.')
parser.add_argument(
    '--data',
    type=str,
    default='./data/ml-arxiv-embeddings.json',
    help='Path to the data file.'
)
args = parser.parse_args()

if __name__ == '__main__':
    populate_database.main(args.data)

database_initialization.py
Genta-Technology/Academic-Assistant
"""
main python file for events and handling communication to gpt and get abstract
"""

import streamlit as st

from .prompt import ask_gpt, generate_search
from .utilities import search_semantics

def trigger_event(token: str):
    """
    Trigger the event for every user input.
    
    Returns:
        list: List of messages.
    """
    # check if the question need source
    session_message = generate_search(token, st.session_state.messages)

    if session_message.lower() != "empty":
        # abstract_list = get_abstract(
        #     input_str=session_message,
        #     weaviate_url='https://genta-academic-assistant-cluster-4vw2zql4.weaviate.network',
        #     openai_api_token=st.session_state.token
        # )
        abstract_list = search_semantics(session_message)
    else:
        abstract_list = []

    # go to function from farrel
    # user_message: list(dict) -> [{"role": "system" or "assistant" or "user", "content": str}]
    new_user_message = ask_gpt(token, st.session_state.messages, abstract_list)

    return new_user_message[-1]['content'], abstract_list

modules/events.py
Genta-Technology/Academic-Assistant
"""
main python file for communicating with GPT and getting the response, main function is the com_gpt

Args:
    - token (str): 
    - messages (dict):
    - docs (list):
"""

import openai

MODEL = "gpt-3.5-turbo"
# PRE_SEARCH = "Based on the recent question, your task is to give me an " + \
#              "appropriate query to search arxiv paper, or only response with " + \
#              "EMPTY If the question does not need a sources to answer. You should " + \
#              "not say more than query. You should not say any words except the query or EMPTY."

PRE_SEARCH_SYSTEM = "You are an expert in finding research papers on the internet. " + \
                    "You are the master in creating perfect search queries. " + \
                    "You will be given the previous chat between you and the user. " + \
                    "Your task is to analyze the latest question and only provide an " + \
                    "appropriate query to search research papers such as Google " + \
                    "Scholar, Arxiv, and Semantic Scholars. Your response should be " + \
                    "only the search query, and you should not say more than the query " + \
                    "itself. If, based on your analysis of the previous chat, the latest " + \
                    "question does not need a search query or is only a small talk chat, " + \
                    "your response should be EMPTY. Once again, you should not say " + \
                    "any words except the query or EMPTY!"


def ask_gpt(token, messages, docs):
    """
    takes in user gpt token, session stage, and required documents (abstract) and return answer
    
    Args:
    - token (str): 
    - message (list):
    - docs (list):
    """
    # Auth Token
    openai.api_key = token

    #Intitial Question
    if len(docs) > 0:
        new_messages = new_question(messages, docs)

    #Continue asking
    else:
        new_messages = con_question(messages)

    return new_messages # String

def generate_search(token, messages):
    """
    generate search querries for abstract or return EMPTY if no question asked

    Args:
    - token (str)
    - messages (list)
    """
    # Auth Token
    openai.api_key = token

    search_system = [{"role":"system", "content":PRE_SEARCH_SYSTEM}]
    output=openai.ChatCompletion.create(
        model=MODEL,
        messages=search_system + messages[1:],
        temperature=1.3,
        max_tokens=256,
        )
    output = output['choices'][0]['message']['content']
    output = output if 'as an ai language' not in output.lower() else 'EMPTY'
    return output

def new_question(messages, docs):
    """
    first time asking gpt with docs/sources included

    Args:
    - messages (dict)
    - docs (dict)
    """
    messages = mod_new_messages(messages, docs)
    output=openai.ChatCompletion.create(
        model=MODEL,
        messages=messages,
        temperature=0.2,
        max_tokens=512,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
        )
    # output = add_source(output, docs)
    messages += [{"role": "assistant", "content":output['choices'][0]['message']['content']}]
    return messages


def create_user_context(question, sources):
    """
    create user context for GPT

    Args:
    - question (str):
    - sources (list):
    """
    user_context = "My question is: " + question + ". " + \
                   "My sources are: " + sources + ". " + \
                   "Answer the question only if you can " + \
                   "find the answer in the sources. " + \
                   "If you can't find the answer in any of " + \
                   "the sources, response with " +\
                   "'i don't know the answer to this question " + \
                   "based on my current knowledge'. " + \
                   "Add the source of your answer as (authors, dOI) " + \
                   "for every text you get from the sources."

    return user_context

def con_question(messages):
    """
    continue asking gpt

    Args:
    - messages (dict)
    """
    output = openai.ChatCompletion.create(
        model=MODEL,
        messages=messages,
        temperature=0.2,
        max_tokens=512,
        top_p=1,
        frequency_penalty=0,
        presence_penalty=0
    )
    return messages + [{"role": "assistant", "content":output['choices'][0]['message']['content']}]

def mod_new_messages(messages, docs):
    """
    modify the new messages for adding context

    Args:
    - messages (list)
    - docs (list)
    """
    prompt = docs_compile(docs) + "\n Question: " + messages[-1]["content"]
    messages = messages[:-1]

    return messages + [{"role":"user", "content":prompt}]

def docs_compile(docs):
    """
    combined all abstracts and other supporting file into one

    Args:
    - docs (list)
    """
    combined_docs = '"""'

    for doc in docs:
        abstract, author, doi, date = abstract_to_string(doc)
        abstract_context = abstract +\
                            "\n {citation: " + f"{author}, {date}, {doi}" + "}\n"
        combined_docs += abstract_context
    return combined_docs + '"""'

def abstract_to_string(abstract_dict):
    """
    combined the dict of abstract into one string

    Args:
    - abstract (dict):
    """
    return (abstract_dict["abstract"],
            abstract_dict["authors"],
            abstract_dict["dOI"],
            abstract_dict["date"])

modules/prompt.py
Genta-Technology/Academic-Assistant
"""
A collection of utility functions.
"""

import os
import openai
import weaviate
import requests

from dotenv import load_dotenv, find_dotenv


def validate_openai_api_key(api_key: str) -> bool:
    """
    Validate OpenAI API Key

    Returns:
        bool: True if valid, False otherwise.
    """

    openai_api_endpoint = "https://api.openai.com/v1/engines"

    headers = {"Authorization": f"Bearer {api_key}"}

    response = requests.get(openai_api_endpoint, headers=headers, timeout=10)

    # Check the status code of the response
    return response.status_code == 200


def open_ai_embeddings(input_str: str, api_token: str):
    """Get the OpenAI embeddings for a given input string.

    :param input_str: Input string.
    :type input_str: str
    :param api_token: OpenAI API token.
    :type api_token: str
    :param input_str: str:
    :param api_token: str:
    :returns: List of embeddings.
    :rtype: list

    """

    # use the api token
    openai.api_key = api_token

    # convert text to embedding
    model_id = "text-embedding-ada-002"
    embedding = openai.Embedding.create(input=input_str,
                                        model=model_id)["data"][0]["embedding"]
    return embedding


def get_abstract(input_str: str,
                 weaviate_url: str,
                 openai_api_token: str,
                 top_n: int = 5):
    """Get the abstract of the top n papers similar to the input string.

    :param input_str: Input string.
    :type input_str: str
    :param weaviate_url: Weaviate URL.
    :type weaviate_url: str
    :param openai_api_token: OpenAI API token.
    :type openai_api_token: str
    :param n: Number of papers to return.
    :type n: int
    :param input_str: str:
    :param weaviate_url: str:
    :param openai_api_token: str:
    :param top_n: int:  (Default value = 5)
    :returns: List of papers.
    :rtype: list

    """

    input_emb = open_ai_embeddings(input_str, openai_api_token)

    client = weaviate.Client(
        url=weaviate_url,
        additional_headers={"X-OpenAI-Api-Key": openai_api_token}
    )

    response = (client.query
                .get("Paper", ["dOI", "authors", "abstract", "date", "title"])
                .with_near_vector({"vector": input_emb})
                .with_additional(['certainty'])
                .with_limit(top_n)
                .do())

    return response["data"]["Get"]["Paper"]


class EnvironmentVariables:
    """
    This class is used to load environment variables from a .env file.
    """

    def __init__(self):
        """
        Initialize the class.
        """

        # Load environment variables from .env file
        self.env_file = find_dotenv()
        if self.env_file:
            load_dotenv(self.env_file)

    def get(self, key: str) -> str:
        """
        Get an environment variable.

        Args:
            key (str): Environment variable key.

        Returns:
            str: Environment variable value.
        """

        return os.environ.get(key)

    def __call__(self, key: str) -> str:
        """
        Get an environment variable.

        Args:
            key (str): Environment variable key.

        Returns:
            str: Environment variable value.
        """

        return self.get(key)

    def __getitem__(self, key: str) -> str:
        """
        Get an environment variable.

        Args:
            key (str): Environment variable key.

        Returns:
            str: Environment variable value.
        """

        return self.get(key)

def search_semantics(querry, total=20):
    """
    search the querry using the semantics api

    Args:
    - querry (str)
    - total (int) (between 1-10, set default to 5) [optional]
    """

    url_search = "https://api.semanticscholar.org/graph/v1/paper" + \
                 f"/search?query={querry.replace(' ', '+')}&limit=" + \
                 f"{total}&fields=abstract,authors,year,externalIds" + \
                 ",title,publicationDate"

    result = requests.get(url_search, timeout=10)

    docs = []
    if result.status_code == 200:
        result = result.json()

        # check if the result is empty
        if 'data' not in result:
            return docs

        for i in range(len(result["data"])):
            if "DOI" not in result["data"][i]['externalIds'] or \
                result["data"][i]["abstract"] is None:
                continue

            doc = {
                "dOI": result["data"][i]["externalIds"]["DOI"], 
                "abstract": result["data"][i]["abstract"],
                "title":result["data"][i]["title"],
                "date":result["data"][i]["publicationDate"],
                "authors":get_authors(result["data"][i]["authors"])
            }
            docs += [doc]

    return docs[:5]

def get_authors(list_author):
    """
    consumed list of dictionary author and return just list of authors name

    Args:
    - list_author (list)
    """
    new_list = []

    for i in list_author:
        new_list += [i["name"]]

    return new_list

modules/utilities.py
Genta-Technology/Academic-Assistant