telegram integration

[dmitriyap]

Description

For those familiar with the telegram messaging app I think it would be cool idea to add telegram bot integration support, so that telegram app could be used as a "conversation ui".

What do you think?

---

I made a small proof of concept script for those who want to give it a quick try, it acts as a gateway between telegram bot api and text-generation-webui api.

Requirements:

$ pip install python-telegram-bot

To run the script you have to register a bot on telegram and get api_key (https://core.telegram.org/bots)

Run:

$ GPTBOT_APIKEY=[your key] GPTBOT_SERVER=[server ip of webui instance] GPTBOT_PORT=[server port of webui instance] python gptbot.py

GPTBOT_SERVER/PORT defaults to 127.0.0.1:7860 so you can omit them, but apikey is required.

This script only supports no-stream webui api, so you have to run it with --no-stream flag, for example I use the following command to run 13b llama model as a backend:

$ python server.py --listen --gptq-bits 4 --model llama-13b-hf --no-stream

Once you are chatting with the bot you can use /start command to reset the conversation and /stats command to display basic conversation stats.

gptbot.py:

import logging
import requests
import json
import os
from telegram import Update
from telegram.ext import filters, ApplicationBuilder, ContextTypes, CommandHandler, MessageHandler

HISTORY_FILE        = "gptbot_hst.json"
HISTORY_PRFX_USER   = 0
HISTORY_PRFX_AI     = 1

ai_params = {
    'max_new_tokens': 100,
    'do_sample': True,
    'temperature': 2.0,
    'top_p': 0.18,
    'typical_p': 1,
    'repetition_penalty': 1.15,
    'encoder_repetition_penalty': 1.0,
    'top_k': 30,
    'min_length': 0,
    'no_repeat_ngram_size': 0,
    'num_beams': 1,
    'penalty_alpha': 0,
    'length_penalty': 1,
    'early_stopping': False,
}

ai_character = {
    'name': "John",
    'persona': "John is a helpful AI chatbot that always provides useful and detailed answers to User's requests and questions. John tries to be as informative and friendly as possible.",
    'greeting': "Hello! I am John, your informative assistant. How may I help you today?",
    'dialogue_sample': "You: Hi. Can you help me with something?\nJohn: Hello, this is Johm. How can I help?\nYou: Have you heard of the latest nuclear fusion experiment from South Korea? I heard their experiment got hotter than the sun.\nJohn: Yes, I have heard about the experiment. Scientists in South Korea have managed to sustain a nuclear fusion reaction running at temperatures in excess of 100 million°C for 30 seconds for the first time and have finally been able to achieve a net energy gain when carrying out a nuclear fusion experiment. That's nearly seven times hotter than the core of the Sun, which has a temperature of 15 million degrees kelvins! That's exciting!\nYou: Wow! That's super interesting to know. Change of topic, I plan to change to the iPhone 14 this year.\nJohn: I see. What makes you want to change to iPhone 14?\nYou: My phone right now is too old, so I want to upgrade.\nJohn: That's always a good reason to upgrade. You should be able to save money by trading in your old phone for credit. I hope you enjoy your new phone when you upgrade."
}

ai_history = {}

logging.basicConfig(format='%(asctime)s - %(message)s', level=logging.INFO)

ai_server_ip        = os.getenv("GPTBOT_SERVER", "127.0.0.1")
ai_server_port      = os.getenv("GPTBOT_PORT",   "7860")
telegram_api_key    = os.getenv("GPTBOT_APIKEY", "")

def get_raw_reply(prompt, params):
    try:
        response = requests.post(f"http://{ai_server_ip}:{ai_server_port}/run/textgen", json={
            "data": [   prompt,
                        params['max_new_tokens'],
                        params['do_sample'],
                        params['temperature'],
                        params['top_p'],
                        params['typical_p'],
                        params['repetition_penalty'],
                        params['encoder_repetition_penalty'],
                        params['top_k'],
                        params['min_length'],
                        params['no_repeat_ngram_size'],
                        params['num_beams'],
                        params['penalty_alpha'],
                        params['length_penalty'],
                        params['early_stopping'],
                    ]
            }).json()
    except Exception as e:
        response = {}

    reply = ""
    if "data" in response.keys():
        reply = response["data"][0]

    return reply

def get_ai_preprompt(ai_desc):
    ctx = ""
    ctx += ai_desc['name'] + "'s persona: " + ai_desc['persona'] + "\n"
    ctx += "<START>\n" + ai_desc['dialogue_sample'] + "\n"

    return ctx

def extract_ai_reply(ai_desc, raw_reply):
    reply = raw_reply

    k = reply.find("\nYou:")
    if k != -1:
        reply = reply[:k]

    k = reply.find(f"\n{ai_desc['name']}:")
    if k != -1:
        reply = reply[:k]

    reply = reply.strip()

    return reply

def reset_ai_history(uid):
    if str(uid) in ai_history.keys():
        ai_history[str(uid)] = []
        with open(HISTORY_FILE, "w", encoding='utf-8') as fp:
            json.dump(ai_history, fp)

def get_ai_history_prompt(ai_desc, uid):
    if str(uid) not in ai_history.keys():
        return ""

    res = ""
    for e in ai_history[str(uid)]:
        if e[0] == HISTORY_PRFX_USER:
            res += "You: " + e[1] + "\n"
        else:
            res += ai_desc['name'] + ": " + e[1] + "\n"

    return res

def add_ai_history(uid, user_id, user_txt):
    if str(uid) not in ai_history.keys():
        ai_history[str(uid)] = []

    ai_history[str(uid)].append([user_id, user_txt])

    with open(HISTORY_FILE, "w", encoding='utf-8') as fp:
        json.dump(ai_history, fp)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    reset_ai_history(update.message.from_user.id)
    add_ai_history(update.message.from_user.id, HISTORY_PRFX_AI, ai_character['greeting'])

    await context.bot.send_message(chat_id=update.effective_chat.id, text=ai_character['greeting'])

async def stats(update: Update, context: ContextTypes.DEFAULT_TYPE):
    uid = update.message.from_user.id
    resp = "-== Statistics ==-\n"

    ctx = get_ai_preprompt(ai_character)
    ctx_hst = get_ai_history_prompt(ai_character, uid)

    resp += f"preprompt: {len(ctx)} characters (~{(len(ctx.split()) / 0.7):.2f} tokens)\n"
    resp += f"history: {len(ctx_hst)} characters (~{(len(ctx_hst.split()) / 0.7):.2f} tokens)\n"

    if str(uid) in ai_history.keys():
        resp += "messages in history: " + str(len(ai_history[str(uid)])) + "\n"

    await context.bot.send_message(chat_id=update.effective_chat.id, text=resp)

async def process_user_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    uid = update.message.from_user.id
    logging.info(f"{update.message.chat.username} ({uid}): {update.message.text}")

    ctx = get_ai_preprompt(ai_character)
    ctx += get_ai_history_prompt(ai_character, uid)

    mreq = ctx + "You: " + update.message.text + "\n" + ai_character['name'] + ":"

    r = get_raw_reply(mreq, ai_params)

    ai_reply = ""

    if len(r) > len(mreq):
        ai_reply = extract_ai_reply(ai_character, r[len(mreq):])
    else:
        ai_reply = "[No connection to webui API]"

    logging.info(f"AI ({uid}): {ai_reply}")

    add_ai_history(uid, HISTORY_PRFX_USER, update.message.text)
    add_ai_history(uid, HISTORY_PRFX_AI, ai_reply)

    await context.bot.send_message(chat_id=update.effective_chat.id, text=ai_reply)

if __name__ == '__main__':
    if not telegram_api_key:
        logging.info("Telegram bot api key env variable is not set (GPTBOT_APIKEY)")
        exit()

    try:
        with open(HISTORY_FILE, "r", encoding='utf-8') as fp:
            ai_history = json.load(fp)
    except:
        pass

    application = ApplicationBuilder().token(telegram_api_key).build()

    start_handler = CommandHandler('start', start)
    application.add_handler(start_handler)

    stats_handler = CommandHandler('stats', stats)
    application.add_handler(stats_handler)

    message_handler = MessageHandler(filters.TEXT & (~filters.COMMAND), process_user_message)
    application.add_handler(message_handler)

    application.run_polling()

[SantygoHH]

[No connection to webui API] in telegram

[dvoidus]

I think this is due to API change, seed param is missing, replace response call with:

        response = requests.post(f"http://{ai_server_ip}:{ai_server_port}/run/textgen", json={
            "data": [   prompt,
                        params['max_new_tokens'],
                        params['do_sample'],
                        params['temperature'],
                        params['top_p'],
                        params['typical_p'],
                        params['repetition_penalty'],
                        params['encoder_repetition_penalty'],
                        params['top_k'],
                        params['min_length'],
                        params['no_repeat_ngram_size'],
                        params['num_beams'],
                        params['penalty_alpha'],
                        params['length_penalty'],
                        params['early_stopping'],
                        params['seed'],
                    ]
            }).json()

If you are interested in telegram integration, I would suggest to look into https://github.com/oobabooga/text-generation-webui/pull/725

[github-actions[bot]]

This issue has been closed due to inactivity for 6 weeks. If you believe it is still relevant, please leave a comment below. You can tag a developer in your comment.