Aiogram数学练习机器人FSM状态异常求助：无法进入下一状态

我正在用Aiogram开发一款数学练习聊天机器人，流程如下：

• 发送/start命令后，用户可选择3种模式：乘法（输入1）、加减混合（输入2）、全模式（输入3）
• 接着选择数字位数
• 机器人出题，用户回答正确判定为对，错误则判定为错
  目前仅接入乘法模式，另外两种模式已准备完毕，后续会接入。

测试时遇到wait_count（数字位数选择）状态的问题：

1. 控制台无报错，但出现FSMStorageWarning提示未设置存储，状态和数据无法保存，建议使用MemoryStorage
2. 忽略警告测试：进入wait_count状态后输入2，机器人会重置状态并重复询问数字位数
3. 添加MemoryStorage后机器人完全无响应

附上原代码：

import random
from aiogram import Bot,Dispatcher,executor,types
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup

bot = Bot("...")
dp = Dispatcher(bot=bot)

def umn(ras):
    n1=random.randint(int("1"+"0"*(ras-1)),int("10"+"0"*(ras-1)))
    n2=random.randint(int("1"+"0"*(ras-1)),int("10"+"0"*(ras-1)))
    inp = str(n1)+"*"+str(n2)
    rest = n1*n2
    return inp,rest
class number(StatesGroup):
    wait_type = State()
    wait_count = State()
    wait_answer = State()
@dp.message_handler(commands=['start'])
async def start(mes:types.Message,state:FSMContext):
    await bot.send_message(mes.from_user.id,"""Welcome to the Mission Impossible math game. Here you can train your brains for multiplication, addition, subtraction. Select the mode:
Enter 1 for multiplication.
Enter 2 for addition - subtraction.
Enter 3 for the unrealistically difficult mode all in the mix""")
    await state.set_state(number.wait_type.state)
@dp.message_handler()
async def type(mes: types.Message,state: FSMContext):
    if mes.text=="":
        await mes.answer("The field is empty, try again:")
        return
    await state.update_data(type=int(mes.text))
    await mes.answer("How many digits do you need for the first time? ")
    await state.set_state(number.wait_count.state)
@dp.message_handler()
async def count(mes: types.Message,state: FSMContext):
    if mes.text=="":
        await mes.answer("The field is empty, try again")
        return
    await state.update_data(count=int(mes.text))
    global inp
    user_type = state.get_data()["type"]
    if user_type == 1:
        inp = umn(mes.text)
    else:
        inp= "Sorry"
    await mes.answer(inp[0])
    await state.set_state(number.wait_answer.state)
@dp.message_handler()
async def answer(mes: types.Message,state: FSMContext):
    if mes.text=="":
        await mes.answer("The field is empty, try again")
        return
    if int(mes.text)==inp[1]:
        await mes.answer("T")
    else:
        await mes.answer("F")
    await state.finish()
if __name__ == '__main__':
    executor.start_polling(dp)

核心问题 #

1. 无状态存储导致状态丢失：Aiogram的FSM必须配置存储才能持久化用户状态，默认不配置的话状态无法保存，导致消息处理器无法识别用户当前状态，触发错误逻辑。
2. 消息处理器无状态过滤：所有非命令消息都会被第一个无过滤的@dp.message_handler()捕获，打乱状态流转。
3. 异步方法调用错误：state.get_data()是异步方法，未加await会导致无法正确获取状态数据。
4. 全局变量不安全：全局变量inp在多用户场景下会互相覆盖，导致答案匹配错误。
5. MemoryStorage初始化错误：原代码未正确将存储传入Dispatcher，导致添加后机器人无响应。

修复后的完整代码 #

import random
from aiogram import Bot, Dispatcher, executor, types
from aiogram.dispatcher import FSMContext
from aiogram.dispatcher.filters.state import State, StatesGroup
from aiogram.contrib.fsm_storage.memory import MemoryStorage

# 初始化Bot和Dispatcher，配置MemoryStorage
bot = Bot("你的Bot Token")
storage = MemoryStorage()
dp = Dispatcher(bot=bot, storage=storage)

def umn(ras):
    # 修正数字范围：确保生成ras位数的随机数
    min_num = 10 ** (ras - 1)
    max_num = (10 ** ras) - 1
    n1 = random.randint(min_num, max_num)
    n2 = random.randint(min_num, max_num)
    inp = f"{n1}*{n2}"
    rest = n1 * n2
    return inp, rest

class number(StatesGroup):
    wait_type = State()
    wait_count = State()
    wait_answer = State()

@dp.message_handler(commands=['start'])
async def start(mes: types.Message, state: FSMContext):
    await state.finish()  # 重置之前的状态，避免残留
    await mes.answer("""Welcome to the Mission Impossible math game. Here you can train your brains for multiplication, addition, subtraction. Select the mode:
Enter 1 for multiplication.
Enter 2 for addition - subtraction.
Enter 3 for the unrealistically difficult mode all in the mix""")
    await state.set_state(number.wait_type.state)

# 仅处理wait_type状态下的消息
@dp.message_handler(state=number.wait_type)
async def type_handler(mes: types.Message, state: FSMContext):
    if not mes.text.strip():
        await mes.answer("The field is empty, try again:")
        return
    # 验证模式输入有效性
    if mes.text not in ["1", "2", "3"]:
        await mes.answer("Invalid input! Please enter 1, 2 or 3.")
        return
    await state.update_data(type=int(mes.text))
    await mes.answer("How many digits do you need for the first time? ")
    await state.set_state(number.wait_count.state)

# 仅处理wait_count状态下的消息
@dp.message_handler(state=number.wait_count)
async def count_handler(mes: types.Message, state: FSMContext):
    if not mes.text.strip():
        await mes.answer("The field is empty, try again")
        return
    # 验证数字位数输入有效性
    if not mes.text.isdigit():
        await mes.answer("Please enter a valid number!")
        return
    count = int(mes.text)
    await state.update_data(count=count)
    # 异步获取用户状态数据
    user_data = await state.get_data()
    user_type = user_data["type"]
    
    if user_type == 1:
        inp, rest = umn(count)
        # 将正确答案存入状态，替代全局变量
        await state.update_data(answer=rest)
        await mes.answer(inp)
    else:
        await mes.answer("Sorry, this mode is not available yet.")
        await state.finish()
        return
    await state.set_state(number.wait_answer.state)

# 仅处理wait_answer状态下的消息
@dp.message_handler(state=number.wait_answer)
async def answer_handler(mes: types.Message, state: FSMContext):
    if not mes.text.strip():
        await mes.answer("The field is empty, try again")
        return
    # 验证答案输入有效性
    if not mes.text.isdigit():
        await mes.answer("Please enter a valid number!")
        return
    user_answer = int(mes.text)
    user_data = await state.get_data()
    correct_answer = user_data["answer"]
    
    if user_answer == correct_answer:
        await mes.answer("T")
    else:
        await mes.answer(f"F, the correct answer is {correct_answer}")
    await state.finish()

if __name__ == '__main__':
    executor.start_polling(dp, skip_updates=True)

关键修复点说明 #

1. 正确配置MemoryStorage：初始化Dispatcher时传入storage=MemoryStorage()，解决状态无法保存的问题。
2. 给处理器添加状态过滤：每个消息处理器通过state=xxx指定仅处理对应状态的消息，避免消息被错误捕获。
3. 修正异步方法调用：给state.get_data()添加await，确保正确获取状态数据。
4. 替换全局变量：将题目答案存入FSM状态数据中，避免多用户场景下的数据冲突。
5. 增加输入验证：检查输入是否为空、是否为有效数字/模式，提升机器人鲁棒性。
6. 修复数字范围计算：原代码生成的数字位数错误，修正后确保生成指定位数的随机数。
7. 启动时跳过历史更新：添加skip_updates=True，避免机器人启动时处理积压的历史消息。

内容的提问来源于stack exchange，提问作者Johnny-kiv