feat: added comprehensive tests, default model qwen

Author: jasperan

agentic_rag/OraDBVectorStore.py

@@ -338,6 +338,72 @@ def query_repo_collection(self, query: str, n_results: int = 3) -> List[Dict[str
             formatted_results.append(result)
 
         return formatted_results
+        
+    def get_collection_count(self, collection_name: str) -> int:
+        """Get the total number of chunks in a collection
+        
+        Args:
+            collection_name: Name of the collection (pdf_documents, web_documents, repository_documents, general_knowledge)
+            
+        Returns:
+            Number of chunks in the collection
+        """
+        # Map collection names to table names
+        collection_map = {
+            "pdf_documents": "PDFCollection",
+            "web_documents": "WebCollection",
+            "repository_documents": "RepoCollection", 
+            "general_knowledge": "GeneralCollection"
+        }
+        
+        table_name = collection_map.get(collection_name)
+        if not table_name:
+            raise ValueError(f"Unknown collection name: {collection_name}")
+        
+        # Count the rows in the table
+        sql = f"SELECT COUNT(*) FROM {table_name}"
+        self.cursor.execute(sql)
+        count = self.cursor.fetchone()[0]
+        
+        return count
+    
+    def get_latest_chunk(self, collection_name: str) -> Dict[str, Any]:
+        """Get the most recently inserted chunk from a collection
+        
+        Args:
+            collection_name: Name of the collection (pdf_documents, web_documents, repository_documents, general_knowledge)
+            
+        Returns:
+            Dictionary containing the content and metadata of the latest chunk
+        """
+        # Map collection names to table names
+        collection_map = {
+            "pdf_documents": "PDFCollection",
+            "web_documents": "WebCollection",
+            "repository_documents": "RepoCollection", 
+            "general_knowledge": "GeneralCollection"
+        }
+        
+        table_name = collection_map.get(collection_name)
+        if not table_name:
+            raise ValueError(f"Unknown collection name: {collection_name}")
+        
+        # Get the most recently inserted row (using ID as a proxy for insertion time)
+        # This assumes IDs are assigned sequentially or have a timestamp component
+        sql = f"SELECT Id, Text, MetaData FROM {table_name} ORDER BY ROWID DESC FETCH FIRST 1 ROW ONLY"
+        self.cursor.execute(sql)
+        row = self.cursor.fetchone()
+        
+        if not row:
+            raise ValueError(f"No chunks found in collection: {collection_name}")
+        
+        result = {
+            "id": row[0],
+            "content": row[1],
+            "metadata": json.loads(row[2]) if isinstance(row[2], str) else row[2]
+        }
+        
+        return result
 
 def main():
     parser = argparse.ArgumentParser(description="Manage Oracle DB vector store")

agentic_rag/README.md

@@ -206,6 +206,43 @@ python store.py --query "your search query"
 python local_rag_agent.py --query "your search query"
 ```
 
+#### Test Oracle DB Vector Store
+
+The system includes a test script to verify Oracle DB connectivity and examine the contents of your collections. This is useful for:
+- Checking if Oracle DB is properly configured
+- Viewing statistics about your collections
+- Inspecting the content stored in each collection
+- Testing basic vector search functionality
+
+To run the test:
+
+```bash
+# Basic test - checks connection and runs a test query
+python test_oradb.py
+
+# Show only collection statistics without inserting test data
+python test_oradb.py --stats-only
+
+# Specify a custom query for testing
+python test_oradb.py --query "artificial intelligence"
+```
+
+The script will:
+1. Verify Oracle DB credentials in your `config.yaml` file
+2. Test connection to the Oracle DB
+3. Display the total number of chunks in each collection (PDF, Web, Repository, General Knowledge)
+4. Show content and metadata from the most recently inserted chunk in each collection
+5. Unless running with `--stats-only`, insert test data and run a sample vector search
+
+Requirements:
+- Oracle DB credentials properly configured in `config.yaml`:
+  ```yaml
+  ORACLE_DB_USERNAME: ADMIN
+  ORACLE_DB_PASSWORD: your_password_here
+  ORACLE_DB_DSN: your_connection_string_here
+  ```
+- The `oracledb` Python package installed
+
 #### Use RAG Agent
 
 To query documents using either OpenAI or a local model, run:

agentic_rag/test_oradb.py

@@ -51,6 +51,59 @@ def test_connection():
         print(f"✗ Connection failed: {str(e)}")
         return None
 
+def check_collection_stats(store):
+    """Check statistics for each collection including total chunks and latest insertion"""
+    if not store:
+        print("Skipping collection stats check as connection failed")
+        return
+    
+    print("\n=== Collection Statistics ===")
+    
+    collections = {
+        "PDF Collection": "pdf_documents",
+        "Repository Collection": "repository_documents",
+        "Web Knowledge Base": "web_documents",
+        "General Knowledge": "general_knowledge"
+    }
+    
+    for name, collection in collections.items():
+        try:
+            # Get total count
+            count = store.get_collection_count(collection)
+            print(f"\n{name}:")
+            print(f"Total chunks: {count}")
+            
+            # Get latest insertion if collection is not empty
+            if count > 0:
+                latest = store.get_latest_chunk(collection)
+                print("Latest chunk:")
+                print(f"  Content: {latest['content'][:150]}..." if len(latest['content']) > 150 else f"  Content: {latest['content']}")
+                
+                # Print metadata
+                if isinstance(latest['metadata'], str):
+                    try:
+                        metadata = json.loads(latest['metadata'])
+                    except:
+                        metadata = {"source": latest['metadata']}
+                else:
+                    metadata = latest['metadata']
+                
+                source = metadata.get('source', 'Unknown')
+                print(f"  Source: {source}")
+                
+                # Print other metadata based on collection type
+                if collection == "pdf_documents" and 'page' in metadata:
+                    print(f"  Page: {metadata['page']}")
+                elif collection == "repository_documents" and 'file_path' in metadata:
+                    print(f"  File: {metadata['file_path']}")
+                elif collection == "web_documents" and 'title' in metadata:
+                    print(f"  Title: {metadata['title']}")
+            else:
+                print("No chunks found in this collection.")
+                
+        except Exception as e:
+            print(f"Error checking {name}: {str(e)}")
+
 def test_add_and_query(store, query_text="machine learning"):
     """Test adding simple data and querying it"""
     if not store:
@@ -109,6 +162,7 @@ def test_add_and_query(store, query_text="machine learning"):
 def main():
     parser = argparse.ArgumentParser(description="Test Oracle DB Vector Store")
     parser.add_argument("--query", default="machine learning", help="Query to use for testing")
+    parser.add_argument("--stats-only", action="store_true", help="Only show collection statistics without inserting test data")
 
     args = parser.parse_args()
 
@@ -146,8 +200,13 @@ def main():
     # Test connection
     store = test_connection()
 
-    # Test add and query functionality
-    test_add_and_query(store, args.query)
+    # Check collection statistics
+    check_collection_stats(store)
+    
+    # If stats-only flag is not set, also test add and query functionality
+    if not args.stats_only:
+        # Test add and query functionality
+        test_add_and_query(store, args.query)
 
     print("\n=== Test Completed ===")