You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何存储python-telegram-bot的ConversationHandler状态解决Heroku重启后需重发命令问题

将ConversationHandler状态存储到数据库自动恢复的方案完全可行,是该场景下的最优解,具体实现方案如下:

方案1:基于Firebase实现会话状态持久化

python-telegram-bot框架原生提供了BasePersistence抽象类,用于对接外部存储实现会话、用户、聊天数据的持久化,你已经有Firebase账号,直接实现适配类即可:

  • 第一步:安装依赖,需要python-telegram-bot、firebase-admin两个库
  • 第二步:实现Firebase持久化类,参考代码如下:
from typing import Dict, Tuple, Any, Optional
from telegram.ext import BasePersistence
import firebase_admin
from firebase_admin import credentials, firestore

class FirebasePersistence(BasePersistence):
    def __init__(self, firebase_cred_path: str):
        # 配置是否存储用户、聊天、机器人全局数据
        super().__init__(store_user_data=True, store_chat_data=True, store_bot_data=True)
        # 初始化Firebase连接
        cred = credentials.Certificate(firebase_cred_path)
        firebase_admin.initialize_app(cred)
        self.db = firestore.client()
        self.conv_collection = self.db.collection("conversation_states")
        self.user_data_collection = self.db.collection("user_data")
        self.chat_data_collection = self.db.collection("chat_data")
    
    def get_conversations(self, name: str) -> Dict[Tuple[int, int], Optional[Any]]:
        # 启动时拉取对应会话的所有状态
        doc = self.conv_collection.document(name).get()
        raw_data = doc.to_dict() if doc.exists else {}
        # 将Firebase存储的字符串key转回(chat_id, user_id)元组
        return {tuple(map(int, k.split("_"))): v for k, v in raw_data.items()}
    
    def update_conversation(self, name: str, key: Tuple[int, int], new_state: Optional[Any]) -> None:
        # 会话状态变更时写入Firebase
        str_key = f"{key[0]}_{key[1]}"
        self.conv_collection.document(name).set({str_key: new_state}, merge=True)

    # 以下为用户数据、聊天数据持久化的实现,不需要可以直接返回空字典
    def get_user_data(self) -> Dict[int, Any]:
        doc = self.user_data_collection.document("all").get()
        return {int(k): v for k, v in doc.to_dict().items()} if doc.exists else {}
    
    def update_user_data(self, user_id: int, data: Any) -> None:
        self.user_data_collection.document("all").set({str(user_id): data}, merge=True)
    
    def get_chat_data(self) -> Dict[int, Any]:
        doc = self.chat_data_collection.document("all").get()
        return {int(k): v for k, v in doc.to_dict().items()} if doc.exists else {}
    
    def update_chat_data(self, chat_id: int, data: Any) -> None:
        self.chat_data_collection.document("all").set({str(chat_id): data}, merge=True)

    def get_bot_data(self) -> Dict:
        return {}
    
    def update_bot_data(self, data: Dict) -> None:
        pass
  • 第三步:初始化机器人时传入持久化实例即可
from telegram.ext import ApplicationBuilder, CommandHandler, ConversationHandler, MessageHandler, filters

# 初始化持久化实例,填入你的Firebase凭证文件路径
persistence = FirebasePersistence("your_firebase_cred.json")

# 构建Application时传入persistence参数
application = ApplicationBuilder().token("YOUR_BOT_TOKEN").persistence(persistence).build()

# 后续正常注册ConversationHandler即可,不需要修改原有业务逻辑
conv_handler = ConversationHandler(
    entry_points=[CommandHandler("start", start)],
    states={...},
    fallbacks=[...]
)
application.add_handler(conv_handler)

完成上述配置后,每次会话状态变更都会自动同步到Firebase,脚本重启时框架会自动拉取所有保存的会话状态恢复,完全不需要用户重新发送触发命令。

方案2:更换无自动重启的托管平台

如果不需要保留Heroku部署,也可以选择支持长期驻留进程的托管服务,不会定期自动重启脚本:

  • 云服务器类:阿里云ECS、腾讯云轻量应用服务器、DigitalOcean Droplet、Vultr VPS,自行部署Python环境后运行脚本即可长期稳定运行
  • 常驻Serverless类:Google Cloud Run 配置常驻实例、AWS Lambda配置预发并发,也可以避免冷启动自动重启

内容的提问来源于stack exchange,提问作者Rahasyar Desh

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.25 03:15:09