Spaces:
Running
Running
"""Answer questions about my resume.""" | |
# %% IMPORTS | |
import logging | |
import gradio as gr | |
import lib | |
# %% LOGGING | |
logging.basicConfig( | |
level=logging.INFO, | |
format="[%(asctime)s][%(levelname)s] %(message)s", | |
) | |
# %% CONFIGS | |
# %% - Frontend | |
THEME = "soft" | |
TITLE = "Fmind AI Assistant" | |
EXAMPLES = [ | |
"Who is Médéric Hurier (Fmind)?", | |
"Is Fmind open to new opportunities?", | |
"Can you share details about Médéric PhD?", | |
"Elaborate on Médéric current work position", | |
"Describe his proficiency with Python programming", | |
"What is the answer to life, the universe, and everything?", | |
] | |
# %% - Backend | |
MODEL = lib.get_language_model() | |
CLIENT = lib.get_database_client(path=lib.DATABASE_PATH) | |
ENCODING = lib.get_encoding_function() | |
EMBEDDING = lib.get_embedding_function() | |
COLLECTION = CLIENT.get_collection( | |
name=lib.DATABASE_COLLECTION, | |
embedding_function=EMBEDDING, | |
) | |
# %% - Answer | |
PROMPT_CONTEXT = """ | |
You are Fmind AI Assistant, specialized in providing information regarding Médéric Hurier's (known as Fmind) professional background. | |
Médéric is a Lead MLOps engineer based in Luxembourg. He is currently working at Decathlon. His calendar is booked until the conclusion of 2024. | |
Your responses should be succinct and maintain a professional tone. If inquiries deviate from Médéric's professional sphere, courteously decline to engage. | |
You may find more information about Médéric below (markdown format): | |
""" | |
PROMPT_MAX_TOKENS = lib.MODEL_INPUT_LIMIT | |
QUERY_MAX_DISTANCE = 0.4 | |
QUERY_N_RESULTS = 20 | |
# %% FUNCTIONS | |
def answer(message: str, history: list[str]) -> str: | |
"""Answer questions about my resume.""" | |
# counters | |
n_tokens = 0 | |
# messages | |
messages = [] | |
# - context | |
n_tokens += len(ENCODING(PROMPT_CONTEXT)) | |
messages += [{"role": "system", "content": PROMPT_CONTEXT}] | |
# - history | |
for user_content, assistant_content in history: | |
n_tokens += len(ENCODING(user_content)) | |
n_tokens += len(ENCODING(assistant_content)) | |
messages += [{"role": "user", "content": user_content}] | |
messages += [{"role": "assistant", "content": assistant_content}] | |
# - message | |
n_tokens += len(ENCODING(message)) | |
messages += [{"role": "user", "content": message}] | |
# database | |
results = COLLECTION.query(query_texts=message, n_results=QUERY_N_RESULTS) | |
logging.info("Results: %s", results) | |
distances = results["distances"][0] | |
documents = results["documents"][0] | |
for distance, document in zip(distances, documents): | |
# - distance | |
logging.debug("Doc distance: %f", distance) | |
if distance > QUERY_MAX_DISTANCE: | |
break | |
# - document | |
n_document_tokens = len(ENCODING(document)) | |
logging.debug("Doc tokens: %f", n_document_tokens) | |
if (n_tokens + n_document_tokens) >= PROMPT_MAX_TOKENS: | |
break | |
n_tokens += n_document_tokens | |
messages[0]["content"] += document | |
# response | |
logging.info("Tokens: %d", n_tokens) | |
logging.info("Messages: %s", messages) | |
api_response = MODEL(messages=messages) | |
logging.info("Response: %s", api_response.to_dict_recursive()) | |
# content | |
content = api_response["choices"][0]["message"]["content"] | |
# return | |
return content | |
# %% INTERFACES | |
interface = gr.ChatInterface( | |
fn=answer, | |
theme=THEME, | |
title=TITLE, | |
examples=EXAMPLES, | |
clear_btn=None, | |
retry_btn=None, | |
undo_btn=None, | |
) | |
if __name__ == "__main__": | |
interface.queue(concurrency_count=20).launch() | |