File size: 3,651 Bytes
74f2af5 | 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 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 | #!/usr/bin/env python3
"""Phase 7 Integration Validation — Test bridge + orchestrator together
Quick test to verify Phase 7 works with actual CodetteOrchestrator (without full web server).
Tests the complete Path A validation.
Usage:
python validate_phase7_integration.py
"""
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "inference"))
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
print("\n" + "=" * 70)
print("Phase 7 Integration Validation Test")
print("=" * 70 + "\n")
# Test 1: Import all required modules
print("[1/4] Importing modules...")
try:
from inference.codette_orchestrator import CodetteOrchestrator
from reasoning_forge.query_classifier import QueryClassifier, QueryComplexity
from inference.codette_forge_bridge import CodetteForgeBridge
from reasoning_forge.executive_controller import ExecutiveController
print(" [OK] All imports successful\n")
except Exception as e:
print(f" [FAIL] Import error: {e}")
sys.exit(1)
# Test 2: Initialize Executive Controller
print("[2/4] Initializing Executive Controller...")
try:
exec_ctrl = ExecutiveController(verbose=False)
print(" [OK] Executive Controller initialized\n")
except Exception as e:
print(f" [FAIL] {e}")
sys.exit(1)
# Test 3: Test routing decisions with classifier
print("[3/4] Testing routing decisions...")
try:
classifier = QueryClassifier()
test_cases = [
("What is the speed of light?", QueryComplexity.SIMPLE, "SIMPLE factual"),
("How does X relate to Y?", QueryComplexity.MEDIUM, "MEDIUM conceptual"),
("Is AI conscious?", QueryComplexity.MEDIUM, "MEDIUM philosophical"),
]
for query, expected, desc in test_cases:
complexity = classifier.classify(query)
decision = exec_ctrl.route_query(query, complexity)
status = "OK" if decision.query_complexity == complexity else "MISMATCH"
latency = decision.estimated_latency_ms
cost = decision.estimated_compute_cost
print(f" [{status}] {desc:25s} - {latency:5.0f}ms, {cost:3.0f} units")
print(" [OK] All routing decisions correct\n")
except Exception as e:
print(f" [FAIL] {e}")
import traceback
traceback.print_exc()
sys.exit(1)
# Test 4: Test CodetteForgeBridge can initialize
print("[4/4] Testing CodetteForgeBridge initialization...")
try:
# Don't load full orchestrator (slow), just test bridge can be imported and instantiated
# We'll use a mock orchestrator for this test
class MockOrchestrator:
"""Mock for testing bridge initialization without loading real model."""
available_adapters = ["test"]
def route_and_generate(self, query, **kwargs):
return {"response": "test", "adapter": "test"}
mock_orch = MockOrchestrator()
bridge = CodetteForgeBridge(mock_orch, use_phase6=True, use_phase7=True, verbose=False)
if bridge.executive_controller is None:
print(" [WARN] Phase 7 Executive Controller not initialized")
print(" (This is expected if Phase 6 is disabled)")
else:
print(" [OK] Phase 7 Executive Controller initialized in bridge")
print(" [OK] CodetteForgeBridge can initialize\n")
except Exception as e:
print(f" [FAIL] {e}")
import traceback
traceback.print_exc()
sys.exit(1)
print("=" * 70)
print("PASS: Phase 7 integration validation complete!")
print("\nNext steps:")
print(" 1. Run: python run_phase7_demo.py")
print(" 2. Run: codette_web.bat")
print(" 3. Test queries in web UI at http://localhost:7860")
print("=" * 70 + "\n")
|