feat: add repository processing and language fix - Add RepoProcessor for GitHub repositories - Update VectorStore with repository collection - Enhance RAG agents to use repository context - Fix language selection in Gradio app

Author: jasperan

agentic_rag/gradio_app.py

@@ -68,9 +68,12 @@ def chat(message: str, history: List[List[str]], agent_type: str, use_cot: bool,
         if not agent:
             return history + [[message, "Agent not available. Please check your configuration."]]
 
+        # Convert language selection to language code
+        lang_code = "es" if language == "Spanish" else "en"
+        
         # Set CoT option and language
         agent.use_cot = use_cot
-        agent.language = language
+        agent.language = lang_code
 
         # Process query
         response = agent.process_query(message)

agentic_rag/local_rag_agent.py

@@ -165,27 +165,40 @@ def process_query(self, query: str) -> Dict[str, Any]:
 
         # First try to get context from PDF documents
         logger.info("Querying PDF collection...")
-        context = self.vector_store.query_pdf_collection(query)
-        logger.info(f"Retrieved {len(context)} context chunks")
+        pdf_context = self.vector_store.query_pdf_collection(query)
+        logger.info(f"Retrieved {len(pdf_context)} PDF context chunks")
 
-        if context:
-            # If we found relevant PDF context, use it
-            for i, ctx in enumerate(context):
+        # Then try repository documents
+        logger.info("Querying repository collection...")
+        repo_context = self.vector_store.query_repo_collection(query)
+        logger.info(f"Retrieved {len(repo_context)} repository context chunks")
+        
+        # Combine and sort context by relevance
+        all_context = pdf_context + repo_context
+        
+        if all_context:
+            # Log context sources
+            for i, ctx in enumerate(all_context):
                 source = ctx["metadata"].get("source", "Unknown")
-                pages = ctx["metadata"].get("page_numbers", [])
-                logger.info(f"Context chunk {i+1}:")
-                logger.info(f"- Source: {source}")
-                logger.info(f"- Pages: {pages}")
+                if "page_numbers" in ctx["metadata"]:
+                    pages = ctx["metadata"].get("page_numbers", [])
+                    logger.info(f"Context chunk {i+1} (PDF):")
+                    logger.info(f"- Source: {source}")
+                    logger.info(f"- Pages: {pages}")
+                else:
+                    file_path = ctx["metadata"].get("file_path", "Unknown")
+                    logger.info(f"Context chunk {i+1} (Repository):")
+                    logger.info(f"- Source: {source}")
+                    logger.info(f"- File: {file_path}")
                 logger.info(f"- Content preview: {ctx['content'][:100]}...")
 
-            logger.info("Generating response with PDF context...")
-            response = self._generate_response(query, context)
+            logger.info("Generating response with context...")
+            response = self._generate_response(query, all_context)
             logger.info("Response generated successfully")
             return response
 
-        # If no PDF context found or if it's a general knowledge query,
-        # use general knowledge
-        logger.info("No relevant PDF context found or general knowledge query detected")
+        # If no context found, use general knowledge
+        logger.info("No relevant context found")
         logger.info("Using general knowledge response...")
         return self._generate_general_response(query)
 

agentic_rag/rag_agent.py

@@ -63,15 +63,18 @@ def process_query(self, query: str) -> Dict[str, Any]:
         if analysis.query_type == "unsupported":
             return self._generate_general_response(query)
 
-        # Retrieve relevant context based on query type
-        if analysis.query_type == "pdf_documents":
-            context = self.vector_store.query_pdf_collection(query)
-        else:
-            context = self.vector_store.query_general_collection(query)
+        # First try to get context from PDF documents
+        pdf_context = self.vector_store.query_pdf_collection(query)
+        
+        # Then try repository documents
+        repo_context = self.vector_store.query_repo_collection(query)
+        
+        # Combine all context
+        all_context = pdf_context + repo_context
 
         # Generate response using context if available, otherwise use general knowledge
-        if context and analysis.requires_context:
-            response = self._generate_response(query, context)
+        if all_context and analysis.requires_context:
+            response = self._generate_response(query, all_context)
         else:
             response = self._generate_general_response(query)
 

agentic_rag/repo_processor.py

@@ -0,0 +1,112 @@
+from pathlib import Path
+from typing import List, Dict, Any, Tuple
+import json
+import argparse
+from urllib.parse import urlparse
+import warnings
+import uuid
+from gitingest import ingest
+
+def is_github_url(url: str) -> bool:
+    """Check if a string is a valid GitHub URL"""
+    try:
+        parsed = urlparse(url)
+        return parsed.netloc.lower() == "github.com"
+    except:
+        return False
+
+class RepoProcessor:
+    def __init__(self):
+        """Initialize repository processor"""
+        pass
+    
+    def _extract_metadata(self, summary: Dict[str, Any], tree: Dict[str, Any]) -> Dict[str, Any]:
+        """Extract metadata from repository summary and tree"""
+        return {
+            "repo_name": summary.get("name", ""),
+            "description": summary.get("description", ""),
+            "language": summary.get("language", ""),
+            "topics": summary.get("topics", []),
+            "stars": summary.get("stars", 0),
+            "forks": summary.get("forks", 0),
+            "last_updated": summary.get("updated_at", ""),
+            "file_count": len(tree) if tree else 0
+        }
+    
+    def process_repo(self, repo_path: str | Path) -> Tuple[List[Dict[str, Any]], str]:
+        """Process a repository and return chunks of content with metadata"""
+        try:
+            # Generate a unique document ID
+            document_id = str(uuid.uuid4())
+            
+            # Check if it's a GitHub URL
+            if isinstance(repo_path, str) and is_github_url(repo_path):
+                print(f"Processing GitHub repository: {repo_path}")
+            else:
+                print(f"Processing local repository: {repo_path}")
+            
+            # Ingest repository
+            summary, tree, content = ingest(str(repo_path))
+            
+            # Print repository information
+            print("\nRepository Summary:")
+            print(json.dumps(summary, indent=2))
+            print("\nFile Tree:")
+            print(json.dumps(tree, indent=2))
+            
+            # Extract metadata
+            metadata = self._extract_metadata(summary, tree)
+            
+            # Process content into chunks
+            processed_chunks = []
+            for file_path, file_content in content.items():
+                chunk = {
+                    "text": file_content,
+                    "metadata": {
+                        **metadata,
+                        "file_path": file_path,
+                        "source": str(repo_path),
+                        "document_id": document_id
+                    }
+                }
+                processed_chunks.append(chunk)
+            
+            return processed_chunks, document_id
+        
+        except Exception as e:
+            raise Exception(f"Error processing repository {repo_path}: {str(e)}")
+
+def main():
+    parser = argparse.ArgumentParser(description="Process GitHub repositories and extract content")
+    parser.add_argument("--input", required=True, 
+                       help="Input repository path or GitHub URL")
+    parser.add_argument("--output", required=True, help="Output JSON file for chunks")
+    
+    args = parser.parse_args()
+    processor = RepoProcessor()
+    
+    try:
+        # Create output directory if it doesn't exist
+        output_dir = Path(args.output).parent
+        output_dir.mkdir(parents=True, exist_ok=True)
+        
+        print(f"\nProcessing repository: {args.input}")
+        print("=" * 50)
+        
+        chunks, doc_id = processor.process_repo(args.input)
+        
+        # Save chunks to JSON
+        with open(args.output, 'w', encoding='utf-8') as f:
+            json.dump(chunks, f, ensure_ascii=False, indent=2)
+        
+        print("\nSummary:")
+        print(f"✓ Processed {len(chunks)} chunks")
+        print(f"✓ Document ID: {doc_id}")
+        print(f"✓ Saved to {args.output}")
+        
+    except Exception as e:
+        print(f"\n✗ Error: {str(e)}")
+        exit(1)
+
+if __name__ == "__main__":
+    main() 

agentic_rag/store.py

@@ -21,6 +21,10 @@ def __init__(self, persist_directory: str = "embeddings"):
             name="web_documents",
             metadata={"hnsw:space": "cosine"}
         )
+        self.repo_collection = self.client.get_or_create_collection(
+            name="repository_documents",
+            metadata={"hnsw:space": "cosine"}
+        )
         self.general_collection = self.client.get_or_create_collection(
             name="general_knowledge",
             metadata={"hnsw:space": "cosine"}
@@ -94,6 +98,23 @@ def add_general_knowledge(self, chunks: List[Dict[str, Any]], source_id: str):
             ids=ids
         )
 
+    def add_repo_chunks(self, chunks: List[Dict[str, Any]], document_id: str):
+        """Add chunks from a repository to the vector store"""
+        if not chunks:
+            return
+        
+        # Prepare data for ChromaDB
+        texts = [chunk["text"] for chunk in chunks]
+        metadatas = [self._sanitize_metadata(chunk["metadata"]) for chunk in chunks]
+        ids = [f"{document_id}_{i}" for i in range(len(chunks))]
+        
+        # Add to collection
+        self.repo_collection.add(
+            documents=texts,
+            metadatas=metadatas,
+            ids=ids
+        )
+    
     def query_pdf_collection(self, query: str, n_results: int = 3) -> List[Dict[str, Any]]:
         """Query the PDF documents collection"""
         results = self.pdf_collection.query(
@@ -147,6 +168,24 @@ def query_general_collection(self, query: str, n_results: int = 3) -> List[Dict[
             formatted_results.append(result)
 
         return formatted_results
+    
+    def query_repo_collection(self, query: str, n_results: int = 3) -> List[Dict[str, Any]]:
+        """Query the repository documents collection"""
+        results = self.repo_collection.query(
+            query_texts=[query],
+            n_results=n_results
+        )
+        
+        # Format results
+        formatted_results = []
+        for i in range(len(results["documents"][0])):
+            result = {
+                "content": results["documents"][0][i],
+                "metadata": results["metadatas"][0][i]
+            }
+            formatted_results.append(result)
+        
+        return formatted_results
 
 def main():
     parser = argparse.ArgumentParser(description="Manage vector store")