app.py

import logging
import os
from subprocess import check_call, STDOUT
from time import asctime

import gradio as gr
from llama_index.core import Document, VectorStoreIndex

from generate_response import generate_chat_response_with_history, set_llm, is_search_query, condense_question, \
    generate_chat_response_with_history_rag_return_response
from web_search import search

API_KEY_PATH = "../keys/gpt_api_key.txt"
logger = logging.getLogger("agent_logger")
sourced = False
query = False
rag_similarity = False


def google_search_chat(message, history):
    condensed_question = condense_question(message, history)
    if is_search_query(condensed_question):
        search_results = search(message, condensed_question)
        relevant_content = ""
        sources = ""
        for index, result in enumerate(search_results):
            relevant_content = relevant_content + "\n" + ''.join(result['text'])
            sources = sources + f'\n {index + 1}. ' + result['url']  # python is zero-indexed

        if relevant_content != "":
            documents = [Document(text=relevant_content)]
            index = VectorStoreIndex.from_documents(documents)

            response = generate_chat_response_with_history_rag_return_response(index, message, history)

            # similar_str = "not calculated"
            # faithfulness_str = "not calculated"
            #
            # if rag_similarity:
            #     sim_evaluator = SemanticSimilarityEvaluator()
            #     faith_evaluator = FaithfulnessEvaluator(llm=get_llm())
            #     # condensed_context = condense_context(relevant_content)
            #     # logger.info("Calculating similarity...")
            #     # similar = sim_evaluator.evaluate(response=str(response),
            #     #                                   reference=condensed_context)
            #     logger.info("Calculating faithfulness...")
            #     faithfulness = faith_evaluator.evaluate_response(query=condensed_question, response=response)
            #     # similar_str = str(round((similar.score * 100), 2)) + "%"
            #     faithfulness_str = "Yes" if faithfulness.passing else "No"
            #
            # logger.info(f'**Search Query:** {condensed_question} \n **Faithfulness:** {faithfulness_str} \n '
            #             f'**Similarity:** {similar_str} \n **Sources used:** \n {sources}')

            response_text = []
            string_output = ""

            for text in response.response_gen:
                response_text.append(text)
                string_output = ''.join(response_text)
                yield string_output

            # if not sourced:
            #     pass
            # if sourced and not query and not rag_similarity:
            #     yield string_output + f'\n\n --- \n **Sources used:** \n {sources}'
            # if sourced and query and not rag_similarity:
            #     yield (string_output
            #            + f'\n\n --- \n **Search Query:** {condensed_question} '
            #              f'\n **Sources used:** \n {sources}')
            # if rag_similarity:
            #     yield (string_output
            #            + f'\n\n --- \n **Search Query:** {condensed_question} \n '
            #            # f'**Similarity of response to the sources [ℹ️]'
            #            # f'(https://en.wikipedia.org/wiki/Semantic_similarity):** {similar_str} \n'
            #              f'**Is response in source documents?**: {faithfulness_str}'
            #              f'\n **Sources used:** \n {sources}')

            logger.info(f'Assistant Response: {string_output}')
        else:
            logger.info(
                f'Assistant Response: Sorry, no search results found.')
            yield "Sorry, no search results found."

    else:
        yield from generate_chat_response_with_history(message, history)


if __name__ == '__main__':
    logging.root.setLevel(logging.INFO)
    filehandler = logging.FileHandler(f'agent_log_{asctime().replace(" ", "").lower().replace(":", "")}.log',
                                      'a')
    formatter = logging.Formatter('%(asctime)-15s::%(levelname)s::%(filename)s::%(funcName)s::%(lineno)d::%(message)s')
    filehandler.setFormatter(formatter)
    logger = logging.getLogger("agent_logger")
    for hdlr in logger.handlers[:]:  # remove the existing file handlers
        if isinstance(hdlr, logging.FileHandler):
            logger.removeHandler(hdlr)
    logger.addHandler(filehandler)  # set the new handler
    logger.setLevel(logging.INFO)

    logging.getLogger('selenium.webdriver.common').setLevel(logging.DEBUG)

    api_key = os.getenv('gpt_api_key')

    # GPT - 4 Turbo. The latest GPT - 4 model intended to reduce cases of “laziness” where the model doesn’t complete
    # a task. Returns a maximum of 4,096 output tokens. Link:
    # https://openai.com/blog/new-embedding-models-and-api-updates
    set_llm(key=api_key, model="gpt-4-0125-preview", temperature=0)
    # apt-get install -y libglib2.0-0=2.50.3-2 \
    #         libnss3=2:3.26.2-1.1+deb9u1 \
    #         libgconf-2-4=3.2.6-4+b1 \
    #         libfontconfig1=2.11.0-6.7+b1
    # check_call(['apt-get', 'install', '-y', 'libglib2.0-0 libnss3 libgconf-2-4 libfontconfig1'],
    #            stdout=open(os.devnull, 'wb'), stderr=STDOUT)

    logger.info("Launching Gradio ChatInterface for searchbot...")

    demo = gr.ChatInterface(fn=google_search_chat,
                            title="Search Assistant", retry_btn=None, undo_btn=None, clear_btn=None,
                            theme="soft")
    demo.launch(auth=('convo', 'session2024'))