如何将Google Colab上的Ooga Booga Web UI接入Telegram Bot打造个人AI助手

一、在Google Colab部署Ooga Booga Web UI（搭载Vicuna/Pygmalion） #

• 新建Colab笔记本，先安装基础依赖：

  !apt-get update && apt-get install -y python3 python3-pip git
  !git clone https://github.com/oobabooga/text-generation-webui
  %cd text-generation-webui
  !pip install -r requirements.txt
  

• 下载目标模型（以Vicuna-7B为例，替换链接可安装Pygmalion）：

  !python download-model.py lmsys/vicuna-7b-v1.5
  

• 启动Web UI并开启API服务（必须开启API才能被Telegram Bot调用）：

  !python server.py --api --listen
  

  记录Colab生成的公网访问地址（如https://xxxx-xxxx-xxxx-xxxx.ngrok.io），后续会用到。

二、创建并配置Telegram Bot #

• 打开Telegram联系@BotFather，发送/newbot按指引创建机器人，获取专属API Token。
• 安装Telegram Bot开发依赖：

  pip install python-telegram-bot==13.7
  

• 编写基础聊天逻辑，对接Ooga Booga API：

  import requests
  from telegram import Update
  from telegram.ext import Updater, CommandHandler, MessageHandler, Filters, CallbackContext
  
  OOGA_API_URL = "你的Colab Ooga Booga API地址/v1/completions"
  TELEGRAM_TOKEN = "你的Telegram Bot Token"
  
  def generate_ai_response(prompt):
      payload = {
          "prompt": prompt,
          "max_tokens": 200,
          "temperature": 0.7,
          "top_p": 0.9,
          "stop": ["\nUser:", "\nAI:"]
      }
      response = requests.post(OOGA_API_URL, json=payload)
      return response.json()["choices"][0]["text"].strip()
  
  def start_cmd(update: Update, context: CallbackContext):
      update.message.reply_text("嗨！我是你的AI助手，有什么需求随时说~")
  
  def handle_user_msg(update: Update, context: CallbackContext):
      user_input = update.message.text
      ai_reply = generate_ai_response(f"User: {user_input}\nAI:")
      update.message.reply_text(ai_reply)
  
  def main():
      updater = Updater(TELEGRAM_TOKEN)
      dp = updater.dispatcher
      dp.add_handler(CommandHandler("start", start_cmd))
      dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_user_msg))
      updater.start_polling()
      updater.idle()
  
  if __name__ == "__main__":
      main()
  

  替换代码中的API地址和Token，运行脚本即可实现基础聊天功能。

三、实现主动对话功能 #

• 安装定时任务依赖：

  pip install apscheduler
  

• 修改主函数添加定时主动消息逻辑：

  from apscheduler.schedulers.background import BackgroundScheduler
  import datetime
  
  def send_proactive_msg(bot):
      # 替换为你的Telegram聊天ID（可通过/start命令的update.effective_chat.id获取）
      target_chat_id = "你的聊天ID"
      # 可根据时间/场景自定义主动消息内容
      proactive_content = "嗨，要不要聊聊今天的计划？或者我帮你查下日历安排？"
      bot.send_message(chat_id=target_chat_id, text=proactive_content)
  
  def main():
      updater = Updater(TELEGRAM_TOKEN)
      dp = updater.dispatcher
      # 保留之前的handler配置
      dp.add_handler(CommandHandler("start", start_cmd))
      dp.add_handler(MessageHandler(Filters.text & ~Filters.command, handle_user_msg))
      
      # 配置定时任务，比如每天上午10点主动发消息
      scheduler = BackgroundScheduler()
      scheduler.add_job(send_proactive_msg, 'cron', hour=10, args=[updater.bot])
      scheduler.start()
      
      updater.start_polling()
      updater.idle()
  

四、集成Google Calendar管理活动 #

• 在Google Cloud控制台创建项目，启用Calendar API，下载服务账号密钥（JSON格式）。
• 安装Google Calendar依赖：

  pip install google-api-python-client google-auth-httplib2 google-auth-oauthlib
  

• 编写日历操作工具函数：

  from google.oauth2 import service_account
  from googleapiclient.discovery import build
  import datetime
  
  SCOPES = ['https://www.googleapis.com/auth/calendar']
  SERVICE_ACCOUNT_FILE = '你的服务账号密钥文件.json'
  CALENDAR_ID = '你的日历ID'
  
  def get_calendar_service():
      credentials = service_account.Credentials.from_service_account_file(
          SERVICE_ACCOUNT_FILE, scopes=SCOPES)
      return build('calendar', 'v3', credentials=credentials)
  
  def add_calendar_event(summary, start_time, end_time):
      service = get_calendar_service()
      event = {
          'summary': summary,
          'start': {'dateTime': start_time, 'timeZone': 'Asia/Shanghai'},
          'end': {'dateTime': end_time, 'timeZone': 'Asia/Shanghai'},
      }
      event = service.events().insert(calendarId=CALENDAR_ID, body=event).execute()
      return f"活动已添加：{event.get('summary')}"
  
  def get_today_schedule():
      service = get_calendar_service()
      now = datetime.datetime.utcnow().isoformat() + 'Z'
      end_of_day = (datetime.datetime.utcnow() + datetime.timedelta(days=1)).isoformat() + 'Z'
      events_result = service.events().list(calendarId=CALENDAR_ID, timeMin=now, timeMax=end_of_day,
                                           singleEvents=True, orderBy='startTime').execute()
      events = events_result.get('items', [])
      if not events:
          return "今天没有日程安排"
      schedule_msg = "今日日程：\n"
      for event in events:
          start_time = event['start'].get('dateTime', event['start'].get('date'))
          schedule_msg += f"- {start_time.split('T')[1][:5]} {event['summary']}\n"
      return schedule_msg
  

• 在Telegram消息处理中添加日历命令支持，比如：

  def today_schedule_cmd(update: Update, context: CallbackContext):
      schedule = get_today_schedule()
      update.message.reply_text(schedule)
  
  def add_event_cmd(update: Update, context: CallbackContext):
      # 示例：/add 下午3点开会 2024-05-20T15:00:00+08:00 2024-05-20T16:00:00+08:00
      args = context.args
      if len(args) < 3:
          update.message.reply_text("格式错误，请输入：/add 活动名称 开始时间 结束时间")
          return
      summary = args[0]
      start_time = args[1]
      end_time = args[2]
      result = add_calendar_event(summary, start_time, end_time)
      update.message.reply_text(result)
  

  记得在main函数中注册这两个命令handler。

五、用Hyper DB存储对话记录 #

• 安装Hyper DB：

  pip install hyperdb
  

• 修改对话处理函数，添加存储逻辑：

  import hyperdb
  
  # 初始化对话数据库
  conv_db = hyperdb.HyperDB('conversation_history.db')
  
  def handle_user_msg(update: Update, context: CallbackContext):
      user_input = update.message.text
      chat_id = update.effective_chat.id
      ai_reply = generate_ai_response(f"User: {user_input}\nAI:")
      # 存储对话记录
      conv_db.add({
          "chat_id": chat_id,
          "timestamp": str(datetime.datetime.now()),
          "user_message": user_input,
          "ai_response": ai_reply
      })
      conv_db.save()
      update.message.reply_text(ai_reply)
  

  可扩展查询历史记录的命令，比如/history返回当前用户的对话记录。

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