Spaces:
Runtime error
Runtime error
# Import libraries. | |
import gradio as gr | |
import os | |
import pprint | |
import requests | |
import sys | |
import weaviate | |
from openai import OpenAI | |
from gradio.themes.base import Base | |
from icecream import ic | |
from langchain.callbacks.tracers import ConsoleCallbackHandler | |
from langchain.chat_models import ChatOpenAI | |
from langchain_community.embeddings import HuggingFaceEmbeddings | |
from langchain_community.llms import HuggingFacePipeline | |
from langchain_core.messages import HumanMessage, SystemMessage | |
from langchain.document_loaders import TextLoader | |
from langchain.embeddings import OpenAIEmbeddings | |
from langchain.prompts import ChatPromptTemplate | |
from langchain.schema.runnable import RunnablePassthrough | |
from langchain.schema.output_parser import StrOutputParser | |
from langchain.text_splitter import RecursiveCharacterTextSplitter | |
from langchain.vectorstores import Weaviate | |
from transformers import AutoModelForCausalLM, AutoTokenizer, pipeline | |
from weaviate.embedded import EmbeddedOptions | |
def AIAimAdvice(query): | |
# OPENAI_API_KEY = os.getenv("OPENAI_API_KEY") | |
directory_path = "ApexDocs" | |
my_txts = os.listdir(directory_path) | |
# Load the TXT. | |
loaders = [] | |
for my_txt in my_txts: | |
my_txt_path = os.path.join(directory_path, my_txt) | |
loaders.append(TextLoader(my_txt_path)) | |
print("len(loaders) =", len(loaders)) | |
# Load the TXT. | |
# data = [loader.load() for loader in loaders] | |
data = [] | |
for loader in loaders: | |
data.append(loader.load()) | |
print("len(data) =", len(data), "\n") | |
# Initialize the text splitter | |
# Uses a text splitter to split the data into smaller documents. | |
text_splitter = RecursiveCharacterTextSplitter(chunk_size=500, chunk_overlap=50) | |
# docs = [text_splitter.split_documents(doc) for doc in data] | |
docs = [] | |
for doc in data: | |
chunk = text_splitter.split_documents(doc) | |
docs.append(chunk) | |
# Debugging purposes. | |
# Print the number of total documents to be stored in the vector database. | |
total = 0 | |
for i in range(len(docs)): | |
if i == len(docs) - 1: | |
print(len(docs[i]), end="") | |
else: | |
print(len(docs[i]), "+ " ,end="") | |
total += len(docs[i]) | |
print(" =", total, " total documents\n") | |
# Print the first document. | |
print(docs[0], "\n\n\n") | |
# Print the total number of PDF files. | |
# docs is a list of lists where each list stores all the documents for one PDF file. | |
print(len(docs)) | |
# Merge the documents to be embededed and store them in the vector database. | |
merged_documents = [] | |
for doc in docs: | |
merged_documents.extend(doc) | |
# Print the merged list of all the documents. | |
print("len(merged_documents) =", len(merged_documents)) | |
print(merged_documents) | |
# Hugging Face model for embeddings. | |
model_name = "sentence-transformers/all-MiniLM-L6-v2" | |
model_kwargs = {'device': 'cpu'} | |
# model_kwargs = {'device': 'cuda'} | |
embeddings = HuggingFaceEmbeddings( | |
model_name=model_name, | |
model_kwargs=model_kwargs, | |
) | |
# Create Weaviate vector store (database). | |
client = weaviate.Client( | |
embedded_options = EmbeddedOptions() | |
) | |
# Initialize the Weaviate vector search with the document segments. | |
# Create a vector store (database) named vector_search from the sample documents. | |
vector_search = Weaviate.from_documents( | |
client = client, | |
documents = merged_documents, | |
# embedding = OpenAIEmbeddings(), | |
embedding = embeddings, | |
by_text = False | |
) | |
# Instantiate Weaviate Vector Search as a retriever | |
# Basic RAG. | |
# k to search for only the 10 most relevant documents. | |
# score_threshold to use only documents with a relevance score above 0.89. | |
retriever = vector_search.as_retriever( | |
search_type = "similarity", # similarity, mmr, similarity_score_threshold. https://api.python.langchain.com/en/latest/vectorstores/langchain_core.vectorstores.VectorStore.html#langchain_core.vectorstores.VectorStore.as_retriever | |
search_kwargs = {"k": 10, "score_threshold": 0.89} | |
) | |
# Define a prompt template. | |
# Define a LangChain prompt template to instruct the LLM to use our documents as the context. | |
# LangChain passes these documents to the {context} input variable and the user's query to the {question} variable. | |
template = """ | |
You are an aim tutor for providing advice on improving aim specifically on Apex Legends. | |
Use the following pieces of retrieved context to answer the question at the end. | |
If you don't know the answer, just say that you don't know. | |
Context: {context} | |
Question: {question} | |
""" | |
custom_rag_prompt = ChatPromptTemplate.from_template(template) | |
# custom_rag_prompt = ChatPromptTemplate( | |
# template=template, input_variables=["context", "question"] | |
# ) | |
llm = ChatOpenAI(model_name="gpt-3.5-turbo", temperature=0.2) # Increasing the temperature, the model becomes more creative and takes longer for inference. | |
# Construct a chain to answer questions on our data: | |
# 1. Weaviate Vector Search as the retriever to search for documents that are used as context by the LLM. | |
# 2. The prompt template that we constructed. | |
# 3. OpenAI's/Hugging Face chat model as the LLM used to generate a context-aware response. | |
# Regular chain format: chain = prompt | model | output_parser | |
rag_chain = ( | |
{"context": retriever, "question": RunnablePassthrough()} | |
| custom_rag_prompt | |
| llm | |
# | hf | |
| StrOutputParser() | |
) | |
# First TXT file. | |
return rag_chain.invoke(query) | |
ApexAiming = gr.Interface(fn=AIAimAdvice, inputs=["text"], outputs="text", title="Apex Aiming advice") | |
ApexAiming.launch() |