如何在Aiogram中实现含循环的交互式功能？控制台转Bot遇困

我正在把一个控制台运行的单词复习程序迁移到基于Aiogram的Telegram Bot，但不知道如何处理其中的check函数。原函数代码如下：

def check():
    print('Here are your dictionaries:')
    retrieve_dictionaries()
    dict_name = input('Select a dictionary: ''')
    cur = conn.cursor()
    cur.execute('''SELECT * from {tab};'''.format(tab = dict_name))
    columns = list(map(lambda x: x[0], cur.description))
    print('Here are possible directions:','
','1.',columns[1],'--',columns[2],'
','2.',columns[2],'--',columns[1])    
    direction = input ('From where do we start? Choose 1 or 2: ')
    if direction == '1':
        box = int(input('From which box do we take the cards? Choose from 1 to 4: '))
        cur.execute('''SELECT * from {tab} WHERE box12 == {box_value} ORDER BY RANDOM()'''.format(tab = dict_name, box_value = box))
        checkwords = cur.fetchall()
        checked = len(checkwords)
        print(checked)
        print('Excellent! There are',checked, 'words to learn. Let us start!')
        print(checkwords)
        id_correct = []
        id_forgotten = []
        id_remembered = []
        while checked > 0:
            for i in checkwords:
                if i[0] in id_correct:
                    continue
                elif i[0] in id_forgotten:
                    print(i[1])
                    answer = input('Insert answer: ')
                    if answer == i[2]:
                        print('Correct!')
                        checked = checked - 1
                        id_remembered.append(i[0])
                    else:
                        print('Nope, the correct answer is:', i[2])
                else:
                    print(i[1])
                    answer = input('Insert answer: ')
                    if answer == i[2]:
                        print('Correct!')
                        checked = checked - 1
                        id_correct.append(i[0])
                    else:
                        print('Nope, the correct answer is:', i[2])
                        id_forgotten.append(i[0])
                        
        print(id_correct)
        print(id_forgotten)
        for i in id_correct:
            cur.execute('''UPDATE {tab} SET box12 = box12 + 1 WHERE id == {word_id}'''.format(tab = dict_name, word_id = i))
    if direction == '2':
        box = int(input('From which box do we take the cards? Choose from 1 to 4: '))
        cur.execute('''SELECT * from {tab} WHERE box21 == {box_value} ORDER BY RANDOM()'''.format(tab = dict_name, box_value = box))
        checkwords = cur.fetchall()
        checked = len(checkwords)
        print(checked)
        print('Excellent! There are',checked, 'words to learn. Let us start!')
        print(checkwords)
        id_correct = []
        id_forgotten = []
        id_remembered = []
        while checked > 0:
            for i in checkwords:
                if i[0] in id_correct:
                    continue
                elif i[0] in id_forgotten:
                    print(i[2])
                    answer = input('Insert answer: ')
                    if answer == i[1]:
                        print('Correct!')
                        checked = checked - 1
                        id_remembered.append(i[0])
                    else:
                        print('Nope, the correct answer is:', i[2])
                else:
                    print(i[2])
                    answer = input('Insert answer: ')
                    if answer == i[1]:
                        print('Correct!')
                        checked = checked - 1
                        id_correct.append(i[0])
                    else:
                        print('Nope, the correct answer is:', i[1])
                        id_forgotten.append(i[0])
        print(id_correct)
        print(id_forgotten)
        for i in id_correct:
            cur.execute('''UPDATE {tab} SET box21 = box21 + 1 WHERE id == {word_id}'''.format(tab = dict_name, word_id = i))

    conn.commit()
    cur.close()

这个函数包含多个for和while循环，我不确定FSM（有限状态机）是否适用于实现这个功能，如果适用希望得到相关建议。另外我想知道有没有其他方法确保函数的顺序执行，我尝试过把它放在单个状态或拆分到多个状态中，但都没成功。还有，能不能在Aiogram中结合机器人和用户消息使用while和for循环？

一、FSM是完全适配的，这是处理这类多步骤交互的标准方案 #

你的复习流程是典型的多步骤顺序交互：选择词典→选择方向→选择盒子→逐个复习单词，每一步都依赖上一步的输入，FSM正好用来跟踪用户当前处于哪个步骤，避免逻辑混乱。

拆分状态的正确思路：

把流程拆成几个明确的状态，每个状态对应一个交互节点：

• ReviewStates.choose_dict：等待用户选择词典
• ReviewStates.choose_direction：等待用户选择复习方向
• ReviewStates.choose_box：等待用户选择盒子
• ReviewStates.review_word：等待用户回答当前单词

同时需要用上下文存储（Aiogram的FSMContext）来保存中间数据：词典名、方向、待复习单词列表、已答对/答错的ID集合、当前正在复习的单词索引等。

二、不要直接在Aiogram中用while/for循环阻塞执行 #

Telegram Bot是异步架构，阻塞式的while/for循环会占用事件循环，导致机器人无法处理其他请求。必须把循环逻辑拆成异步的状态跳转：

1. 进入复习状态时，从上下文取出待复习单词列表，发送第一个单词
2. 用户回复后，判断对错，更新上下文里的统计数据
3. 如果还有未复习的单词，发送下一个；如果全部完成，更新数据库并结束流程

三、具体实现示例（核心逻辑） #

1. 定义状态

from aiogram.fsm.state import StatesGroup, State

class ReviewStates(StatesGroup):
    choose_dict = State()
    choose_direction = State()
    choose_box = State()
    review_word = State()

2. 启动复习流程的命令

@router.message(Command("review"))
async def start_review(message: Message, state: FSMContext):
    # 调用原retrieve_dictionaries获取词典列表，发送给用户
    dicts = retrieve_dictionaries()
    await message.answer(f"你的词典列表：{'、'.join(dicts)}\n请输入要复习的词典名称：")
    await state.set_state(ReviewStates.choose_dict)

3. 处理词典选择

@router.message(ReviewStates.choose_dict)
async def process_dict_choice(message: Message, state: FSMContext):
    dict_name = message.text.strip()
    # 验证词典是否存在（省略数据库校验逻辑）
    cur = conn.cursor()
    cur.execute(f"SELECT * FROM {dict_name};")
    columns = [x[0] for x in cur.description]
    await state.update_data(dict_name=dict_name, columns=columns)
    await message.answer(
        f"可选复习方向：\n1. {columns[1]} → {columns[2]}\n2. {columns[2]} → {columns[1]}\n请选择1或2："
    )
    await state.set_state(ReviewStates.choose_direction)

4. 处理方向选择和盒子选择（类似逻辑）

@router.message(ReviewStates.choose_direction)
async def process_direction_choice(message: Message, state: FSMContext):
    direction = message.text.strip()
    if direction not in ["1", "2"]:
        await message.answer("请输入1或2！")
        return
    data = await state.get_data()
    await state.update_data(direction=direction)
    await message.answer("请选择复习的盒子（1-4）：")
    await state.set_state(ReviewStates.choose_box)

@router.message(ReviewStates.choose_box)
async def process_box_choice(message: Message, state: FSMContext):
    try:
        box = int(message.text.strip())
        if box not in range(1,5):
            raise ValueError
    except ValueError:
        await message.answer("请输入1-4之间的数字！")
        return
    data = await state.get_data()
    dict_name = data["dict_name"]
    direction = data["direction"]
    # 查询待复习单词
    cur = conn.cursor()
    if direction == "1":
        cur.execute(f"SELECT * FROM {dict_name} WHERE box12 = {box} ORDER BY RANDOM();")
    else:
        cur.execute(f"SELECT * FROM {dict_name} WHERE box21 = {box} ORDER BY RANDOM();")
    checkwords = cur.fetchall()
    if not checkwords:
        await message.answer("这个盒子里没有单词！")
        await state.clear()
        return
    # 初始化复习数据
    await state.update_data(
        checkwords=checkwords,
        current_idx=0,
        id_correct=[],
        id_forgotten=[],
        id_remembered=[]
    )
    # 发送第一个单词
    await send_next_word(message, state)
    await state.set_state(ReviewStates.review_word)

5. 处理单词回答和循环复习

async def send_next_word(message: Message, state: FSMContext):
    data = await state.get_data()
    idx = data["current_idx"]
    word = data["checkwords"][idx]
    direction = data["direction"]
    columns = data["columns"]
    # 根据方向显示单词
    if direction == "1":
        prompt = f"请输入对应的{columns[2]}：\n{word[1]}"
    else:
        prompt = f"请输入对应的{columns[1]}：\n{word[2]}"
    await message.answer(prompt)

@router.message(ReviewStates.review_word)
async def process_word_answer(message: Message, state: FSMContext):
    data = await state.get_data()
    idx = data["current_idx"]
    word = data["checkwords"][idx]
    direction = data["direction"]
    answer = message.text.strip()
    # 判断对错
    is_correct = False
    if direction == "1":
        is_correct = (answer == word[2])
    else:
        is_correct = (answer == word[1])
    # 更新统计数据
    if is_correct:
        await message.answer("回答正确！")
        data["id_correct"].append(word[0])
    else:
        await message.answer(f"回答错误，正确答案是：{word[2] if direction == '1' else word[1]}")
        data["id_forgotten"].append(word[0])
    # 处理下一个单词
    next_idx = idx + 1
    if next_idx < len(data["checkwords"]):
        data["current_idx"] = next_idx
        await state.set_data(data)
        await send_next_word(message, state)
    else:
        # 复习结束，更新数据库
        dict_name = data["dict_name"]
        direction = data["direction"]
        cur = conn.cursor()
        if direction == "1":
            for word_id in data["id_correct"]:
                cur.execute(f"UPDATE {dict_name} SET box12 = box12 + 1 WHERE id = {word_id};")
        else:
            for word_id in data["id_correct"]:
                cur.execute(f"UPDATE {dict_name} SET box21 = box21 + 1 WHERE id = {word_id};")
        conn.commit()
        cur.close()
        await message.answer(f"复习完成！共答对{len(data['id_correct'])}个单词，答错{len(data['id_forgotten'])}个单词。")
        await state.clear()

四、关键注意事项 #

• 所有同步数据库操作最好用asyncio.to_thread包装，避免阻塞事件循环
• 要处理用户输入错误的情况（比如输入非数字、不存在的词典），引导用户重新输入
• 上下文存储的数据要尽量精简，避免占用过多内存

内容的提问来源于stack exchange，提问作者Vlad F