ganireddikumar/AI_Debate_Club

1# knowledge_base.py2import os3from typing import List, Optional4from loaders import process_documents5from langchain.text_splitter import RecursiveCharacterTextSplitter6from langchain.docstore.document import Document7import chromadb8 9client = chromadb.PersistentClient(path="chroma_db")  # Persist data to disk10 11def process_uploaded_docs(file_obj: Optional[str] = None):12    """Process uploaded documents from Gradio13    14    Args:15        file_obj: File object from Gradio upload16    """17    collection = client.get_or_create_collection(name="debate_docs")18    valid_docs = []19    20    if file_obj:21        # Handle single file or list of files22        files = [file_obj] if isinstance(file_obj, str) else [file_obj]  # Ensure files is always a list23        24        for file_path in files:25            if not isinstance(file_path, str):26                print(f"Skipping invalid file object: {file_path}")27                continue28                29            try:30                content = process_documents(file_path)31                if content:32                    valid_docs.append(content)33            except Exception as e:34                print(f"Error processing file {file_path}: {str(e)}")35 36        if valid_docs:37            text_splitter = RecursiveCharacterTextSplitter(38                chunk_size=500,  # Reduced chunk size39                chunk_overlap=100,   # Reduced overlap40                length_function=len,  # Explicitly set the length function41            )42            43            # Create Document objects and split into chunks44            docs = [Document(page_content=doc) for doc in valid_docs]45            chunks = text_splitter.split_documents(docs)46            47            # Add to ChromaDB48            collection.add(49                documents=[chunk.page_content for chunk in chunks],50                ids=[f"doc_{i}" for i in range(len(chunks))]51            )52            53    return collection54 55def retrieve_evidence(query: str, n_results: int = 3):56    """Retrieve relevant evidence from the knowledge base"""57    collection = client.get_or_create_collection(name="debate_docs")58    results = collection.query(59        query_texts=[query],60        n_results=n_results61    )62    63    if not results or "documents" not in results or not results["documents"]:64        return "No relevant evidence found"65    66    # Format for T5 compatibility67    return " ".join([68        f"[Source {i+1}] {doc}" 69        for i, doc in enumerate(results["documents"][0][:n_results])70    ])