基于aiogram的测验机器人开发问题：实现题目逐个发送而非批量发送

阿华AIGC实验室

2026-4-27

解决Aiogram测验机器人逐题发送的问题

你的问题根源在于：当前代码里的循环会一次性把所有发送请求都提交给Telegram API，没有等待用户的任何响应，所以用户会瞬间收到所有题目。要实现"发一题等回复再发下一题"的逻辑，我们需要用aiogram的**有限状态机（FSM）**来跟踪测验进度，每次用户回复后再触发下一题的发送。

具体实现步骤

1. 定义测验状态

首先创建一个状态类，用来保存当前的题目索引和完整的题目列表：

from aiogram.fsm.state import State, StatesGroup
from aiogram.fsm.context import FSMContext

class QuizStates(StatesGroup):
    answering = State()  # 标记用户正在回答测验的状态

2. 初始化测验并发送第一题

替换你原来的循环代码，改成初始化测验的逻辑——当用户触发开始测验的命令/回调时，先获取题目列表，存入状态，然后发送第一题：

# 假设这是用户触发开始测验的回调处理器（根据你的实际触发方式调整）
async def start_quiz(call: CallbackQuery, state: FSMContext):
    # 从数据库获取全部10道题目（替换成你实际的数据库查询代码）
    quest = await your_db_query_function()
    
    # 将当前进度（第0题）和题目列表存入用户状态
    await state.update_data(current_index=0, questions=quest)
    
    # 发送第一题（和你原来的发送逻辑一致，只是只发送第一题）
    first_question = quest[0]
    markup = await your_markup_generation_function(first_question)  # 替换成你生成选项按钮的逻辑
    
    if len(first_question['img']) >= 1:
        await call.message.answer_photo(
            photo=open(f'backends/{first_question["img"]}', 'rb'),
            caption=f'{first_question["question"]}',
            reply_markup=markup
        )
    elif len(first_question['video_gif']) >= 1:
        await call.message.answer_video(
            video=open(f'backends/{first_question["video_gif"]}', 'rb'),
            caption=f'{first_question["question"]}',
            reply_markup=markup
        )
    else:
        await call.message.answer(
            text=f'{first_question["question"]}',
            reply_markup=markup
        )
    
    # 将用户状态设置为"正在回答"
    await state.set_state(QuizStates.answering)

3. 处理用户回答并发送下一题

创建一个处理器，专门处理用户在测验状态下的回复（这里假设用按钮回调，如果你用普通消息回复，改成MessageHandler即可）：

async def handle_quiz_answer(call: CallbackQuery, state: FSMContext):
    # 从状态中取出当前进度和题目列表
    state_data = await state.get_data()
    current_index = state_data.get('current_index', 0)
    questions = state_data.get('questions', [])
    
    # 这里可以添加记录用户答案的逻辑（比如存入数据库）
    # user_selected_option = call.data
    # await save_user_answer(call.from_user.id, questions[current_index]['id'], user_selected_option)
    
    # 计算下一题的索引
    next_index = current_index + 1
    
    if next_index < len(questions):
        # 还有题目未完成，发送下一题
        next_question = questions[next_index]
        markup = await your_markup_generation_function(next_question)
        
        if len(next_question['img']) >= 1:
            await call.message.answer_photo(
                photo=open(f'backends/{next_question["img"]}', 'rb'),
                caption=f'{next_question["question"]}',
                reply_markup=markup
            )
        elif len(next_question['video_gif']) >= 1:
            await call.message.answer_video(
                video=open(f'backends/{next_question["video_gif"]}', 'rb'),
                caption=f'{next_question["question"]}',
                reply_markup=markup
            )
        else:
            await call.message.answer(
                text=f'{next_question["question"]}',
                reply_markup=markup
            )
        
        # 更新状态中的当前题目索引
        await state.update_data(current_index=next_index)
    else:
        # 所有题目完成，结束测验
        await call.message.answer("🎉 测验全部完成！感谢你的参与！")
        # 清除用户的测验状态
        await state.clear()
    
    # 必须调用这行，否则用户的按钮会一直处于加载状态
    await call.answer()

4. 注册处理器

最后别忘了把这两个处理器注册到你的Dispatcher中：

# 注册开始测验的回调处理器（假设你的回调数据是"start_quiz"，根据实际调整）
dp.callback_query.register(start_quiz, lambda c: c.data == "start_quiz")

# 注册回答处理器，只有在用户处于QuizStates.answering状态时才触发
dp.callback_query.register(handle_quiz_answer, state=QuizStates.answering)