Merge branch 'bugfix/chatbotworking'

.vscode/settings.json

@@ -8,5 +8,6 @@
         "tests"
     ],
     "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true
+    "python.testing.pytestEnabled": true,
+    "git.ignoreLimitWarning": true
 }

src/agents/rag_agent.py

@@ -3,9 +3,9 @@ from dataclasses import dataclass
 from typing import List, Optional
 
 from ..llms.base_llm import BaseLLM
-from ..embeddings import BaseEmbedding
-from ..vectorstores.base_vectorstore import BaseVectorStore
-from ..utils import split_text
+from src.embeddings.base_embedding import BaseEmbedding
+from src.vectorstores.base_vectorstore import BaseVectorStore
+from src.utils.text_splitter import split_text
 
 @dataclass
 class RAGResponse:

src/embeddings/__init__.py

@@ -0,0 +1,4 @@
+# src/embeddings/__init__.py
+from .base_embedding import BaseEmbedding
+
+__all__ = ['BaseEmbedding']

src/llms/bert_llm.py

@@ -0,0 +1,44 @@
+# src/llms/bert_llm.py
+from transformers import AutoModelForSequenceClassification, AutoTokenizer, pipeline
+from typing import Optional, List
+from .base_llm import BaseLLM
+
+class BERTLanguageModel(BaseLLM):
+    def __init__(
+        self,
+        model_name: str = "bert-base-uncased",
+        max_length: int = 512
+    ):
+        """Initialize BERT model"""
+        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+        self.model = AutoModelForSequenceClassification.from_pretrained(model_name)
+        self.generator = pipeline(
+            "text-generation",
+            model=self.model,
+            tokenizer=self.tokenizer
+        )
+        self.max_length = max_length
+
+    def generate(
+        self,
+        prompt: str,
+        max_tokens: Optional[int] = None,
+        temperature: float = 0.7,
+        **kwargs
+    ) -> str:
+        """Generate text using BERT"""
+        output = self.generator(
+            prompt,
+            max_length=max_tokens or self.max_length,
+            temperature=temperature,
+            **kwargs
+        )
+        return output[0]['generated_text']
+
+    def tokenize(self, text: str) -> List[str]:
+        """Tokenize text using BERT tokenizer"""
+        return self.tokenizer.tokenize(text)
+
+    def count_tokens(self, text: str) -> int:
+        """Count tokens in text"""
+        return len(self.tokenizer.encode(text))

src/llms/falcon_llm.py

@@ -0,0 +1,39 @@
+# src/llms/falcon_llm.py
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import torch
+from typing import Optional, List
+from .base_llm import BaseLLM
+
+class FalconLanguageModel(BaseLLM):
+    def __init__(
+        self,
+        model_name: str = "tiiuae/falcon-7b",
+        device: str = "cuda" if torch.cuda.is_available() else "cpu"
+    ):
+        """Initialize Falcon model"""
+        self.tokenizer = AutoTokenizer.from_pretrained(model_name)
+        self.model = AutoModelForCausalLM.from_pretrained(
+            model_name,
+            device_map=device,
+            torch_dtype=torch.float16
+        )
+        self.device = device
+
+    def generate(
+        self,
+        prompt: str,
+        max_tokens: Optional[int] = None,
+        temperature: float = 0.7,
+        **kwargs
+    ) -> str:
+        """Generate text using Falcon"""
+        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
+        
+        outputs = self.model.generate(
+            **inputs,
+            max_length=max_tokens if max_tokens else 100,
+            temperature=temperature,
+            **kwargs
+        )
+        
+        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)

src/llms/llama_llm.py

@@ -0,0 +1,39 @@
+# src/llms/llama_llm.py
+from transformers import LlamaTokenizer, LlamaForCausalLM
+import torch
+from typing import Optional, List
+from .base_llm import BaseLLM
+
+class LlamaLanguageModel(BaseLLM):
+    def __init__(
+        self,
+        model_name: str = "meta-llama/Llama-2-7b",
+        device: str = "cuda" if torch.cuda.is_available() else "cpu"
+    ):
+        """Initialize Llama model"""
+        self.tokenizer = LlamaTokenizer.from_pretrained(model_name)
+        self.model = LlamaForCausalLM.from_pretrained(
+            model_name,
+            device_map=device,
+            torch_dtype=torch.float16
+        )
+        self.device = device
+
+    def generate(
+        self,
+        prompt: str,
+        max_tokens: Optional[int] = None,
+        temperature: float = 0.7,
+        **kwargs
+    ) -> str:
+        """Generate text using Llama"""
+        inputs = self.tokenizer(prompt, return_tensors="pt").to(self.device)
+        
+        outputs = self.model.generate(
+            **inputs,
+            max_length=max_tokens if max_tokens else 100,
+            temperature=temperature,
+            **kwargs
+        )
+        
+        return self.tokenizer.decode(outputs[0], skip_special_tokens=True)

src/llms/openai_llm.py

@@ -1,15 +1,12 @@
 # src/llms/openai_llm.py
 import openai
 from typing import Optional, List
+from openai import OpenAI  # Import the new client
 
 from .base_llm import BaseLLM
 
 class OpenAILanguageModel(BaseLLM):
-    def __init__(
-        self, 
-        api_key: str, 
-        model: str = 'gpt-3.5-turbo'
-    ):
+    def __init__(self, api_key: str, model: str = "gpt-3.5-turbo"):
         """
         Initialize OpenAI Language Model
         
@@ -17,7 +14,7 @@ class OpenAILanguageModel(BaseLLM):
             api_key (str): OpenAI API key
             model (str): Name of the OpenAI model to use
         """
-        openai.api_key = api_key
+        self.client = OpenAI(api_key=api_key)  # Use the new client
         self.model = model
     
     def generate(
@@ -38,7 +35,7 @@ class OpenAILanguageModel(BaseLLM):
         Returns:
             str: Generated response
         """
-        response = openai.ChatCompletion.create(
+        response = self.client.chat.completions.create(
             model=self.model,
             messages=[{"role": "user", "content": prompt}],
             max_tokens=max_tokens,
@@ -58,8 +55,6 @@ class OpenAILanguageModel(BaseLLM):
         Returns:
             List[str]: List of tokens
         """
-        # Note: This is a placeholder. OpenAI doesn't provide a direct 
-        # tokenization method without making an API call.
         return text.split()
     
     def count_tokens(self, text: str) -> int:
@@ -72,5 +67,4 @@ class OpenAILanguageModel(BaseLLM):
         Returns:
             int: Number of tokens
         """
-        # Approximate token counting
         return len(self.tokenize(text))

src/main.py

@@ -1,66 +1,381 @@
 # src/main.py
-from fastapi import FastAPI, Depends, HTTPException
+from fastapi import FastAPI, UploadFile, File, HTTPException, Depends, BackgroundTasks
+from fastapi.responses import StreamingResponse, JSONResponse
 from pydantic import BaseModel
-from typing import List, Optional
+from typing import List, Optional, AsyncGenerator, Dict
+import asyncio
+import json
+import uuid
+from datetime import datetime
+import aiosqlite
+from pathlib import Path
+import shutil
+import os
 
+# Import custom modules
 from .agents.rag_agent import RAGAgent
 from .llms.openai_llm import OpenAILanguageModel
 from .llms.ollama_llm import OllamaLanguageModel
+from .llms.bert_llm import BERTLanguageModel
+from .llms.falcon_llm import FalconLanguageModel
+from .llms.llama_llm import LlamaLanguageModel
 from .embeddings.huggingface_embedding import HuggingFaceEmbedding
 from .vectorstores.chroma_vectorstore import ChromaVectorStore
+from .utils.document_processor import DocumentProcessor
+from .utils.conversation_summarizer import ConversationSummarizer
+from .utils.logger import logger
 from config.config import settings
 
 app = FastAPI(title="RAG Chatbot API")
 
+# Initialize core components
+doc_processor = DocumentProcessor(
+    chunk_size=1000,
+    chunk_overlap=200,
+    max_file_size=10 * 1024 * 1024
+)
+summarizer = ConversationSummarizer()
+
+# Pydantic models
 class ChatRequest(BaseModel):
     query: str
-    context_docs: Optional[List[str]] = None
     llm_provider: str = 'openai'
+    max_context_docs: int = 3
+    temperature: float = 0.7
+    stream: bool = False
+    conversation_id: Optional[str] = None
 
 class ChatResponse(BaseModel):
     response: str
     context: Optional[List[str]] = None
+    sources: Optional[List[Dict[str, str]]] = None
+    conversation_id: str
+    timestamp: datetime
+    relevant_doc_scores: Optional[List[float]] = None
 
-@app.post("/chat", response_model=ChatResponse)
-async def chat_endpoint(request: ChatRequest):
+class DocumentResponse(BaseModel):
+    message: str
+    document_id: str
+    status: str
+    document_info: Optional[dict] = None
+
+class BatchUploadResponse(BaseModel):
+    message: str
+    processed_files: List[DocumentResponse]
+    failed_files: List[dict]
+
+class SummarizeRequest(BaseModel):
+    conversation_id: str
+    include_metadata: bool = True
+
+class SummaryResponse(BaseModel):
+    summary: str
+    key_insights: Dict
+    metadata: Optional[Dict] = None
+
+class FeedbackRequest(BaseModel):
+    rating: int
+    feedback: Optional[str] = None
+
+# Database initialization
+async def init_db():
+    async with aiosqlite.connect('chat_history.db') as db:
+        await db.execute('''
+            CREATE TABLE IF NOT EXISTS chat_history (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                conversation_id TEXT,
+                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
+                query TEXT,
+                response TEXT,
+                context TEXT,
+                sources TEXT,
+                llm_provider TEXT,
+                feedback TEXT,
+                rating INTEGER
+            )
+        ''')
+        await db.commit()
+
+# Utility functions
+def get_llm_instance(provider: str):
+    """Get LLM instance based on provider"""
+    llm_map = {
+        'openai': lambda: OpenAILanguageModel(api_key=settings.OPENAI_API_KEY),
+        'ollama': lambda: OllamaLanguageModel(base_url=settings.OLLAMA_BASE_URL),
+        'bert': lambda: BERTLanguageModel(),
+        'falcon': lambda: FalconLanguageModel(),
+        'llama': lambda: LlamaLanguageModel(),
+    }
+    
+    if provider not in llm_map:
+        raise ValueError(f"Unsupported LLM provider: {provider}")
+    return llm_map[provider]()
+
+async def get_vector_store():
+    """Initialize and return vector store with embedding model."""
     try:
-        # Select LLM based on provider
-        if request.llm_provider == 'openai':
-            llm = OpenAILanguageModel(api_key=settings.OPENAI_API_KEY)
-        elif request.llm_provider == 'ollama':
-            llm = OllamaLanguageModel(base_url=settings.OLLAMA_BASE_URL)
-        else:
-            raise HTTPException(status_code=400, detail="Unsupported LLM provider")
-        
-        # Initialize embedding and vector store
         embedding = HuggingFaceEmbedding(model_name=settings.EMBEDDING_MODEL)
         vector_store = ChromaVectorStore(
-            embedding_function=embedding.embed_documents, 
+            embedding_function=embedding.embed_documents,
             persist_directory=settings.CHROMA_PATH
         )
+        return vector_store, embedding
+    except Exception as e:
+        logger.error(f"Error initializing vector store: {str(e)}")
+        raise HTTPException(status_code=500, detail="Failed to initialize vector store")
+
+async def process_and_store_document(
+    file_path: Path,
+    vector_store: ChromaVectorStore,
+    document_id: str
+):
+    """Process document and store in vector database."""
+    try:
+        processed_doc = await doc_processor.process_document(file_path)
+        
+        vector_store.add_documents(
+            documents=processed_doc['chunks'],
+            metadatas=[{
+                'document_id': document_id,
+                'chunk_id': i,
+                'source': str(file_path.name),
+                'metadata': processed_doc['metadata']
+            } for i in range(len(processed_doc['chunks']))],
+            ids=[f"{document_id}_chunk_{i}" for i in range(len(processed_doc['chunks']))]
+        )
+        
+        return processed_doc
+    finally:
+        if file_path.exists():
+            file_path.unlink()
+
+async def store_chat_history(
+    conversation_id: str,
+    query: str,
+    response: str,
+    context: List[str],
+    sources: List[Dict],
+    llm_provider: str
+):
+    """Store chat history in database"""
+    async with aiosqlite.connect('chat_history.db') as db:
+        await db.execute(
+            '''INSERT INTO chat_history 
+               (conversation_id, query, response, context, sources, llm_provider)
+               VALUES (?, ?, ?, ?, ?, ?)''',
+            (conversation_id, query, response, json.dumps(context),
+             json.dumps(sources), llm_provider)
+        )
+        await db.commit()
+
+# Endpoints
+@app.post("/documents/upload", response_model=BatchUploadResponse)
+async def upload_documents(
+    files: List[UploadFile] = File(...),
+    background_tasks: BackgroundTasks = BackgroundTasks()
+):
+    """Upload and process multiple documents"""
+    try:
+        vector_store, _ = await get_vector_store()
+        upload_dir = Path("temp_uploads")
+        upload_dir.mkdir(exist_ok=True)
+        
+        processed_files = []
+        failed_files = []
+        
+        for file in files:
+            try:
+                document_id = str(uuid.uuid4())
+                
+                if not any(file.filename.lower().endswith(ext) 
+                          for ext in doc_processor.supported_formats):
+                    failed_files.append({
+                        "filename": file.filename,
+                        "error": "Unsupported file format"
+                    })
+                    continue
+                
+                temp_path = upload_dir / f"{document_id}_{file.filename}"
+                with open(temp_path, "wb") as buffer:
+                    shutil.copyfileobj(file.file, buffer)
+                
+                background_tasks.add_task(
+                    process_and_store_document,
+                    temp_path,
+                    vector_store,
+                    document_id
+                )
+                
+                processed_files.append(
+                    DocumentResponse(
+                        message="Document queued for processing",
+                        document_id=document_id,
+                        status="processing",
+                        document_info={
+                            "original_filename": file.filename,
+                            "size": os.path.getsize(temp_path),
+                            "content_type": file.content_type
+                        }
+                    )
+                )
+                
+            except Exception as e:
+                logger.error(f"Error processing file {file.filename}: {str(e)}")
+                failed_files.append({
+                    "filename": file.filename,
+                    "error": str(e)
+                })
+        
+        return BatchUploadResponse(
+            message=f"Processed {len(processed_files)} documents with {len(failed_files)} failures",
+            processed_files=processed_files,
+            failed_files=failed_files
+        )
+        
+    except Exception as e:
+        logger.error(f"Error in document upload: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+    
+    finally:
+        if upload_dir.exists() and not any(upload_dir.iterdir()):
+            upload_dir.rmdir()
+
+@app.post("/chat", response_model=ChatResponse)
+async def chat_endpoint(
+    request: ChatRequest,
+    background_tasks: BackgroundTasks
+):
+    """Chat endpoint with RAG support"""
+    try:
+        vector_store, embedding_model = await get_vector_store()
+        llm = get_llm_instance(request.llm_provider)
         
-        # Create RAG agent
         rag_agent = RAGAgent(
-            llm=llm, 
-            embedding=embedding, 
+            llm=llm,
+            embedding=embedding_model,
             vector_store=vector_store
         )
         
-        # Process query
-        response = rag_agent.generate_response(
-            query=request.query, 
-            context_docs=request.context_docs
+        if request.stream:
+            return StreamingResponse(
+                rag_agent.generate_streaming_response(request.query),
+                media_type="text/event-stream"
+            )
+        
+        response = await rag_agent.generate_response(
+            query=request.query,
+            temperature=request.temperature
+        )
+        
+        conversation_id = request.conversation_id or str(uuid.uuid4())
+        
+        background_tasks.add_task(
+            store_chat_history,
+            conversation_id,
+            request.query,
+            response.response,
+            response.context_docs,
+            response.sources,
+            request.llm_provider
         )
         
         return ChatResponse(
-            response=response.response, 
-            context=response.context_docs
+            response=response.response,
+            context=response.context_docs,
+            sources=response.sources,
+            conversation_id=conversation_id,
+            timestamp=datetime.now(),
+            relevant_doc_scores=response.scores if hasattr(response, 'scores') else None
         )
-    
+        
     except Exception as e:
+        logger.error(f"Error in chat endpoint: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.get("/chat/history/{conversation_id}")
+async def get_conversation_history(conversation_id: str):
+    """Get complete conversation history"""
+    async with aiosqlite.connect('chat_history.db') as db:
+        db.row_factory = aiosqlite.Row
+        async with db.execute(
+            'SELECT * FROM chat_history WHERE conversation_id = ? ORDER BY timestamp',
+            (conversation_id,)
+        ) as cursor:
+            history = await cursor.fetchall()
+            
+        if not history:
+            raise HTTPException(status_code=404, detail="Conversation not found")
+            
+        return {
+            "conversation_id": conversation_id,
+            "messages": [dict(row) for row in history]
+        }
+
+@app.post("/chat/summarize", response_model=SummaryResponse)
+async def summarize_conversation(request: SummarizeRequest):
+    """Generate a summary of a conversation"""
+    try:
+        async with aiosqlite.connect('chat_history.db') as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                'SELECT * FROM chat_history WHERE conversation_id = ? ORDER BY timestamp',
+                (request.conversation_id,)
+            ) as cursor:
+                history = await cursor.fetchall()
+        
+        if not history:
+            raise HTTPException(status_code=404, detail="Conversation not found")
+        
+        messages = [{
+            'role': 'user' if msg['query'] else 'assistant',
+            'content': msg['query'] or msg['response'],
+            'timestamp': msg['timestamp'],
+            'sources': json.loads(msg['sources']) if msg['sources'] else None
+        } for msg in history]
+        
+        summary = await summarizer.summarize_conversation(
+            messages,
+            include_metadata=request.include_metadata
+        )
+        
+        return SummaryResponse(**summary)
+        
+    except Exception as e:
+        logger.error(f"Error generating summary: {str(e)}")
+        raise HTTPException(status_code=500, detail=str(e))
+
+@app.post("/chat/feedback/{conversation_id}")
+async def submit_feedback(
+    conversation_id: str,
+    feedback_request: FeedbackRequest
+):
+    """Submit feedback for a conversation"""
+    try:
+        async with aiosqlite.connect('chat_history.db') as db:
+            await db.execute(
+                '''UPDATE chat_history 
+                   SET feedback = ?, rating = ? 
+                   WHERE conversation_id = ?''',
+                (feedback_request.feedback, feedback_request.rating, conversation_id)
+            )
+            await db.commit()
+            
+        return {"status": "Feedback submitted successfully"}
+        
+    except Exception as e:
+        logger.error(f"Error submitting feedback: {str(e)}")
         raise HTTPException(status_code=500, detail=str(e))
 
-# Optional: Health check endpoint
 @app.get("/health")
 async def health_check():
-    return {"status": "healthy"}
+    """Health check endpoint"""
+    return {"status": "healthy"}
+
+# Startup event
+@app.on_event("startup")
+async def startup_event():
+    await init_db()
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

src/utils/conversation_summarizer.py

@@ -0,0 +1,128 @@
+# src/utils/conversation_summarizer.py
+from typing import List, Dict
+from transformers import pipeline
+import numpy as np
+from datetime import datetime
+
+class ConversationSummarizer:
+    def __init__(
+        self,
+        model_name: str = "facebook/bart-large-cnn",
+        max_length: int = 130,
+        min_length: int = 30
+    ):
+        """Initialize the summarizer"""
+        self.summarizer = pipeline(
+            "summarization",
+            model=model_name,
+            device=-1  # CPU
+        )
+        self.max_length = max_length
+        self.min_length = min_length
+
+    async def summarize_conversation(
+        self,
+        messages: List[Dict],
+        include_metadata: bool = True
+    ) -> Dict:
+        """
+        Summarize a conversation and provide key insights
+        """
+        # Format conversation for summarization
+        formatted_convo = self._format_conversation(messages)
+        
+        # Generate summary
+        summary = self.summarizer(
+            formatted_convo,
+            max_length=self.max_length,
+            min_length=self.min_length,
+            do_sample=False
+        )[0]['summary_text']
+        
+        # Extract key insights
+        insights = self._extract_insights(messages)
+        
+        # Generate metadata if requested
+        metadata = self._generate_metadata(messages) if include_metadata else {}
+        
+        return {
+            'summary': summary,
+            'key_insights': insights,
+            'metadata': metadata
+        }
+
+    def _format_conversation(self, messages: List[Dict]) -> str:
+        """Format conversation for summarization"""
+        formatted = []
+        for msg in messages:
+            role = msg.get('role', 'unknown')
+            content = msg.get('content', '')
+            formatted.append(f"{role}: {content}")
+        
+        return "\n".join(formatted)
+
+    def _extract_insights(self, messages: List[Dict]) -> Dict:
+        """Extract key insights from conversation"""
+        # Count message types
+        message_counts = {
+            'user': len([m for m in messages if m.get('role') == 'user']),
+            'assistant': len([m for m in messages if m.get('role') == 'assistant'])
+        }
+        
+        # Calculate average message length
+        avg_length = np.mean([len(m.get('content', '')) for m in messages])
+        
+        # Extract main topics (simplified)
+        topics = self._extract_topics(messages)
+        
+        return {
+            'message_distribution': message_counts,
+            'average_message_length': int(avg_length),
+            'main_topics': topics,
+            'total_messages': len(messages)
+        }
+
+    def _extract_topics(self, messages: List[Dict]) -> List[str]:
+        """Extract main topics from conversation"""
+        # Combine all messages
+        full_text = " ".join([m.get('content', '') for m in messages])
+        
+        # Use the summarizer to extract main points
+        topics = self.summarizer(
+            full_text,
+            max_length=50,
+            min_length=10,
+            do_sample=False
+        )[0]['summary_text'].split('. ')
+        
+        return topics
+
+    def _generate_metadata(self, messages: List[Dict]) -> Dict:
+        """Generate conversation metadata"""
+        if not messages:
+            return {}
+            
+        return {
+            'start_time': messages[0].get('timestamp', None),
+            'end_time': messages[-1].get('timestamp', None),
+            'duration_minutes': self._calculate_duration(messages),
+            'sources_used': self._extract_sources(messages)
+        }
+
+    def _calculate_duration(self, messages: List[Dict]) -> float:
+        """Calculate conversation duration in minutes"""
+        try:
+            start_time = datetime.fromisoformat(messages[0].get('timestamp', ''))
+            end_time = datetime.fromisoformat(messages[-1].get('timestamp', ''))
+            return (end_time - start_time).total_seconds() / 60
+        except:
+            return 0
+
+    def _extract_sources(self, messages: List[Dict]) -> List[str]:
+        """Extract unique sources used in conversation"""
+        sources = set()
+        for message in messages:
+            if message.get('sources'):
+                for source in message['sources']:
+                    sources.add(source.get('filename', ''))
+        return list(sources)

src/utils/document_processor.py

@@ -0,0 +1,262 @@
+# src/utils/document_processor.py
+from typing import List, Dict, Optional, Union
+import PyPDF2
+import docx
+import pandas as pd
+import json
+from pathlib import Path
+import hashlib
+import magic  # python-magic library for file type detection
+from bs4 import BeautifulSoup
+import requests
+import csv
+from datetime import datetime
+import threading
+from queue import Queue
+import tiktoken
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+
+class DocumentProcessor:
+    def __init__(
+        self,
+        chunk_size: int = 1000,
+        chunk_overlap: int = 200,
+        max_file_size: int = 10 * 1024 * 1024,  # 10MB
+        supported_formats: Optional[List[str]] = None
+    ):
+        self.chunk_size = chunk_size
+        self.chunk_overlap = chunk_overlap
+        self.max_file_size = max_file_size
+        self.supported_formats = supported_formats or [
+            '.txt', '.pdf', '.docx', '.csv', '.json', 
+            '.html', '.md', '.xml', '.rtf'
+        ]
+        self.processing_queue = Queue()
+        self.processed_docs = {}
+        self._initialize_text_splitter()
+
+    def _initialize_text_splitter(self):
+        """Initialize the text splitter with custom settings"""
+        self.text_splitter = RecursiveCharacterTextSplitter(
+            chunk_size=self.chunk_size,
+            chunk_overlap=self.chunk_overlap,
+            length_function=len,
+            separators=["\n\n", "\n", " ", ""]
+        )
+
+    async def process_document(
+        self,
+        file_path: Union[str, Path],
+        metadata: Optional[Dict] = None
+    ) -> Dict:
+        """
+        Process a document with metadata and content extraction
+        """
+        file_path = Path(file_path)
+        
+        # Basic validation
+        if not self._validate_file(file_path):
+            raise ValueError(f"Invalid file: {file_path}")
+
+        # Extract content based on file type
+        content = self._extract_content(file_path)
+        
+        # Generate document metadata
+        doc_metadata = self._generate_metadata(file_path, content, metadata)
+        
+        # Split content into chunks
+        chunks = self.text_splitter.split_text(content)
+        
+        # Calculate embeddings chunk hashes
+        chunk_hashes = [self._calculate_hash(chunk) for chunk in chunks]
+        
+        return {
+            'content': content,
+            'chunks': chunks,
+            'chunk_hashes': chunk_hashes,
+            'metadata': doc_metadata,
+            'statistics': self._generate_statistics(content, chunks)
+        }
+
+    def _validate_file(self, file_path: Path) -> bool:
+        """
+        Validate file type, size, and content
+        """
+        if not file_path.exists():
+            raise FileNotFoundError(f"File not found: {file_path}")
+            
+        if file_path.suffix.lower() not in self.supported_formats:
+            raise ValueError(f"Unsupported file format: {file_path.suffix}")
+            
+        if file_path.stat().st_size > self.max_file_size:
+            raise ValueError(f"File too large: {file_path}")
+            
+        # Check if file is not empty
+        if file_path.stat().st_size == 0:
+            raise ValueError(f"Empty file: {file_path}")
+            
+        return True
+
+    def _extract_content(self, file_path: Path) -> str:
+        """
+        Extract content from different file formats
+        """
+        suffix = file_path.suffix.lower()
+        
+        try:
+            if suffix == '.pdf':
+                return self._extract_pdf(file_path)
+            elif suffix == '.docx':
+                return self._extract_docx(file_path)
+            elif suffix == '.csv':
+                return self._extract_csv(file_path)
+            elif suffix == '.json':
+                return self._extract_json(file_path)
+            elif suffix == '.html':
+                return self._extract_html(file_path)
+            elif suffix == '.txt':
+                return file_path.read_text(encoding='utf-8')
+            else:
+                raise ValueError(f"Unsupported format: {suffix}")
+        except Exception as e:
+            raise Exception(f"Error extracting content from {file_path}: {str(e)}")
+
+    def _extract_pdf(self, file_path: Path) -> str:
+        """Extract text from PDF with advanced features"""
+        text = ""
+        with open(file_path, 'rb') as file:
+            reader = PyPDF2.PdfReader(file)
+            metadata = reader.metadata
+            
+            for page in reader.pages:
+                text += page.extract_text() + "\n\n"
+                
+                # Extract images if available
+                if '/XObject' in page['/Resources']:
+                    for obj in page['/Resources']['/XObject'].get_object():
+                        if page['/Resources']['/XObject'][obj]['/Subtype'] == '/Image':
+                            # Process images if needed
+                            pass
+                            
+        return text.strip()
+
+    def _extract_docx(self, file_path: Path) -> str:
+        """Extract text from DOCX with formatting"""
+        doc = docx.Document(file_path)
+        full_text = []
+        
+        for para in doc.paragraphs:
+            full_text.append(para.text)
+            
+        # Extract tables if present
+        for table in doc.tables:
+            for row in table.rows:
+                row_text = [cell.text for cell in row.cells]
+                full_text.append(" | ".join(row_text))
+                
+        return "\n\n".join(full_text)
+
+    def _extract_csv(self, file_path: Path) -> str:
+        """Convert CSV to structured text"""
+        df = pd.read_csv(file_path)
+        return df.to_string()
+
+    def _extract_json(self, file_path: Path) -> str:
+        """Convert JSON to readable text"""
+        with open(file_path) as f:
+            data = json.load(f)
+        return json.dumps(data, indent=2)
+
+    def _extract_html(self, file_path: Path) -> str:
+        """Extract text from HTML with structure preservation"""
+        with open(file_path) as f:
+            soup = BeautifulSoup(f, 'html.parser')
+            
+        # Remove script and style elements
+        for script in soup(["script", "style"]):
+            script.decompose()
+            
+        text = soup.get_text(separator='\n')
+        lines = [line.strip() for line in text.splitlines() if line.strip()]
+        return "\n\n".join(lines)
+
+    def _generate_metadata(
+        self,
+        file_path: Path,
+        content: str,
+        additional_metadata: Optional[Dict] = None
+    ) -> Dict:
+        """Generate comprehensive metadata"""
+        file_stat = file_path.stat()
+        
+        metadata = {
+            'filename': file_path.name,
+            'file_type': file_path.suffix,
+            'file_size': file_stat.st_size,
+            'created_at': datetime.fromtimestamp(file_stat.st_ctime),
+            'modified_at': datetime.fromtimestamp(file_stat.st_mtime),
+            'content_hash': self._calculate_hash(content),
+            'mime_type': magic.from_file(str(file_path), mime=True),
+            'word_count': len(content.split()),
+            'character_count': len(content),
+            'processing_timestamp': datetime.now().isoformat()
+        }
+        
+        if additional_metadata:
+            metadata.update(additional_metadata)
+            
+        return metadata
+
+    def _generate_statistics(self, content: str, chunks: List[str]) -> Dict:
+        """Generate document statistics"""
+        return {
+            'total_chunks': len(chunks),
+            'average_chunk_size': sum(len(chunk) for chunk in chunks) / len(chunks),
+            'token_estimate': len(content.split()),
+            'unique_words': len(set(content.lower().split())),
+            'sentences': len([s for s in content.split('.') if s.strip()]),
+        }
+
+    def _calculate_hash(self, text: str) -> str:
+        """Calculate SHA-256 hash of text"""
+        return hashlib.sha256(text.encode()).hexdigest()
+
+    async def batch_process(
+        self,
+        file_paths: List[Union[str, Path]],
+        parallel: bool = True
+    ) -> Dict[str, Dict]:
+        """
+        Process multiple documents in parallel
+        """
+        results = {}
+        
+        if parallel:
+            threads = []
+            for file_path in file_paths:
+                thread = threading.Thread(
+                    target=self._process_and_store,
+                    args=(file_path, results)
+                )
+                threads.append(thread)
+                thread.start()
+                
+            for thread in threads:
+                thread.join()
+        else:
+            for file_path in file_paths:
+                await self._process_and_store(file_path, results)
+                
+        return results
+
+    async def _process_and_store(
+        self,
+        file_path: Union[str, Path],
+        results: Dict
+    ):
+        """Process a single document and store results"""
+        try:
+            result = await self.process_document(file_path)
+            results[str(file_path)] = result
+        except Exception as e:
+            results[str(file_path)] = {'error': str(e)}

src/vectorstores/__init__.py

@@ -0,0 +1,3 @@
+from .base_vectorstore import BaseVectorStore
+
+__all__ = ['BaseVectorStore']