如何为Telegram Bot实现用户独立对话上下文隔离？

问题概述 #

当前Telegram Bot存在所有用户共享同一对话上下文的问题，导致不同用户的应答无法匹配各自的对话场景。尝试通过SQLite存储用户上下文但代码未能正常运行，以下是经过验证的正确实现方案。

核心思路 #

以用户的chat_id作为唯一标识，为每个用户单独维护对话上下文数据，通过SQLite数据库持久化存储。每次用户发起交互时，先从数据库读取该用户的专属上下文，处理完成后再将更新后的上下文保存回数据库，确保上下文的用户隔离性。

代码示例 #

1. 原共享上下文的可运行代码 #

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes

# 全局共享的上下文变量
global_context = {}

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    await update.message.reply_text('你好！开始对话吧')

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    user_input = update.message.text
    # 使用全局上下文处理所有用户的对话
    if 'last_input' in global_context:
        await update.message.reply_text(f'你上次说的是：{global_context["last_input"]}，这次说的是：{user_input}')
    else:
        await update.message.reply_text(f'这是你第一次说话：{user_input}')
    global_context['last_input'] = user_input

if __name__ == '__main__':
    application = ApplicationBuilder().token('你的Bot Token').build()
    
    application.add_handler(CommandHandler('start', start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    application.run_polling()

2. 基于SQLite的正确实现代码 #

from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, filters, ContextTypes
import sqlite3
from contextlib import closing
import json

# 初始化SQLite数据库表
def init_db():
    with closing(sqlite3.connect('user_contexts.db')) as conn:
        with closing(conn.cursor()) as cur:
            cur.execute('''
                CREATE TABLE IF NOT EXISTS user_contexts (
                    chat_id INTEGER PRIMARY KEY,
                    context_data TEXT NOT NULL
                )
            ''')
            conn.commit()

# 获取指定用户的上下文，支持复杂结构（字典/列表）
def get_user_context(chat_id):
    with closing(sqlite3.connect('user_contexts.db')) as conn:
        with closing(conn.cursor()) as cur:
            cur.execute('SELECT context_data FROM user_contexts WHERE chat_id = ?', (chat_id,))
            result = cur.fetchone()
            if result:
                return json.loads(result[0])
            return {}

# 保存/更新指定用户的上下文
def save_user_context(chat_id, context_data):
    serialized_data = json.dumps(context_data)
    with closing(sqlite3.connect('user_contexts.db')) as conn:
        with closing(conn.cursor()) as cur:
            cur.execute('''
                INSERT OR REPLACE INTO user_contexts (chat_id, context_data)
                VALUES (?, ?)
            ''', (chat_id, serialized_data))
            conn.commit()

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    # 初始化用户上下文
    save_user_context(chat_id, {'last_input': None, 'conversation_step': 0})
    await update.message.reply_text('你好！开始对话吧')

async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    chat_id = update.effective_chat.id
    user_input = update.message.text
    
    # 获取当前用户的专属上下文
    user_ctx = get_user_context(chat_id)
    last_input = user_ctx.get('last_input')
    
    if last_input:
        await update.message.reply_text(f'你上次说的是：{last_input}，这次说的是：{user_input}')
    else:
        await update.message.reply_text(f'这是你第一次说话：{user_input}')
    
    # 更新用户上下文
    user_ctx['last_input'] = user_input
    save_user_context(chat_id, user_ctx)

if __name__ == '__main__':
    # 初始化数据库
    init_db()
    application = ApplicationBuilder().token('你的Bot Token').build()
    
    application.add_handler(CommandHandler('start', start))
    application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message))
    
    application.run_polling()

关键细节说明 #

• 用chat_id作为用户唯一标识，确保上下文与用户绑定
• 使用json序列化复杂上下文数据（如字典），解决SQLite无法直接存储非文本类型数据的问题
• 通过INSERT OR REPLACE语句实现上下文的初始化和更新，避免重复创建用户记录
• 用contextlib.closing管理数据库连接和游标，确保资源自动释放，避免连接泄漏

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