基于pyTelegramBotAPI实现用户与客服同会话对接的技术方案问询

核心实现步骤 #

1. 会话状态管理 #

需要维护每个用户的会话状态，标记其是否处于客服对接中。内存字典可快速实现原型，生产环境建议用Redis、SQLite等持久化存储，避免机器人重启后丢失状态：

# 内存存储示例，生产环境替换为数据库
user_support_state = {}  # key: 用户ID, value: True/False（是否在客服会话中）
SUPPORT_AGENTS = [123456789, 987654321]  # 替换为实际客服的Telegram ID

2. 处理用户/support指令 #

当用户发送/support时，机器人先检查状态，未在会话中则标记状态，并给所有客服发送通知（包含用户聊天入口）：

async def handle_support(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    chat_id = update.effective_chat.id

    if user_id in user_support_state and user_support_state[user_id]:
        await update.message.reply_text("你当前已在客服会话中，发送/stop可结束会话")
        return

    # 标记用户进入客服会话
    user_support_state[user_id] = True
    await update.message.reply_text("已通知客服，将尽快为你接入，请稍等")

    # 生成用户聊天链接（优先用用户名，无则用ID）
    user_contact = f"https://t.me/{update.effective_user.username}" if update.effective_user.username else f"用户ID: {user_id}"
    notify_text = f"新的客服请求：{user_contact}\n点击链接直接进入聊天：{user_contact}" if update.effective_user.username else f"新的客服请求，用户ID：{user_id}\n请通过用户ID发起聊天"

    # 通知所有客服
    for agent_id in SUPPORT_AGENTS:
        try:
            await context.bot.send_message(chat_id=agent_id, text=notify_text)
        except Exception as e:
            print(f"通知客服失败 {agent_id}: {str(e)}")

3. 拦截机器人指令响应 #

在消息处理环节，先判断用户是否处于客服会话中，若是则仅响应/stop指令，忽略其他所有指令和消息：

async def intercept_messages(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id

    # 处于客服会话中时，仅处理/stop
    if user_id in user_support_state and user_support_state[user_id]:
        if update.message.text.strip() == "/stop":
            await handle_stop_support(update, context)
        return  # 跳过其他消息处理，机器人不响应

    # 非会话状态下的正常消息/指令处理逻辑
    # ...

async def handle_stop_support(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_id = update.effective_user.id
    if user_id in user_support_state and user_support_state[user_id]:
        user_support_state[user_id] = False
        await update.message.reply_text("客服会话已结束，机器人恢复正常响应")

        # 通知客服会话结束
        for agent_id in SUPPORT_AGENTS:
            try:
                await context.bot.send_message(chat_id=agent_id, text=f"用户{user_id}的客服会话已结束")
            except Exception as e:
                print(f"通知客服会话结束失败 {agent_id}: {str(e)}")

4. 配置机器人处理器 #

将上述函数注册到机器人的调度器中，确保命令和消息按逻辑处理：

def main():
    app = ApplicationBuilder().token("你的机器人Token").build()

    # 注册命令处理器
    app.add_handler(CommandHandler("support", handle_support))
    app.add_handler(CommandHandler("stop", handle_stop_support))

    # 注册消息拦截处理器（优先级高于其他消息处理器）
    app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, intercept_messages), group=1)

    # 其他正常指令处理器（比如/start等）
    # app.add_handler(CommandHandler("start", handle_start))

    app.run_polling()

if __name__ == "__main__":
    main()

关键优化点 #

• 状态持久化：用Redis替代内存字典，避免机器人重启后丢失会话状态，示例：

  import redis
  r = redis.Redis(host='localhost', port=6379, db=0)
  # 设置状态：r.set(f"support:{user_id}", "1")
  # 获取状态：r.get(f"support:{user_id}") == b"1"
  

• 客服身份校验：如果需要限制仅客服能发起某些操作（比如主动结束会话），可以在消息处理时验证发送者ID是否在SUPPORT_AGENTS列表中。
• 会话超时：添加定时任务，对超过N分钟无交互的会话自动重置状态，避免资源占用。
• 消息通知增强：在客服通知中增加用户的历史消息摘要，帮助客服快速了解上下文。

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