File size: 2,798 Bytes
9d29c62
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
import json
import re
import logging
import asyncio
from strategy_manager import StrategyManager
from pedagogical_builder import build_pedagogical_response
from orchestrator import verify_math_consistency

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

async def test_v5_pipeline():
    print("๐Ÿš€ Starting V5.0 Pipeline Verification Test")
    
    # 1. Mock Raw LLM Output (with Prefix Leakage Risk #1)
    raw_response = """
    Here is the structured solution for you:
    {
      "confidence_score": 0.95,
      "uncertainty_flag": false,
      "is_valid": true,
      "blocks": [
        {"type": "text", "content": "ื‘ื•ื ื ืคืชื•ืจ ืืช ื”ืžืฉื•ื•ืื” ื”ื ืคืœืื” ื”ื–ื•!"},
        {"type": "math", "content": "x^2 - 4 = 0"},
        {"type": "text", "content": "ื ืขื‘ื™ืจ ืื’ืฃ:"},
        {"type": "math", "content": "x^2 = 4"},
        {"type": "text", "content": "ื•ื ื•ืฆื™ื ืฉื•ืจืฉ:"},
        {"type": "math", "content": "x = \\\\pm 2"}
      ]
    }
    """
    
    # 2. Test Strategy Manager Parsing & Hardening
    sm = StrategyManager(None) # Model not needed for parsing test
    parsed = sm._parse_v5_json(raw_response)
    
    print("โœ… Step 1: Prefix Leakage Protection - PASSED")
    print(f"๐Ÿ“Š Parsed block count: {len(parsed.get('blocks', []))}")
    
    assert len(parsed['blocks']) == 6, "Block count mismatch"
    assert sm._validate_v5_schema(parsed), "Schema validation failed"
    print("โœ… Step 2: Schema Validation - PASSED")

    # 3. Test Pedagogical Builder Mapping
    response = build_pedagogical_response(
        topic_id="GENERAL",
        llm_output=parsed,
        data_anchor={}
    )
    
    steps = response['sections'][0]['steps']
    print(f"๐Ÿ“Š Pedagogical steps count: {len(steps)}")
    # Template 'GENERAL' adds: Intro step + 'Approach' step + 6 blocks + 'Solution' step = 9 steps
    assert len(steps) == 9, f"Pedagogical mapping failed, expected 9 steps, got {len(steps)}"
    print("โœ… Step 3: Pedagogical Mapping - PASSED")

    # 4. Test Orchestrator Anchor Protection (Risk #3)
    # Simulate orchestrator extraction
    blocks = parsed.get("blocks", [])
    math_blocks = [b for b in blocks if b.get("type") == "math"]
    llm_fn = math_blocks[-1]["content"] if math_blocks else "None"
    
    print(f"โš–๏ธ Extracted final result for verification: {llm_fn}")
    assert llm_fn == "x = \\pm 2", "Anchor protection extraction failed"
    
    anchor_fn = "x^2 = 4" # Simple anchor for test
    is_identical, scale = verify_math_consistency(anchor_fn, "2")
    print(f"โš–๏ธ Symbolic verification (x^2=4 vs 2): {is_identical}, scale: {scale}")
    
    print("\n๐Ÿ† V5.0 End-to-End Pipeline Verified!")

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