operations-granite/HuggingFace-Granite-AI-Practice-Agent-Dev

1#!/usr/bin/env python32"""3Quick test to verify RAG system is working4Tests: embedding generation, storage, and semantic search5"""6 7import asyncio8import sys9from pathlib import Path10 11# Add backend to path12sys.path.insert(0, str(Path(__file__).parent / "backend"))13 14from database.db_manager import DatabaseManager15from services.knowledge_embedding_service import embed_any_source16from utils.semantic_search import semantic_search17 18 19async def test_rag_system():20    """Test the RAG system end-to-end"""21 22    print("🧪 Testing RAG System")23    print("=" * 70)24 25    db = DatabaseManager()26 27    # Test 1: Check if knowledge_embeddings table exists28    print("\n1️⃣ Checking knowledge_embeddings table...")29    try:30        async with db.get_connection() as conn:31            count = await conn.fetchval("SELECT COUNT(*) FROM knowledge_embeddings")32        print(f"   ✅ Table exists with {count} embeddings")33    except Exception as e:34        print(f"   ❌ Table check failed: {e}")35        return False36 37    # Test 2: Check for existing meetings to embed38    print("\n2️⃣ Looking for existing meetings...")39    try:40        async with db.get_connection() as conn:41            meetings = await conn.fetch("""42                SELECT meeting_id, title43                FROM meetings44                WHERE analysis IS NOT NULL45                LIMIT 346            """)47 48        if not meetings:49            print("   ⚠️  No analyzed meetings found")50            print("   💡 Upload a meeting or analyze one to test embedding")51            return True52 53        print(f"   ✅ Found {len(meetings)} analyzed meeting(s)")54 55        # Test 3: Try embedding the first meeting56        print(f"\n3️⃣ Testing auto-embedding for meeting: {meetings[0]['title'][:50]}...")57        meeting_id = str(meetings[0]['meeting_id'])58 59        success = await embed_any_source('meeting', meeting_id, db)60 61        if success:62            print(f"   ✅ Successfully embedded meeting!")63        else:64            print(f"   ❌ Embedding failed")65            return False66 67        # Test 4: Verify embedding was stored68        print("\n4️⃣ Verifying embedding storage...")69        async with db.get_connection() as conn:70            stored = await conn.fetchrow("""71                SELECT source_type, source_id, metadata72                FROM knowledge_embeddings73                WHERE source_type = 'meeting' AND source_id = $174            """, meeting_id)75 76        if stored:77            print(f"   ✅ Embedding stored successfully!")78            print(f"   📊 Metadata: {stored['metadata'].get('title', 'N/A')[:60]}")79        else:80            print(f"   ❌ Embedding not found in database")81            return False82 83        # Test 5: Try semantic search84        print("\n5️⃣ Testing semantic search...")85        try:86            results = await semantic_search(87                query="budget discussions",88                limit=389            )90            print(f"   ✅ Semantic search returned {len(results)} result(s)")91 92            if results:93                print(f"   📋 Top result: {results[0].get('title', 'N/A')[:60]}")94                print(f"   🎯 Similarity: {results[0].get('similarity', 0):.3f}")95        except Exception as e:96            print(f"   ❌ Semantic search failed: {e}")97            return False98 99        # Success summary100        print("\n" + "=" * 70)101        print("✅ RAG SYSTEM FULLY OPERATIONAL!")102        print("=" * 70)103        print("\n🎯 What's working:")104        print("   ✅ Vector embeddings table created")105        print("   ✅ Auto-embedding for meetings")106        print("   ✅ Semantic search queries")107        print("\n🔥 Next steps:")108        print("   1. Analyze a new meeting → auto-embeds automatically")109        print("   2. Upload an RFP → auto-embeds automatically")110        print("   3. Try semantic search: 'budget', 'healthcare', etc.")111        print("   4. Add API endpoints for frontend integration")112        print("\n📊 Current knowledge base: {} embeddings".format(count + 1))113 114        return True115 116    except Exception as e:117        print(f"   ❌ Test failed: {e}")118        import traceback119        traceback.print_exc()120        return False121 122 123if __name__ == "__main__":124    result = asyncio.run(test_rag_system())125    sys.exit(0 if result else 1)126