AsyncTeleBot无register_next_step_handler属性，异步Telegram机器人开发受阻

我用pyTelegramBot开发问答记忆类Telegram机器人，同步版TeleBot的register_next_step_handler方法能正常实现等待用户输入的逻辑，示例代码如下：

...

@bot.message_handler(func=lambda msg: msg.text is not None and '/start' in msg.text)
def send_welcome(msg):
    global cur_user
    cur_user = msg.from_user.id
    keyboard = types.InlineKeyboardMarkup()
    keyboard.add(types.InlineKeyboardButton('Да', callback_data='old'),
                 types.InlineKeyboardButton('Нет', callback_data='new'))
    keyboard.add(types.InlineKeyboardButton('Пора остановиться', callback_data='stop_bot'))
    greet = 'Бла...бла...бла\nМы знакомы?'
    bot.send_message(msg.chat.id, greet, reply_markup=keyboard)

...

@bot.callback_query_handler(func=lambda call: True)
def query_processing(call):
    global user
    global cur_user
    if call.data == 'new':
        user = dict.fromkeys(user, None)
        nxt = bot.send_message(call.message.chat.id, 'Как звать?')
        bot.register_next_step_handler(nxt, get_name_ask_goal)

...

但由于需要处理消息等待、支持多用户同时操作，切换到AsyncTeleBot后发现没有register_next_step_handler方法，无法实现等待输入的逻辑，且找不到解决方案。另外尝试的异步代码运行时报错TypeError: An asyncio.Future, a coroutine or an awaitable is required，代码如下：

...
bot = AsyncTeleBot(os.getenv('token2'))
...

async def beep(chat_id) -> None:
    """Send the beep message."""
    await bot.send_message(chat_id, text='Beep!')
    aioschedule.clear(chat_id)

async def scheduler():
    global chat_id
    aioschedule.every(5).seconds.do(beep, chat_id).tag(chat_id)
    while True:
        await aioschedule.run_pending()
        await asyncio.sleep(1)

async def main():
    await asyncio.gather(scheduler(), bot.polling(non_stop=True))

if __name__ == "__main__":
    asyncio.run(main(), debug=True)
...

1. 适合需求的其他库 #

• python-telegram-bot：v13及以上版本为异步架构，内置ConversationHandler组件，专门用于管理多用户对话流程，完美替代register_next_step_handler的功能，支持状态持久化，文档完善。
• aiogram：纯异步的Telegram机器人框架，原生支持对话状态管理，用FSMContext（有限状态机）来追踪用户当前对话阶段，轻量高效，适合处理多用户并发场景。

2. 保存用户消息的简单方案 #

内存存储（适合测试/小型机器人） #

用字典存储用户对话状态和消息内容，键为用户ID，值为当前状态或已保存的信息：

# 示例：存储用户状态和输入的信息
user_data = {}
# 结构：{user_id: {"state": "waiting_for_name", "name": None, "goal": None}}

@bot.message_handler(...)
async def handle_input(msg):
    user_id = msg.from_user.id
    current_state = user_data.get(user_id, {}).get("state")
    
    if current_state == "waiting_for_name":
        user_data.setdefault(user_id, {})["name"] = msg.text
        user_data[user_id]["state"] = "waiting_for_goal"
        await bot.send_message(user_id, "你的目标是什么？")
    elif current_state == "waiting_for_goal":
        user_data[user_id]["goal"] = msg.text
        await bot.send_message(user_id, f"已保存：姓名{msg.text}，目标{user_data[user_id]['goal']}")

持久化存储（适合生产环境） #

• SQLite：轻量文件型数据库，无需额外服务，用sqlite3或asyncio兼容的aiosqlite操作，适合存储用户对话数据。
• Redis：内存型键值数据库，适合高并发场景，快速读写用户状态，支持过期时间，适合临时对话状态存储。

3. 能否将asyncio与线程式的TeleBot配合使用？ #

可以，但不推荐。同步版TeleBot并非线程安全，多用户并发操作时容易出现数据竞争或异常。如果非要这么做，需要：

• 用asyncio.run_in_executor将同步TeleBot的操作放到线程池中执行，避免阻塞事件循环；
• 对共享资源（如全局用户数据）加线程锁，防止并发冲突。

但这种方式维护成本高，不如直接迁移到异步框架（如python-telegram-bot v13+或aiogram）更可靠。另外你之前的aioschedule报错，是因为do方法传入了异步函数但未正确处理，需要用asyncio.create_task包装：

# 修改scheduler中的代码
aioschedule.every(5).seconds.do(lambda: asyncio.create_task(beep(chat_id))).tag(chat_id)

同时要注意避免使用全局chat_id，应该根据实际场景传递正确的用户ID。

内容的提问来源于stack exchange，提问作者Pavel Stepanov