Detect and analyze emergence behavior in multi-agent RL systems through information-theoretic metrics and their dynamics.
This module detects emergence in multi-agent systems by analyzing:
Phase 1 - Exploration: High entropy (exploring), low MI (independent) Phase 2 - Emergence: Large |dS/dt|, large |dMI/dt|, rapid coordination forming Phase 3 - Exploitation: Low entropy (converged), high MI (coordinated)
Calculate Shannon entropy of system states.
Methods:
histogram: Fast, requires discretizationkde: Smooth, better for continuousplugin: Principled estimatorKey Functions:
calc = EntropyCalculator(method='histogram', bins=10)
# Single entropy value
s = calc.calculate_entropy(states)
# Trajectory of entropies
s_t = calc.calculate_trajectory_entropy(trajectory)
# Joint/conditional entropy
h_xy = calc.calculate_joint_entropy(x_states, y_states)
Calculate mutual information between agents.
Key Functions:
calc = MutualInformationCalculator(bins=10)
# Pairwise MI
mi = calc.calculate_mutual_information(agent_a_states, agent_b_states)
# Collective MI (all agents)
mi_collective = calc.calculate_collective_mi([agent1, agent2, agent3])
# Pairwise matrix
mi_matrix = calc.calculate_pairwise_mi([agent1, agent2, agent3])
# System properties
redundancy = calc.calculate_redundancy(agent_states)
synergy = calc.calculate_synergy(agent_states)
Analyze speed and acceleration of metrics.
Key Functions:
analyzer = DynamicsAnalyzer(smoothing_window=5)
# Derivatives
dS_dt = analyzer.calculate_speed(entropy_ts)
d2S_dt2 = analyzer.calculate_acceleration(entropy_ts)
# Emergence detection
signature = analyzer.calculate_emergence_signature(entropy_ts, mi_ts)
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
# Phase transitions
transitions = analyzer.detect_phase_transitions(speed, acceleration)
pip install -r requirements.txt
python test_core_modules.py
Output:
emergence_analysis.pngimport numpy as np
from entropy_calculator import EntropyCalculator
from mutual_information import MutualInformationCalculator
from dynamics_analyzer import DynamicsAnalyzer
# Create synthetic data
entropy_ts = np.random.randn(200).cumsum()
mi_ts = -entropy_ts + np.random.randn(200) * 0.2
# Calculate emergence
analyzer = DynamicsAnalyzer()
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
# Detect emergence windows
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
print(f"Emergence windows: {windows}")
import numpy as np
from entropy_calculator import EntropyCalculator
from mutual_information import MutualInformationCalculator
from dynamics_analyzer import DynamicsAnalyzer
# Your RL environment
env = YourMultiAgentEnv()
agents = [agent1, agent2, agent3]
# Collect state history (from training)
state_history = [] # shape: (n_timesteps, n_agents, state_dim)
for t in range(10000):
actions = [agent.act() for agent in agents]
obs, rewards, done, info = env.step(actions)
state_history.append(obs)
state_history = np.array(state_history)
# Calculate entropy and MI over time
entropy_calc = EntropyCalculator()
mi_calc = MutualInformationCalculator()
entropy_ts = entropy_calc.calculate_state_entropy_over_time(state_history, window_size=50)
mi_ts = mi_calc.calculate_collective_mi_over_time([state_history[:, i] for i in range(n_agents)], window_size=50)
# Analyze emergence
analyzer = DynamicsAnalyzer()
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
print(f"Emergence detected in windows: {windows}")
print(f"Peak emergence score: {em_score.max():.3f}")
S = -Σ p_i * log₂(p_i)
MI(X;Y) = H(X) + H(Y) - H(X,Y)
= Σ p(x,y) * log₂(p(x,y) / (p(x)*p(y)))
dS/dt ≈ (S(t+1) - S(t-1)) / 2
d²S/dt² ≈ (dS/dt(t+1) - dS/dt(t-1)) / 2
E(t) = (|dS/dt| + |dMI/dt|) / (1 + d²S/dt² + d²MI/dt²)
Emergence detected when E(t) > 0.5
Phase 2 will add:
emergence_detector.py - Main detection enginemulti_agent_rl_wrapper.py - RL integrationvisualizer.py - Advanced plottingbenchmark_scenarios.py - Test environments✓ Multiple entropy calculation methods (histogram, KDE, plugin) ✓ Pairwise and collective mutual information ✓ System redundancy and synergy measures ✓ Speed and acceleration analysis ✓ Phase transition detection ✓ Emergence scoring and windowing ✓ Stability and baseline comparison ✓ Transfer entropy for directed information flow
python test_core_modules.pyemergence_analysis.pngThis emergence detection framework is designed for:
The three core modules provide robust foundation for detecting and characterizing emergence through information-theoretic metrics.
MIT License - See LICENSE file
For questions or issues, refer to the main project documentation.
28 commits
1 commits
Python
72.6%
TeX
19.7%
Shell
6.3%
Detect and analyze emergence behavior in multi-agent RL systems through information-theoretic metrics and their dynamics.
This module detects emergence in multi-agent systems by analyzing:
Phase 1 - Exploration: High entropy (exploring), low MI (independent) Phase 2 - Emergence: Large |dS/dt|, large |dMI/dt|, rapid coordination forming Phase 3 - Exploitation: Low entropy (converged), high MI (coordinated)
Calculate Shannon entropy of system states.
Methods:
histogram: Fast, requires discretizationkde: Smooth, better for continuousplugin: Principled estimatorKey Functions:
calc = EntropyCalculator(method='histogram', bins=10)
# Single entropy value
s = calc.calculate_entropy(states)
# Trajectory of entropies
s_t = calc.calculate_trajectory_entropy(trajectory)
# Joint/conditional entropy
h_xy = calc.calculate_joint_entropy(x_states, y_states)
Calculate mutual information between agents.
Key Functions:
calc = MutualInformationCalculator(bins=10)
# Pairwise MI
mi = calc.calculate_mutual_information(agent_a_states, agent_b_states)
# Collective MI (all agents)
mi_collective = calc.calculate_collective_mi([agent1, agent2, agent3])
# Pairwise matrix
mi_matrix = calc.calculate_pairwise_mi([agent1, agent2, agent3])
# System properties
redundancy = calc.calculate_redundancy(agent_states)
synergy = calc.calculate_synergy(agent_states)
Analyze speed and acceleration of metrics.
Key Functions:
analyzer = DynamicsAnalyzer(smoothing_window=5)
# Derivatives
dS_dt = analyzer.calculate_speed(entropy_ts)
d2S_dt2 = analyzer.calculate_acceleration(entropy_ts)
# Emergence detection
signature = analyzer.calculate_emergence_signature(entropy_ts, mi_ts)
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
# Phase transitions
transitions = analyzer.detect_phase_transitions(speed, acceleration)
pip install -r requirements.txt
python test_core_modules.py
Output:
emergence_analysis.pngimport numpy as np
from entropy_calculator import EntropyCalculator
from mutual_information import MutualInformationCalculator
from dynamics_analyzer import DynamicsAnalyzer
# Create synthetic data
entropy_ts = np.random.randn(200).cumsum()
mi_ts = -entropy_ts + np.random.randn(200) * 0.2
# Calculate emergence
analyzer = DynamicsAnalyzer()
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
# Detect emergence windows
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
print(f"Emergence windows: {windows}")
import numpy as np
from entropy_calculator import EntropyCalculator
from mutual_information import MutualInformationCalculator
from dynamics_analyzer import DynamicsAnalyzer
# Your RL environment
env = YourMultiAgentEnv()
agents = [agent1, agent2, agent3]
# Collect state history (from training)
state_history = [] # shape: (n_timesteps, n_agents, state_dim)
for t in range(10000):
actions = [agent.act() for agent in agents]
obs, rewards, done, info = env.step(actions)
state_history.append(obs)
state_history = np.array(state_history)
# Calculate entropy and MI over time
entropy_calc = EntropyCalculator()
mi_calc = MutualInformationCalculator()
entropy_ts = entropy_calc.calculate_state_entropy_over_time(state_history, window_size=50)
mi_ts = mi_calc.calculate_collective_mi_over_time([state_history[:, i] for i in range(n_agents)], window_size=50)
# Analyze emergence
analyzer = DynamicsAnalyzer()
em_score = analyzer.calculate_emergence_score(entropy_ts, mi_ts)
windows = analyzer.identify_emergence_windows(em_score, threshold=0.5)
print(f"Emergence detected in windows: {windows}")
print(f"Peak emergence score: {em_score.max():.3f}")
S = -Σ p_i * log₂(p_i)
MI(X;Y) = H(X) + H(Y) - H(X,Y)
= Σ p(x,y) * log₂(p(x,y) / (p(x)*p(y)))
dS/dt ≈ (S(t+1) - S(t-1)) / 2
d²S/dt² ≈ (dS/dt(t+1) - dS/dt(t-1)) / 2
E(t) = (|dS/dt| + |dMI/dt|) / (1 + d²S/dt² + d²MI/dt²)
Emergence detected when E(t) > 0.5
Phase 2 will add:
emergence_detector.py - Main detection enginemulti_agent_rl_wrapper.py - RL integrationvisualizer.py - Advanced plottingbenchmark_scenarios.py - Test environments✓ Multiple entropy calculation methods (histogram, KDE, plugin) ✓ Pairwise and collective mutual information ✓ System redundancy and synergy measures ✓ Speed and acceleration analysis ✓ Phase transition detection ✓ Emergence scoring and windowing ✓ Stability and baseline comparison ✓ Transfer entropy for directed information flow
python test_core_modules.pyemergence_analysis.pngThis emergence detection framework is designed for:
The three core modules provide robust foundation for detecting and characterizing emergence through information-theoretic metrics.
MIT License - See LICENSE file
For questions or issues, refer to the main project documentation.
28 commits
1 commits
Python
72.6%
TeX
19.7%
Shell
6.3%