server/scripts/test_learning.py

import asyncio
import logging
from app.services.unanswered_question_service import get_unanswered_service
from app.services.unified_data_manager import UnifiedDataSourceManager
from app.services.context.query_context import QueryContext
from app.core.database import AsyncSessionLocal
from app.models.unanswered_question import UnansweredQuestion
from app.models.website import Website
from sqlalchemy import select

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)

async def test_learning_loop():
    logger.info("🧪 Starting Learning Loop Verification Test...")
    
    # Initialize Model Registry for embeddings
    from app.core.model_registry import get_model_registry
    get_model_registry().initialize()
    
    async with AsyncSessionLocal() as db:
        # 1. Setup: Find or create a test website and a ticket
        website = (await db.execute(select(Website))).scalars().first()
        if not website:
            logger.error("❌ No website found to test with.")
            return

        test_question = "What is the secret code for testing?"
        test_answer = "The secret code is AL-2026-LEARN."
        
        # Create a ticket
        ticket = UnansweredQuestion(
            website_id=website.id,
            question=test_question,
            confidence_score=0.1,
            is_resolved=False
        )
        db.add(ticket)
        await db.commit()
        await db.refresh(ticket)
        ticket_id = ticket.id
        logger.info(f"✅ Created test ticket {ticket_id}")

        # 2. Simulate Admin Resolution
        service = get_unanswered_service()
        logger.info(f"📝 Resolving ticket with manual answer: {test_answer}")
        success = await service.resolve_ticket(ticket_id, test_answer)
        
        if not success:
            logger.error("❌ Failed to resolve ticket via service.")
            return
            
        # Wait a moment for any async/IO
        await asyncio.sleep(1)
        
        # 3. Verify AI Retrieval (The "Learning" part)
        manager = UnifiedDataSourceManager(db)
        
        # Create a context for a similar query
        context = QueryContext(
            original_query="Tell me the secret code for testing",
            website_id=website.id
        )
        # Mock pre-processing (usually done by orchestrator)
        from app.services.language_detector import Language
        context.english_query = context.original_query
        context.language = Language.ENGLISH
        
        logger.info(f"🔎 Querying AI for learned knowledge...")
        result = await manager.query(context)
        
        logger.info(f"🤖 AI Answer: {result['answer']}")
        logger.info(f"📊 Source: {result['source']}")
        
        if test_answer in result['answer']:
            logger.info("🎉 SUCCESS: AI correctly learned from the manual answer!")
        else:
            logger.error("❌ FAILURE: AI did not retrieve the manual answer.")

if __name__ == "__main__":
    asyncio.run(test_learning_loop())