You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

使用transitions库跨线程触发分层状态机转换出现TypeError故障

错误根本原因

你出现这个报错是两个核心误用导致的:

  • 首先是transitions分层状态机的嵌套规则使用错误:children配置项接收的是子状态的定义列表,不能传入已经实例化的LockedHierarchicalMachine对象。你当前的代码创建了ProductionMachine、PrepareMachine、FlashMachine三个完全独立的状态机实例,各自的上下文、状态队列完全隔离,当你在子状态回调中触发跳转时,状态路径解析逻辑会因为上下文错乱拿到None值,最终抛出你遇到的TypeError。
  • 其次存在时序风险:你在Controller的__init__方法中就直接启动了工作线程,有可能工作线程第一次调用to_connected()时,主线程还没执行controller.machine.to_prepare(),状态机还处于初始态,根本找不到prepare下的connected子状态。
修正方案

调整子状态的封装方式,不要让子状态继承LockedHierarchicalMachine,只需要封装状态定义和回调逻辑,由父状态机统一初始化嵌套结构,同时调整线程启动时机,确保状态机进入prepare状态后再处理事件:

from transitions.extensions import LockedHierarchicalMachine
from threading import Thread
from time import sleep
import logging as log

class PrepareMachine:
    def __init__(self):
        self.counter = 3
        # 只定义子状态配置,不初始化状态机
        self.states = [
            {"name": "connected", "on_enter": self.entry_connected},
            {"name": "done"},
        ]

    def entry_connected(self, event_data):
        self.counter -= 1
        if self.counter == 0:
            event_data.model.to_done()

class FlashMachine:
    def __init__(self):
        # 只定义子状态配置,不初始化状态机
        self.states = [
            {"name": "initial", "on_enter": self.entry_initial},
            {"name": "flashing"},
        ]

    def entry_initial(self, event_data):
        event_data.model.to_flashing()

class ProductionMachine(LockedHierarchicalMachine):
    def __init__(self):
        prep = PrepareMachine()
        flash = FlashMachine()
        states = [
            # children传子状态的配置列表,而非状态机实例
            {"name": "prepare", "children": prep.states, "remap": {"done": "flash"}},
            {"name": "flash", "children": flash.states},
        ]
        # 将子状态的回调实例作为model传入,共享上下文
        super().__init__(model=[self, prep, flash], states=states, queued=True, send_event=True, initial="prepare")

class Controller:
    def __init__(self):
        self.machine = ProductionMachine()
        # 先确保状态机初始化完成,再启动工作线程
        self.worker_thread = Thread(target=self.worker, name="controller")
        self.worker_thread.start()

    def worker(self):
        for i in range(3):
            sleep(0.2)
            self.machine.to_connected()

log.basicConfig(level=log.INFO)
controller = Controller()
controller.worker_thread.join()
# 测试最终状态是否正确
print(f"最终状态:{controller.machine.state}")
关键调整说明
  1. 子状态类PrepareMachine、FlashMachine不再继承LockedHierarchicalMachine,仅作为状态配置和回调的载体,所有状态逻辑由顶层的ProductionMachine统一调度,共享同一个状态上下文。
  2. 父状态机初始化时将子状态实例也传入model列表,确保子状态的回调方法能正确获取到事件上下文和状态跳转方法。
  3. 父状态机指定初始状态为prepare,不需要主线程额外调用to_prepare(),避免时序问题。

内容的提问来源于stack exchange,提问作者Jan Krejci

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.05 08:54:02