RSK World - RAG Chatbot - Project Files Browser | RSK World - Free Programming Resources & Source Code

prepare_knowledge_base.py

"""
RAG Chatbot - Knowledge Base Preparation
Project: RAG Chatbot
Developer: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277
Year: 2026
Description: Script to prepare and load knowledge base documents into the vector database
"""

import os
from pathlib import Path
from vector_store import VectorStoreManager
from langchain.schema import Document
try:
    from langchain_community.document_loaders import PyPDFLoader, TextLoader
except ImportError:
    # Fallback for older LangChain versions
    from langchain.document_loaders import PyPDFLoader, TextLoader
from dotenv import load_dotenv

# Load environment variables
load_dotenv()


def load_documents_from_directory(directory_path: str):
    """
    Load all documents from a directory.
    
    Args:
        directory_path: Path to the directory containing documents
        
    Returns:
        List of Document objects
    """
    documents = []
    directory = Path(directory_path)
    
    if not directory.exists():
        print(f"Directory {directory_path} does not exist. Creating it...")
        directory.mkdir(parents=True, exist_ok=True)
        return documents
    
    # Supported file types
    supported_extensions = {'.pdf', '.txt', '.md'}
    
    for file_path in directory.rglob('*'):
        if file_path.is_file() and file_path.suffix.lower() in supported_extensions:
            try:
                if file_path.suffix.lower() == '.pdf':
                    loader = PyPDFLoader(str(file_path))
                else:
                    loader = TextLoader(str(file_path))
                
                docs = loader.load()
                documents.extend(docs)
                print(f"Loaded {len(docs)} documents from {file_path.name}")
            except Exception as e:
                print(f"Error loading {file_path}: {str(e)}")
    
    return documents


def create_sample_knowledge_base():
    """
    Create a sample knowledge base with default content.
    """
    sample_documents = [
        Document(page_content="""
        RAG (Retrieval-Augmented Generation) is a technique that combines the power of 
        information retrieval with language generation. It works by first retrieving 
        relevant documents from a knowledge base, then using those documents as context 
        for generating accurate and informed responses.
        
        The main components of a RAG system include:
        1. A knowledge base containing documents
        2. A vector database for efficient similarity search
        3. An embedding model to convert text to vectors
        4. A language model for generating responses
        5. A retrieval mechanism to find relevant context
        """),
        Document(page_content="""
        Vector databases are specialized databases designed to store and query high-dimensional 
        vectors efficiently. They use techniques like approximate nearest neighbor search to 
        quickly find similar vectors.
        
        Popular vector databases include:
        - ChromaDB: Lightweight and easy to use
        - Pinecone: Managed vector database service
        - Weaviate: Open-source vector search engine
        - FAISS: Facebook's similarity search library
        
        Vector databases are essential for RAG systems as they enable fast retrieval of 
        relevant context from large knowledge bases.
        """),
        Document(page_content="""
        Embeddings are numerical representations of text that capture semantic meaning. 
        Similar texts have similar embeddings, which allows for semantic search.
        
        OpenAI provides several embedding models:
        - text-embedding-3-small: Fast and efficient
        - text-embedding-3-large: More accurate but slower
        - text-embedding-ada-002: Previous generation model
        
        Embeddings are created by neural networks trained on large text corpora. They 
        convert text into dense vectors that can be compared using cosine similarity.
        """),
        Document(page_content="""
        LangChain is a framework for building applications with large language models. 
        It provides abstractions for:
        - Document loaders and text splitters
        - Vector stores and retrievers
        - Chains for combining LLM calls
        - Agents for complex reasoning
        
        LangChain makes it easy to build RAG applications by providing pre-built components 
        that work together seamlessly. It supports multiple LLM providers including OpenAI, 
        Anthropic, and open-source models.
        """),
        Document(page_content="""
        Best practices for RAG systems:
        
        1. Chunk size: Use appropriate chunk sizes (typically 500-1000 tokens) to balance 
           context and precision
        
        2. Overlap: Include overlap between chunks to maintain context continuity
        
        3. Retrieval: Retrieve multiple relevant chunks (typically 3-5) to provide sufficient 
           context
        
        4. Prompting: Design clear prompts that instruct the LLM to use the provided context
        
        5. Evaluation: Regularly evaluate the system's performance and update the knowledge base
        
        6. Metadata: Store metadata with documents to enable filtering and better retrieval
        """)
    ]
    
    return sample_documents


def main():
    """
    Main function to prepare the knowledge base.
    """
    print("Initializing vector store...")
    vector_store = VectorStoreManager()
    
    # Check if knowledge base directory exists
    kb_directory = "./knowledge_base"
    
    if os.path.exists(kb_directory) and any(Path(kb_directory).iterdir()):
        print(f"Loading documents from {kb_directory}...")
        documents = load_documents_from_directory(kb_directory)
        
        if documents:
            print(f"Found {len(documents)} documents. Adding to vector store...")
            vector_store.add_documents(documents)
            print("Knowledge base prepared successfully!")
        else:
            print("No documents found. Creating sample knowledge base...")
            sample_docs = create_sample_knowledge_base()
            vector_store.add_documents(sample_docs)
            print("Sample knowledge base created successfully!")
    else:
        print("Knowledge base directory is empty. Creating sample knowledge base...")
        sample_docs = create_sample_knowledge_base()
        vector_store.add_documents(sample_docs)
        print("Sample knowledge base created successfully!")
    
    # Display collection info
    count = vector_store.get_collection_count()
    print(f"\nVector store contains {count} document chunks.")


if __name__ == "__main__":
    main()

180 lines•6.8 KB

python

analytics.py

Raw Download

"""
RAG Chatbot - Analytics and Statistics
Project: RAG Chatbot
Developer: RSK World
Website: https://rskworld.in
Email: help@rskworld.in
Phone: +91 93305 39277
Year: 2026
Description: Tracks analytics, statistics, and user feedback
"""

from collections import defaultdict, Counter
from datetime import datetime, timedelta
import json
import os


class Analytics:
    """
    Tracks analytics and statistics for the chatbot.
    """
    
    def __init__(self):
        """
        Initialize analytics tracker.
        """
        self.storage_path = './analytics'
        self.data_file = os.path.join(self.storage_path, 'analytics.json')
        
        # Create storage directory if it doesn't exist
        if not os.path.exists(self.storage_path):
            os.makedirs(self.storage_path)
        
        # Load existing data
        self.data = self._load_data()
    
    def _load_data(self):
        """
        Load analytics data from disk.
        
        Returns:
            Dictionary with analytics data
        """
        if os.path.exists(self.data_file):
            try:
                with open(self.data_file, 'r', encoding='utf-8') as f:
                    return json.load(f)
            except Exception:
                pass
        
        # Return default structure
        return {
            'queries': [],
            'sessions': [],
            'feedback': [],
            'errors': [],
            'sources_used': []
        }
    
    def _save_data(self):
        """
        Save analytics data to disk.
        """
        try:
            with open(self.data_file, 'w', encoding='utf-8') as f:
                json.dump(self.data, f, indent=2, ensure_ascii=False)
        except Exception as e:
            print(f"Error saving analytics: {e}")
    
    def record_query(self, query: str, response_time: float = None, sources: list = None):
        """
        Record a query.
        
        Args:
            query: User query
            response_time: Response time in seconds
            sources: List of sources used
        """
        query_record = {
            'query': query,
            'timestamp': datetime.now().isoformat(),
            'response_time': response_time,
            'sources_count': len(sources) if sources else 0
        }
        
        self.data['queries'].append(query_record)
        
        # Record sources
        if sources:
            for source in sources:
                source_metadata = source.get('metadata', {})
                source_id = source_metadata.get('source', 'unknown')
                self.data['sources_used'].append({
                    'source': source_id,
                    'timestamp': datetime.now().isoformat()
                })
        
        # Keep only last 10000 queries
        if len(self.data['queries']) > 10000:
            self.data['queries'] = self.data['queries'][-10000:]
        
        self._save_data()
    
    def record_session(self):
        """
        Record a new session.
        """
        session_record = {
            'timestamp': datetime.now().isoformat()
        }
        self.data['sessions'].append(session_record)
        
        # Keep only last 10000 sessions
        if len(self.data['sessions']) > 10000:
            self.data['sessions'] = self.data['sessions'][-10000:]
        
        self._save_data()
    
    def record_feedback(self, positive: bool):
        """
        Record user feedback.
        
        Args:
            positive: True for positive feedback, False for negative
        """
        feedback_record = {
            'positive': positive,
            'timestamp': datetime.now().isoformat()
        }
        self.data['feedback'].append(feedback_record)
        
        # Keep only last 10000 feedback entries
        if len(self.data['feedback']) > 10000:
            self.data['feedback'] = self.data['feedback'][-10000:]
        
        self._save_data()
    
    def record_error(self, error_type: str = 'unknown'):
        """
        Record an error.
        
        Args:
            error_type: Type of error
        """
        error_record = {
            'type': error_type,
            'timestamp': datetime.now().isoformat()
        }
        self.data['errors'].append(error_record)
        
        # Keep only last 1000 errors
        if len(self.data['errors']) > 1000:
            self.data['errors'] = self.data['errors'][-1000:]
        
        self._save_data()
    
    def get_stats(self, days: int = 30):
        """
        Get statistics for the last N days.
        
        Args:
            days: Number of days to analyze
            
        Returns:
            Dictionary with statistics
        """
        cutoff_date = datetime.now() - timedelta(days=days)
        cutoff_iso = cutoff_date.isoformat()
        
        # Filter data by date
        recent_queries = [
            q for q in self.data['queries']
            if q['timestamp'] >= cutoff_iso
        ]
        
        recent_sessions = [
            s for s in self.data['sessions']
            if s['timestamp'] >= cutoff_iso
        ]
        
        recent_feedback = [
            f for f in self.data['feedback']
            if f['timestamp'] >= cutoff_iso
        ]
        
        recent_sources = [
            s for s in self.data['sources_used']
            if s['timestamp'] >= cutoff_iso
        ]
        
        # Calculate statistics
        total_queries = len(recent_queries)
        total_sessions = len(recent_sessions)
        
        # Average response time
        response_times = [q['response_time'] for q in recent_queries if q.get('response_time')]
        avg_response_time = sum(response_times) / len(response_times) if response_times else 0
        
        # Feedback score
        positive_feedback = sum(1 for f in recent_feedback if f['positive'])
        total_feedback = len(recent_feedback)
        feedback_score = (positive_feedback / total_feedback * 100) if total_feedback > 0 else 0
        
        # Top queries
        query_texts = [q['query'] for q in recent_queries]
        top_queries = Counter(query_texts).most_common(10)
        
        # Top sources
        source_names = [s['source'] for s in recent_sources]
        top_sources = Counter(source_names).most_common(10)
        
        return {
            'total_queries': total_queries,
            'total_sessions': total_sessions,
            'avg_response_time': round(avg_response_time, 2),
            'feedback_score': round(feedback_score, 1),
            'total_feedback': total_feedback,
            'positive_feedback': positive_feedback,
            'top_queries': [{'query': q, 'count': c} for q, c in top_queries],
            'top_sources': [{'source': s, 'count': c} for s, c in top_sources],
            'period_days': days
        }

223 lines•6.9 KB

python

Theme Settings

Color Scheme

Display Options

Font Size

Theme Settings

Color Scheme

Display Options

Font Size

🚀 Support RSK World