A Telegram chatbot that responds back from a GGUF LLM Intel neural chat running on local host machine

telegram-bot-server.py

# llama_cpp_python==0.2.20
# requests==2.23.0


import json
import requests
import time
import urllib
from llama_cpp import Llama

TOKEN = "<Telegram token that comes from BotFather>"
URL = "https://api.telegram.org/bot{}/".format(TOKEN)

chat_history_map = {}
llm = None

def get_url(url):
    response = requests.get(url)
    content = response.content.decode("utf8")
    return content


def get_json_from_url(url):
    content = get_url(url)
    js = json.loads(content)
    return js


def get_updates(offset=None):
    url = URL + "getUpdates?timeout=100"
    if offset:
        url += "&offset={}".format(offset)
    js = get_json_from_url(url)
    return js


def get_last_update_id(updates):
    update_ids = []
    for update in updates["result"]:
        update_ids.append(int(update["update_id"]))
    return max(update_ids)


def call_llm(message_list):
    global llm

    q = "Q: {{question_without_mark}}? "
    a = "A: {{answer}} <END> "
    a_placeholder = "A: "
    prompt = "Q: What do you do? A: I'm an intelligent machine that responds to questions by being concise and precise. <END> Q: What is 2+2? 4 <END> Q: What is the state of water in room temperature? A: Liquid. <END> "
    mode = 'q'
    for message in message_list:
        if mode == 'q':
            question_without_mark = message
            if question_without_mark.strip()[-1] == "?":
                question_without_mark = question_without_mark.strip()[:-1]
            pq = q.replace("{{question_without_mark}}", question_without_mark)
            prompt = prompt + pq
            mode = 'a'
        else:
            pa = a.replace("{{answer}}", message)
            prompt = prompt + pa
            mode = 'q'
    if mode == 'a':
        prompt = prompt + a_placeholder
    
    output = llm(
        prompt,
        max_tokens=100,
        stop=["Q:", "<END>"],
        echo=True
    )
    
    result = output["choices"][0]["text"][len(prompt):].strip()
    return result


def process_conversation(text_messages):
    prompt = call_llm(text_messages)
    return prompt


def echo_all(updates):
    global chat_history_map
    for update in updates["result"]:
        text = update["message"]["text"]
        chat = update["message"]["chat"]["id"]

        chat_history = [] if update["message"]["chat"]["id"] not in chat_history_map else chat_history_map[update["message"]["chat"]["id"]]
        chat_history.append(text)

        result = process_conversation(chat_history)

        chat_history.append(result)
        chat_history = chat_history[-4:] # Last 2 message pairs.
        chat_history_map[update["message"]["chat"]["id"]] = chat_history

        send_message(result, chat)


def get_last_chat_id_and_text(updates):
    num_updates = len(updates["result"])
    last_update = num_updates - 1
    text = updates["result"][last_update]["message"]["text"]
    chat_id = updates["result"][last_update]["message"]["chat"]["id"]
    return (text, chat_id)


def send_message(text, chat_id):
    text = urllib.parse.quote_plus(text)
    url = URL + "sendMessage?text={}&chat_id={}".format(text, chat_id)
    get_url(url)


def main():
    global llm
    last_update_id = None
    llm = Llama(model_path="./neural-chat-7b-v3-1.Q8_0.gguf")
    while True:
        updates = get_updates(last_update_id)
        if len(updates["result"]) > 0:
            last_update_id = get_last_update_id(updates) + 1
            echo_all(updates)
        time.sleep(0.5)


if __name__ == '__main__':
    main()