Spaces:
Sleeping
Sleeping
| from PyPDF2 import PdfReader | |
| import requests | |
| import json | |
| import os | |
| import concurrent.futures | |
| import random | |
| from langchain_google_genai import ChatGoogleGenerativeAI | |
| from langchain_community.document_loaders import WebBaseLoader | |
| from langchain_community.document_loaders import PyPDFLoader | |
| from langchain.text_splitter import RecursiveCharacterTextSplitter | |
| import google.generativeai as genai | |
| from io import BytesIO | |
| gemini = ChatGoogleGenerativeAI(model="gemini-1.0-pro-001",google_api_key='AIzaSyBmZtXjJgp7yIAo9joNCZGSxK9PbGMcVaA',temperature = 0.1) | |
| gemini1 = ChatGoogleGenerativeAI(model="gemini-1.0-pro-001",google_api_key='AIzaSyABsaDjPujPCBlz4LLxcXDX_bDA9uEL7Xc',temperature = 0.1) | |
| gemini2 = ChatGoogleGenerativeAI(model="gemini-1.0-pro-001",google_api_key='AIzaSyBCIQgt1uK7-sJH5Afg5vUZ99EWkx5gSU0',temperature = 0.1) | |
| gemini3 = ChatGoogleGenerativeAI(model="gemini-1.0-pro-001",google_api_key='AIzaSyBot9W5Q-BKQ66NAYRUmVeloXWEbXOXTmM',temperature = 0.1) | |
| genai.configure(api_key="AIzaSyBmZtXjJgp7yIAo9joNCZGSxK9PbGMcVaA") | |
| def pdf_extractor(link): | |
| text = '' | |
| try: | |
| # Fetch the PDF file from the URL | |
| response = requests.get(link) | |
| response.raise_for_status() # Raise an error for bad status codes | |
| # Use BytesIO to handle the PDF content in memory | |
| pdf_file = BytesIO(response.content) | |
| # Load the PDF file | |
| reader = PdfReader(pdf_file) | |
| for page in reader.pages: | |
| text += page.extract_text() # Extract text from each page | |
| except requests.exceptions.HTTPError as e: | |
| print(f'HTTP error occurred: {e}') | |
| except Exception as e: | |
| print(f'An error occurred: {e}') | |
| return [text] | |
| def web_extractor(link): | |
| text = '' | |
| try: | |
| loader = WebBaseLoader(link) | |
| pages = loader.load_and_split() | |
| for page in pages: | |
| text+=page.page_content | |
| except: | |
| pass | |
| return [text] | |
| def feature_extraction(tag, history , context): | |
| prompt = f''' | |
| You are an intelligent assistant tasked with updating product information. You have two data sources: | |
| 1. Tag_History: Previously gathered information about the product. | |
| 2. Tag_Context: New data that might contain additional details. | |
| Your job is to read the Tag_Context and update the relevant field in the Tag_History with any new details found. The field to be updated is the {tag} FIELD. | |
| Guidelines: | |
| - Only add new details that are relevant to the {tag} FIELD. | |
| - Do not add or modify any other fields in the Tag_History. | |
| - Ensure your response is in coherent sentences, integrating the new details seamlessly into the existing information. | |
| Here is the data: | |
| Tag_Context: {str(context)} | |
| Tag_History: {history} | |
| Respond with the updated Tag_History. | |
| ''' | |
| model = random.choice([gemini,gemini1,gemini2,gemini3]) | |
| result = model.invoke(prompt) | |
| return result.content | |
| def detailed_feature_extraction(find, context): | |
| prompt = f''' | |
| You are an intelligent assistant tasked with finding product information. You have one data source and one output format: | |
| 1. Context: The gathered information about the product. | |
| 2. Format: Details which need to be filled based on Context. | |
| Your job is to read the Context and update the relevant field in Format using Context. | |
| Guidelines: | |
| - Only add details that are relevant to the individual FIELD. | |
| - Do not add or modify any other fields in the Format. | |
| - If nothing found return None. | |
| Here is the data: | |
| The Context is {str(context)} | |
| The Format is {str(find)} | |
| ''' | |
| model = random.choice([gemini,gemini1,gemini2,gemini3]) | |
| result = model.invoke(prompt) | |
| return result.content | |
| def detailed_history(history): | |
| details = { | |
| "Introduction": { | |
| "Product Name": None, | |
| "Overview of the product": None, | |
| "Purpose of the manual": None, | |
| "Audience": None, | |
| "Additional Details": None | |
| }, | |
| "Specifications": { | |
| "Technical specifications": None, | |
| "Performance metrics": None, | |
| "Additional Details": None | |
| }, | |
| "Product Overview": { | |
| "Product features": None, | |
| "Key components and parts": None, | |
| "Additional Details": None | |
| }, | |
| "Safety Information": { | |
| "Safety warnings and precautions": None, | |
| "Compliance and certification information": None, | |
| "Additional Details": None | |
| }, | |
| "Installation Instructions": { | |
| "Unboxing and inventory checklist": None, | |
| "Step-by-step installation guide": None, | |
| "Required tools and materials": None, | |
| "Additional Details": None | |
| }, | |
| "Setup and Configuration": { | |
| "Initial setup procedures": None, | |
| "Configuration settings": None, | |
| "Troubleshooting setup issues": None, | |
| "Additional Details": None | |
| }, | |
| "Operation Instructions": { | |
| "How to use the product": None, | |
| "Detailed instructions for different functionalities": None, | |
| "User interface guide": None, | |
| "Additional Details": None | |
| }, | |
| "Maintenance and Care": { | |
| "Cleaning instructions": None, | |
| "Maintenance schedule": None, | |
| "Replacement parts and accessories": None, | |
| "Additional Details": None | |
| }, | |
| "Troubleshooting": { | |
| "Common issues and solutions": None, | |
| "Error messages and their meanings": None, | |
| "Support Information": None, | |
| "Additional Details": None | |
| }, | |
| "Warranty Information": { | |
| "Terms and Conditions": None, | |
| "Service and repair information": None, | |
| "Additional Details": None | |
| }, | |
| "Legal Information": { | |
| "Copyright information": None, | |
| "Trademarks and patents": None, | |
| "Disclaimers": None, | |
| "Additional Details": None | |
| } | |
| } | |
| for key,val in history.items(): | |
| find = details[key] | |
| details[key] = str(detailed_feature_extraction(find,val)) | |
| return details | |
| def get_embeddings(link): | |
| print(f"\nCreating Embeddings ----- {link}") | |
| history = { | |
| "Introduction": "", | |
| "Specifications": "", | |
| "Product Overview": "", | |
| "Safety Information": "", | |
| "Installation Instructions": "", | |
| "Setup and Configuration": "", | |
| "Operation Instructions": "", | |
| "Maintenance and Care": "", | |
| "Troubleshooting": "", | |
| "Warranty Information": "", | |
| "Legal Information": "" | |
| } | |
| # Extract Text ----------------------------- | |
| print("Extracting Text") | |
| if link[-3:] == '.md' or link[8:11] == 'en.': | |
| text = web_extractor(link) | |
| else: | |
| text = pdf_extractor(link) | |
| # Create Chunks ---------------------------- | |
| print("Writing Tag Data") | |
| chunks = text_splitter.create_documents(text) | |
| for chunk in chunks: | |
| with concurrent.futures.ThreadPoolExecutor() as executor: | |
| future_to_key = { | |
| executor.submit( | |
| feature_extraction, f"Product {key}", history[key], chunk.page_content | |
| ): key for key in history | |
| } | |
| for future in concurrent.futures.as_completed(future_to_key): | |
| key = future_to_key[future] | |
| try: | |
| response = future.result() | |
| history[key] = response | |
| except Exception as e: | |
| print(f"Error processing {key}: {e}") | |
| # history = detailed_history(history) | |
| print("Creating Vectors") | |
| genai_embeddings=[] | |
| for tag in history: | |
| result = genai.embed_content( | |
| model="models/embedding-001", | |
| content=history[tag], | |
| task_type="retrieval_document") | |
| genai_embeddings.append(result['embedding']) | |
| return history,genai_embeddings | |
| global text_splitter | |
| global data | |
| global history | |
| text_splitter = RecursiveCharacterTextSplitter( | |
| chunk_size = 10000, | |
| chunk_overlap = 100, | |
| separators = ["",''," "] | |
| ) | |
| if __name__ == '__main__': | |
| pass | |