Telegram bot 使用 Finite State Machine 时无法等待用户回复

2026-10-24支付老李

结论

问题不是 FSM “不会等待”，而是消息处理器的匹配条件和状态转换时机有误。

pyTelegramBotAPI 只会在收到消息时调用 handler。要实现逐步问答，流程应当是：

发送当前问题；
将 FSM 切换到等待该问题答案的状态；
结束当前 handler；
用户发送新消息后，根据当前状态处理答案；
保存答案并切换状态，然后发送下一个问题。

原代码中的 @bot.message_handler(content_types=['text']) 几乎会匹配所有文本消息。它与下面的状态过滤器叠加后，并不会形成”同时满足两个条件”的关系，而是将同一个函数注册为两个独立的 handler，因此可能绕过状态判断。

另外，第一个处理器发送尺寸问题后调用了：

fsm.asked_for_payment_method(message.text)

但此时应该进入等待尺寸答案的状态，而不是保存付款方式。

重新定义状态

状态名称可以直接描述机器人正在等待的内容：

from transitions import Machine


class TelegramBotFSM:
    states = [
        'idle',
        'waiting_size',
        'waiting_payment',
        'waiting_flavour',
        'waiting_confirmation',
    ]

    transitions = [
        {
            'trigger': 'start_order',
            'source': 'idle',
            'dest': 'waiting_size',
        },
        {
            'trigger': 'select_size',
            'source': 'waiting_size',
            'dest': 'waiting_payment',
            'before': 'update_size',
        },
        {
            'trigger': 'select_payment',
            'source': 'waiting_payment',
            'dest': 'waiting_flavour',
            'before': 'update_payment',
        },
        {
            'trigger': 'select_flavour',
            'source': 'waiting_flavour',
            'dest': 'waiting_confirmation',
            'before': 'update_flavour',
        },
        {
            'trigger': 'finish',
            'source': 'waiting_confirmation',
            'dest': 'idle',
            'after': 'clear_order',
        },
    ]

    def __init__(self):
        self.size = ''
        self.pay_method = ''
        self.flavour = ''

        self.machine = Machine(
            model=self,
            states=self.states,
            transitions=self.transitions,
            initial='idle',
        )

    def update_size(self, value):
        self.size = value

    def update_payment(self, value):
        self.pay_method = value

    def update_flavour(self, value):
        self.flavour = value

    def clear_order(self):
        self.size = ''
        self.pay_method = ''
        self.flavour = ''

Machine 中引用的类名必须与实际定义一致。原示例定义的是 telegram_bot_state，初始化时却引用了 TelegramBot.states。如果没有其他相关代码，这里会触发 NameError。

使用一个文本入口按状态分发

这种线性流程更适合使用一个文本 handler，再根据状态分发消息。这样也不会受到多个 handler 匹配顺序的影响：

import telebot
import config

from telebot import types
from FSM import TelegramBotFSM


bot = telebot.TeleBot(config.TOKEN)
fsm = TelegramBotFSM()


@bot.message_handler(commands=['start'])
def welcome(message):
    markup = types.ReplyKeyboardMarkup(resize_keyboard=True)
    markup.add(
        types.KeyboardButton("Заказать пиццу"),
        types.KeyboardButton("Посмотреть меню"),
    )

    bot.send_message(
        message.chat.id,
        (
            f"Добро пожаловать, {message.from_user.first_name}!\n"
            "Я помогу вам сделать заказ."
        ),
        reply_markup=markup,
    )


@bot.message_handler(content_types=['text'])
def handle_text(message):
    text = message.text.strip()

    if fsm.state == 'idle':
        if text != 'Заказать пиццу':
            bot.send_message(
                message.chat.id,
                'Чтобы начать заказ, нажмите «Заказать пиццу».',
            )
            return

        fsm.start_order()
        bot.send_message(
            message.chat.id,
            'Какую вы хотите пиццу? Большую или маленькую?',
        )
        return

    if fsm.state == 'waiting_size':
        fsm.select_size(text)
        bot.send_message(
            message.chat.id,
            'Как вы будете платить?',
        )
        return

    if fsm.state == 'waiting_payment':
        fsm.select_payment(text)
        bot.send_message(
            message.chat.id,
            'Какой вкус вы хотите?',
        )
        return

    if fsm.state == 'waiting_flavour':
        fsm.select_flavour(text)

        markup = types.InlineKeyboardMarkup(row_width=2)
        markup.add(
            types.InlineKeyboardButton("Да", callback_data='order_yes'),
            types.InlineKeyboardButton("Нет", callback_data='order_no'),
        )

        bot.send_message(
            message.chat.id,
            (
                f'Вы хотите пиццу размера «{fsm.size}», '
                f'вкус: «{fsm.flavour}», '
                f'способ оплаты: «{fsm.pay_method}». Всё верно?'
            ),
            reply_markup=markup,
        )
        return

每个分支末尾都要使用 return，明确结束当前消息的处理。后续步骤只能由用户发送的新消息触发。

处理确认按钮

确认按钮只应在 waiting_confirmation 状态下生效：

@bot.callback_query_handler(
    func=lambda call: call.data in ('order_yes', 'order_no')
)
def handle_confirmation(call):
    if fsm.state != 'waiting_confirmation':
        bot.answer_callback_query(
            call.id,
            text='这个订单已经失效，请重新下单。',
        )
        return

    summary = (
        f'Пицца: {fsm.size}, '
        f'вкус: {fsm.flavour}, '
        f'оплата: {fsm.pay_method}'
    )

    if call.data == 'order_yes':
        bot.edit_message_text(
            summary,
            chat_id=call.message.chat.id,
            message_id=call.message.message_id,
        )
        bot.send_message(
            call.message.chat.id,
            'Спасибо за заказ',
        )
        bot.answer_callback_query(
            call.id,
            text='Заказ оформлен',
            show_alert=True,
        )
    else:
        bot.edit_message_text(
            summary,
            chat_id=call.message.chat.id,
            message_id=call.message.message_id,
        )
        bot.send_message(
            call.message.chat.id,
            'Заказ отменён.',
        )
        bot.answer_callback_query(call.id)

    fsm.finish()


bot.infinity_polling()

这里必须先调用 edit_message_text()，再调用 fsm.finish()。finish() 会清空订单数据，如果提前执行，编辑后的订单摘要中只会剩下空字符串。

如果继续使用多个 handler

使用多个 handler 也可以，但不要在状态处理器上再叠加一个无条件的 content_types=['text']：

@bot.message_handler(
    func=lambda message: (
        message.content_type == 'text'
        and fsm.state == 'waiting_size'
    )
)
def handle_size(message):
    fsm.select_size(message.text.strip())
    bot.send_message(message.chat.id, 'Как вы будете платить?')


@bot.message_handler(
    func=lambda message: (
        message.content_type == 'text'
        and fsm.state == 'waiting_payment'
    )
)
def handle_payment(message):
    fsm.select_payment(message.text.strip())
    bot.send_message(message.chat.id, 'Какой вкус вы хотите?')

不要这样写：

@bot.message_handler(content_types=['text'])
@bot.message_handler(func=lambda message: fsm.state == 'waiting_size')

这两个装饰器不会组成”文本类型并且状态正确”的匹配条件，而会分别注册一次 handler。仅 content_types=['text'] 就足以匹配所有文本消息。

多用户场景必须分别保存 FSM

上面的单个 fsm 对象只适合本地测试：

fsm = TelegramBotFSM()

在真实 bot 中，所有聊天都会共享这个对象。用户 A 选择尺寸后，用户 B 发来的消息可能被当成用户 A 的付款方式。

至少需要按 chat.id 分别保存状态：

from collections import defaultdict


sessions = defaultdict(TelegramBotFSM)


def get_fsm(message):
    return sessions[message.chat.id]

然后在 handler 中获取当前聊天对应的 FSM：

@bot.message_handler(content_types=['text'])
def handle_text(message):
    fsm = get_fsm(message)
    text = message.text.strip()

    if fsm.state == 'idle':
        # ...
        return

    if fsm.state == 'waiting_size':
        # ...
        return

处理回调按钮时，也要获取对应聊天的状态：

@bot.callback_query_handler(
    func=lambda call: call.data in ('order_yes', 'order_no')
)
def handle_confirmation(call):
    fsm = sessions[call.message.chat.id]
    # ...

如果同一个私聊用户可能同时从多个入口操作，可以使用 (chat.id, from_user.id) 作为会话键。生产环境还要考虑进程重启后的状态持久化。多进程部署时，则需要使用 Redis、数据库等共享存储。

原则很简单：发送问题时只切换到等待状态，然后结束 handler。等下一条消息到达后，再保存答案、执行状态转换并发送下一个问题。