feat: Add probabilistic transitions with weighted random selection

README.md

@@ -25,6 +25,7 @@ machines in sync or asynchonous Python codebases.
 - ✨ **Basic components**: Easily define **States**, **Events**, and **Transitions** to model your logic.
 - ⚙️ **Actions and handlers**: Attach actions and handlers to states, events, and transitions to control behavior dynamically.
 - 🛡️ **Conditional transitions**: Implement **Guards** and **Validators** to conditionally control transitions, ensuring they only occur when specific conditions are met.
+- 🎲 **Probabilistic transitions**: Define weighted transitions for non-deterministic behavior, perfect for game AI, simulations, and randomized workflows.
 - 🚀 **Full async support**: Enjoy full asynchronous support. Await events, and dispatch callbacks asynchronously for seamless integration with async codebases.
 - 🔄 **Full sync support**: Use the same state machine from synchronous codebases without any modifications.
 - 🎨 **Declarative and simple API**: Utilize a clean, elegant, and readable API to define your state machine, making it easy to maintain and understand.

docs/transitions.md

@@ -163,6 +163,77 @@ the event name is used to describe the transition.
 
 ```
 
+### Probabilistic transitions
+
+```{versionadded} 2.5.0
+Probabilistic transitions allow you to define weighted random selection when multiple transitions
+share the same event from the same source state.
+```
+
+Probabilistic transitions are useful for:
+- Game AI with non-deterministic behavior
+- Simulations requiring randomness
+- Idle animations in games
+- Randomized workflows
+
+When you define multiple transitions with the same event and source state, you can assign weights
+to control the probability of each transition being chosen:
+
+```py
+>>> class GameCharacter(StateMachine):
+...     standing = State(initial=True)
+...     shift_weight = State()
+...     adjust_hair = State()
+...     bang_shield = State()
+...
+...     # Weighted transitions: 70/20/10 probability split
+...     idle = (
+...         standing.to(shift_weight, event="idle", weight=70)
+...         | standing.to(adjust_hair, event="idle", weight=20)
+...         | standing.to(bang_shield, event="idle", weight=10)
+...     )
+...
+...     # Return transitions
+...     finish = (
+...         shift_weight.to(standing)
+...         | adjust_hair.to(standing)
+...         | bang_shield.to(standing)
+...     )
+
+```
+
+The `weight` parameter controls the relative probability of each transition. In the example above:
+- `shift_weight` has a 70% chance (70/(70+20+10))
+- `adjust_hair` has a 20% chance (20/(70+20+10))
+- `bang_shield` has a 10% chance (10/(70+20+10))
+
+```{note}
+Weights are relative, not absolute. The actual probability is calculated as `weight / sum(all_weights)`.
+```
+
+**Key behaviors:**
+
+1. **Deterministic testing**: Use `random_seed` parameter for reproducible behavior:
+
+```py
+>>> character = GameCharacter(random_seed=42)
+
+```
+
+2. **Zero/negative weights ignored**: Transitions with weight ≤ 0 are excluded from selection.
+
+3. **Mixed weighted/unweighted**: When any transition has a weight, only weighted transitions are considered.
+
+4. **Conditions still apply**: Guards and validators filter transitions before weight-based selection.
+
+5. **Backward compatibility**: If no weights are specified, the first matching transition is used (original behavior).
+
+```{tip}
+Probabilistic transitions integrate seamlessly with guards and validators. The weight-based selection
+happens first among matching transitions, then conditions are evaluated to determine if the selected
+transition can execute.
+```
+
 ## Events
 
 An event is an external signal that something has happened.

statemachine/contrib/diagram.py

@@ -127,10 +127,31 @@ def _transition_as_edge(self, transition):
         cond = ", ".join([str(cond) for cond in transition.cond])
         if cond:
             cond = f"\n[{cond}]"
+
+        # Calculate probability label if this transition has a weight
+        probability_label = ""
+        if transition.weight is not None and transition.weight > 0:
+            # Find all transitions from the same source with the same event
+            same_event_transitions = [
+                t for t in transition.source.transitions
+                if t.match(transition.event) and t.weight is not None and t.weight > 0
+            ]
+
+            if len(same_event_transitions) > 1:
+                # Calculate probability as percentage
+                total_weight = sum(t.weight for t in same_event_transitions)
+                probability = (transition.weight / total_weight) * 100
+
+                # Format as percentage if it's a clean calculation
+                if probability == int(probability):
+                    probability_label = f" [{int(probability)}%]"
+                else:
+                    probability_label = f" [{probability:.1f}%]"
+
         return pydot.Edge(
             transition.source.id,
             transition.target.id,
-            label=f"{transition.event}{cond}",
+            label=f"{transition.event}{probability_label}{cond}",
             color="blue",
             fontname=self.font_name,
             fontsize=self.transition_font_size,

statemachine/engines/async_.py

@@ -83,10 +83,42 @@ async def _trigger(self, trigger_data: TriggerData):
             return self._sentinel
 
         state = self.sm.current_state
-        for transition in state.transitions:
-            if not transition.match(trigger_data.event):
-                continue
-
+
+        # Collect all matching transitions
+        matching_transitions = [
+            t for t in state.transitions if t.match(trigger_data.event)
+        ]
+
+        if not matching_transitions:
+            if not self.sm.allow_event_without_transition:
+                raise TransitionNotAllowed(trigger_data.event, state)
+            return None
+
+        # Check if any transition has a positive weight
+        weighted_transitions = [
+            t for t in matching_transitions if t.weight is not None and t.weight > 0
+        ]
+
+        # If we have weighted transitions, select one randomly
+        if weighted_transitions:
+            weights = [t.weight for t in weighted_transitions]
+            selected_transition = self.sm._random.choices(weighted_transitions, weights=weights, k=1)[0]
+            executed, result = await self._activate(trigger_data, selected_transition)
+            if executed:
+                return result
+            # If the selected transition failed its conditions, try others
+            for transition in weighted_transitions:
+                if transition == selected_transition:
+                    continue
+                executed, result = await self._activate(trigger_data, transition)
+                if executed:
+                    return result
+            if not self.sm.allow_event_without_transition:
+                raise TransitionNotAllowed(trigger_data.event, state)
+            return None
+
+        # Otherwise, use first-match behavior (backward compatible)
+        for transition in matching_transitions:
             executed, result = await self._activate(trigger_data, transition)
             if not executed:
                 continue

statemachine/engines/sync.py

@@ -84,14 +84,45 @@ def _trigger(self, trigger_data: TriggerData):
             return self._sentinel
 
         state = self.sm.current_state
-        for transition in state.transitions:
-            if not transition.match(trigger_data.event):
-                continue
-
+
+        # Collect all matching transitions
+        matching_transitions = [
+            t for t in state.transitions if t.match(trigger_data.event)
+        ]
+
+        if not matching_transitions:
+            if not self.sm.allow_event_without_transition:
+                raise TransitionNotAllowed(trigger_data.event, state)
+            return None
+
+        # Check if any transition has a positive weight
+        weighted_transitions = [
+            t for t in matching_transitions if t.weight is not None and t.weight > 0
+        ]
+
+        # If we have weighted transitions, select one randomly
+        if weighted_transitions:
+            weights = [t.weight for t in weighted_transitions]
+            selected_transition = self.sm._random.choices(weighted_transitions, weights=weights, k=1)[0]
+            executed, result = self._activate(trigger_data, selected_transition)
+            if executed:
+                return result
+            # If the selected transition failed its conditions, try others
+            for transition in weighted_transitions:
+                if transition == selected_transition:
+                    continue
+                executed, result = self._activate(trigger_data, transition)
+                if executed:
+                    return result
+            if not self.sm.allow_event_without_transition:
+                raise TransitionNotAllowed(trigger_data.event, state)
+            return None
+
+        # Otherwise, use first-match behavior (backward compatible)
+        for transition in matching_transitions:
             executed, result = self._activate(trigger_data, transition)
             if not executed:
                 continue
-
             break
         else:
             if not self.sm.allow_event_without_transition:

statemachine/statemachine.py

@@ -1,3 +1,4 @@
+import random
 import warnings
 from inspect import isawaitable
 from typing import TYPE_CHECKING
@@ -53,6 +54,11 @@ class StateMachine(metaclass=StateMachineMetaclass):
         listeners: An optional list of objects that provies attributes to be used as callbacks.
             See :ref:`listeners` for more details.
 
+        random_seed: An optional seed for the random number generator used in probabilistic
+            transitions. When multiple transitions share the same event from the same source
+            state and have weights assigned, one will be chosen randomly. Setting a seed
+            ensures deterministic behavior for testing and reproducibility. Default: ``None``.
+
     """
 
     TransitionNotAllowed = TransitionNotAllowed
@@ -74,13 +80,15 @@ def __init__(
         rtc: bool = True,
         allow_event_without_transition: bool = False,
         listeners: "List[object] | None" = None,
+        random_seed: Any = None,
     ):
         self.model = model if model is not None else Model()
         self.state_field = state_field
         self.start_value = start_value
         self.allow_event_without_transition = allow_event_without_transition
         self._callbacks = CallbacksRegistry()
         self._states_for_instance: Dict[State, State] = {}
+        self._random = random.Random(random_seed)
 
         self._listeners: Dict[Any, Any] = {}
         """Listeners that provides attributes to be used as callbacks."""
@@ -130,17 +138,22 @@ def __repr__(self):
     def __getstate__(self):
         state = self.__dict__.copy()
         state["_rtc"] = self._engine._rtc
+        state["_random_state"] = self._random.getstate()
         del state["_callbacks"]
         del state["_states_for_instance"]
         del state["_engine"]
+        del state["_random"]
         return state
 
     def __setstate__(self, state):
         listeners = state.pop("_listeners")
         rtc = state.pop("_rtc")
+        random_state = state.pop("_random_state")
         self.__dict__.update(state)
         self._callbacks = CallbacksRegistry()
         self._states_for_instance: Dict[State, State] = {}
+        self._random = random.Random()
+        self._random.setstate(random_state)
 
         self._listeners: Dict[Any, Any] = {}
 

statemachine/transition.py

@@ -34,6 +34,10 @@ class Transition:
             before the transition is executed.
         after (Optional[Union[str, Callable, List[Callable]]]): The callbacks to be invoked
             after the transition is executed.
+        weight (Optional[float]): The weight for probabilistic transition selection. When multiple
+            transitions share the same event from the same source state and at least one has a
+            positive weight, the transition will be chosen randomly based on the weights.
+            Default ``None``.
     """
 
     def __init__(
@@ -48,10 +52,12 @@ def __init__(
         on=None,
         before=None,
         after=None,
+        weight=None,
     ):
         self.source = source
         self.target = target
         self.internal = internal
+        self.weight = weight
 
         if internal and source is not target:
             raise InvalidDefinition("Internal transitions should be self-transitions.")
@@ -75,9 +81,10 @@ def __init__(
         )
 
     def __repr__(self):
+        weight_str = f", weight={self.weight!r}" if self.weight is not None else ""
         return (
             f"{type(self).__name__}({self.source!r}, {self.target!r}, event={self.event!r}, "
-            f"internal={self.internal!r})"
+            f"internal={self.internal!r}{weight_str})"
         )
 
     def __str__(self):
@@ -137,8 +144,9 @@ def _copy_with_args(self, **kwargs):
         target = kwargs.pop("target", self.target)
         event = kwargs.pop("event", self.event)
         internal = kwargs.pop("internal", self.internal)
+        weight = kwargs.pop("weight", self.weight)
         new_transition = Transition(
-            source=source, target=target, event=event, internal=internal, **kwargs
+            source=source, target=target, event=event, internal=internal, weight=weight, **kwargs
         )
         for spec in self._specs:
             new_spec = deepcopy(spec)