Spaces:
Build error
Build error
File size: 6,251 Bytes
8a682b5 |
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 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
#!/usr/bin/env python3
"""
Enhanced FSM Example
====================
A simple working example that demonstrates the Enhanced FSM features.
This can be run immediately to test the implementation.
"""
import sys
import os
import logging
from datetime import datetime
# Add src to path
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'src'))
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
def main():
"""Main example function"""
print("🚀 Enhanced FSM Example")
print("=" * 50)
try:
# Import enhanced FSM components
from src.enhanced_fsm import (
HierarchicalFSM,
AtomicState,
CompositeState,
ProbabilisticTransition,
StateDiscoveryEngine
)
print("✅ Successfully imported Enhanced FSM components")
# Step 1: Create a simple FSM
print("\n📋 Step 1: Creating Simple FSM")
hfsm = HierarchicalFSM("ExampleFSM")
# Create atomic states
planning = AtomicState("PLANNING")
execution = AtomicState("EXECUTION")
synthesis = AtomicState("SYNTHESIS")
# Add states to FSM
hfsm.add_state(planning)
hfsm.add_state(execution)
hfsm.add_state(synthesis)
print(f" Created {len(hfsm.states)} states: {list(hfsm.states.keys())}")
# Step 2: Add probabilistic transitions
print("\n🔄 Step 2: Adding Probabilistic Transitions")
# Create transitions
plan_to_exec = ProbabilisticTransition("PLANNING", "EXECUTION", 0.9)
exec_to_synth = ProbabilisticTransition("EXECUTION", "SYNTHESIS", 0.8)
# Add context modifiers
plan_to_exec.add_context_modifier("confidence<0.5", 0.5)
exec_to_synth.add_context_modifier("errors>0", 0.3)
# Add transitions to FSM
hfsm.add_transition(plan_to_exec)
hfsm.add_transition(exec_to_synth)
print(f" Added {len(hfsm.transitions)} transitions")
# Step 3: Test FSM execution
print("\n▶️ Step 3: Testing FSM Execution")
# Create context
context = {
"query": "What is 2+2?",
"confidence": 0.8,
"errors": 0,
"start_time": datetime.now()
}
# Start FSM
hfsm.start("PLANNING", context)
print(f" Started FSM in state: {hfsm.current_state.name}")
# Get available transitions
available = hfsm.get_available_transitions(context)
print(f" Available transitions: {len(available)}")
for transition in available:
print(f" -> {transition['to_state']} (p={transition['probability']:.3f})")
# Execute transitions
print("\n Executing transitions...")
# Transition to execution
success = hfsm.transition_to("EXECUTION", context)
print(f" PLANNING -> EXECUTION: {'✅' if success else '❌'}")
# Transition to synthesis
success = hfsm.transition_to("SYNTHESIS", context)
print(f" EXECUTION -> SYNTHESIS: {'✅' if success else '❌'}")
print(f" Final state: {hfsm.current_state.name}")
# Step 4: Test state discovery
print("\n🔍 Step 4: Testing State Discovery")
discovery = StateDiscoveryEngine(similarity_threshold=0.8, min_pattern_frequency=2)
# Test contexts
test_contexts = [
{
'recent_tools': ['calculator', 'search'],
'error_types': ['timeout'],
'data_stats': {'result_count': 3, 'error_rate': 0.1},
'metrics': {'execution_time': 2.0, 'confidence': 0.9}
},
{
'recent_tools': ['calculator', 'search'],
'error_types': ['timeout'],
'data_stats': {'result_count': 4, 'error_rate': 0.15},
'metrics': {'execution_time': 2.2, 'confidence': 0.85}
}
]
for i, test_context in enumerate(test_contexts):
print(f" Analyzing context {i+1}...")
pattern = discovery.analyze_context(test_context)
if pattern:
print(f" ✅ Discovered pattern: {pattern.name}")
else:
print(f" ⚠️ No new pattern discovered")
# Step 5: Test metrics and visualization
print("\n📊 Step 5: Testing Metrics and Visualization")
# Get state metrics
metrics = hfsm.get_state_metrics()
print(" State Metrics:")
for state_name, state_metrics in metrics.items():
success_rate = state_metrics.success_count / max(1, state_metrics.exit_count)
print(f" {state_name}: {success_rate:.1%} success rate, {state_metrics.avg_time:.3f}s avg time")
# Generate visualization
print("\n FSM Visualization:")
visualization = hfsm.visualize()
print(visualization)
# Export comprehensive metrics
export_data = hfsm.export_metrics()
print(f"\n Export Summary:")
print(f" FSM Name: {export_data['fsm_name']}")
print(f" Total States: {export_data['total_states']}")
print(f" Total Transitions: {export_data['total_transitions']}")
print(f" Transition Log Entries: {len(export_data['transition_log'])}")
print("\n🎉 Enhanced FSM Example Completed Successfully!")
print("=" * 50)
return True
except ImportError as e:
print(f"❌ Import Error: {e}")
print(" Make sure all dependencies are installed:")
print(" pip install numpy scikit-learn matplotlib networkx")
return False
except Exception as e:
print(f"❌ Error: {e}")
logger.exception("Detailed error information:")
return False
if __name__ == "__main__":
success = main()
sys.exit(0 if success else 1)
|