|
import os |
|
from typing import List, Tuple, Dict, Union, Any |
|
import requests |
|
|
|
import numpy as np |
|
import openai |
|
import pandas as pd |
|
import streamlit as st |
|
from langchain.document_loaders import TextLoader |
|
from langchain.embeddings.sentence_transformer import SentenceTransformerEmbeddings |
|
from langchain.text_splitter import CharacterTextSplitter |
|
from langchain.vectorstores import Chroma |
|
from scipy.spatial.distance import cosine |
|
|
|
openai.api_key = os.environ["OPENAI_API_KEY"] |
|
|
|
|
|
def call_chatgpt(prompt: str) -> str: |
|
""" |
|
Uses the OpenAI API to generate an AI response to a prompt. |
|
|
|
Args: |
|
prompt: A string representing the prompt to send to the OpenAI API. |
|
|
|
Returns: |
|
A string representing the AI's generated response. |
|
|
|
""" |
|
|
|
|
|
response = openai.Completion.create( |
|
model="gpt-3.5-turbo-instruct", |
|
prompt=prompt, |
|
temperature=0.5, |
|
max_tokens=500, |
|
top_p=1, |
|
frequency_penalty=0, |
|
presence_penalty=0, |
|
) |
|
|
|
|
|
ans = response.choices[0]["text"] |
|
|
|
|
|
return ans |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def ai_judge(sentence1: str, sentence2: str) -> float: |
|
|
|
HF_TOKEN = os.environ["HF_TOKEN"] |
|
API_URL = "https://api-inference.huggingface.co/models/sentence-transformers/msmarco-distilbert-base-tas-b" |
|
headers = {"Authorization": f"Bearer {HF_TOKEN}"} |
|
|
|
def helper(payload): |
|
response = requests.post(API_URL, headers=headers, json=payload) |
|
return response.json() |
|
|
|
data = helper( |
|
{ |
|
"inputs": { |
|
"source_sentence": sentence1, |
|
"sentences": [sentence2] |
|
} |
|
} |
|
) |
|
|
|
return data |
|
|
|
|
|
def query(payload: Dict[str, Any]) -> Dict[str, Any]: |
|
""" |
|
Sends a JSON payload to a predefined API URL and returns the JSON response. |
|
Args: |
|
payload (Dict[str, Any]): The JSON payload to be sent to the API. |
|
Returns: |
|
Dict[str, Any]: The JSON response received from the API. |
|
""" |
|
|
|
|
|
API_URL = "https://sks7h7h5qkhoxwxo.us-east-1.aws.endpoints.huggingface.cloud" |
|
|
|
|
|
headers = { |
|
"Accept": "application/json", |
|
"Content-Type": "application/json" |
|
} |
|
|
|
|
|
response = requests.post(API_URL, headers=headers, json=payload) |
|
|
|
|
|
return response.json() |
|
|
|
|
|
def llama2_7b_ysa(prompt: str) -> str: |
|
""" |
|
Queries a model and retrieves the generated text based on the given prompt. |
|
This function sends a prompt to a model (presumably named 'llama2_7b') and extracts |
|
the generated text from the model's response. It's tailored for handling responses |
|
from a specific API or model query structure where the response is expected to be |
|
a list of dictionaries, with at least one dictionary containing a key 'generated_text'. |
|
Parameters: |
|
- prompt (str): The text prompt to send to the model. |
|
Returns: |
|
- str: The generated text response from the model. |
|
Note: |
|
- The function assumes that the 'query' function is previously defined and accessible |
|
within the same scope or module. It should send a request to the model and return |
|
the response in a structured format. |
|
- The 'parameters' dictionary is passed empty but can be customized to include specific |
|
request parameters as needed by the model API. |
|
""" |
|
|
|
|
|
query_payload: Dict[str, Any] = { |
|
"inputs": prompt, |
|
"parameters": {} |
|
} |
|
|
|
|
|
output = query(query_payload) |
|
|
|
|
|
response: str = output[0]['generated_text'] |
|
|
|
return response |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
from datasets import load_dataset |
|
import chromadb |
|
import string |
|
|
|
dataset = load_dataset("eagle0504/youthless-homeless-shelter-web-scrape-dataset-qa-formatted") |
|
client = chromadb.Client() |
|
random_number = np.random.randint(low=1e9, high=1e10) |
|
random_string = ''.join(np.random.choice(list(string.ascii_uppercase + string.digits), size=10)) |
|
combined_string = f"{random_number}{random_string}" |
|
collection = client.create_collection(combined_string) |
|
|
|
|
|
L = len(dataset["train"]['questions']) |
|
collection.add( |
|
ids=[str(i) for i in range(0, L)], |
|
documents=dataset["train"]['questions'], |
|
metadatas=[{"type": "support"} for _ in range(0, L)], |
|
) |
|
|
|
st.title("Youth Homelessness Chatbot") |
|
|
|
|
|
if "messages" not in st.session_state: |
|
st.session_state.messages = [] |
|
|
|
|
|
for message in st.session_state.messages: |
|
with st.chat_message(message["role"]): |
|
st.markdown(message["content"]) |
|
|
|
st.sidebar.markdown( |
|
""" |
|
|
|
### Instructions: |
|
|
|
This app guides you through YSA's website, utilizing a RAG-ready Q&A dataset [here](https://huggingface.co/datasets/eagle0504/youthless-homeless-shelter-web-scrape-dataset-qa-formatted) for chatbot assistance. 🤖 Enter a question, and it finds similar ones in the database, offering answers with a distance score to gauge relevance—the lower the score, the closer the match. 🎯 For better accuracy and to reduce errors, user feedback helps refine the database. ✨ |
|
|
|
""") |
|
st.sidebar.success("Please enter a distance threshold (we advise to set it around 0.2).") |
|
special_threshold = st.sidebar.number_input("Insert a number", value=0.2, placeholder="Type a number...") |
|
clear_button = st.sidebar.button("Clear Conversation", key="clear") |
|
|
|
if clear_button: |
|
st.session_state.messages = [] |
|
|
|
|
|
if prompt := st.chat_input("Tell me about YSA"): |
|
|
|
st.chat_message("user").markdown(prompt) |
|
|
|
st.session_state.messages.append({"role": "user", "content": prompt}) |
|
|
|
question = prompt |
|
|
|
with st.spinner("Wait for it..."): |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
results = collection.query( |
|
query_texts=question, |
|
n_results=5 |
|
) |
|
idx = results["ids"][0] |
|
idx = [int(i) for i in idx] |
|
ref = pd.DataFrame( |
|
{ |
|
"idx": idx, |
|
"questions": [dataset["train"]['questions'][i] for i in idx], |
|
"answers": [dataset["train"]['answers'][i] for i in idx], |
|
"distances": results["distances"][0] |
|
} |
|
) |
|
|
|
filtered_ref = ref[ref["distances"] < special_threshold] |
|
if filtered_ref.shape[0] > 0: |
|
st.success("There are highly relevant information in our database.") |
|
ref_from_db_search = filtered_ref["answers"] |
|
final_ref = filtered_ref |
|
else: |
|
st.warning("The database may not have relevant information to help your question so please be aware of hallucinations.") |
|
ref_from_db_search = ref["answers"] |
|
final_ref = ref |
|
|
|
try: |
|
llm_response = llama2_7b_ysa(question) |
|
except: |
|
llm_response = "Sorry, the inference endpoint is temporarily down. 😔" |
|
|
|
finetuned_llm_guess = ["from_llm", question, llm_response, 0] |
|
final_ref.loc[-1] = finetuned_llm_guess |
|
final_ref.index = final_ref.index + 1 |
|
|
|
|
|
independent_ai_judge_score = [] |
|
for i in range(final_ref.shape[0]): |
|
this_quest = question |
|
this_content = final_ref["answers"][i] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
this_score = ai_judge(question, this_content) |
|
independent_ai_judge_score.append(this_score[0]) |
|
|
|
final_ref["ai_judge"] = independent_ai_judge_score |
|
|
|
engineered_prompt = f""" |
|
Based on the context: {ref_from_db_search}, |
|
answer the user question: {question}. |
|
Answer the question directly (don't say "based on the context, ...") |
|
""" |
|
|
|
answer = call_chatgpt(engineered_prompt) |
|
response = answer |
|
|
|
|
|
with st.chat_message("assistant"): |
|
with st.spinner("Wait for it..."): |
|
st.markdown(response) |
|
with st.expander("See reference:"): |
|
st.table(final_ref) |
|
|
|
st.session_state.messages.append({"role": "assistant", "content": response}) |
|
|
|
|
|
|
|
|