app_small_bug.py

# This block contains the full combined script for testing.
# It includes all the code from the previous successful steps.

# Combined Imports
import spaces
import os
import gradio as gr
from huggingface_hub import InferenceClient
import torch
import re
import warnings
import time
import json
from transformers import AutoTokenizer, AutoModelForCausalLM, GenerationConfig, BitsAndBytesConfig
from sentence_transformers import SentenceTransformer, util, CrossEncoder
import gspread
# from google.colab import auth
from google.auth import default
from tqdm import tqdm
from duckduckgo_search import DDGS
import spacy
from datetime import date, timedelta
from dateutil.relativedelta import relativedelta # Corrected typo
import traceback # Import traceback
import base64 # Import base64

@spaces.GPU
def startup():
    print("GPU function registered for Hugging Face Spaces startup.")
    return "Ready"

startup()


# Suppress warnings
warnings.filterwarnings("ignore", category=UserWarning)

# Define global variables and load secrets
HF_TOKEN = os.getenv("HF_TOKEN")
SHEET_ID = "19ipxC2vHYhpXCefpxpIkpeYdI43a1Ku2kYwecgUULIw"
GOOGLE_BASE64_CREDENTIALS = os.getenv("GOOGLE_BASE64_CREDENTIALS")

# Initialize InferenceClient
client = InferenceClient("google/gemma-2-9b-it", token=HF_TOKEN)

# Initialize InferenceClient
client = InferenceClient("google/gemma-2-9b-it", token=HF_TOKEN)

# Load spacy model for sentence splitting
nlp = None
try:
    nlp = spacy.load("en_core_web_sm")
    print("SpaCy model 'en_core_web_sm' loaded.")
except OSError:
    print("SpaCy model 'en_core_web_sm' not found. Downloading...")
    try:
        os.system("python -m spacy download en_core_web_sm")
        nlp = spacy.load("en_core_web_sm")
        print("SpaCy model 'en_core_web_sm' downloaded and loaded.")
    except Exception as e:
        print(f"Failed to download or load SpaCy model: {e}")


# Load SentenceTransformer for RAG/business info retrieval
embedder = None
try:
    print("Attempting to load Sentence Transformer (sentence-transformers/paraphrase-MiniLM-L6-v2)...")
    embedder = SentenceTransformer("sentence-transformers/paraphrase-MiniLM-L6-v2")
    print("Sentence Transformer loaded.")
except Exception as e:
     print(f"Error loading Sentence Transformer: {e}")


# Load a Cross-Encoder model for re-ranking retrieved documents
reranker = None
try:
    print("Attempting to load Cross-Encoder Reranker (cross-encoder/ms-marco-MiniLM-L6-v2)...")
    reranker = CrossEncoder('cross-encoder/ms-marco-MiniLM-L6-v2')
    print("Cross-Encoder Reranker loaded.")
except Exception as e:
    print(f"Error loading Cross-Encoder Reranker: {e}")
    print("Please ensure the model identifier 'cross-encoder/ms-marco-MiniLM-L6-v2' is correct and accessible on Hugging Face Hub.")
    print(traceback.format_exc())
    reranker = None
# This block contains the full combined script for testing.
# This block contains the full combined script for testing.
# It includes all the code from the previous successful steps.


# Google Sheets Authentication
gc = None # Global variable for gspread client
def authenticate_google_sheets():
    """Authenticates with Google Sheets using base64 encoded credentials."""
    global gc
    print("Authenticating Google Account...")
    if not GOOGLE_BASE64_CREDENTIALS:
        print("Error: GOOGLE_BASE64_CREDENTIALS secret not found.")
        return False

    try:
        # Decode the base64 credentials
        credentials_json = base64.b64decode(GOOGLE_BASE64_CREDENTIALS).decode('utf-8')
        credentials = json.loads(credentials_json)

        # Authenticate using service account from dictionary
        gc = gspread.service_account_from_dict(credentials)
        print("Google Sheets authentication successful via service account.")
        return True
    except Exception as e:
        print(f"Google Sheets authentication failed: {e}")
        print("Please ensure your GOOGLE_BASE64_CREDENTIALS secret is correctly set and contains valid service account credentials.")
        print(traceback.format_exc())
        return False

# Google Sheets Data Loading and Embedding
# business_data = [] # Global variable to store loaded data - This was intended to be global, but needs to be named 'data' to match usage
data = [] # Global variable to store loaded data - Renamed to 'data'
descriptions_for_embedding = []
embeddings = torch.tensor([])
business_info_available = False # Flag to indicate if business info was loaded successfully

def load_business_info():
    """Loads business information from Google Sheet and creates embeddings."""
    global data, descriptions_for_embedding, embeddings, business_info_available # Added 'data' to global
    business_info_available = False # Reset flag

    if gc is None:
        print("Skipping Google Sheet loading: Google Sheets client not authenticated.")
        return

    if not SHEET_ID:
         print("Error: SHEET_ID not set.")
         return

    try:
        sheet = gc.open_by_key(SHEET_ID).sheet1
        print(f"Successfully opened Google Sheet with ID: {SHEET_ID}")
        data_records = sheet.get_all_records()

        if not data_records:
            print(f"Warning: No data records found in Google Sheet with ID: {SHEET_ID}")
            data = [] # Use the global 'data'
            descriptions_for_embedding = []
        else:
            # Filter out rows missing 'Service' or 'Description'
            filtered_data = [row for row in data_records if row.get('Service') and row.get('Description')]
            if not filtered_data:
                print("Warning: Filtered data is empty after checking for 'Service' and 'Description'.")
                data = [] # Use the global 'data'
                descriptions_for_embedding = []
            else:
                data = filtered_data # Assign to the global 'data'
                # Use BOTH Service and Description for embedding
                descriptions_for_embedding = [f"Service: {row['Service']}. Description: {row['Description']}" for row in data]

                # Only encode if descriptions_for_embedding are found and embedder is available
                if descriptions_for_embedding and embedder is not None:
                    print("Encoding descriptions...")
                    try:
                        embeddings = embedder.encode(descriptions_for_embedding, convert_to_tensor=True)
                        print("Encoding complete.")
                        business_info_available = True # Set flag if successful
                    except Exception as e:
                        print(f"Error during description encoding: {e}")
                        embeddings = torch.tensor([]) # Ensure embeddings is an empty tensor on error
                        business_info_available = False # Encoding failed
                else:
                    print("Skipping encoding descriptions: No descriptions found or embedder not available.")
                    embeddings = torch.tensor([]) # Ensure embeddings is an empty tensor
                    business_info_available = False # Cannot use RAG without descriptions or embedder

        print(f"Loaded {len(descriptions_for_embedding)} entries from Google Sheet for embedding/RAG.")
        if not business_info_available:
            print("Business information retrieval (RAG) is NOT available.")

    except gspread.exceptions.SpreadsheetNotFound:
        print(f"Error: Google Sheet with ID '{SHEET_ID}' not found.")
        print("Please check the SHEET_ID and ensure your authenticated Google Account has access to this sheet.")
        business_info_available = False # Sheet not found
    except Exception as e:
        print(f"An error occurred while accessing the Google Sheet: {e}")
        print(traceback.format_exc())
        business_info_available = False # Other sheet access error

# Business Info Retrieval (RAG)
def retrieve_business_info(query: str, top_n: int = 3) -> list:
    """
    Retrieves relevant business information from loaded data based on a query.

    Args:
        query: The user's query string.
        top_n: The number of top relevant entries to retrieve.

    Returns:
        A list of dictionaries, where each dictionary is a relevant row from the
        Google Sheet data. Returns an empty list if RAG is not available or
        no relevant information is found.
    """
    # Access the global 'data' variable
    global data
    if not business_info_available or embedder is None or not descriptions_for_embedding or not data: # Added check for data
        print("Business information retrieval is not available or data is empty.")
        return []

    try:
        # Compute the embedding for the query
        query_embedding = embedder.encode(query, convert_to_tensor=True)

        # Compute cosine similarity between the query embedding and all description embeddings
        cosine_scores = util.cos_sim(query_embedding, embeddings)[0]

        # Get the top N indices based on cosine similarity
        top_results_indices = torch.topk(cosine_scores, k=min(top_n, len(data)))[1].tolist() # Use len(data)

        # Retrieve the actual data entries corresponding to the top indices
        top_results = [data[i] for i in top_results_indices] # Use data[i]

        # Optional: Re-rank the top results using the Cross-Encoder
        if reranker is not None and top_results:
            print("Re-ranking top results...")
            # Create pairs of (query, description) for the Cross-Encoder
            rerank_pairs = [(query, descriptions_for_embedding[i]) for i in top_results_indices]
            rerank_scores = reranker.predict(rerank_pairs)

            # Sort the top results based on the re-ranker scores
            reranked_indices = sorted(range(len(rerank_scores)), key=lambda i: rerank_scores[i], reverse=True)
            reranked_results = [top_results[i] for i in reranked_indices]
            print("Re-ranking complete.")
            return reranked_results
        else:
            return top_results

    except Exception as e:
        print(f"Error during business information retrieval: {e}")
        print(traceback.format_exc())
        return []


# Function to perform DuckDuckGo Search and return results with URLs
def perform_duckduckgo_search(query: str, max_results: int = 5):
    """
    Performs a search using DuckDuckGo and returns a list of dictionaries.
    Includes a delay to avoid rate limits.
    Returns an empty list and prints an error if search fails.
    """
    print(f"Executing Tool: perform_duckduckgo_search with query='{query}')")
    search_results_list = []
    try:
        # Add a delay before each search
        time.sleep(1) # Sleep for 1 second

        with DDGS() as ddgs:
            if not query or len(query.split()) < 2:
                 print(f"Skipping search for short query: '{query}'")
                 return []

            # Use text() method for general text search
            results_generator = ddgs.text(query, max_results=max_results)
            results_found = False
            for r in results_generator:
                search_results_list.append(r)
                results_found = True

            if not results_found and max_results > 0:
                 print(f"DuckDuckGo search for '{query}' returned no results.")


    except Exception as e:
        print(f"Error during Duckduckgo search for '{query}': {e}")
        return []

    return search_results_list

# Function to perform date calculation if needed
def perform_date_calculation(query: str):
    """
    Analyzes query for date calculation requests and performs the calculation.
    Returns a dict describing the calculation and result, or None.
    Handles formats like 'X days ago', 'X days from now', 'X weeks ago', 'X weeks from now', 'what is today's date'.
    Uses dateutil for slightly more flexibility (though core logic remains simple).
    """
    print(f"Executing Tool: perform_date_calculation with query='{query}')")
    query_lower = query.lower()
    today = date.today()
    result_date = None
    calculation_description = None

    if re.search(r"\btoday'?s date\b|what is today'?s date\b|what day is it\b", query_lower):
        result_date = today
        calculation_description = f"The current date is: {today.strftime('%Y-%m-%d')}"
        print(f"Identified query for today's date.")
        return {"query": query, "description": calculation_description, "result": result_date.strftime('%Y-%m-%d'), "success": True}

    match = re.search(r"(\d+)\s+(day|week|month|year)s?\s+(ago|from now)", query_lower)

    if match:
        value = int(match.group(1))
        unit = match.group(2)
        direction = match.group(3)

        try:
            if unit == 'day':
                delta = timedelta(days=value)
            elif unit == 'week':
                delta = timedelta(weeks=value)
            elif unit == 'month':
                 delta = relativedelta(months=value)
            elif unit == 'year':
                 delta = relativedelta(years=value)
            else:
                 desc = f"Could not understand the time unit '{unit}' in '{query}'."
                 print(desc)
                 return {"query": query, "description": desc, "result": None, "success": False, "error": desc}

            if direction == 'ago':
                result_date = today - delta
                calculation_description = f"Calculating date {value} {unit}s ago from {today.strftime('%Y-%m-%d')}: {result_date.strftime('%Y-%m-%d')}"
            elif direction == 'from now':
                result_date = today + delta
                calculation_description = f"Calculating date {value} {unit}s from now from {today.strftime('%Y-%m-%d')}: {result_date.strftime('%Y-%m-%d')}"

            print(f"Performed date calculation: {calculation_description}")
            return {"query": query, "description": calculation_description, "result": result_date.strftime('%Y-%m-%d'), "success": True}

        except OverflowError:
             desc = f"Date calculation overflow for query: {query}"
             print(f"Date calculation overflow for query: {query}")
             return {"query": query, "description": desc, "result": None, "success": False, "error": desc}
        except Exception as e:
             desc = f"An error occurred during date calculation for query '{query}': {e}"
             print(desc)
             return {"query": query, "description": desc, "result": None, "success": False, "error": str(e)}

    desc = "No specific date calculation pattern recognized."
    print(f"No specific date calculation pattern found in query: '{query}'")
    return {"query": query, "description": desc, "result": None, "success": False}


# ──────────────────────────
# 2  Chat handler
# ──────────────────────────
def respond(
    message: str,
    history: list[tuple[str, str]],
    system_message: str,
    max_tokens: int,
    temperature: float,
    top_p: float,
):
    # Retrieve relevant business information based on the user's message
    retrieved_info = retrieve_business_info(message)

    # Build ChatML conversation
    messages = [{"role": "system", "content": system_message}]

    # Include retrieved information as context if available
    if retrieved_info:
        # Modified context formatting
        context_message = "Use the following business information to help answer the user's question if relevant:\n"
        for i, info in enumerate(retrieved_info):
            # Use a clear delimiter between entries
            context_message += f"--- Business Info Entry {i+1} ---\n"
            # Include all key-value pairs from the dictionary
            for key, value in info.items():
                 # Ensure values are strings
                 context_message += f"{key}: {str(value)}\n"
            context_message += "---\n" # Delimiter after each entry
        # Add the formatted context as a user message right after the initial system message
        # This format might help the model see it as explicit information provided for the current turn
        messages.append({"role": "user", "content": context_message})
        print("Added retrieved business info to messages in a new format.") # Debug print


    # Add conversation history
    for user_msg, bot_msg in history:
        if user_msg:
            messages.append({"role": "user", "content": user_msg})
        if bot_msg:
            messages.append({"role": "assistant", "content": bot_msg})

    # Add the current user message
    messages.append({"role": "user", "content": message})

    # Stream tokens
    response = ""
    try:
        for chunk in client.chat_completion(
            messages=messages,
            max_tokens=max_tokens,
            stream=True,
            temperature=temperature,
            top_p=top_p,
        ):
            token = chunk.choices[0].delta.content or ""
            response += token
            yield response
    except Exception as e:
        print(f"Error during chat completion: {e}")
        print(traceback.format_exc())
        yield f"An error occurred: {e}"


# ──────────────────────────
# 3  Gradio interface
# ──────────────────────────

# The Gradio interface definition remains the same as it correctly
# uses the updated respond function.

print(f"RAG functionality available: {business_info_available}")

demo = gr.ChatInterface(
    fn=respond,
    additional_inputs=[
        gr.Textbox(value="You are a friendly Chatbot. Use the provided business information to answer questions when relevant.", label="System message"),
        gr.Slider(1, 2048, value=512, step=1, label="Max new tokens"),
        gr.Slider(0.1, 4.0, value=0.7, step=0.1, label="Temperature"),
        gr.Slider(0.1, 1.0, value=0.95, step=0.05, label="Top‑p (nucleus sampling)"),
    ],
    title="Gemma‑2‑9B‑IT Chat with RAG",
    description="Chat with Google Gemma‑2‑9B‑IT via Hugging Face Inference API, with business info retrieved from Google Sheets.",
)

# Enable request queueing (concurrency handled automatically on Gradio ≥ 4)
demo.queue()

if __name__ == "__main__":
    # Authenticate and load data before launching the demo
    if authenticate_google_sheets():
        load_business_info()
    else:
        print("Google Sheets authentication failed. RAG functionality will not be available.")

    # The print statement for RAG status is added here, before launching the demo.
    print(f"RAG functionality available: {business_info_available}")

    demo.launch()