import gradio as gr from transformers import pipeline import requests import os import pandas as pd import time from dotenv import load_dotenv from textblob import TextBlob load_dotenv() hf_token = os.getenv("HF_TOKEN") # ========================= # ENV VARIABLES (use HF Secrets) # ========================= client_id = "sb-cap1-3c4588e0trial-dev!t617058" client_secret = "acbe78be-ead5-4b12-b3b4-32fdb27d0f5f$hFj-hDXxwHkNHC-CAvv-OKSr3KH96nLL4KqwIg7M8D8=" token_url = "https://3c4588e0trial.authentication.us10.hana.ondemand.com/oauth/token" cap_service_url_customers = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/Customers?$top=2" cap_service_url_products = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/Products?$top=2" cap_service_url_saleorders = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/SalesOrders?$top=2" cap_service_url_saleorderitems = "https://3c4588e0trial-dev-cap1-srv.cfapps.us10-001.hana.ondemand.com/odata/v4/sales/SalesOrderItems?$top=2" # ========================= # GLOBAL VARIABLES # ========================= access_token = None cached_customers = None cached_products = None cached_salesorders = None cached_salesorderitems = None last_refresh = 0 # ========================= # LOAD MODEL (once) # ========================= print("Loading model...") pipe = pipeline( "text-generation", # model="Qwen/Qwen2.5-0.5B-Instruct", model="Qwen/Qwen2.5-1.5B-Instruct", device="cpu", torch_dtype="auto" ) # ========================= # TOKEN FUNCTION # ========================= def generate_sap_xsuaa_token(): global access_token print("Generating SAP token...") auth_response = requests.post( token_url, data={"grant_type": "client_credentials"}, auth=(client_id, client_secret) ) if auth_response.status_code != 200: print("Token Error:", auth_response.text) return None access_token = auth_response.json().get("access_token") print("Token generated!") return access_token # ========================= # FETCH SAP DATA # ========================= def fetch_sap_data(): global access_token if not access_token: generate_sap_xsuaa_token() headers = { "Authorization": f"Bearer {access_token}", "Accept": "application/json" } res1 = requests.get(cap_service_url_customers, headers=headers) res2 = requests.get(cap_service_url_products, headers=headers) res3 = requests.get(cap_service_url_saleorders, headers=headers) res4 = requests.get(cap_service_url_saleorderitems, headers=headers) # Retry if token expired if res1.status_code in [400,401,403]: print("Token expired. Regenerating...") access_token = None generate_sap_xsuaa_token() headers["Authorization"] = f"Bearer {access_token}" res1 = requests.get(cap_service_url_customers, headers=headers) res2 = requests.get(cap_service_url_products, headers=headers) res3 = requests.get(cap_service_url_saleorders, headers=headers) res4 = requests.get(cap_service_url_saleorderitems, headers=headers) df_customers = pd.DataFrame(res1.json()["value"]) df_products = pd.DataFrame(res2.json()["value"]) df_saleorders = pd.DataFrame(res3.json()["value"]) df_saleorderitems = pd.DataFrame(res4.json()["value"]) # Keep only important columns df_customers = df_customers[["ID","name","country","industry"]] df_products = df_products[["ID","name","category","price","currency"]] df_saleorders = df_saleorders[["ID","customer_ID","orderDate","status"]] df_saleorderitems = df_saleorderitems[["ID","parent_ID","product_ID","quantity","netAmount"]] return df_customers, df_products, df_saleorders, df_saleorderitems # ========================= # CACHE LOGIC # ========================= def get_cached_data(): global cached_customers, cached_products,cached_salesorders,cached_salesorderitems, last_refresh # Refresh every 5 minutes if time.time() - last_refresh > 3000 or cached_customers is None: print("Refreshing SAP data...") cached_customers, cached_products,cached_salesorders,cached_salesorderitems = fetch_sap_data() last_refresh = time.time() return cached_customers, cached_products,cached_salesorders,cached_salesorderitems # ========================= # MAIN FUNCTION (LLM) # ========================= # def generate_response(user_prompt): # try: # # Get cached SAP data # df_customers, df_products, df_saleorders, df_saleorderitems = get_cached_data() # # Reduce size (IMPORTANT) # # customers_text = str(df_customers)[:500] # # products_text = str(df_products)[:500] # # saleorders_text = str(df_saleorders)[:500] # # saleorderitems_text = str(df_saleorderitems)[:500] # customers_text = df_customers.to_string(index=False) # products_text = df_products.to_string(index=False) # saleorders_text = df_saleorders.to_string(index=False) # saleorderitems_text = df_saleorderitems.to_string(index=False) # # Build prompt # system_prompt = f""" # You are an intelligent Corporate SAP Assistant bot. # Your sole purpose is to answer the user's questions based strictly on the database records provided to you. # Customers Data: {customers_text} # Products Data: {products_text} # Sale orders Data: {saleorders_text} # Sale order items Data: {saleorderitems_text} # CRITICAL RULES: # 1. NO HALLUCINATIONS: You must base your answer ONLY on the data provided. # 2. MISSING DATA: If the provided data does not contain the answer, do not guess. Say: "I could not find that information in the current SAP database." # 3. FORMATTING: You must output your response in Markdown. Use bold text for important nouns and bullet points for lists to make it easy to read. # 4. TONE: Be concise, highly professional, and helpful. # """ # prompt = f""" # {system_prompt} # User: {user_prompt} # Assistant: # """ # # Generate response # result = pipe( # prompt, # max_new_tokens=100, # # temperature=0.3, # Temperature controls randomness # do_sample=False, # controls HOW the next word is selected. If False then Always picks most probable next word and No randomness in answer. And if you use True then Picks from multiple possible words # # repetition_penalty=1.1, # return_full_text=False # If return_full_text=False ensure that the model output contains only the newly generated text. # ) # generated_text = result[0]["generated_text"] # # Clean output # response = generated_text.replace(prompt, "").strip() # return response # except Exception as e: # return f"Error: {str(e)}" def generate_response(user_prompt): try: # Get cached SAP data df_customers, df_products, df_saleorders, df_saleorderitems = get_cached_data() # Reduce size customers_text = df_customers.to_string(index=False) products_text = df_products.to_string(index=False) saleorders_text = df_saleorders.to_string(index=False) saleorderitems_text = df_saleorderitems.to_string(index=False) # Build system prompt system_prompt = f""" Your purpose is to answer the user's questions based strictly on the database records provided to you. Customers Data: {customers_text} Products Data: {products_text} Sale orders Data: {saleorders_text} Sale order items Data: {saleorderitems_text} CRITICAL RULES: 1. NO HALLUCINATIONS: You must base your answer ONLY on the data provided. 2. MISSING DATA: If the provided data does not contain the answer, do not guess. Say: "I could not find that information in the current SAP database." 3. FORMATTING: You must output your response in Markdown. Use bold text for important nouns and bullet points for lists to make it easy to read. 4. TONE: Be concise, highly professional, and helpful. """ user_corrected_prompt = str(TextBlob(user_prompt).correct()) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_corrected_prompt} ] prompt = pipe.tokenizer.apply_chat_template( messages, tokenize=False, add_generation_prompt=True ) # Generate response result = pipe( prompt, max_new_tokens=100, do_sample=True, temperature = 0.3, top_k = 3, top_p = 0.7, # use_cache = True, return_full_text=False ) generated_text = result[0]["generated_text"] # Clean output and strip any leftover end-of-turn tokens response = generated_text.replace(prompt, "").replace("<|im_end|>", "").strip() return response except Exception as e: return f"Error: {str(e)}" # ========================= # GRADIO UI + API # ========================= with gr.Blocks() as demo: user_input = gr.Textbox(label="User Question") output = gr.Textbox(label="Response") btn = gr.Button("Generate") btn.click( fn=generate_response, inputs=[user_input], outputs=output, api_name="predict" ) # REQUIRED for API exposure demo.queue() demo.launch()