|
import os |
|
from dotenv import load_dotenv |
|
from langchain_community.vectorstores import Qdrant |
|
from langchain_huggingface import HuggingFaceEmbeddings |
|
from langchain.prompts import ChatPromptTemplate |
|
from langchain.schema.runnable import RunnablePassthrough |
|
from langchain.schema.output_parser import StrOutputParser |
|
from qdrant_client import QdrantClient, models |
|
from langchain_openai import ChatOpenAI |
|
import gradio as gr |
|
import logging |
|
|
|
|
|
logging.basicConfig(level=logging.INFO) |
|
logger = logging.getLogger(__name__) |
|
|
|
|
|
load_dotenv() |
|
|
|
|
|
HF_TOKEN = os.getenv("HF_TOKEN") |
|
if not HF_TOKEN: |
|
logger.error("HF_TOKEN is not set in the environment variables.") |
|
exit(1) |
|
|
|
|
|
embeddings = HuggingFaceEmbeddings(model_name="BAAI/bge-large-en-v1.5") |
|
|
|
|
|
try: |
|
client = QdrantClient( |
|
url=os.getenv("QDRANT_URL"), |
|
api_key=os.getenv("QDRANT_API_KEY"), |
|
prefer_grpc=True |
|
) |
|
except Exception as e: |
|
logger.error("Failed to connect to Qdrant. Ensure QDRANT_URL and QDRANT_API_KEY are correctly set.") |
|
exit(1) |
|
|
|
|
|
collection_name = "mawared" |
|
|
|
|
|
try: |
|
client.create_collection( |
|
collection_name=collection_name, |
|
vectors_config=models.VectorParams( |
|
size=768, |
|
distance=models.Distance.COSINE |
|
) |
|
) |
|
logger.info(f"Created new collection: {collection_name}") |
|
except Exception as e: |
|
if "already exists" in str(e): |
|
logger.info(f"Collection {collection_name} already exists, continuing...") |
|
else: |
|
logger.error(f"Error creating collection: {e}") |
|
exit(1) |
|
|
|
|
|
db = Qdrant( |
|
client=client, |
|
collection_name=collection_name, |
|
embeddings=embeddings, |
|
) |
|
|
|
|
|
retriever = db.as_retriever( |
|
search_type="similarity", |
|
search_kwargs={"k": 5} |
|
) |
|
|
|
|
|
llm = ChatOpenAI( |
|
base_url="https://api-inference.huggingface.co/v1/", |
|
temperature=0, |
|
api_key=HF_TOKEN, |
|
model="meta-llama/Llama-3.3-70B-Instruct" |
|
) |
|
|
|
|
|
template = """ |
|
You are an expert assistant specializing in the Mawared HR System. Your task is to provide accurate and contextually relevant answers strictly based on the provided context. If the context lacks sufficient information, ask targeted clarifying questions to gather specific details required for a precise response. Always ensure clarity, brevity, and relevance in your answers. |
|
Context: |
|
{context} |
|
Question: |
|
{question} |
|
Answer: |
|
""" |
|
|
|
prompt = ChatPromptTemplate.from_template(template) |
|
|
|
|
|
rag_chain = ( |
|
{"context": retriever, "question": RunnablePassthrough()} |
|
| prompt |
|
| llm |
|
| StrOutputParser() |
|
) |
|
|
|
|
|
def ask_question_gradio(question): |
|
try: |
|
response = "" |
|
for chunk in rag_chain.stream(question): |
|
response += chunk |
|
return response |
|
except Exception as e: |
|
logger.error(f"Error during question processing: {e}") |
|
return "An error occurred. Please try again later." |
|
|
|
|
|
iface = gr.Interface( |
|
fn=ask_question_gradio, |
|
inputs=gr.Textbox(label="Ask a question about Mawared HR System:"), |
|
outputs=gr.Textbox(label="Answer:"), |
|
title="Mawared HR Assistant", |
|
description="Ask questions about the Mawared HR system, and this assistant will provide answers based on the available context." |
|
) |
|
|
|
|
|
if __name__ == "__main__": |
|
iface.launch() |