python-telegram-bot：如何无需触发ConversationHandler重复调用对话方法

我在用python-telegram-bot开发对话机器人时遇到以下问题：

• 现有对话流程中，用户输入邮箱格式错误时，提示错误后希望直接引导重新输入邮箱，不用等用户输入触发ConversationHandler再调用gender方法。
• 需要实现基于数据库的多步骤对话状态控制，将用户步骤信息存在数据库里，能在单个方法内灵活切换步骤，比如验证不通过时回退到上一步重新请求输入。
• 我写了更新后的示例代码，但目前必须等用户输入才能触发ConversationHandler，不符合需求，求解决办法。

初始代码 #

import re
 
from telegram import __version__ as TG_VER
 
try:
    from telegram import __version_info__
except ImportError:
    __version_info__ = (0, 0, 0, 0, 0)
 
if __version_info__ < (20, 0, 0, "alpha", 1):
    raise RuntimeError(
        f"This example is not compatible with your current PTB version {TG_VER}. To view the "
        f"{TG_VER} version of this example, "
        f"visit https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
    )
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    ConversationHandler,
    MessageHandler,
    filters,
)
 
GENDER, EMAIL = range(2)
 
async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    reply_keyboard = [["Boy", "Girl", "Other"]]
    await update.message.reply_text(
        "Hi! My name is Professor Bot. I will hold a conversation with you. "
        "Are you a boy or a girl?",
        reply_markup=ReplyKeyboardMarkup(
            reply_keyboard, one_time_keyboard=True, input_field_placeholder="Boy or Girl?"
        ),
    )
    return GENDER
 
async def gender(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    user = update.message.from_user
    await update.message.reply_text(
        "Please Enter your email:",
        reply_markup=ReplyKeyboardRemove(),
    )
 
    return EMAIL
 
async def email(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    if update.message.text != "" and re.match(r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+', update.message.text):
        await update.message.reply_text(
            "Thanks for your registration",
            reply_markup=ReplyKeyboardRemove(),
        )
    else:
        await update.message.reply_text(
            "Your Email address was wrong!",
            reply_markup=ReplyKeyboardRemove(),
        )
        return GENDER
 
    return ConversationHandler.END
 
async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    await update.message.reply_text(
        "Bye! I hope we can talk again some day.", reply_markup=ReplyKeyboardRemove()
    )
    return ConversationHandler.END
 
def main() -> None:
    application = Application.builder().token("***************").build()
 
    # Add conversation handler with the states GENDER, PHOTO, LOCATION and BIO
    conv_handler = ConversationHandler(
        entry_points=[CommandHandler("start", start)],
        states={
            GENDER: [MessageHandler(filters.TEXT, gender)],
            EMAIL: [MessageHandler(filters.TEXT & ~filters.COMMAND, email)],
        },
        fallbacks=[CommandHandler("cancel", cancel)],
    )
 
    application.add_handler(conv_handler)
 
    # Run the bot until the user presses Ctrl-C
    application.run_polling()
 

if __name__ == "__main__":
    main()

我希望提示Your Email address was wrong!后，直接显示Please Enter your email:，不用等用户输入触发ConversationHandler，直接重新引导输入。

此外，我需要实现如下结构的多步骤对话，基于数据库控制用户状态，可在单个方法内灵活切换步骤：

async def register(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:

if user.user_status == 'start':

    if user.step == 0:
        //Show Accept rules
        // if it accepted
            //go to step 1
    elif user.step == 1:
        //get name and family
        //goes to step 2
    elif user.step == 2:
        // check name and family
            // if it was correct
                // save in DB
                // get mobile number
            // if it was not correct
                // show error message 
                // back to step 1 and ask again name and family
    elif user.step == 3:
        // check mobile number 
            // if it was correct
                    // save in DB
                    //get email
            // if it was not correct
                    //show error message 
                    // back to step 2 and ask again mobile number
        // get email
        ....
    elif user.step == 4:
        ...
    elif user.step == 5:

        await update.message.reply_text(
                "you successfully register",
                reply_markup=await _reply_keyboard('GoHome'),parse_mode='HTML'
        )

return Register

更新示例代码（展示核心问题） #

import re

from telegram import __version__ as TG_VER

try:
    from telegram import __version_info__
except ImportError:
    __version_info__ = (0, 0, 0, 0, 0)

if __version_info__ < (20, 0, 0, "alpha", 1):
    raise RuntimeError(
        f"This example is not compatible with your current PTB version {TG_VER}. To view the "
        f"{TG_VER} version of this example, "
        f"visit https://docs.python-telegram-bot.org/en/v{TG_VER}/examples.html"
    )
from telegram import ReplyKeyboardMarkup, ReplyKeyboardRemove, Update
from telegram.ext import (
    Application,
    CommandHandler,
    ContextTypes,
    ConversationHandler,
    MessageHandler,
    filters,
)
step = 0
prev_step = 0
START,RULES,STEPS = range(3)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    reply_keyboard = [["Yes", "No"]]
    await update.message.reply_text(
        "Hi! My name is Professor Bot. I will hold a conversation with you. "
        "Are you accept rules?",
        reply_markup=ReplyKeyboardMarkup(
            reply_keyboard, one_time_keyboard=True, input_field_placeholder="Yes or No?"
        ),
    )
    return STEPS

async def steps(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    global step
    global prev_step

    if step == 0:
        print(step,prev_step)
        if update.message.text == "Yes" or prev_step == 1:
            step=1
            prev_step=0
            await update.message.reply_text(
                "Please enter your email address:",
                reply_markup=ReplyKeyboardRemove(),
            )
        else:
            await update.message.reply_text(
                "You should accept rules to continue!",
                reply_markup=ReplyKeyboardRemove(),
            )
            return START

    elif step == 1:
        prev_step = 1
        if update.message.text != "" and re.match(r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+', update.message.text):
            await update.message.reply_text(
                "Please enter your name:",
                reply_markup=ReplyKeyboardRemove(),
            )
            step = 2
        else:
            step = 0
            await update.message.reply_text(
                "Your Email address was wrong!",
                reply_markup=ReplyKeyboardRemove(),
            )

    return STEPS

async def cancel(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    await update.message.reply_text(
        "Bye! I hope we can talk again some day.", reply_markup=ReplyKeyboardRemove()
    )
    return ConversationHandler.END

def main() -> None:
    application = Application.builder().token("*****").build()

    # Add conversation handler with the states GENDER, PHOTO, LOCATION and BIO
    conv_handler = ConversationHandler(
        entry_points=[CommandHandler("start", start)],
        states={
            START: [MessageHandler(filters.TEXT, start)],
            RULES: [MessageHandler(filters.TEXT, steps)],
            STEPS: [MessageHandler(filters.TEXT & ~filters.COMMAND, steps)],
        },
        fallbacks=[CommandHandler("cancel", cancel)],
    )

    application.add_handler(conv_handler)

    # Run the bot until the user presses Ctrl-C
    application.run_polling()


if __name__ == "__main__":
    main()

解决方案 #

1. 邮箱验证失败后直接重新请求输入 #

原代码问题在于验证失败后返回了GENDER状态，导致需要用户输入才能触发下一步。修改思路：验证失败时直接在email函数内发送重新输入提示，保持EMAIL状态，无需跳转其他步骤。

修改后的email函数：

async def email(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    email_input = update.message.text.strip()
    if re.match(r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+', email_input):
        await update.message.reply_text(
            "Thanks for your registration",
            reply_markup=ReplyKeyboardRemove(),
        )
        return ConversationHandler.END
    else:
        await update.message.reply_text(
            "Your Email address was wrong!\nPlease Enter your email:",
            reply_markup=ReplyKeyboardRemove(),
        )
        # 停留在EMAIL状态，等待用户重新输入
        return EMAIL

2. 基于数据库的多步骤状态控制 #

核心逻辑是用数据库存储每个用户的对话步骤，在单个处理函数内根据数据库记录的步骤执行对应操作，验证失败时更新步骤并重新发送输入提示。以下是基于SQLAlchemy的实现框架：

# 定义用户模型
from sqlalchemy import Column, Integer, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    telegram_id = Column(Integer, unique=True)
    step = Column(Integer, default=0)
    name = Column(String(100))
    mobile = Column(String(20))
    email = Column(String(100))

# 初始化数据库
engine = create_engine('sqlite:///users.db')
Base.metadata.create_all(engine)
Session = sessionmaker(bind=engine)

# 对话状态仅保留一个REGISTER
REGISTER = 0

async def register(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    user_id = update.effective_user.id
    session = Session()
    
    # 获取或创建用户记录
    user = session.query(User).filter_by(telegram_id=user_id).first()
    if not user:
        user = User(telegram_id=user_id, step=0)
        session.add(user)
        session.commit()
    
    if user.step == 0:
        await update.message.reply_text("请阅读规则：xxx，是否接受？(Yes/No)")
        user.step = 1
        session.commit()
    elif user.step == 1:
        if update.message.text == "Yes":
            await update.message.reply_text("请输入您的姓名和姓氏：")
            user.step = 2
        else:
            await update.message.reply_text("您必须接受规则才能继续，请回复Yes/No")
        session.commit()
    elif user.step == 2:
        name_input = update.message.text.strip()
        if len(name_input) >= 2:
            user.name = name_input
            await update.message.reply_text("请输入您的手机号：")
            user.step = 3
        else:
            await update.message.reply_text("姓名格式错误，请重新输入您的姓名和姓氏：")
        session.commit()
    elif user.step == 3:
        mobile_input = update.message.text.strip()
        if mobile_input.isdigit() and len(mobile_input) == 11:
            user.mobile = mobile_input
            await update.message.reply_text("请输入您的邮箱：")
            user.step = 4
        else:
            await update.message.reply_text("手机号格式错误，请重新输入：")
        session.commit()
    elif user.step == 4:
        email_input = update.message.text.strip()
        if re.match(r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+', email_input):
            user.email = email_input
            await update.message.reply_text("您已成功注册！")
            session.commit()
            return ConversationHandler.END
        else:
            await update.message.reply_text("邮箱格式错误，请重新输入：")
    
    session.close()
    return REGISTER

# 更新ConversationHandler配置
conv_handler = ConversationHandler(
    entry_points=[CommandHandler("start", register)],
    states={
        REGISTER: [MessageHandler(filters.TEXT & ~filters.COMMAND, register)],
    },
    fallbacks=[CommandHandler("cancel", cancel)],
)

3. 避免等待用户输入触发的问题 #

原更新代码使用全局变量存储步骤会导致多用户冲突，且逻辑冗余。改用context.user_data（内存临时存储）或数据库存储步骤，在单个处理函数内完成所有步骤逻辑：

REGISTER = 0

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    context.user_data['step'] = 0
    await update.message.reply_text(
        "Hi! My name is Professor Bot. I will hold a conversation with you. Are you accept rules?", 
        reply_markup=ReplyKeyboardMarkup([["Yes", "No"]], one_time_keyboard=True)
    )
    return REGISTER

async def steps(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    current_step = context.user_data.get('step', 0)
    
    if current_step == 0:
        if update.message.text == "Yes":
            context.user_data['step'] = 1
            await update.message.reply_text("Please enter your email address:", reply_markup=ReplyKeyboardRemove())
        else:
            await update.message.reply_text("You should accept rules to continue!")
    elif current_step == 1:
        email_input = update.message.text.strip()
        if re.match(r'([A-Za-z0-9]+[.-_])*[A-Za-z0-9]+@[A-Za-z0-9-]+(\.[A-Z|a-z]{2,})+', email_input):
            context.user_data['step'] = 2
            await update.message.reply_text("Please enter your name:")
        else:
            await update.message.reply_text("Your Email address was wrong!\nPlease enter your email address:")
    
    return REGISTER

# 更新ConversationHandler
conv_handler = ConversationHandler(
    entry_points=[CommandHandler("start", start)],
    states={
        REGISTER: [MessageHandler(filters.TEXT & ~filters.COMMAND, steps)],
    },
    fallbacks=[CommandHandler("cancel", cancel)],
)

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