| import os |
| import uuid |
| import re |
| import string |
| import pandas as pd |
| import gradio as gr |
| from sentence_transformers import SentenceTransformer |
| import chromadb |
| from chromadb.config import Settings |
| from llama_cpp import Llama |
| import requests |
| from pathlib import Path |
|
|
|
|
| |
| MODEL_URL = "https://huggingface.co/datasets/psy7743/llama3-8b-instruct-Q8_0.gguf/resolve/main/llama3-8b-instruct-Q8_0.gguf" |
| MODEL_PATH = "llama3-8b-instruct-Q8_0.gguf" |
|
|
| |
| |
| |
|
|
| if not Path(MODEL_PATH).exists(): |
| print("📥 Downloading Meta LLaMA 3 8B Q8_0 model from tensorblock...") |
| response = requests.get(MODEL_URL, stream=True) |
| with open(MODEL_PATH, "wb") as f: |
| for chunk in response.iter_content(chunk_size=8192): |
| if chunk: |
| f.write(chunk) |
| print("✅ Download complete!") |
|
|
| |
|
|
| df = pd.read_csv("jupiter_faqs.csv") |
|
|
| def clean_text(text): |
| text = text.lower() |
| text = text.translate(str.maketrans('', '', string.punctuation)) |
| text = text.replace('\n', ' ').replace('\t', ' ') |
| return re.sub(r'\s+', ' ', text).strip() |
|
|
| df['clean_question'] = df['question'].apply(clean_text) |
| df['clean_answer'] = df['answer'].apply(clean_text) |
| df['document'] = df.apply(lambda row: f"Question: {row['clean_question']}\nAnswer: {row['clean_answer']}", axis=1) |
|
|
| |
|
|
| embedding_model = SentenceTransformer('all-mpnet-base-v2') |
| df['embedding'] = df['clean_question'].apply(lambda x: embedding_model.encode(x).tolist()) |
| df['uid'] = [str(uuid.uuid4()) for _ in range(len(df))] |
|
|
| |
|
|
| persist_dir = "chroma_qa_db" |
| chroma_client = chromadb.PersistentClient(path=persist_dir, settings=Settings()) |
| collection_name = "qa_collection" |
|
|
| if collection_name in [c.name for c in chroma_client.list_collections()]: |
| chroma_client.delete_collection(name=collection_name) |
|
|
| collection = chroma_client.get_or_create_collection(name=collection_name) |
|
|
| if len(collection.get()["ids"]) == 0: |
| collection.add( |
| documents=df['document'].tolist(), |
| embeddings=df['embedding'].tolist(), |
| ids=df['uid'].astype(str).tolist() |
| ) |
|
|
| |
|
|
| llm = Llama( |
| model_path=MODEL_PATH, |
| n_ctx=4096, |
| n_threads=8, |
| n_gpu_layers=-1, |
| ) |
|
|
| |
|
|
| def search_chroma(query, n_results=5): |
| query_embedding = embedding_model.encode(query).tolist() |
| results = collection.query( |
| query_embeddings=[query_embedding], |
| n_results=n_results, |
| include=['documents'] |
| ) |
| return results["documents"][0] |
|
|
| def generate_response(message, history=[]): |
| docs = search_chroma(message) |
| context_str = "\n\n".join(docs) |
|
|
| sys_prompt = """You are a helpful assistant. Based on the context and user query, provide an accurate and concise answer. |
| |
| Rules: |
| - If context is relevant, answer based on it. |
| - If context is insufficient, ask for clarification or say it's not possible. |
| - Do not hallucinate or make up answers. |
| - Keep the tone friendly.""" |
|
|
| prompt = f"{sys_prompt}\n\ncontext:\n{context_str}\n\nQuestion: {message}\nAnswer:" |
|
|
| response = llm( |
| prompt, |
| max_tokens=512, |
| temperature=0.3, |
| top_p=0.8, |
| stop=["Q:", "\n"], |
| echo=True |
| ) |
|
|
| return response["choices"][0]["text"].split("Answer:")[-1].strip() |
|
|
| |
|
|
| def gradio_chat_interface(message, history): |
| reply = generate_response(message) |
| history = history + [(message, reply)] |
| return history, history |
|
|
| demo = gr.ChatInterface( |
| fn=generate_response, |
| title="🦙 LLaMA-3 FAQ Chatbot", |
| chatbot=gr.Chatbot(label="Ask me anything about Jupiter Money!"), |
| examples=[ |
| "What is Jupiter Edge credit card?", |
| "What happens if I miss a payment?", |
| "How to change billing address?" |
| ], |
| cache_examples=False |
| ) |
|
|
| if __name__ == "__main__": |
| demo.launch(share=True) |
|
|
