Подключаем ChatGPT к Telegram-боту: умный бот своими руками

сообщение в Telegram → хендлер (PTB) → история диалога (в памяти/SQLite)
                                            ↓
                                    запрос к ChatGPT (openai SDK)
                                            ↓
                         ответ → (опционально стриминг) → сообщение пользователю

import os

from openai import OpenAI
from telegram import Update
from telegram.ext import ApplicationBuilder, CommandHandler, ContextTypes, MessageHandler, filters

TOKEN = os.environ["BOT_TOKEN"]
openai = OpenAI()  # OPENAI_API_KEY и OPENAI_BASE_URL из окружения
MODEL = os.environ.get("OPENAI_MODEL", "gpt-4o-mini")

SYSTEM_PROMPT = (
    "Ты — ассистент канала @PythonSimple о Python, Django и AI. "
    "Отвечай по-русски, коротко и по делу, с примерами кода, когда уместно."
)

MAX_HISTORY = 10        # сколько сообщений помнит бот
MAX_TOKENS = 600        # потолок ответа — защита бюджета

# Простое хранилище: chat_id -> история. В проде — SQLite или Redis.
histories: dict[int, list[dict]] = {}


async def chat(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    chat_id = update.effective_chat.id
    history = histories.setdefault(chat_id, [{"role": "system", "content": SYSTEM_PROMPT}])
    history.append({"role": "user", "content": update.message.text})
    history[:] = history[-MAX_HISTORY:]  # держим окно, иначе счёт растёт с каждым сообщением

    response = openai.chat.completions.create(
        model=MODEL,
        messages=history,
        max_tokens=MAX_TOKENS,
        temperature=0.4,
    )
    answer = response.choices[0].message.content
    history.append({"role": "assistant", "content": answer})
    await update.message.reply_text(answer)


async def reset(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    histories.pop(update.effective_chat.id, None)
    await update.message.reply_text("Контекст очищен.")


app = ApplicationBuilder().token(TOKEN).build()
app.add_handler(CommandHandler("reset", reset))
app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, chat))
app.run_polling()

async def chat_streaming(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    chat_id = update.effective_chat.id
    history = histories.setdefault(chat_id, [{"role": "system", "content": SYSTEM_PROMPT}])
    history.append({"role": "user", "content": update.message.text})
    history[:] = history[-MAX_HISTORY:]

    placeholder = await update.message.reply_text("…")
    answer, chunk_count = "", 0

    stream = openai.chat.completions.create(
        model=MODEL, messages=history, max_tokens=MAX_TOKENS, temperature=0.4, stream=True,
    )
    for chunk in stream:
        delta = chunk.choices[0].delta.content
        if not delta:
            continue
        answer += delta
        chunk_count += 1
        if chunk_count % 12 == 0:  # Telegram: не чаще ~1 редактирования в секунду
            await placeholder.edit_text(answer + " ▍")

    await placeholder.edit_text(answer)
    history.append({"role": "assistant", "content": answer})

async def compress_if_needed(history: list[dict]) -> None:
    if len(history) < MAX_HISTORY + 6:
        return
    summary_resp = openai.chat.completions.create(
        model=MODEL,
        messages=[
            {"role": "system", "content": "Сожми диалог в 3–4 предложения: о чём спрашивал пользователь, что отвечали."},
            *history[1:],  # без старого system-промпта
        ],
        max_tokens=200,
    )
    summary = summary_resp.choices[0].message.content
    history[:] = [history[0], {"role": "system", "content": f"Контекст диалога: {summary}"}]

import aiosqlite

DB = "bot.db"

async def load_history(chat_id: int) -> list[dict]:
    async with aiosqlite.connect(DB) as db:
        row = await (await db.execute(
            "SELECT messages FROM history WHERE chat_id = ?", (chat_id,)
        )).fetchone()
    import json
    return json.loads(row[0]) if row else [{"role": "system", "content": SYSTEM_PROMPT}]

async def save_history(chat_id: int, history: list[dict]) -> None:
    import json
    async with aiosqlite.connect(DB) as db:
        await db.execute(
            "INSERT INTO history (chat_id, messages) VALUES (?, ?) "
            "ON CONFLICT(chat_id) DO UPDATE SET messages = excluded.messages",
            (chat_id, json.dumps(history, ensure_ascii=False)),
        )
        await db.commit()