Self hosted ChatGPT with Cloudflare tunnel and Mattermost

version: "3.7"

services:

  mattermost:
    container_name: mattermost
    image: mattermost/mattermost-preview:${MATTERMOST_VER}
    restart: unless-stopped
    environment:
      - TZ=Europe/Warsaw
      - MM_SERVICESETTINGS_SITEURL=http://localhost:8065
    ports:
      - 8065:8065
    volumes:
      - ./mattermost/data:/mm/mattermost-data
      - ./mattermost/sql:/var/lib/mysql
      - ./mattermost/config.json:/mm/mattermost/config/config.json
      - ./mattermost/logs:/mm/mattermost/logs
    networks:
      - chatops

  chatops-ai:
    container_name: chatops-ai
    image: chatops-ai:${ALPINE_PY_VER}
    restart: unless-stopped
    build:
      context: ./chatops_bot/
      dockerfile: Dockerfile
      args:
        - ALPINE_PY_VER=${ALPINE_PY_VER}
    volumes:
      - ./chatops_bot:/app
    depends_on:
      - mattermost
    networks:
      - chatops

  cloudflared:
    image: cloudflare/cloudflared:${CLOUDFLARED_VER}
    container_name: cloudflare_tunnel
    restart: unless-stopped
    command: tunnel run
    environment:
      - TUNNEL_TOKEN=${CLOUDFLARED_TOKEN}
    depends_on:
      - chatops-ai
    networks:
      - chatops


networks:
  chatops:
    name: chatops

ARG ALPINE_PY_VER
FROM python:${ALPINE_PY_VER}
WORKDIR /app
COPY requirements.txt .
RUN \
    apk add tzdata && \
    cp /usr/share/zoneinfo/Europe/Warsaw /etc/localtime && \
    echo "Europe/Warsaw" >  /etc/timezone && \
    apk del tzdata && \
    pip install -r requirements.txt
ENTRYPOINT [ "python", "chatops_bot.py" ]

from mmpy_bot import Bot, Settings
from dotenv import load_dotenv
from mmpy_bot import Plugin, listen_to
from mmpy_bot import Message
import os, sys, re
from time import sleep
import json
import logging
from pathlib import Path
from redis import Redis
from dataclasses import dataclass


cwd = Path(__file__).parent
if not load_dotenv(dotenv_path=str(cwd) + '/.env'):
    sys.exit(120)

r = Redis(os.environ['REDIS_HOST'])
logging.basicConfig(level=logging.INFO, format='%(asctime)s :: %(levelname)s :: %(message)s')
logger = logging.getLogger('chatops-ai')


@dataclass
class OpenAiModel():
    gpt4o: str = 'gpt-4o-2024-05-13'          # 128,000 tokens, Up to Oct 2023:
    gpt4turbo: str = 'gpt-4-turbo-2024-04-09' # 128,000 tokens, Up to Dec 2023:


class ChatGPT():
    def __init__(self, model: str, temperature: float = 0.5):
        self.model = model
        self.temperature = temperature
    
    from openai import AsyncOpenAI

    async def askgpt(self, question: str, chat_log: list[dict[str, str]] = None):
        client = self.AsyncOpenAI(api_key=os.environ['OPEN_AI_KEY'], max_retries=3)
        if not chat_log:
            chat_log = [{
                'role': 'system',
                'content': 'Hello ChatGPT.',
            }]
        chat_log.append({'role': 'user', 'content': question})
        completion = await client.chat.completions.create(model=self.model, messages=chat_log, temperature=self.temperature)
        answer = completion.choices[0].message.content
        chat_log.append({'role': 'assistant', 'content': answer})
        return answer, chat_log


class MyPlugin(Plugin):
    chat = ChatGPT(model=OpenAiModel.gpt4o)
    pattern = '^--help|^--use-gpt4o|^--use-gpt4-turbo|^--clean-db|^--set-temp=[0-9]+.[0-9]+'
    regexp = re.compile(pattern)

    @listen_to(".*", direct_only=True, allowed_users=["krzysiek"])
    async def bot_respond(self, message: Message):
        if m := self.regexp.match(message.text):
            match m.group():
                case '--help':
                    msg = f'to use specific GPT model use following parameters:\n```--use-gpt4o```\n```--use-gpt4-turbo```\ncurrent model: {self.chat.model}, '
                    msg += f'temperature: {self.chat.temperature}\n'
                    msg += f'to change model temperature (from 0.0 to 2.0) use:\n```--set-temp=0.2```\n'
                    msg += f'to remove old context entries from database use:\n```--clean-db```\n\n'
                    self.driver.reply_to(message, msg)
                case '--use-gpt4o':
                    self.chat = ChatGPT(model=OpenAiModel.gpt4o)
                    self.driver.reply_to(message, f'you are now using: {self.chat.model}')
                case '--use-gpt4-turbo':
                    self.chat = ChatGPT(model=OpenAiModel.gpt4turbo)
                    self.driver.reply_to(message, f'you are now using: {self.chat.model}')
                case '--clean-db':
                    keys_removed = init_db_cleanup()
                    if keys_removed:
                        r.delete(*keys_removed)
                    self.driver.reply_to(message, f'number of keys cleaned: {len(keys_removed)}')
                case x if x.startswith('--set-temp='):
                    self.chat.temperature = float(x.split('=')[-1])
                    self.driver.reply_to(message,
                        f'temperature set to: {self.chat.temperature}, current model: {self.chat.model}'
                    )

        else:
            conv_thread = r.get(f'chatgpt:thread:{message.reply_id}')
            conv_log = json.loads(conv_thread) if conv_thread else None
            try:
                answer, conv_log = await self.chat.askgpt(question=message.text, chat_log=conv_log)
            except Exception as e:
                logger.error(e)
                self.driver.reply_to(message, "can you please repeat your question, I did not get it")
            else:
                log_line = dict(conv_log[-1])
                log_line['question'] = message.text
                logger.info(json.dumps(log_line, ensure_ascii=False))
                r.set(f'chatgpt:thread:{message.reply_id}', json.dumps(conv_log, ensure_ascii=False))
                self.driver.reply_to(message, answer)


def init_db_cleanup() -> list[str]:
    import db_utils
    thread_keys = [ i.decode('utf-8') for i in r.keys('chatgpt:thread:*') ]
    to_remove = db_utils.get_old_keys(thread_keys)
    _ = [logger.info(f'removing: {i}') for i in to_remove ]
    return to_remove


def init_bot():
    bot = Bot(
        settings=Settings(
            MATTERMOST_URL = os.environ['MATTERMOST_URL'],
            MATTERMOST_PORT = os.environ['MATTERMOST_PORT'],
            BOT_TOKEN = os.environ['BOT_TOKEN'],
            BOT_TEAM = os.environ['BOT_TEAM'],
            MATTERMOST_API_PATH = '',
            SSL_VERIFY = False
        ),
        plugins=[MyPlugin()]
    )
    bot.run()


def connect(timeout: float, retry: int):
    cnt = 0
    while True:
        try:
            init_bot()
        except Exception as e:
            if cnt > retry:
                raise TimeoutError('too many retries exceeded', e)
            sleep(timeout)
            timeout *= 1.5
            cnt += 1
            logger.warning(f'retrying ({cnt}) connection')



if __name__ == '__main__':
    logger.info('connecting to mattermost api...')
    connect(timeout=1.5, retry=15)

from urllib.request import Request, urlopen
import os, json

username = os.environ['MM_USER']
passwd = os.environ['MM_PASS']
api_url = os.environ['MATTERMOST_URL']
api_port = os.environ['MATTERMOST_PORT']
mm_user_id = os.environ['MM_USER_ID']
mm_team_id = os.environ['MM_TEAM_ID']

def get_old_keys(redis_threads: list[str]) -> list[str]:
    token_url = f'{api_url}:{api_port}/api/v4/users/login'
    body = json.dumps({
        "login_id": username,
        "password": passwd
    }).encode('utf-8')

    req = urlopen(Request(url=token_url, data=body, headers={'Content-Type': 'application/json'}))
    token = req.headers['token']

    threads_url = f'{api_url}:{api_port}/api/v4/users/{mm_user_id}/teams/{mm_team_id}/threads'
    r = urlopen(Request(threads_url, headers={'Authorization': f'Bearer {token}'}))
    content = json.loads(r.read())
    thread_ids = [ f"chatgpt:thread:{i['id']}" for i in content['threads'] ]
    return [ k for k in redis_threads if k not in thread_ids ]