|
from fastapi import FastAPI, UploadFile, File, HTTPException
|
|
from pydantic import BaseModel
|
|
from llama_index.core import StorageContext, load_index_from_storage, VectorStoreIndex, SimpleDirectoryReader, ChatPromptTemplate
|
|
from llama_index.llms.huggingface import HuggingFaceInferenceAPI
|
|
from llama_index.embeddings.huggingface import HuggingFaceEmbedding
|
|
from llama_index.core import Settings
|
|
import os
|
|
from dotenv import load_dotenv
|
|
import shutil
|
|
|
|
|
|
load_dotenv()
|
|
|
|
app = FastAPI()
|
|
|
|
|
|
Settings.llm = HuggingFaceInferenceAPI(
|
|
model_name="meta-llama/Meta-Llama-3-8B-Instruct",
|
|
tokenizer_name="meta-llama/Meta-Llama-3-8B-Instruct",
|
|
context_window=3900,
|
|
token=os.getenv("HF_TOKEN"),
|
|
max_new_tokens=1000,
|
|
generate_kwargs={"temperature": 0.5},
|
|
)
|
|
Settings.embed_model = HuggingFaceEmbedding(
|
|
model_name="BAAI/bge-small-en-v1.5"
|
|
)
|
|
|
|
|
|
PERSIST_DIR = "./db"
|
|
DATA_DIR = "data"
|
|
|
|
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
os.makedirs(PERSIST_DIR, exist_ok=True)
|
|
|
|
class Query(BaseModel):
|
|
question: str
|
|
|
|
def data_ingestion():
|
|
documents = SimpleDirectoryReader(DATA_DIR).load_data()
|
|
storage_context = StorageContext.from_defaults()
|
|
index = VectorStoreIndex.from_documents(documents)
|
|
index.storage_context.persist(persist_dir=PERSIST_DIR)
|
|
|
|
def handle_query(query):
|
|
storage_context = StorageContext.from_defaults(persist_dir=PERSIST_DIR)
|
|
index = load_index_from_storage(storage_context)
|
|
chat_text_qa_msgs = [
|
|
(
|
|
"user",
|
|
"""You are Q&A assistant named CHAT-DOC. Your main goal is to provide answers as accurately as possible, based on the instructions and context you have been given. If a question does not match the provided context or is outside the scope of the document, kindly advise the user to ask questions within the context of the document.
|
|
Context:
|
|
{context_str}
|
|
Question:
|
|
{query_str}
|
|
"""
|
|
)
|
|
]
|
|
text_qa_template = ChatPromptTemplate.from_messages(chat_text_qa_msgs)
|
|
query_engine = index.as_query_engine(text_qa_template=text_qa_template)
|
|
answer = query_engine.query(query)
|
|
|
|
if hasattr(answer, 'response'):
|
|
return answer.response
|
|
elif isinstance(answer, dict) and 'response' in answer:
|
|
return answer['response']
|
|
else:
|
|
return "Sorry, I couldn't find an answer."
|
|
|
|
@app.post("/upload")
|
|
async def upload_file(file: UploadFile = File(...)):
|
|
file_extension = os.path.splitext(file.filename)[1].lower()
|
|
if file_extension not in [".pdf", ".docx", ".txt"]:
|
|
raise HTTPException(status_code=400, detail="Invalid file type. Only PDF, DOCX, and TXT are allowed.")
|
|
|
|
file_path = os.path.join(DATA_DIR, file.filename)
|
|
with open(file_path, "wb") as buffer:
|
|
shutil.copyfileobj(file.file, buffer)
|
|
|
|
data_ingestion()
|
|
return {"message": "File uploaded and processed successfully"}
|
|
|
|
@app.post("/query")
|
|
async def query_document(query: Query):
|
|
if not os.listdir(DATA_DIR):
|
|
raise HTTPException(status_code=400, detail="No document has been uploaded yet.")
|
|
|
|
response = handle_query(query.question)
|
|
return {"response": response}
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
uvicorn.run(app, host="0.0.0.0", port=8000) |
