Spaces:
Sleeping
Sleeping
File size: 3,707 Bytes
7268351 979706a 7268351 be89ae1 8d3fcda b83a640 be89ae1 7268351 be89ae1 7268351 be89ae1 f5b718b 7268351 be89ae1 7268351 5a94c8e f763dd0 7268351 8d3fcda 7268351 5a94c8e f763dd0 8d3fcda 5a94c8e f763dd0 5a94c8e f763dd0 8d3fcda 979706a 8d3fcda 6e2dc41 7268351 5a94c8e 979706a 8d3fcda 979706a 8d3fcda 6e2dc41 8d3fcda e332fa0 8d3fcda 979706a 8d3fcda 979706a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 |
import os
import re
import streamlit as st
import google.generativeai as genai
from transformers import pipeline, AutoModelForSequenceClassification, AutoTokenizer
from db import get_entry_by_index
# Configure Gemini API key
GEMINI_API_KEY = os.getenv("gemini_api")
if GEMINI_API_KEY:
genai.configure(api_key=GEMINI_API_KEY)
else:
st.error("⚠️ Google API key is missing! Set it in Hugging Face Secrets.")
# Load pre-trained sentiment analysis model
MODEL_NAME = "cardiffnlp/twitter-roberta-base-sentiment"
try:
tokenizer = AutoTokenizer.from_pretrained(MODEL_NAME)
model = AutoModelForSequenceClassification.from_pretrained(MODEL_NAME)
sentiment_pipeline = pipeline("sentiment-analysis", model=model, tokenizer=tokenizer)
except Exception as e:
st.error(f"❌ Error loading sentiment model: {e}")
# Load Topic Extraction Model
try:
topic_pipeline = pipeline("zero-shot-classification", model="facebook/bart-large-mnli")
except Exception as e:
st.error(f"❌ Error loading topic extraction model: {e}")
# Predefined topic labels for classification
TOPIC_LABELS = [
"Technology", "Politics", "Business", "Sports", "Entertainment",
"Health", "Science", "Education", "Finance", "Travel", "Food"
]
def analyze_sentiment(text):
try:
result = sentiment_pipeline(text)[0]
label = result['label']
score = result['score']
mapping = {"LABEL_0": "Negative", "LABEL_1": "Neutral", "LABEL_2": "Positive"}
return mapping.get(label, "Unknown"), score
except Exception as e:
return f"Error analyzing sentiment: {e}", None
def extract_topic(text):
try:
result = topic_pipeline(text, TOPIC_LABELS)
top_topic = result["labels"][0]
confidence = result["scores"][0]
return top_topic, confidence
except Exception as e:
return f"Error extracting topic: {e}", None
# Detect queries like "data entry 1" or "entry 3" (case-insensitive)
def is_entry_query(prompt):
pattern = r"(?:data entry|entry)\s*(\d+)"
match = re.search(pattern, prompt, re.IGNORECASE)
if match:
# Convert to index (assuming user numbering starts at 1)
index = int(match.group(1)) - 1
return True, index
return False, None
def chatbot_response(user_prompt):
if not user_prompt:
return None, None, None, None, None
try:
entry_query, index = is_entry_query(user_prompt)
if entry_query:
entry = get_entry_by_index(index)
if entry is None:
return "❌ No entry found for the requested index.", None, None, None, None
entry_text = entry.get("text", "No text available.")
# Fixed AI response for entry queries (as per instructions)
ai_response_text = "Let's break down this tweet-like MongoDB entry:"
# Analyze the entry's text
sentiment_label, sentiment_confidence = analyze_sentiment(entry_text)
topic_label, topic_confidence = extract_topic(entry_text)
return ai_response_text, sentiment_label, sentiment_confidence, topic_label, topic_confidence
else:
# For non-entry queries, fallback to the generative model as usual.
model_gen = genai.GenerativeModel("gemini-1.5-pro")
ai_response = model_gen.generate_content(user_prompt)
sentiment_label, sentiment_confidence = analyze_sentiment(user_prompt)
topic_label, topic_confidence = extract_topic(user_prompt)
return ai_response.text, sentiment_label, sentiment_confidence, topic_label, topic_confidence
except Exception as e:
return f"❌ Error: {e}", None, None, None, None
|