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

1#!/usr/bin/env python32"""Quick test for RAG system using direct PostgreSQL connection"""3 4import psycopg25from psycopg2.extras import RealDictCursor6import os7 8def test_rag():9    """Test RAG system with direct connection"""10 11    print("🧪 Testing RAG System")12    print("=" * 70)13 14    # Connect to database15    conn = psycopg2.connect(16        dbname="granite_rfp",17        user=os.environ.get("USER"),18        host="localhost",19        port=543220    )21 22    try:23        cursor = conn.cursor(cursor_factory=RealDictCursor)24 25        # Test 1: Check table exists26        print("\n1️⃣ Checking knowledge_embeddings table...")27        cursor.execute("SELECT COUNT(*) as count FROM knowledge_embeddings")28        result = cursor.fetchone()29        print(f"   ✅ Table exists with {result['count']} embeddings")30 31        # Test 2: Check if pgvector extension is loaded32        print("\n2️⃣ Checking pgvector extension...")33        cursor.execute("SELECT * FROM pg_extension WHERE extname='vector'")34        result = cursor.fetchone()35        if result:36            print(f"   ✅ pgvector extension loaded (version: {result['extversion']})")37        else:38            print(f"   ❌ pgvector extension not found")39            return False40 41        # Test 3: Check for meetings to test with42        print("\n3️⃣ Looking for meetings to test...")43        cursor.execute("""44            SELECT meeting_id, title45            FROM meetings46            WHERE analysis IS NOT NULL47            LIMIT 348        """)49        meetings = cursor.fetchall()50 51        if not meetings:52            print("   ⚠️  No analyzed meetings found")53            print("   💡 Analyze a meeting to test auto-embedding")54            print("\n✅ RAG system setup complete but no data to embed yet")55            return True56 57        print(f"   ✅ Found {len(meetings)} analyzed meeting(s)")58        for m in meetings:59            print(f"      - {m['title'][:60]}")60 61        # Test 4: Check for RFPs62        print("\n4️⃣ Looking for RFPs to test...")63        cursor.execute("""64            SELECT rfp_id, client_name, project_title65            FROM rfp_documents66            LIMIT 367        """)68        rfps = cursor.fetchall()69 70        if rfps:71            print(f"   ✅ Found {len(rfps)} RFP(s)")72            for r in rfps:73                print(f"      - {r['client_name']}: {r['project_title'][:50]}")74        else:75            print("   ⚠️  No RFPs found")76 77        # Success summary78        print("\n" + "=" * 70)79        print("✅ RAG SYSTEM READY!")80        print("=" * 70)81        print("\n🎯 What's ready:")82        print("   ✅ pgvector extension installed and loaded")83        print("   ✅ knowledge_embeddings table created with indexes")84        print("   ✅ Auto-embedding code integrated")85        print("\n🔥 Auto-embedding will trigger for:")86        print("   ✅ New meetings (after analysis)")87        print("   ✅ New RFPs (after processing)")88        print("   ⏳ Zoho deals (need to add integration)")89        print("   ⏳ Client briefs (need to add integration)")90        print("   ⏳ Important emails (need to add integration)")91 92        if meetings or rfps:93            print(f"\n💡 Next: Analyze a new meeting or upload an RFP to see auto-embedding in action!")94 95        return True96 97    except Exception as e:98        print(f"\n❌ Error: {e}")99        import traceback100        traceback.print_exc()101        return False102    finally:103        conn.close()104 105if __name__ == "__main__":106    success = test_rag()107    exit(0 if success else 1)108