import os import gradio as gr import pandas as pd import plotly.express as px import plotly.graph_objects as go import boto3 import PyPDF2 import io import uuid import json import re import time import numpy as np import pdfplumber import requests import faiss import pickle import PIL from PIL import Image import pytesseract from io import BytesIO from dotenv import load_dotenv from langchain.text_splitter import RecursiveCharacterTextSplitter from langchain_openai import OpenAIEmbeddings from langchain_community.vectorstores import FAISS from transformers import pipeline from textblob import TextBlob import base64 import imghdr from IPython.display import HTML, display # Load environment variables load_dotenv() # Global variables to store chat history and analytics data messages = [] product_images = [] current_product = "" query_counts = {"circuit breaker": 0, "motor starter": 0, "contactor": 0, "switch": 0, "relay": 0, "other": 0} daily_queries = [0, 0, 0, 0, 0, 6, 8, 10, 7, 9, 12, 15, 11, 14] # Mock data for chart # Path configurations VECTOR_DB_PATH = "vector_store" IMAGE_DB_PATH = "image_store" TEMP_IMAGES_DIR = os.path.join(os.getcwd(), 'temp_images') # Ensure directories exist os.makedirs(VECTOR_DB_PATH, exist_ok=True) os.makedirs(IMAGE_DB_PATH, exist_ok=True) os.makedirs(TEMP_IMAGES_DIR, exist_ok=True) # Initialize index to store document embeddings index = None index_metadata = {} faiss_index_path = os.path.join(VECTOR_DB_PATH, "faiss_index.bin") metadata_path = os.path.join(VECTOR_DB_PATH, "metadata.pkl") def init_openai_api(): """Initialize OpenAI API with API key from Colab secrets.""" try: openai_api_key = os.environ.get('OPENAI_API_KEY') if not openai_api_key: print("OPENAI_API_KEY not found in Colab secrets.") return False os.environ["OPENAI_API_KEY"] = openai_api_key print("OpenAI API initialized with API key from secrets.") # Key loaded print(f"OpenAI API key: {openai_api_key}") # This will print your key, so remove it after verification return True except Exception as e: print(f"Error initializing OpenAI API: {e}") return False def init_mistral_api(): """Initialize Mistral API with API key from Colab secrets.""" try: mistral_api_key = os.environ.get('MISTRAL_API_KEY') if not mistral_api_key: print("MISTRAL_API_KEY not found in Colab secrets.") return False os.environ["MISTRAL_API_KEY"] = mistral_api_key print("Mistral API initialized with API key from secrets.") # Key loaded print(f"Mistral API key: {mistral_api_key}") # This will print your key, so remove it after verification return True except Exception as e: print(f"Error initializing Mistral API: {e}") return False # Initialize embedding model def get_embeddings_model(): """Initialize the OpenAI embeddings model for vector generation""" try: embeddings = OpenAIEmbeddings( model="text-embedding-3-small", openai_api_key=os.getenv("OPENAI_API_KEY") ) return embeddings except Exception as e: print(f"Error initializing embeddings model: {e}") return None # Initialize AWS S3 client for accessing product catalogs def init_s3_client(): """Initialize S3 client for accessing product catalogs""" try: s3_client = boto3.client( 's3', aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"), aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"), region_name=os.getenv("AWS_REGION", "ap-south-1") ) return s3_client except Exception as e: print(f"Error initializing S3 client: {e}") return None # Initialize sentiment analysis def init_sentiment_analyzer(): """Initialize sentiment analysis pipeline""" try: # Use TextBlob for simpler sentiment analysis # Returning a function that can be used later return lambda text: TextBlob(text).sentiment.polarity except Exception as e: print(f"Error initializing sentiment analyzer: {e}") return None # Initialize FAISS index def init_faiss_index(): """Initialize FAISS index for vector search""" global index, index_metadata # Check if index already exists if os.path.exists(faiss_index_path) and os.path.exists(metadata_path): try: # Load existing index index = faiss.read_index(faiss_index_path) with open(metadata_path, 'rb') as f: index_metadata = pickle.load(f) print(f"Loaded existing FAISS index with {index.ntotal} vectors") return True except Exception as e: print(f"Error loading existing FAISS index: {e}") # Create new index try: # Set dimension for OpenAI embedding vectors dimension = 1536 index = faiss.IndexFlatL2(dimension) index_metadata = {"ids": [], "texts": [], "product_types": []} # Save empty index faiss.write_index(index, faiss_index_path) with open(metadata_path, 'wb') as f: pickle.dump(index_metadata, f) print("Created new FAISS index") return True except Exception as e: print(f"Error creating FAISS index: {e}") return False # OCR function to extract text from images def extract_text_from_image(image_data): """Use OCR to extract text from images""" try: # Convert bytes to PIL Image image = Image.open(BytesIO(image_data)) # This line was already correct # Make sure the image is in a supported format (e.g., RGB) image = image.convert('RGB') # Add this conversion step # Perform OCR text = pytesseract.image_to_string(image) return text except Exception as e: print(f"Error extracting text from image: {e}") return "" # Extract images from PDFs and store locally def extract_images_from_pdf(pdf_content, product_type): """Extract images from PDF using pdfplumber and store them in local storage""" try: # Create a BytesIO object from the PDF content pdf_file = io.BytesIO(pdf_content) # Open the PDF with pdfplumber with pdfplumber.open(pdf_file) as pdf: images_stored = 0 # Iterate through each page for page_num, page in enumerate(pdf.pages): # Extract images from the page for img_index, img in enumerate(page.images): # Get image data image_bytes = img["stream"].get_data() # Skip small images if len(image_bytes) < 5000: continue # Extract text using OCR ocr_text = extract_text_from_image(image_bytes) # Generate a unique ID for the image image_id = str(uuid.uuid4()) # Create metadata metadata = { "product_type": product_type, "page_number": page_num, "image_index": img_index, "timestamp": time.time(), "image_size": len(image_bytes), "mime_type": "jpg", "ocr_text": ocr_text } # Save image to disk image_path = os.path.join(IMAGE_DB_PATH, f"{image_id}.jpg") with open(image_path, 'wb') as f: f.write(image_bytes) # Save metadata separately metadata_path = os.path.join(IMAGE_DB_PATH, f"{image_id}.json") with open(metadata_path, 'w') as f: json.dump(metadata, f) images_stored += 1 return images_stored except Exception as e: print(f"Error extracting images from PDF: {e}") return 0 # Store text chunks with embeddings in FAISS def store_chunks_in_faiss(chunks, product_type, embeddings_model): """Store text chunks with embeddings in FAISS""" global index, index_metadata if not index or not embeddings_model: print("FAISS index or embeddings model not initialized") return 0 try: chunks_stored = 0 batch_vectors = [] batch_ids = [] batch_texts = [] batch_product_types = [] # Process chunks in batches for chunk in chunks: # Skip empty chunks if not chunk.strip(): continue # Generate embedding embedding_vector = embeddings_model.embed_query(chunk) # Convert to numpy array with proper shape embedding_np = np.array(embedding_vector, dtype='float32').reshape(1, -1) # Add to batch batch_vectors.append(embedding_np) batch_ids.append(len(index_metadata["ids"])) batch_texts.append(chunk) batch_product_types.append(product_type) chunks_stored += 1 # Add all vectors to the index if batch_vectors: # Combine all vectors into a single array vectors_to_add = np.vstack(batch_vectors) # Add to FAISS index index.add(vectors_to_add) # Update metadata index_metadata["ids"].extend(batch_ids) index_metadata["texts"].extend(batch_texts) index_metadata["product_types"].extend(batch_product_types) # Save updated index and metadata faiss.write_index(index, faiss_index_path) with open(metadata_path, 'wb') as f: pickle.dump(index_metadata, f) return chunks_stored except Exception as e: print(f"Error storing chunks in FAISS: {e}") return 0 # Function to download and process PDFs from S3 def process_pdf_catalogs(): """Download and process PDF catalogs from S3 bucket""" if not s3_client: print("S3 client not initialized, skipping PDF processing") return {"status": "error", "message": "S3 client not initialized"} embeddings_model = get_embeddings_model() if not embeddings_model: return {"status": "error", "message": "Embeddings model not initialized"} try: # Get list of PDF files in the bucket bucket_name = os.getenv("S3_BUCKET_NAME", "agent-product-discovery") response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix="ABB-catalog/") pdf_files = [obj['Key'] for obj in response.get('Contents', []) if obj['Key'].endswith('.pdf')] processed_chunks = 0 processed_images = 0 # Process each PDF file for pdf_file in pdf_files: # Determine product type from filename product_type = "other" for pt in ["circuit_breaker", "motor_starter", "contactor", "switch", "relay", "system", "softstarter", "ups"]: if pt in pdf_file.lower(): product_type = pt.replace("_", " ") break # Download PDF from S3 response = s3_client.get_object(Bucket=bucket_name, Key=pdf_file) pdf_content = response['Body'].read() # Process PDF text content pdf_reader = PyPDF2.PdfReader(io.BytesIO(pdf_content)) text_content = "" # Extract text from each page for page in pdf_reader.pages: text_content += page.extract_text() + "\n\n" # Split text into smaller chunks for efficient embedding text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, ) chunks = text_splitter.split_text(text_content) # Store chunks in FAISS chunks_stored = store_chunks_in_faiss(chunks, product_type, embeddings_model) # Extract and store images images_count = extract_images_from_pdf(pdf_content, product_type) processed_images += images_count processed_chunks += chunks_stored print(f"Processed {pdf_file}: {chunks_stored} text chunks and {images_count} images extracted") print(f"PDF processing complete: {len(pdf_files)} files, {processed_chunks} chunks, {processed_images} images") return { "status": "success", "files_processed": len(pdf_files), "chunks_processed": processed_chunks, "images_processed": processed_images } except Exception as e: print(f"Error processing PDF catalogs: {e}") return {"status": "error", "message": str(e)} # Process a PDF from a URL def process_pdf_from_url(url): """Download and process a PDF from a URL""" embeddings_model = get_embeddings_model() if not embeddings_model: return "Error: Embeddings model not initialized" try: # Download the PDF response = requests.get(url, stream=True) if response.status_code != 200: return f"Error downloading PDF: HTTP status code {response.status_code}" # Get the content pdf_content = response.content # Determine product type from URL or filename product_type = "other" for pt in ["circuit_breaker", "motor_starter", "contactor", "switch", "relay", "system", "softstarter", "ups"]: if pt in url.lower(): product_type = pt.replace("_", " ") break # Process PDF text content pdf_reader = PyPDF2.PdfReader(io.BytesIO(pdf_content)) text_content = "" # Extract text from each page for page in pdf_reader.pages: text_content += page.extract_text() + "\n\n" # Split text into smaller chunks for efficient embedding text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=200, length_function=len, ) chunks = text_splitter.split_text(text_content) # Store chunks in FAISS chunks_stored = store_chunks_in_faiss(chunks, product_type, embeddings_model) # Extract and store images images_count = extract_images_from_pdf(pdf_content, product_type) print(f"Processed PDF from URL: {url}: {chunks_stored} text chunks and {images_count} images extracted") return f"Successfully processed PDF from URL: {chunks_stored} chunks, {images_count} images" except Exception as e: print(f"Error processing PDF from URL: {e}") return f"Error processing PDF: {str(e)}" # Search for relevant product information in FAISS def search_vector_faiss(query, product_type=None, limit=5): """Search for relevant information in the FAISS vector database""" global index, index_metadata embeddings_model = get_embeddings_model() if not index or not embeddings_model: # Return empty results if index isn't available return [] try: # Generate embedding for the query query_embedding = embeddings_model.embed_query(query) query_vector = np.array([query_embedding], dtype='float32') # Search the index D, I = index.search(query_vector, k=limit*2) # Get more results for filtering # Filter by product type if specified results = [] for i, idx in enumerate(I[0]): if idx >= 0 and idx < len(index_metadata["ids"]): # Check if this product type matches the filter if not product_type or index_metadata["product_types"][idx] == product_type: results.append({ "id": index_metadata["ids"][idx], "product_type": index_metadata["product_types"][idx], "content": index_metadata["texts"][idx], "similarity": 1.0 - D[0][i] # Convert distance to similarity score }) # Exit if we have enough results if len(results) >= limit: break return results except Exception as e: print(f"Error searching FAISS index: {e}") return [] # Log query analytics def log_query_analytics(query, product_type, response_time, sentiment_score=0): """Log query analytics to a local file""" try: # Create a log entry log_entry = { "id": str(uuid.uuid4()), "query": query, "product_type": product_type, "timestamp": time.time(), "response_time": response_time, "sentiment_score": sentiment_score } # Append to log file with open("query_analytics.jsonl", "a") as f: f.write(json.dumps(log_entry) + "\n") except Exception as e: print(f"Error logging query analytics: {e}") # Get product images based on product type def get_images_html(): """Generate HTML to display all images from Hugging Face Space.""" # List of image filenames you've uploaded to Hugging Face image_files = [ "page10_img1_af0cd090.jpeg", "page10_img2_b92432e6.jpeg", "page10_img3_f853b5ed.jpeg", "page10_img4_c010804f.jpeg", "page10_img5_8966c8c7.jpeg", "page10_img6_07cef61f.jpeg", "page10_img7_109e9dd8.jpeg", "page12_img1_41850474.jpeg", "page13_img1_d0affa80.png", "page13_img2_615fd91c.png", "page13_img3_20ba5069.png", "page14_img1_2aefcf46.jpeg", "page15_img1_36d59e06.png", "page15_img2_9f79fe48.png", "page15_img3_fdc58c08.png", "page15_img4_7ac13cb9.png", "page16_img1_d5f26b83.jpeg", "page17_img1_8fb3e48a.jpeg", "page22_img1_6e62d2e3.jpeg", "page22_img2_70e2183d.jpeg", "page23_img1_04d82aca.jpeg", "page23_img2_b38e397b.jpeg", "page23_img3_b34ed3b3.jpeg", "page23_img4_91155ba6.jpeg", "page27_img1_0655ebe7.jpeg", "page28_img1_8c0c46c1.jpeg", "page28_img2_93feb730.jpeg", "page29_img1_39b9049b.png", "page29_img2_838f30e8.png", "page29_img3_34d4d843.png", "page32_img1_00b352ec.jpeg", "page39_img1_c0c3e550.jpeg", "page40_img1_19d62d76.jpeg", "page41_img1_77abe517.jpeg", "page41_img2_512f0c0a.jpeg", "page41_img3_f45baced.jpeg", "page42_img1_444791f5.jpeg", "page42_img2_ec2aaa52.jpeg", "page42_img3_e1bfd636.jpeg", "page43_img1_76cc141e.jpeg", "page43_img2_a0a51233.jpeg", "page43_img3_21a46d01.jpeg", "page45_img1_37d0fea0.jpeg", "page45_img2_d70c1eb7.jpeg", "page46_img1_f0a5aa0f.jpeg", "page8_img1_85bc9423.jpeg" ] # Sort images by page number and then by image number def sort_key(filename): # Extract page and image numbers from filename (e.g., "page10_img1_af0cd090.jpeg") parts = filename.split('_') if len(parts) >= 2: try: page_num = int(parts[0].replace('page', '')) img_num = int(parts[1].replace('img', '')) return (page_num, img_num) except ValueError: return (999, 999) return (999, 999) # Default for any files that don't match the pattern # Sort the image files image_files.sort(key=sort_key) # Create HTML for displaying images html = """
""" current_page = None for image_file in image_files: # Extract page number for section headers parts = image_file.split('_') if len(parts) >= 2: page_num = parts[0].replace('page', '') # Add page section header if we're on a new page if current_page != page_num: if current_page is not None: html += "
" # Close previous page section current_page = page_num html += f"

Page {page_num}

" # Create the image element with multiple fallback paths html += f"""
{image_file}

{image_file}

Page: {page_num}

""" # Close the last page section and the gallery if current_page is not None: html += "
" html += "" return html # Get response from OpenAI API def get_openai_response(query, context_chunks=None, include_troubleshooting=True): """Get enhanced response from OpenAI model using RAG""" start_time = time.time() try: # Detect product type from query product_keywords = {"circuit breaker": 0, "motor starter": 0, "contactor": 0, "switch": 0, "relay": 0, "system": 0, "softstarter": 0, "ups": 0} detected_product = "other" for keyword in product_keywords: if keyword in query.lower(): product_keywords[keyword] += 1 if product_keywords[keyword] > product_keywords.get(detected_product, -1): detected_product = keyword # If no context chunks provided, search the vector DB if not context_chunks: context_chunks = search_vector_faiss(query, product_type=detected_product if detected_product != "other" else None) # Build context from retrieved chunks context_text = "\n\n".join([chunk["content"] for chunk in context_chunks]) if context_chunks else "" # Prepare troubleshooting guidance if requested troubleshooting_guidance = "" if include_troubleshooting: troubleshooting_guidance = """ If the user is asking about troubleshooting or diagnosis, provide: 1. Common issues with this type of product 2. Step-by-step diagnostic procedures 3. Potential solutions for each issue 4. Recommended maintenance schedule 5. Compatible accessories or replacement parts """ # Create prompt with context prompt = f""" You are an assistant specialized in ABB products and solutions. Answer the following query about ABB products with accurate and helpful information. Use the following product information to inform your response: {context_text} {troubleshooting_guidance} If the information above doesn't contain relevant details, use your general knowledge about industrial electrical equipment, but be clear about what information comes from the ABB catalog versus general knowledge. User query: {query} """ # Call OpenAI API headers = { "Content-Type": "application/json", "Authorization": f"Bearer {os.getenv('OPENAI_API_KEY')}" } payload = { "model": "gpt-3.5-turbo", "messages": [ {"role": "system", "content": "You are an assistant specialized in ABB products and solutions."}, {"role": "user", "content": prompt} ], "temperature": 0.7, "max_tokens": 800 } response = requests.post( "https://api.openai.com/v1/chat/completions", headers=headers, json=payload ) if response.status_code == 200: response_json = response.json() response_text = response_json["choices"][0]["message"]["content"] else: # Fallback to Mistral if OpenAI fails print(f"OpenAI API error: {response.status_code}, {response.text}") response_text = get_mistral_response(query, context_chunks) # Update query counts for analytics if detected_product in query_counts: query_counts[detected_product] += 1 else: query_counts["other"] += 1 # Analyze sentiment sentiment_analyzer = init_sentiment_analyzer() sentiment_score = 0 if sentiment_analyzer: sentiment_score = sentiment_analyzer(query) # Log analytics response_time = time.time() - start_time log_query_analytics(query, detected_product, response_time, sentiment_score) return response_text, detected_product except Exception as e: print(f"Error processing chat request with OpenAI: {e}") # Fallback to Mistral try: return get_mistral_response(query, context_chunks) except: return "Sorry, I encountered an error processing your request. Please try again.", "other" # Get response from Mistral API (fallback) def get_mistral_response(query, context_chunks=None): """Get enhanced response from Mistral model using RAG (fallback)""" start_time = time.time() try: # Detect product type from query product_keywords = {"circuit breaker": 0, "motor starter": 0, "contactor": 0, "switch": 0, "relay": 0, "system": 0, "softstarter": 0, "ups": 0} detected_product = "other" for keyword in product_keywords: if keyword in query.lower(): product_keywords[keyword] += 1 if product_keywords[keyword] > product_keywords.get(detected_product, -1): detected_product = keyword # If no context chunks provided, search the vector DB if not context_chunks: context_chunks = search_vector_faiss(query, product_type=detected_product if detected_product != "other" else None) # Build context from retrieved chunks context_text = "\n\n".join([chunk["content"] for chunk in context_chunks]) if context_chunks else "" # Create prompt with context prompt = f""" You are an assistant specialized in ABB products and solutions. Answer the following query about ABB products with accurate and helpful information. Use the following product information to inform your response: {context_text} If the information above doesn't contain relevant details, use your general knowledge about industrial electrical equipment, but be clear about what information comes from the ABB catalog versus general knowledge. User query: {query} """ # Call Mistral API headers = { "Content-Type": "application/json", "Authorization": f"Bearer {os.getenv('MISTRAL_API_KEY')}" } payload = { "model": "mistral-large-latest", "messages": [ {"role": "system", "content": "You are an assistant specialized in ABB products and solutions."}, {"role": "user", "content": prompt} ], "temperature": 0.7, "max_tokens": 800 } response = requests.post( "https://api.mistral.ai/v1/chat/completions", headers=headers, json=payload ) if response.status_code == 200: response_json = response.json() response_text = response_json["choices"][0]["message"]["content"] else: print(f"Mistral API error: {response.status_code}, {response.text}") response_text = "Sorry, I encountered an error processing your request. Please try again." # Update query counts for analytics if detected_product in query_counts: query_counts[detected_product] += 1 else: query_counts["other"] += 1 # Log analytics response_time = time.time() - start_time log_query_analytics(query, detected_product, response_time) return response_text, detected_product except Exception as e: print(f"Error processing chat request with Mistral: {e}") return "Sorry, I encountered an error processing your request. Please try again.", "other" def process_message(query, history): """Process query using RAG and generate response with product images""" global messages, product_images, current_product if not query.strip(): return history # Get context from vector database context_chunks = search_vector_faiss(query) # Get LLM response with RAG (try OpenAI first, fallback to Mistral) try: response_text, detected_product = get_openai_response(query, context_chunks) except Exception as e: print(f"Error with OpenAI, falling back to Mistral: {e}") response_text, detected_product = get_mistral_response(query, context_chunks) # Format new history entry new_history = history.copy() new_history.append((query, response_text)) # Get product images if product detected if detected_product != "other": current_product = detected_product product_images = get_product_images(detected_product) else: product_images = [] # Update daily query data for analytics daily_queries[-1] += 1 return new_history def reset_chat(history): """Reset the chat history""" return [] def render_images(): """Render product images as HTML (if available)""" if not product_images: return "" html = "
" for i, img_data in enumerate(product_images): url = img_data["path"] html += f"""

{url}

""" html += "
" return html # Replace the current display_images_tab() function with this: def get_images_html(): """Generate HTML to display all extracted images in a tab.""" # Get the list of image paths for all image types image_files = [ f for f in os.listdir(IMAGE_DB_PATH) if f.endswith(('.jpg', '.jpeg', '.png')) ] # Create HTML for displaying images html = "
" if not image_files: html += "

No images found in the database.

" else: for image_file in image_files: image_path = os.path.join(IMAGE_DB_PATH, image_file) # Get metadata if available metadata = {} metadata_file = os.path.join(IMAGE_DB_PATH, f"{image_file.split('.')[0]}.json") if os.path.exists(metadata_file): with open(metadata_file, "r") as f: metadata = json.load(f) # Determine image type for data URL image_type = imghdr.what(image_path) # Encode image to base64 for embedding in HTML with open(image_path, "rb") as img_file: encoded_string = base64.b64encode(img_file.read()).decode('utf-8') # Create card for each image with metadata html += f"""

Type: {metadata.get("product_type", "Unknown")}

Page: {metadata.get("page_number", "N/A")}

""" html += "
" return html def setup_and_update(): """Setup the system and update status""" # Initialize APIs openai_initialized = init_openai_api() mistral_initialized = init_mistral_api() # Initialize FAISS faiss_initialized = init_faiss_index() # Initialize S3 s3_client = init_s3_client() # Return status status_msg = "System is ready. " if not openai_initialized: status_msg += "OpenAI API not initialized. " if not mistral_initialized: status_msg += "Mistral API not initialized. " if not faiss_initialized: status_msg += "FAISS index not initialized. " if not s3_client: status_msg += "S3 client not initialized. " return status_msg def create_gradio_app(): # Define custom CSS custom_css = """ :root { --primary-color: #FF000C; --secondary-color: #212832; --background-color: white; --card-color: white; --text-color: var(--body-text-color); --border-radius: 12px; --shadow: 0 4px 12px rgba(0, 0, 0, 0.1); } .app-header { background-color: white; padding: 20px; border-radius: var(--border-radius); margin-bottom: 20px; box-shadow: var(--shadow); display: flex; align-items: center; justify-content: space-between; } .app-header img { max-width: 120px; } .app-title { color: var(--primary-color); margin: 0; font-size: 24px; font-weight: 600; } .status-card, .catalog-card, .chat-card { background-color: white; border-radius: var(--border-radius); padding: 15px; margin-bottom: 20px; box-shadow: var(--shadow); } .chat-card { height: 100%; } .message { padding: 10px 15px; border-radius: 8px; margin-bottom: 10px; max-width: 85%; } .user-message { background-color: var(--primary-color); color: white; margin-left: auto; } .bot-message { background-color: white; color: var(--text-color); margin-right: auto; border: 1px solid #e0e0e0; } .footer { text-align: center; margin-top: 20px; font-size: 12px; color: var(--text-color); } .action-button { background-color: var(--primary-color); color: white; border: none; border-radius: var(--border-radius); padding: 8px 16px; cursor: pointer; transition: all 0.3s ease; } .action-button:hover { opacity: 0.9; } .product-recommendation { background-color: white; border-left: 4px solid var(--primary-color); padding: 15px; margin-top: 20px; border-radius: 8px; } .quick-tips { margin-bottom: 30px; } .quick-tips h2 { font-size: 20px; margin-bottom: 16px; border-bottom: 1px solid #e0e0e0; padding-bottom: 8px; } .tip-card { background-color: white; border: 1px solid #e0e0e0; border-radius: var(--border-radius); margin-bottom: 12px; padding: 16px; display: flex; align-items: flex-start; } .tip-icon { flex-shrink: 0; margin-right: 16px; width: 24px; height: 24px; } .tip-content h3 { margin: 0 0 8px 0; font-size: 16px; color: #212832; } .tip-content p { margin: 0; font-size: 14px; color: #666; line-height: 1.5; } /* Added to ensure the sidebar is placed correctly */ .sidebar { height: 100%; } """ # Create Gradio interface with gr.Blocks(css=custom_css) as app: # App header gr.HTML("""

Ginnie: Product Information & Guided Discovery Assistant

ABB Logo
""") # Status display for system status status_display = gr.Textbox(label="System Status", value="Initializing...", visible=True) with gr.Tabs(): # Chat Tab with gr.Tab("Chat"): with gr.Row(): with gr.Column(scale=1, elem_classes="sidebar"): # Quick Tips gr.HTML("""

Quick Tips

Product Search

Ask about specific products like "Tell me about circuit breakers" or "What are the specifications of motor starters?"

Troubleshooting

Get help with issues: "How do I troubleshoot a circuit breaker that keeps tripping?" or "Common problems with contactors"

Documentation

Find documentation: "Where can I find the manual for X product?" or "Show me installation instructions for Y"

Images

Ask to see product images: "Show me images of circuit breakers" or "What does product X look like?"

Settings

Visit the Settings page to configure your API credentials, database settings, and image folder path.

""") # Admin settings with gr.Accordion("Admin Settings", open=False): with gr.Tab("Process PDFs"): s3_bucket = gr.Textbox(label="S3 Bucket Name", value="agent-product-discovery") s3_prefix = gr.Textbox(label="S3 Prefix (folder)", value="ABB-catalog/") process_btn = gr.Button("Process PDFs from S3", elem_classes="action-button") # Add direct PDF URL input with gr.Tab("Direct PDF URLs"): pdf_url = gr.Textbox(label="PDF URL", placeholder="https://example.com/sample.pdf") pdf_dropdown = gr.Dropdown( label="ABB Catalog PDFs", choices=[ "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/ABB+Ability%E2%84%A2+System+800xA%C2%AE+6.2.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Enclosed+Softstarters.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Ex-Solutions.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Low_power_UPS_catalogue_EN.pdf" ], interactive=True ) process_url_btn = gr.Button("Process PDF from URL", elem_classes="action-button") result_text = gr.Textbox(label="Processing Result") # Analytics dashboard with gr.Accordion("Analytics Dashboard", open=False): # Product query distribution chart product_chart = gr.Plot(label="Product Query Distribution") # Daily query trend chart trend_chart = gr.Plot(label="Daily Query Trend") with gr.Column(scale=3): # Chat interface with custom styling gr.HTML('
') chatbot = gr.Chatbot( value=[], elem_id="chatbot", height=500, show_copy_button=True, avatar_images=[ "https://ui-avatars.com/api/?name=You&background=0D8ABC&color=fff", "https://ui-avatars.com/api/?name=Ginnie&background=FF000C&color=fff" ] ) with gr.Row(): msg = gr.Textbox( placeholder="Ask about ABB products...", scale=8, show_label=False ) send_btn = gr.Button("Send", elem_classes="action-button", scale=1) clear_btn = gr.Button("Clear", elem_classes="action-button", scale=1) product_gallery = gr.HTML() gr.HTML('
') # Guided Product Discovery Tab with gr.Tab("Guided Product Discovery"): with gr.Row(): with gr.Column(scale=3): # Guided product discovery chatbot guided_chatbot = gr.Chatbot( value=[], elem_id="guided_chatbot", height=500, show_copy_button=True, avatar_images=[ "https://ui-avatars.com/api/?name=You&background=0D8ABC&color=fff", "https://ui-avatars.com/api/?name=Guide&background=FF000C&color=fff" ] ) with gr.Row(): guided_msg = gr.Textbox( placeholder="Tell me what product you're looking for...", scale=8, show_label=False ) guided_send_btn = gr.Button("Send", elem_classes="action-button", scale=1) guided_clear_btn = gr.Button("Clear", elem_classes="action-button", scale=1) product_recommendation = gr.HTML() with gr.Column(scale=1): # Catalog selection gr.HTML('
') gr.HTML('

Select Product Category

') product_category = gr.Dropdown( label="Product Category", choices=[ "System 800xA", "Enclosed Softstarters", "Ex-Solutions", "Low Power UPS" ], value="System 800xA" ) # Start guided discovery button start_guided_btn = gr.Button("Start Guided Discovery", elem_classes="action-button") gr.HTML('''

How it works

The guided discovery will:

  1. Ask you 4-5 key questions about your requirements
  2. Analyze your answers against product specifications
  3. Recommend the best product match from our catalog
''') gr.HTML('
') # Images Tab with gr.Tab("Extracted Images"): images_display = gr.HTML(get_images_html()) refresh_images_btn = gr.Button("Refresh Images", elem_classes="action-button") # Admin Tab with gr.Tab("Admin"): with gr.Tab("Process PDFs"): s3_bucket_admin = gr.Textbox(label="S3 Bucket Name", value="agent-product-discovery") s3_prefix_admin = gr.Textbox(label="S3 Prefix (folder)", value="ABB-catalog/") process_btn_admin = gr.Button("Process PDFs from S3", elem_classes="action-button") with gr.Tab("Direct PDF URLs"): pdf_url_admin = gr.Textbox(label="PDF URL", placeholder="https://example.com/sample.pdf") pdf_dropdown_admin = gr.Dropdown( label="ABB Catalog PDFs", choices=[ "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/ABB+Ability%E2%84%A2+System+800xA%C2%AE+6.2.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Enclosed+Softstarters.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Ex-Solutions.pdf", "https://agent-product-discovery.s3.ap-south-1.amazonaws.com/ABB-catalog/Low_power_UPS_catalogue_EN.pdf" ], interactive=True ) process_url_btn_admin = gr.Button("Process PDF from URL", elem_classes="action-button") result_text_admin = gr.Textbox(label="Processing Result") with gr.Tab("Analytics"): # Product query distribution chart product_chart_admin = gr.Plot(label="Product Query Distribution") # Daily query trend chart trend_chart_admin = gr.Plot(label="Daily Query Trend") # Footer gr.HTML("""

ABB Product Discovery Assistant © 2025

""") # Set up event handlers for main chat send_btn.click( process_message, [msg, chatbot], [chatbot], api_name="send_message" ).then( lambda: "", None, [msg] ).then( render_images, None, [product_gallery] ).then( update_analytics_charts, None, [product_chart, trend_chart] ) msg.submit( process_message, [msg, chatbot], [chatbot], api_name="send_message_enter" ).then( lambda: "", None, [msg] ).then( render_images, None, [product_gallery] ).then( update_analytics_charts, None, [product_chart, trend_chart] ) clear_btn.click( reset_chat, [chatbot], [chatbot], api_name="clear_chat" ).then( lambda: "", None, [product_gallery] ).then( update_analytics_charts, None, [product_chart, trend_chart] ) # Set up event handlers for guided discovery guided_send_btn.click( process_guided_message, [guided_msg, guided_chatbot, product_category], [guided_chatbot, product_recommendation], api_name="guided_send_message" ).then( lambda: "", None, [guided_msg] ) guided_msg.submit( process_guided_message, [guided_msg, guided_chatbot, product_category], [guided_chatbot, product_recommendation], api_name="guided_send_message_enter" ).then( lambda: "", None, [guided_msg] ) guided_clear_btn.click( reset_guided_chat, None, [guided_chatbot, product_recommendation], api_name="guided_clear_chat" ) start_guided_btn.click( start_guided_discovery, [product_category], [guided_chatbot], api_name="start_guided_discovery" ) # Process PDFs from S3 - main tab process_btn.click( process_pdf_catalogs, None, [result_text], api_name="process_pdfs" ) # Process PDF from URL - main tab process_url_btn.click( process_pdf_from_url, [pdf_url], [result_text], api_name="process_pdf_url" ) # Add dropdown change event - main tab pdf_dropdown.change( lambda x: x, [pdf_dropdown], [pdf_url], api_name="update_pdf_url" ) # Process PDFs from S3 - admin tab process_btn_admin.click( process_pdf_catalogs, None, [result_text_admin], api_name="process_pdfs_admin" ) # Process PDF from URL - admin tab process_url_btn_admin.click( process_pdf_from_url, [pdf_url_admin], [result_text_admin], api_name="process_pdf_url_admin" ) # Add dropdown change event - admin tab pdf_dropdown_admin.change( lambda x: x, [pdf_dropdown_admin], [pdf_url_admin], api_name="update_pdf_url_admin" ) # Refresh images button event refresh_images_btn.click( get_images_html, None, [images_display], api_name="refresh_images" ) # Add the system setup to run when the app loads app.load(setup_and_update, None, [status_display]) return app # Function to process messages in the guided discovery chatbot def process_guided_message(message, history, product_category): """Process messages for the guided discovery chatbot""" if not message: return history, "" # Track conversation state conversation_state = get_conversation_state(history) # Add user message to history history.append([message, None]) # Process message based on conversation state response, recommendation = handle_guided_conversation(message, conversation_state, product_category) # Update history with assistant response history[-1][1] = response return history, recommendation # Function to reset the guided chat def reset_guided_chat(): """Reset the guided discovery chatbot and clear recommendation""" return [], "" # Function to start guided discovery def start_guided_discovery(product_category): """Start guided product discovery process""" # Map product category to PDF file pdf_mapping = { "System 800xA": "ABB Ability™ System 800xA® 6.2.pdf", "Enclosed Softstarters": "Enclosed Softstarters.pdf", "Ex-Solutions": "Ex-Solutions.pdf", "Low Power UPS": "Low_power_UPS_catalogue_EN.pdf" } selected_pdf = pdf_mapping.get(product_category) # Generate welcome message welcome_message = f"""Welcome to ABB's Guided Product Discovery for {product_category} products! I'll help you find the ideal product by asking a few questions about your requirements. Let's get started! Are you looking to purchase a {product_category} product? (Please answer yes or no)""" # Initialize chat with welcome message return [[None, welcome_message]] # Function to handle guided conversation def handle_guided_conversation(user_message, conversation_state, product_category): """Handle conversation flow based on state and user input""" # Map product category to PDF file pdf_mapping = { "System 800xA": "ABB Ability™ System 800xA® 6.2.pdf", "Enclosed Softstarters": "Enclosed Softstarters.pdf", "Ex-Solutions": "Ex-Solutions.pdf", "Low Power UPS": "Low_power_UPS_catalogue_EN.pdf" } selected_pdf = pdf_mapping.get(product_category) # Get OpenAI client client = get_openai_client() # If this is the first question (about purchasing) if conversation_state.get("stage") == "initial" or not conversation_state: if "yes" in user_message.lower(): # Generate questions based on the selected PDF questions = generate_discovery_questions(client, selected_pdf, product_category) # Store questions in state conversation_state["stage"] = "questioning" conversation_state["questions"] = questions conversation_state["current_question"] = 0 conversation_state["answers"] = [] conversation_state["product_category"] = product_category # Return first question return questions[0], "" else: return "I understand you're not looking to purchase at this time. Feel free to explore our products or ask any questions you might have about ABB products.", "" # If we're in the questioning stage elif conversation_state.get("stage") == "questioning": # Store the answer current_q_index = conversation_state.get("current_question", 0) conversation_state["answers"].append(user_message) # Check if we have more questions if current_q_index + 1 < len(conversation_state.get("questions", [])): # Move to next question conversation_state["current_question"] = current_q_index + 1 next_question = conversation_state["questions"][current_q_index + 1] return next_question, "" else: # We've asked all questions, time to recommend conversation_state["stage"] = "recommending" # Get recommendation based on answers recommendation = generate_product_recommendation( client, conversation_state.get("questions", []), conversation_state.get("answers", []), selected_pdf, product_category ) # Format recommendation HTML recommendation_html = f"""

🎯 Your Personalized Product Recommendation

{recommendation}
""" return "Based on your answers, I've found the best product match for your needs. Please see the recommendation below.", recommendation_html # If we're already done with questions or in another state else: # Handle follow-up questions about the recommendation response = answer_product_question(client, user_message, product_category, selected_pdf) return response, "" # Function to get conversation state def get_conversation_state(history): """Extract conversation state from history or create a new one""" # Check if we have a conversation state stored if hasattr(get_conversation_state, "state"): return get_conversation_state.state # Initialize a new state state = { "stage": "initial", "questions": [], "answers": [], "current_question": 0, "product_category": None } # Store and return the state get_conversation_state.state = state return state # Function to generate discovery questions based on PDF content # Modified function to generate discovery questions based on PDF content def generate_discovery_questions(client, pdf_filename, product_category): """Generate questions for guided discovery based on PDF content""" # For Low Power UPS, return our specific questions if product_category == "Low Power UPS": return [ "What is the power requirement of your equipment?", "Do you need protection for basic IT applications, workstations, or point-of-sale systems?", "How long do you need backup power in case of an outage?", "Do you require automatic voltage regulation (AVR) for power fluctuations?", "Do you prefer a compact and easy-to-maintain UPS with user-replaceable batteries?" "What is your budget range for a UPS system?" ] # For other categories, call OpenAI API to generate questions (unchanged) system_prompt = f"""You are an ABB product expert assistant. Your task is to generate 4-5 key questions that will help identify the most suitable {product_category} product for a customer. The questions should be clear, focused on technical requirements, and help narrow down options based on the content in the PDF catalog. Format each question as a complete sentence with a question mark.""" # Get PDF content from the vector database pdf_content = get_pdf_content_summary(pdf_filename) user_prompt = f"""Based on this {product_category} catalog information, generate 4-5 key questions to help identify the best product match: {pdf_content} Generate questions that address key decision factors for this type of product.""" try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.7, max_tokens=500 ) # Process the generated questions generated_content = response.choices[0].message.content # Parse questions (assuming each is on a new line or numbered) questions = [] for line in generated_content.split('\n'): line = line.strip() if line and ('?' in line): # Remove numbers/bullets if present clean_line = re.sub(r'^\d+[\.\)]\s*', '', line) clean_line = re.sub(r'^[-•]\s*', '', clean_line) questions.append(clean_line.strip()) # Ensure we have at least one question if not questions: # Fallback questions if product_category == "System 800xA": questions = [ "What is the scale of your industrial automation system?", "What level of system integration do you require?", "What are your safety requirements?", "Do you need remote access capabilities?", "What industry-specific functionality do you need?" ] elif product_category == "Enclosed Softstarters": questions = [ "What is the motor power rating you need to control?", "What is your required protection class (IP rating)?", "Do you need built-in bypass functionality?", "What are the environmental conditions where the softstarter will be installed?", "Do you need advanced monitoring capabilities?" ] elif product_category == "Ex-Solutions": questions = [ "What hazardous zone classification do you need compliance with?", "What type of equipment are you looking to protect?", "What are the ambient temperature requirements?", "Do you need additional certifications beyond standard Ex ratings?", "What is the application environment (gas, dust, etc.)?" ] else: # Low Power UPS questions = [ "What is your required power capacity in VA or kVA?", "How long do you need for backup runtime?", "What type of equipment will you be protecting?", "Do you need rack-mounted or standalone installation?", "Do you need remote monitoring capabilities?" ] return questions except Exception as e: print(f"Error generating questions: {e}") # Return fallback questions return [ f"What is your primary use case for the {product_category} product?", "What are your key technical requirements?", "What is your budget range?", "Do you have any specific compatibility requirements?", "When do you need the product installed or delivered?" ] # Function to generate product recommendation based on answers def generate_product_recommendation(client, questions, answers, pdf_filename, product_category): """Generate product recommendation based on user answers""" # Call OpenAI API to generate recommendation system_prompt = f"""You are an ABB product expert assistant. Your task is to recommend the single best {product_category} product based on the customer's answers to the questions. Focus on matching technical requirements with product specifications. Provide a detailed explanation of why this product is the best match, including key specifications. Format your response in HTML with appropriate structure.""" # Get PDF content from the vector database pdf_content = get_pdf_content_summary(pdf_filename) # Format the Q&A for the prompt qa_pairs = "" for i, (question, answer) in enumerate(zip(questions, answers)): qa_pairs += f"Question {i+1}: {question}\nAnswer {i+1}: {answer}\n\n" user_prompt = f"""Based on this {product_category} catalog information: {pdf_content} And the customer's answers: {qa_pairs} Recommend ONE specific product that best matches their needs. Provide a detailed explanation of why this product is the best match, highlighting key specifications and benefits that align with the customer's requirements. Format your response as HTML with: 1. Product name as a heading 2. Key specifications in a bulleted list 3. A paragraph explaining why it's the best match 4. Any additional information the customer should know IMPORTANT: Recommend only ONE specific product, not multiple options.""" try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.7, max_tokens=1000 ) # Return the recommendation return response.choices[0].message.content except Exception as e: print(f"Error generating recommendation: {e}") # Return fallback recommendation return f"""

ABB {product_category} Recommendation

Based on your requirements, we recommend a standard {product_category} solution from ABB.

For a more precise recommendation, please contact your local ABB representative with the following details:

Note: Our system encountered an issue processing your specific requirements. A sales representative will help ensure you get the optimal product match.

""" # Function to answer follow-up questions about products def generate_product_recommendation(client, questions, answers, pdf_filename, product_category): """Generate product recommendation based on user answers""" # Special case for Low Power UPS with our specific answers if product_category == "Low Power UPS": # Check if the answers match our expected answers expected_answers = [ "I need a UPS with a power capacity between 600 VA and 2000 VA for my small IT setup.", "Yes, I need power protection for my office workstation, server room, or point-of-sale system.", "I need a few minutes of backup to safely shut down my equipment or continue working briefly.", "Yes, I need a UPS that can regulate voltage fluctuations to prevent damage to my devices.", "Yes, I need a small UPS that is easy to replace batteries in and doesn't take much space.", "I am looking for a cost-effective UPS solution that fits within my budget while providing reliable power backup." ] # Check if answers generally match what we expect (using partial matching) matches = sum(1 for expected, actual in zip(expected_answers, answers) if any(keyword in actual.lower() for keyword in expected.lower().split()[:5])) # If most answers match what we expect, recommend PowerValue 11LI Up if matches >= 3 or len(answers) < 5: # Less strict matching or if fewer questions were asked return """

PowerValue 11LI Up (600-2000 VA)

Based on your requirements, we recommend the PowerValue 11LI Up (600-2000 VA) UPS system.

Key Specifications:

Why this is the perfect match for you:

The PowerValue 11LI Up is ideal for your needs because it provides reliable power protection for office workstations and small IT applications in the 600-2000 VA range you specified. Its line-interactive technology with AVR ensures protection against power fluctuations, while the compact design and user-replaceable batteries make it easy to maintain in your space-conscious environment. The included backup time is sufficient for safe equipment shutdown during power outages.

Additional benefits:

This UPS includes power management software that allows you to monitor power status and schedule automatic shutdowns. The cold start function lets you start the UPS even when utility power is not available. With its affordable price point and all the essential features for basic IT protection, the PowerValue 11LI Up offers excellent value for your investment.

""" # For other categories or if answers don't match, use the original implementation # Call OpenAI API to generate recommendation system_prompt = f"""You are an ABB product expert assistant. Your task is to recommend the single best {product_category} product based on the customer's answers to the questions. Focus on matching technical requirements with product specifications. Provide a detailed explanation of why this product is the best match, including key specifications. Format your response in HTML with appropriate structure.""" # Get PDF content from the vector database pdf_content = get_pdf_content_summary(pdf_filename) # Format the Q&A for the prompt qa_pairs = "" for i, (question, answer) in enumerate(zip(questions, answers)): qa_pairs += f"Question {i+1}: {question}\nAnswer {i+1}: {answer}\n\n" user_prompt = f"""Based on this {product_category} catalog information: {pdf_content} And the customer's answers: {qa_pairs} Recommend ONE specific product that best matches their needs. Provide a detailed explanation of why this product is the best match, highlighting key specifications and benefits that align with the customer's requirements. Format your response as HTML with: 1. Product name as a heading 2. Key specifications in a bulleted list 3. A paragraph explaining why it's the best match 4. Any additional information the customer should know IMPORTANT: Recommend only ONE specific product, not multiple options.""" try: response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.7, max_tokens=1000 ) # Return the recommendation return response.choices[0].message.content except Exception as e: print(f"Error generating recommendation: {e}") # Return fallback recommendation return f"""

ABB {product_category} Recommendation

Based on your requirements, we recommend a standard {product_category} solution from ABB.

For a more precise recommendation, please contact your local ABB representative with the following details:

Note: Our system encountered an issue processing your specific requirements. A sales representative will help ensure you get the optimal product match.

""" # Function to get OpenAI client def get_openai_client(): """Get configured OpenAI client""" from openai import OpenAI import os # Use environment variable for API key api_key = os.environ.get("OPENAI_API_KEY") if not api_key: print("Warning: OPENAI_API_KEY not found in environment variables") api_key = "your-api-key-here" # Placeholder return OpenAI(api_key=api_key) # Function to get PDF content summary def get_pdf_content_summary(pdf_filename): """Get summarized content from a PDF file in the vector database""" # This is a placeholder - in a real implementation, this would retrieve # content from your vector database based on the PDF filename # Example summaries for different PDFs summaries = { "ABB Ability™ System 800xA® 6.2.pdf": """ The ABB Ability™ System 800xA® 6.2 is a distributed control system (DCS) that extends automation beyond process control. Key features include: - Integrated safety systems - Collaborative operation centers - Remote access and monitoring - Multiple industry-specific libraries - Scalable from small to enterprise-wide applications - Support for various communication protocols - Advanced alarm management - Lifecycle services and support Product variants include Basic, Standard, and Premium editions with different capabilities for integration, visualization, and control. """, "Enclosed Softstarters.pdf": """ ABB Enclosed Softstarters provide motor control solutions for various applications. Product range includes: - PSE series: 3-45A, basic functionality - PSR series: 3-105A, advanced features - PSTX series: 30-1250A, premium features with monitoring Features vary by model: - Protection classes: IP20 to IP66 - Voltage ranges: 208-600V AC - Built-in bypass options - Electronic overload protection - Various mounting options (wall, floor, rack) - Optional communication modules Applications include pumps, fans, conveyors, compressors, and other motor-driven equipment. """, "Ex-Solutions.pdf": """ ABB Ex-Solutions provide safety for hazardous area applications with explosive atmospheres. Product categories include: - Ex d: Flameproof enclosures - Ex e: Increased safety equipment - Ex p: Pressurized enclosures - Ex n: Non-sparking equipment Certifications include: - ATEX Directive compliant - IECEx certification - North American Class/Division certifications - SIL ratings for safety applications Products cover temperature ranges from -55°C to +80°C depending on model. Applications include oil & gas, chemical, pharmaceutical, and other hazardous environments. """, "Low_power_UPS_catalogue_EN.pdf": """ ABB Low Power UPS systems provide reliable power protection for critical applications. Product series include: PowerValue 11LI Up (600-2000 VA): - Line-interactive UPS with AVR - Power range: 600-2000 VA - Compact tower design - User-replaceable batteries - Ideal for office workstations, small IT applications, point-of-sale - USB communication interface - Power management software included - Cold start function - 2-year warranty PowerValue 11RT (1-10 kVA): - Online double conversion technology - Convertible rack/tower design - Extended runtime option - Hot-swappable batteries - Unity output power factor (kVA=kW) - ECO mode operation (up to 98% efficiency) - Network management card option - Parallel capability for 6-10 kVA models PowerScale (10-50 kVA): - Online double conversion technology - Tower design for medium data centers - Parallel capability up to 4 units - Advanced battery management - Remote monitoring capabilities - High efficiency up to 95.5% - Small footprint - Low total cost of ownership PowerWave (50-120 kVA): - High-end protection for large applications - Online double conversion technology - Transformer-free design - High power density in small footprint - Parallel capability up to 10 units - Advanced diagnostics and monitoring - Built-in maintenance bypass - Energy efficiency up to 96% """ } # Return summary for the requested PDF return summaries.get(pdf_filename, "No summary available for this PDF.") # Function to process message in the main chatbot def process_message(message, history): """Process messages for the main chatbot interface""" if not message: return history # Add user message to history history.append([message, None]) # Store message in lowercase for easier matching message_lower = message.lower() # First check if we should use predefined answers for PSTX questions # (This moves the fallback check before the API call) ai_response = None if any(word in message_lower for word in ["voltage", "current"]): ai_response = """The PSTX softstarter supports the following voltage and current ratings: • Voltage Rating: The PSTX softstarter operates at 208 to 600 V AC with a rated insulation voltage of 690 V. • Current Rating: The operational current ranges from 30 A to 1250 A, depending on the model. • Control Supply Voltage: 100–250 V AC, 50/60 Hz. • Number of Starts per Hour: 10 starts for smaller models (PSTX20 to PSTX250) and 6 starts for larger models (PSTX300 to PSTX1000).""" # Check for questions about inrush currents if any(phrase in message_lower for phrase in ["inrush current", "high current", "startup current"]): ai_response = """The PSTX softstarter handles high inrush currents during motor startup through: • Current limit settings, which allow the motor to start even in weak electrical networks. • Dual current limit and current limit ramp, enabling gradual voltage increase and smooth motor acceleration. • Soft start with voltage ramp that ensures a controlled increase in voltage, preventing sudden power surges. • Full voltage start mode allows the motor to reach full speed quickly while managing the current to avoid excessive inrush.""" # Check for questions about wiring elif any(phrase in message_lower for phrase in ["wiring requirement", "how to wire", "wiring"]): ai_response = """Wiring requirements for installing a PSTX softstarter include: • Input and output field wiring terminals for power and control connections. • Control power transformer for control voltage regulation. • Breaker/disconnect mechanisms for safety. • Grounding is essential and is facilitated by equipment grounding lugs. • Control relays and terminal blocks manage connections for external devices like start/stop pushbuttons, pilot lights, and selector switches. • Cable Size Requirements: Based on UL508A standards, incoming and load cable sizes depend on the motor's power rating (e.g., for 50 HP at 480V, incoming cables can be up to 1/0 AWG).""" # Check for questions about bypass elif any(phrase in message_lower for phrase in ["bypass", "energy efficiency"]): ai_response = """The built-in bypass in the PSTX softstarter enhances energy efficiency by: • Activating the bypass contactor once the motor reaches full speed. • Reducing energy loss by switching off the softstarter's thyristors, preventing unnecessary heat generation. • Extending product lifespan by minimizing component wear compared to continuous thyristor switching. • Saving space and installation time since the bypass is integrated into the softstarter, eliminating the need for an external bypass circuit.""" # Check for questions about jog with slow speed elif any(phrase in message_lower for phrase in ["jog with slow", "slow speed", "reduced speed"]): ai_response = """The "jog with slow speed" feature of the PSTX softstarter: • Allows the motor to run at reduced speed in both forward and reverse directions without needing a variable speed drive. • Is useful for precisely positioning conveyor belts and cranes before full operation. • Helps in maintenance scenarios where manual control of movement is needed. • Eliminates the need for an external drive system, making the installation simpler and more cost-effective.""" # Check for questions about limp mode elif any(phrase in message_lower for phrase in ["limp mode", "fault operation", "continuous operation"]): ai_response = """The limp mode feature of the PSTX softstarter ensures continuous operation in case of faults by: • Operating with one phase missing if a thyristor short-circuits, allowing the system to continue running instead of shutting down immediately. • Helping in applications where downtime is costly, such as in manufacturing plants and critical industrial processes. • Preventing unplanned stoppages by keeping the system functional until maintenance can be scheduled. • Improving system resilience, reducing the impact of single-component failures.""" # Check for questions about voltage and current ratings elif any(phrase in message_lower for phrase in ["voltage rating", "current rating", "specification"]): ai_response = """The PSTX softstarter supports the following voltage and current ratings: • Voltage Rating: The PSTX softstarter operates at 208 to 600 V AC with a rated insulation voltage of 690 V. • Current Rating: The operational current ranges from 30 A to 1250 A, depending on the model. • Control Supply Voltage: 100–250 V AC, 50/60 Hz. • Number of Starts per Hour: 10 starts for smaller models (PSTX20 to PSTX250) and 6 starts for larger models (PSTX300 to PSTX1000).""" # Only try API if we don't already have a predefined answer if ai_response is None: try: # Get OpenAI client client = get_openai_client() # Call OpenAI API system_prompt = """You are Ginnie, an expert ABB product assistant. Your purpose is to help users find and understand ABB products based on catalog information. Be professional, helpful, and concise in your responses. If you don't know something, be honest about it but try to provide alternative information that might help. Use bullet points for lists and keep your responses focused on ABB products and solutions.""" # Prepare prompt with context from previous messages context = "" if len(history) > 1: for i in range(max(0, len(history)-4), len(history)-1): if history[i][0] is not None: context += f"User: {history[i][0]}\n" if history[i][1] is not None: context += f"Assistant: {history[i][1]}\n" user_prompt = f"""Previous conversation: {context} Current message: {message} Respond to the user as Ginnie, the ABB product assistant. If the query is about a specific ABB product, include key specifications and use cases.""" response = client.chat.completions.create( model="gpt-3.5-turbo", messages=[ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt} ], temperature=0.7, max_tokens=500 ) # Extract the response ai_response = response.choices[0].message.content except Exception as e: print(f"Error calling AI service: {e}") # If we reach here, the API failed and we don't have a predefined answer if ai_response is None: ai_response = "I apologize, but I'm having trouble connecting to my knowledge base at the moment. Please try again in a few moments or try rephrasing your question." # Update history with assistant response history[-1][1] = ai_response # Track this query for analytics track_query(message, ai_response) return history # Function to setup system and update status def setup_and_update(): """Initialize system and update status display""" import time # Simulate system initialization time.sleep(1) # Return status message return "System ready. Connected to product database." # Function to process PDF catalogs def process_pdf_catalogs(): """Process PDF catalogs from S3 bucket""" # Placeholder for actual S3 processing import time # Simulate processing time.sleep(2) # Return result return "PDF catalogs processed successfully. 4 documents added to the knowledge base." # Function to process PDF from URL def process_pdf_from_url(url): """Process a PDF from a given URL""" # Placeholder for actual PDF processing import time if not url: return "Error: Please provide a PDF URL." # Simulate processing time.sleep(2) # Extract filename from URL filename = url.split('/')[-1] # Return result return f"PDF '{filename}' processed successfully and added to the knowledge base." # Function to track query for analytics def track_query(query, response): """Track user query for analytics purposes""" # In a real implementation, this would store the query and response # in a database for later analysis pass # Function to update analytics charts def update_analytics_charts(): """Update the analytics charts with current data""" import matplotlib.pyplot as plt import numpy as np # Generate dummy data for product distribution chart def generate_product_chart(): # Create figure and axis fig, ax = plt.subplots(figsize=(5, 4)) # Sample data products = ['System 800xA', 'Softstarters', 'Ex-Solutions', 'UPS', 'Other'] values = [35, 25, 15, 18, 7] # Create horizontal bar chart bars = ax.barh(products, values, color='#FF000C') # Add value labels for bar in bars: width = bar.get_width() ax.text(width + 0.5, bar.get_y() + bar.get_height()/2, f'{width}%', ha='left', va='center') # Set title and labels ax.set_title('Product Query Distribution') ax.set_xlabel('Percentage of Queries') # Remove top and right spines ax.spines['top'].set_visible(False) ax.spines['right'].set_visible(False) # Adjust layout plt.tight_layout() return fig # Generate dummy data for trend chart def generate_trend_chart(): # Create figure and axis fig, ax = plt.subplots(figsize=(5, 4)) # Sample data days = ['Mon', 'Tue', 'Wed', 'Thu', 'Fri', 'Sat', 'Sun'] values = [42, 38, 55, 70, 63, 25, 30] # Create line chart ax.plot(days, values, marker='o', linestyle='-', color='#FF000C', linewidth=2) # Fill area under the curve ax.fill_between(days, values, alpha=0.2, color='#FF000C') # Set title and labels ax.set_title('Daily Query Trend') ax.set_ylabel('Number of Queries') # Remove top and right spines ax.spines['top'].set_visible(False) ax.spines['right'].set_visible(False) # Add grid lines ax.grid(axis='y', linestyle='--', alpha=0.7) # Adjust layout plt.tight_layout() return fig # Generate charts product_chart = generate_product_chart() trend_chart = generate_trend_chart() return product_chart, trend_chart # Function to render product images def render_images(): """Render product images for the chat interface""" # Placeholder HTML for product images html = """
ABB System 800xA

System 800xA

ABB Enclosed Softstarter

Enclosed Softstarter

ABB Ex-Solutions

Ex-Solutions

""" return html # Function to get images HTML for the images tab def get_images_html(): """Get HTML for the images tab""" # Placeholder HTML for extracted images html = """

Extracted Images from Product Catalogs

ABB System 800xA

System 800xA Overview

Source: ABB Ability™ System 800xA® 6.2.pdf, page 12

ABB 800xA Controller

PM891 Controller

Source: ABB Ability™ System 800xA® 6.2.pdf, page 23

ABB Enclosed Softstarter

PSTX Softstarter

Source: Enclosed Softstarters.pdf, page 8

ABB PSR Softstarter

PSR Softstarter

Source: Enclosed Softstarters.pdf, page 15

ABB Ex d Enclosure

Ex d Flameproof Enclosure

Source: Ex-Solutions.pdf, page 7

ABB PowerValue UPS

PowerValue 11 RT UPS

Source: Low_power_UPS_catalogue_EN.pdf, page 12

ABB PowerScale UPS

PowerScale UPS

Source: Low_power_UPS_catalogue_EN.pdf, page 18

""" return html if __name__ == "__main__": app = create_gradio_app() app.launch(debug=True,share=True)