如何用 Python 和 API 创建 Telegram AI 机器人

准备 Python 和两种访问凭据

安装 Python 3.11 或更新版本。只要 Python 进程保持运行，并且能访问 Telegram API 和 https://void-api.tech/v1，机器人就能处理消息。首次运行可以使用自己的电脑。

向 Telegram 官方 @BotFather 发送 /newbot，设置名称，并选择一个以 bot 结尾且尚未被占用的用户名。它返回的令牌用于连接 Telegram。

在 VOID 账户中创建一个单独的 API 密钥，并检查余额。本教程通过 Chat Completions 调用 claude-haiku-4.5。BotFather 令牌和 VOID API 密钥用途不同，请分别填入 .env 中对应的字段。

通过 BotFather 创建机器人

创建 VOID API 密钥

查看模型与价格

安装依赖

创建独立的项目文件夹，将依赖列表保存为 requirements.txt。文件内容下方给出了 Linux 和 macOS 的终端命令。

在 Windows 上，可以用 py -m venv .venv 创建环境，然后在 PowerShell 中运行 .\.venv\Scripts\Activate.ps1 激活环境。之后同样使用 python -m pip install -r requirements.txt 安装依赖。

requirements.txt

python-telegram-bot==22.8
openai==3.37.0
python-dotenv==1.2.4

终端命令：Linux 和 macOS

python3 -m venv .venv
source .venv/bin/activate
python -m pip install -r requirements.txt

将配置保存到 .env

在 Python 文件所在目录创建 .env，将尖括号中的占位内容替换为自己的凭据。暂时将 TELEGRAM_ALLOWED_USER_ID 保持为 0；此时任何人都不能通过机器人调用模型。

首次启动后，/start 会显示你的 Telegram 数字用户 ID。将它写入 .env 并重启程序。这里不能填写以 @ 开头的用户名。

如果使用 Git，请把下列排除项写入 .gitignore，避免将密钥、虚拟环境和本地配额数据库加入仓库。

.env：替换凭据占位内容

TELEGRAM_BOT_TOKEN=<BotFather 令牌>
VOID_API_KEY=<VOID API 密钥>
VOID_MODEL=claude-haiku-4.5
TELEGRAM_ALLOWED_USER_ID=0

.gitignore

.env
.venv/
__pycache__/
quota.sqlite3*

启动机器人前先检查 VOID API

保存 check_api.py，在已激活的环境中运行 python check_api.py。脚本会向选定模型发送一次付费请求，并输出回复和 Token 数量。这个检查不需要 Telegram。

2026 年 10 月 23 日，原始俄语提示词得到的回复是 МАЯК，输入为 33 个 Token，输出为 7 个 Token。更换模型或提示词后，数量可能不同。

如果这一步报错，先检查 VOID_API_KEY、VOID_MODEL 和账户余额。修改 Telegram 配置无法解决模型 API 的调用失败。

check_api.py（使用实测的原始提示词）

import os
from pathlib import Path

from dotenv import load_dotenv
from openai import APIError, OpenAI

load_dotenv(Path(__file__).resolve().parent / ".env")

with OpenAI(
    api_key=os.environ["VOID_API_KEY"],
    base_url="https://void-api.tech/v1",
    timeout=60,
    max_retries=0,
) as client:
    try:
        response = client.chat.completions.create(
            model=os.environ["VOID_MODEL"],
            messages=[{"role": "user", "content": "Ответь только словом МАЯК."}],
            max_tokens=32,
        )
    except APIError as error:
        print(f"Ошибка API: {type(error).__name__}, HTTP {getattr(error, 'status_code', None)}")
        raise SystemExit(1) from None
    if not response.choices or not response.choices[0].message.content:
        raise SystemExit("API ответил, но текст отсутствует. Проверьте модель и лимит ответа.")
    print(response.choices[0].message.content)
    if response.usage:
        print(f"input={response.usage.prompt_tokens}, output={response.usage.completion_tokens}")

运行检查

python check_api.py

已验证运行的输出

МАЯК
input=33, output=7

将机器人代码保存为 bot.py

这个机器人处理所有者发来的私聊文字消息。示例不处理群聊、图片或语音消息。

每日计数保存在程序旁的 quota.sqlite3 中，重启后仍然有效。对话历史保存在进程内存里。/reset 会清空历史，但保留每日计数。

bot.py（实测示例，消息文本为俄语）

import logging
import os
import sqlite3
from datetime import datetime, timezone
from pathlib import Path

from dotenv import load_dotenv
from openai import APIError, AsyncOpenAI
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters

ROOT = Path(__file__).resolve().parent
load_dotenv(ROOT / ".env")
OWNER_ID = int(os.getenv("TELEGRAM_ALLOWED_USER_ID", "0"))
MODEL = os.environ["VOID_MODEL"]
DAILY_REQUESTS = 20
HISTORY_TURNS = 3
MAX_INPUT_CHARS = 3000
MAX_OUTPUT_TOKENS = 600

logging.basicConfig(level=logging.WARNING)
client = AsyncOpenAI(
    api_key=os.environ["VOID_API_KEY"],
    base_url="https://void-api.tech/v1",
    timeout=60.0,
    max_retries=0,
)
db = sqlite3.connect(ROOT / "quota.sqlite3")
db.execute("CREATE TABLE IF NOT EXISTS quota (day TEXT PRIMARY KEY, used INTEGER NOT NULL)")
db.commit()


def reserve_request():
    day = datetime.now(timezone.utc).date().isoformat()
    with db:
        cursor = db.execute(
            "INSERT INTO quota(day, used) VALUES (?, 1) "
            "ON CONFLICT(day) DO UPDATE SET used = used + 1 WHERE used < ?",
            (day, DAILY_REQUESTS),
        )
        return cursor.rowcount == 1


def keep_history(history, turns):
    if turns < 0:
        raise ValueError("turns must be non-negative")
    if turns == 0:
        return []
    return history[-2 * turns:]


async def start(update, context):
    await update.message.reply_text(
        f"Ваш Telegram ID: {update.effective_user.id}.\n"
        "Укажите его в TELEGRAM_ALLOWED_USER_ID и перезапустите бота.\n"
        "Команда /reset очищает историю диалога."
    )


async def reset(update, context):
    if update.effective_user.id != OWNER_ID:
        return
    context.user_data.clear()
    await update.message.reply_text("История очищена. Дневной счётчик сохранён.")


async def answer(update, context):
    if update.effective_user.id != OWNER_ID:
        await update.message.reply_text("Это личный бот. Доступ к модели закрыт.")
        return
    text = update.message.text.strip()
    if not text or len(text) > MAX_INPUT_CHARS:
        await update.message.reply_text(f"Отправьте от 1 до {MAX_INPUT_CHARS} символов.")
        return
    if not reserve_request():
        await update.message.reply_text("Дневной лимит исчерпан. Обновление в 00:00 UTC.")
        return

    history = context.user_data.get("history", [])
    user_message = {"role": "user", "content": text}
    messages = [
        {"role": "system", "content": "Отвечай по-русски, кратко и по существу."},
        *history,
        user_message,
    ]
    try:
        response = await client.chat.completions.create(
            model=MODEL,
            messages=messages,
            max_tokens=MAX_OUTPUT_TOKENS,
        )
    except APIError as error:
        logging.warning(
            "Void API: %s, HTTP %s", type(error).__name__, getattr(error, "status_code", None)
        )
        await update.message.reply_text(
            "Не удалось получить ответ. Проверьте ключ, баланс и модель в VOID. "
            "Попытка учтена в дневном лимите."
        )
        return

    choice = response.choices[0] if response.choices else None
    reply = (choice.message.content or "").strip() if choice else ""
    if not reply:
        await update.message.reply_text("Модель не вернула текст. Попытка учтена в лимите.")
        return
    context.user_data["history"] = keep_history(
        history + [user_message, {"role": "assistant", "content": reply}],
        HISTORY_TURNS,
    )

    for offset in range(0, len(reply), 2000):
        await update.message.reply_text(reply[offset : offset + 2000])
    if choice.finish_reason == "length":
        await update.message.reply_text("Ответ остановлен по лимиту токенов.")
    if response.usage:
        print(
            f"model={response.model} input={response.usage.prompt_tokens} "
            f"output={response.usage.completion_tokens}",
            flush=True,
        )


async def on_error(update, context):
    logging.warning("Ошибка обработчика: %s", type(context.error).__name__)


async def shutdown(application):
    await client.close()
    db.close()


def build_app():
    app = (
        Application.builder()
        .token(os.environ["TELEGRAM_BOT_TOKEN"])
        .concurrent_updates(False)
        .post_shutdown(shutdown)
        .build()
    )
    private = filters.ChatType.PRIVATE
    app.add_handler(CommandHandler("start", start, filters=private))
    app.add_handler(CommandHandler("reset", reset, filters=private))
    app.add_handler(MessageHandler(private & filters.TEXT & ~filters.COMMAND, answer))
    app.add_error_handler(on_error)
    return app


if __name__ == "__main__":
    build_app().run_polling(allowed_updates=[Update.MESSAGE], drop_pending_updates=True)

授权访问并检查对话记忆

运行 python bot.py，在 Telegram 中打开自己的机器人并发送 /start。把显示的 ID 填入 TELEGRAM_ALLOWED_USER_ID，使用 Ctrl+C 停止程序，然后重新启动。

先让机器人记住暗号 МАЯК，并要求它只回复这个词。接着询问刚才要求它记住的暗号，同样要求只回复一个词。第二次请求会包含前一次的问题和回答。实际使用的俄语消息保留在下方的测试记录中。

发送 /reset 后再提出新问题，程序就不会附带旧历史。模型仍可能碰巧猜中某个词，因此应检查发出的消息，或改用一个无关的新话题；仅凭一次猜中不能判断历史是否已清空。

同一个机器人令牌只运行一个实例。程序通过长轮询接收消息，这个示例不需要公开的 webhook 地址。drop_pending_updates=True 会跳过程序停止期间积压的消息。

启动机器人

python bot.py

调整记忆和用量限制

HISTORY_TURNS 指定保留多少轮历史问答。值为 3 时，新请求最多包含三轮历史。如果不需要记忆，将它设为 0 后重启。keep_history 单独处理了零值，因为普通切片 history[-0:] 会返回整个列表。

MAX_INPUT_CHARS 限制新消息的字符数，MAX_OUTPUT_TOKENS 限制模型的输出预算。字符和 Token 不是同一种单位，不能把这两个数相加来计算价格。

DAILY_REQUESTS 限制每个 UTC 自然日向模型发起请求的尝试次数。程序会在发送前计数，失败的调用也包含在内。网络错误发生后，不一定能确定服务提供方是否已经处理请求。SDK 的自动重试已关闭。

请求次数限制并不等于精确的金额预算。请在账户中查看输入 Token、输出 Token 和实际扣费。更换模型时，需要检查价格，以及对 Chat Completions 和 max_tokens 的支持情况。

bot.py 开头的配置

DAILY_REQUESTS = 20
HISTORY_TURNS = 3
MAX_INPUT_CHARS = 3000
MAX_OUTPUT_TOKENS = 600

查看 API 用量

如何减少一次请求的上下文

通过真实 API 验证了什么

2026 年 10 月 23 日，我们向机器人的处理函数传入了两条包含暗号的消息。处理函数通过 VOID 调用 claude-haiku-4.5，得到下方的回复。Token 数量来自 usage，费用来自 VOID 请求记录。对话记录保留了原始俄语消息。

这次测试用本地模拟对象代替了 Telegram 入站事件和 reply_text 消息发送。它验证了处理函数、历史传递和真实 VOID 回复，但没有测试消息经由 Telegram 服务器送达的过程。

其他本地检查覆盖了拒绝未授权用户、输入长度限制、历史裁剪与关闭、/reset、API 错误处理、长回复分段，以及每日配额的持久化。这些检查针对程序行为，不衡量模型的回答质量。

处理函数对 VOID 的两次真实调用

消息（原始俄语）：Запомни кодовое слово МАЯК. Ответь только этим словом.
模型回复：МАЯК
输入：73 Token。输出：7 Token。
请求扣费：$0.00005701

消息（原始俄语）：Какое кодовое слово я попросил запомнить? Ответь одним словом.
模型回复：МАЯК
输入：110 Token。输出：7 Token。
请求扣费：$0.00007810

机器人不回复时检查什么

如果 /start 没有响应，检查进程是否运行、BotFather 令牌是否正确，以及电脑能否访问 Telegram API。获取更新时出现冲突，可能是另一个实例正在使用同一令牌。

如果 /start 正常，但无法调用模型，核对 TELEGRAM_ALLOWED_USER_ID 是否为正确的数字 ID，并在修改 .env 后重启。

如果 check_api.py 报错，查看输出的错误类型和 HTTP 状态码，再检查 VOID 密钥、模型 ID 和余额。这里需要 API 密钥，不能使用账户密码。

如果提示每日配额已用完，检查请求次数。计数在 UTC 00:00 进入新的一天。删除 quota.sqlite3 会重置限制，因此迁移到服务器时应保留这个文件。

如果回复因 Token 限制而中断，可以要求更短的回复格式，或在关注费用的前提下提高 MAX_OUTPUT_TOKENS。

API 接入与参考资料

运行示例前，请创建 VOID 密钥并查看模型的当前价格。价格和模型可用性可能在测试日期之后发生变化。

创建 API 密钥

模型与价格

VOID API 参考

Telegram 文档

python-telegram-bot 22.8