Create rag_app.py

rag_app.py

@@ -0,0 +1,304 @@
+import os
+from dotenv import load_dotenv
+from langchain_groq import ChatGroq
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+from langchain.embeddings import HuggingFaceBgeEmbeddings
+from langchain_community.vectorstores import Chroma
+from langchain.chains import ConversationalRetrievalChain
+from langchain.schema import Document
+import requests
+from bs4 import BeautifulSoup
+from scrapegraphai.graphs import SmartScraperGraph
+import asyncio
+from functools import partial
+import sys
+from crawl4ai import AsyncWebCrawler, CacheMode, CrawlerRunConfig
+from langchain_community.document_loaders import TextLoader
+
+import chromadb
+from chromadb.config import Settings
+import os
+chroma_setting = Settings(anonymized_telemetry=False)
+persist_directory = "chroma_db"
+collection_metadata = {"hnsw:space": "cosine"}
+client = chromadb.PersistentClient(path=persist_directory, settings=chroma_setting)
+# Set Windows event loop policy
+if sys.platform == "win32":
+    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
+
+# Apply nest_asyncio to allow nested event loops
+import nest_asyncio  # Import nest_asyncio module for asynchronous operations
+nest_asyncio.apply()  # Apply nest_asyncio to resolve any issues with asyncio event loop
+
+# Load environment variables
+load_dotenv()
+print(os.getenv("GROQ_API_KEY"))
+
+class WebRAG:
+    def __init__(self):
+        # Initialize Groq
+        self.llm = ChatGroq(
+            api_key=os.getenv("GROQ_API_KEY"),
+            model_name="mixtral-8x7b-32768"
+        )
+        self.response_llm = ChatGroq(
+            api_key=os.getenv("GROQ_API_KEY"),
+            model_name="DeepSeek-R1-Distill-Llama-70B",
+            temperature=0.6,
+            max_tokens=2048,
+        )
+        # Initialize embeddings
+        model_kwargs = {"device": "cpu"}
+        encode_kwargs = {"normalize_embeddings": True}
+        
+        self.embeddings = HuggingFaceBgeEmbeddings(
+            model_name="BAAI/bge-base-en-v1.5",
+            model_kwargs=model_kwargs,
+            encode_kwargs=encode_kwargs
+        )
+        
+        # Initialize text splitter
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=1000,
+            chunk_overlap=200
+        )
+        
+        self.vector_store =  Chroma(embedding_function= self.embeddings,
+                        client = client,
+                    persist_directory=persist_directory,
+                    client_settings=chroma_setting,
+                    )
+        # self.qa_chain = None
+
+    def crawl_webpage_bs4(self, url):
+        """Crawl webpage using BeautifulSoup"""
+        headers = {
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
+        }
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+        
+        soup = BeautifulSoup(response.text, 'html.parser')
+        
+        # Remove script and style elements
+        for script in soup(["script", "style"]):
+            script.decompose()
+            
+        # Get text content from relevant tags
+        text_elements = soup.find_all(['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'div'])
+        content = ' '.join([elem.get_text(strip=True) for elem in text_elements])
+        
+        # Clean up whitespace
+        content = ' '.join(content.split())
+        return content
+
+    # Crawl4ai
+    async def crawl_webpage_crawl4ai_async(self, url):
+        """Crawl webpage using Crawl4ai asynchronously"""
+        try:
+            crawler_run_config = CrawlerRunConfig(cache_mode=CacheMode.BYPASS)
+            async with AsyncWebCrawler() as crawler:
+                result = await crawler.arun(url=url, config=crawler_run_config)
+                return result.markdown
+        except Exception as e:
+            raise Exception(f"Error in Crawl4ai async: {str(e)}")
+
+    def crawl_webpage_crawl4ai(self, url):
+        """Synchronous wrapper for crawl4ai"""
+        try:
+            loop = asyncio.get_event_loop()
+        except RuntimeError:
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            
+        try:
+            return loop.run_until_complete(self.crawl_webpage_crawl4ai_async(url))
+        except Exception as e:
+            raise Exception(f"Error in Crawl4ai: {str(e)}")
+
+    def crawl_webpage_scrapegraph(self, url):
+        """Crawl webpage using ScrapeGraphAI"""
+        try:
+            # First try with Groq
+            graph_config = {
+                "llm": {
+                    "api_key": os.getenv("GROQ_API_KEY"),
+                    "model": "groq/mixtral-8x7b-32768",
+                },
+                "verbose": True,
+                "headless": True,
+                "disable_async": True  # Use synchronous mode
+            }
+            
+            scraper = SmartScraperGraph(
+                prompt="Extract all the useful textual content from the webpage",
+                source=url,
+                config=graph_config
+            )
+            
+            # Use synchronous run
+            result = scraper.run()
+            print("Groq scraping successful")
+            return str(result)
+            
+        except Exception as e:
+            print(f"Groq scraping failed, falling back to Ollama: {str(e)}")
+            try:
+                # Fallback to Ollama
+                graph_config = {
+                    "llm": {
+                        "model": "ollama/deepseek-r1:8b",
+                        "temperature": 0,
+                        "max_tokens": 2048,
+                        "format": "json",
+                        "base_url": "http://localhost:11434",
+                    },
+                    "embeddings": {
+                        "model": "ollama/nomic-embed-text",
+                        "base_url": "http://localhost:11434",
+                    },
+                    "verbose": True,
+                    "disable_async": True  # Use synchronous mode
+                }
+                
+                scraper = SmartScraperGraph(
+                    prompt="Extract all the useful textual content from the webpage",
+                    source=url,
+                    config=graph_config
+                )
+                
+                result = scraper.run()
+                print("Ollama scraping successful")
+                return str(result)
+                
+            except Exception as e2:
+                raise Exception(f"Both Groq and Ollama scraping failed: {str(e2)}")
+
+    def crawl_and_process(self, url, scraping_method="beautifulsoup"):
+        """Crawl the URL and process the content"""
+        try:
+            # Validate URL
+            if not url.startswith(('http://', 'https://')):
+                raise ValueError("Invalid URL. Please include http:// or https://")
+            
+            # Crawl the website using selected method
+            if scraping_method == "beautifulsoup":
+                content = self.crawl_webpage_bs4(url)
+            elif scraping_method == "crawl4ai":
+                content = self.crawl_webpage_crawl4ai(url)
+            else:  # scrapegraph
+                content = self.crawl_webpage_scrapegraph(url)
+            
+            if not content:
+                raise ValueError("No content found at the specified URL")
+            
+            # Clean the content of any problematic characters
+            content = content.encode('utf-8', errors='ignore').decode('utf-8')
+            
+            # Create a temporary file with proper encoding
+            import tempfile
+            with tempfile.NamedTemporaryFile(mode='w', encoding='utf-8', delete=False, suffix='.txt') as temp_file:
+                temp_file.write(content)
+                temp_path = temp_file.name
+            
+            try:
+                # Load and process the document
+                docs = TextLoader(temp_path, encoding='utf-8').load()
+                docs = [Document(page_content=doc.page_content, metadata={"source": url}) for doc in docs]
+                chunks = self.text_splitter.split_documents(docs)
+                print(f"Length of chunks: {len(chunks)}")
+                print(f"First chunk: {chunks[0].metadata['source']}")
+                
+                # Check if path exists
+                data_exists = False
+                existing_urls = []
+                
+                if os.path.exists("chroma_db"):
+                    # Check if the URL is already in the metadata
+                    print(f"Checking if URL {url} is already in the metadata")
+                    try:
+                        self.vectorstore = Chroma(
+                        embedding_function=self.embeddings,
+                        client=client,
+                        persist_directory=persist_directory
+                        )
+                        entities = self.vector_store.get(include=["metadatas"])
+                        print(f"Entities: {len(entities['metadatas'])}")
+                        if len(entities['metadatas']) > 0:
+                            for entry in entities['metadatas']:
+                                #print(f"Entry: {entry}")
+                                existing_urls.append(entry["source"])
+                    except Exception as e:
+                        print(f"Error checking existing URLs: {str(e)}")
+                print(f"Existing URLs: {set(existing_urls)}")
+                if url in set(existing_urls):
+                    data_exists = True
+                    print(f"URL {url} already exists in the vector store")
+                    # Load the existing vector store
+                else:
+                    # Add new documents to the vector store
+                    MAX_BATCH_SIZE = 100
+                    for i in range(0,len(chunks),MAX_BATCH_SIZE):
+                        #print(f"start of processing: {i}")
+                        i_end = min(len(chunks),i+MAX_BATCH_SIZE)
+                        #print(f"end of processing: {i_end}")
+                        batch = chunks[i:i_end]
+                        #
+                        self.vectorstore.add_documents(batch)
+                        print(f"vectors for batch {i} to {i_end} stored successfully...")
+                    
+                
+                # Create QA chain
+                self.qa_chain = ConversationalRetrievalChain.from_llm(
+                    llm=self.response_llm,
+                    retriever=self.vector_store.as_retriever(search_type="similarity",
+                                                             search_kwargs={"k": 5,"filter":{"source": url}}),
+                    return_source_documents=True
+                )
+            
+            finally:
+                # Clean up the temporary file
+                try:
+                    os.unlink(temp_path)
+                except:
+                    pass
+                    
+        except Exception as e:
+            raise Exception(f"Error processing URL: {str(e)}")
+
+    def ask_question(self, question, chat_history=[]):
+        """Ask a question about the processed content"""
+        try:
+            if not self.qa_chain:
+                raise ValueError("Please crawl and process a URL first")
+            
+            response = self.qa_chain.invoke({"question": question, "chat_history": chat_history[:4000]})
+            print(f"Response: {response}")
+            final_answer = response["answer"].split("</think>\n\n")[-1]
+            return final_answer 
+        except Exception as e:
+            raise Exception(f"Error generating response: {str(e)}")
+
+def main():
+    # Initialize the RAG system
+    rag = WebRAG()
+    
+    # Get URL from user
+    url = input("Enter the URL to process: ")
+    print("Processing URL... This may take a moment.")
+    scraping_method = input("Choose scraping method (beautifulsoup or scrapegraph or crawl4ai): ")
+    rag.crawl_and_process(url, scraping_method)
+    
+    # Interactive Q&A loop
+    chat_history = []
+    while True:
+        question = input("\nEnter your question (or 'quit' to exit): ")
+        if question.lower() == 'quit':
+            break
+            
+        answer = rag.ask_question(question, chat_history)
+        print("\nAnswer:", answer)
+        chat_history.append((question, answer))
+
+if __name__ == "__main__":
+    main()