【同款】免费chatgpt telegram bot源码-盛世博客

今天给搭建分享一个tg bot源码，可以做一个免费的chatgpt机器人，支持问答以及绘图

最近 Cloudflare Workers 已经正式开始商用了，配置了挺多大模型的，而且只要是 Beta 模型都免费使用，而目前大多数模型 (不论 GPT / 绘图等) 都是 Beta 状态，所以基本上可以免费试用

就算是收费了，Cloudflare 也贴心提供了每天 10000 个神经元，大约可以进行 500 次对话

你需要准备：1.一台服务器可以连接tg服务器 2.一个cloudflare账号 3.一个tg账号

第一步，创建一个机器人，https://t.me/BotFather

跟着要求创建就行，不会的google一下

第二部，登录你的cloudflare账号，获取Cloudflare 的 Account ID

获取方式就是打开 Cloudflare Dash，URL 中的那串就是，比如 41810b51b9f7521da5fea96d12xxxxxx

然后获取api key，点这里创建，记得要开放work ai权限https://dash.cloudflare.com/profile/api-tokens

第三部，下载源码，傻瓜式一键安装

git clone https://github.com/GenshinMinecraft/Cloudflare-Workers-Ai-Telegram-Bot.git
cd Cloudflare-Workers-Ai-Telegram-Bot
pip install requests pyTelegramBotAPI

程序文件是main.py，你需要打开它做一些修改

ACCOUNT_ID = "" # CloudFlare Account  账号 id，就是上面获取的
AUTH_TOKEN = "" # CloudFlare API Token  api token
Chat_MODEL = "@cf/qwen/qwen1.5-14b-chat-awq" # Text-Generation Model  语言模型
Image_MODEL = "@cf/bytedance/stable-diffusion-xl-lightning" # Text-to-Image Model  绘图模型
Telegram_Bot_Token = "" # Telegram Bot Token  从tg创建的机器人token
ADMIN_ID = xxxxx # Telegram Admin ID  就是管理员ID

至于管理员ID，就是你自己tg账号的ID，获取方法 @sswc_dc1_bot 发送 /getme 即可看到ID，复制过来就行

最后 python3 main.py 启动即可

项目是全开源的，没啥成本可以自己搞一个玩玩，支持群组、私聊

原项目地址 https://blogcdn.blog.highp.ing/p/workersaibot/

然后下面放一个我的修改版，区别就是，让gpt去掉个性化部分（在他的源码里有设定某些个性化语言），并且更换了语言模型，我认为相较于默认模型更加实用

import requests
import telebot
import os
import time

ACCOUNT_ID = "" # CloudFlare Account
AUTH_TOKEN = "" # CloudFlare API Token
Chat_MODEL = "@cf/mistral/mistral-7b-instruct-v0.1"
Image_MODEL = "@cf/bytedance/stable-diffusion-xl-lightning"
Audio2Text_MODEL = "@cf/openai/whisper"
Telegram_Bot_Token = "" # Telegram Bot Token
ADMIN_ID = xxxxx # Telegram Admin ID

bot = telebot.TeleBot(f"{Telegram_Bot_Token}")

def ChangeChat_MODEL(MODEL):
    global Chat_MODEL
    Chat_MODEL = MODEL

def GPT(prompt):
    response = requests.post(
        f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/ai/run/{Chat_MODEL}",
        headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
        json={
            "max_tokens": 1024,
            "messages": [
            {"role": "user", "content": prompt}
            ]
        }
    )
    result = response.json()
    answer = result.get("result").get("response")
    print(answer)
    return answer

def Image(prompt):
    response = requests.post(
        f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/ai/run/{Image_MODEL}",
        headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
        json={ "prompt": prompt }
    )
    return response.content

def Audio2Text(audio):
    response = requests.post(
        f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT_ID}/ai/run/{Audio2Text_MODEL}",
        headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
        data=audio
    )
    req = response.content
    print(req)
    audiotext = req.get("result").get("text")
    return audiotext

@bot.message_handler(commands=['start', 'image', 'ai', 'changegptmodel'])
def handle_command(message):
    command = message.text.split()[0]
    print(f"用户 {message.from_user.id} 使用了 {command} 功能，命令是 {message.text}")
    if command == "/start":
        print('start')
        bot.reply_to(message, 
            """
            基础命令:
            /ai 问题: 群组内使用
            /image 关键词: 画图
            """)

    elif command == "/image":
        imageword = (message.text[7:len(message.text)])
        if imageword == '':
            bot.reply_to(message, "绘画提示词不能为空")
            return 0
        try:
            try:
                bot.reply_to(message, "Drawing...", parse_mode='Markdown')
            except:
                bot.send_message(message.chat.id, "Drawing...", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
            png = Image(imageword)
        except:
            print("获取失败")
            try:
                bot.reply_to(message, "连接服务器失败", parse_mode='Markdown')
                return 1
            except:
                bot.send_message(message.chat.id, "连接服务器失败", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
                return 1
        bot.send_photo(message.chat.id, png, caption=imageword)
        print(f"绘制完成，提示词 {imageword}")

    elif command == "/ai":
        question = (message.text[4:len(message.text)])
        print(f"用户 {message.from_user.id} 使用了 Ask GPT 功能，问题是 {message.text}")
        try:
            try:
                bot.reply_to(message, "Thinking...", parse_mode='Markdown')
            except:
                bot.send_message(message.chat.id, "Thinking...", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
            replytxt = GPT(message.text)
        except:
            print("获取失败")
            print(Chat_MODEL)
            try:
                bot.reply_to(message, "连接服务器失败", parse_mode='Markdown')
                return 1
            except:
                bot.send_message(message.chat.id, "连接服务器失败", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
                return 1
        try:
            bot.reply_to(message, replytxt, parse_mode='Markdown')
        except:
            bot.send_message(message.chat.id, replytxt, parse_mode='Markdown')
            print("为什么有人会删消息啊...")

    elif command == "/changegptmodel":
        if message.from_user.id == ADMIN_ID:
            if (message.text[15:len(message.text)]) != '':
                ChangeChat_MODEL(message.text[16:len(message.text)])
                bot.reply_to(message, "GPT 模型已经更改为 "+Chat_MODEL, parse_mode='Markdown')
            else:
                bot.reply_to(message, "模型名不得为空", parse_mode='Markdown')
        else:
            bot.reply_to(message, "非管理员不可使用该命令", parse_mode='Markdown')

    elif command == "/getgptmodel":
        bot.reply_to(message, Chat_MODEL)

# 由于 Cloudflare API 原因，语音转文字暂不开放
#@bot.message_handler(content_types=['audio'])
#def handle_audio(message):
#    print(f"用户 {message.from_user.id} 使用了 Audio2Text 功能")
#    file_id = message.audio.file_id
#    file_info = bot.get_file(file_id)
#    file = bot.download_file(file_info.file_path)
#    print(f"用户 {message.from_user.id} 的 Audio2Text 文件已下载完成")
#    audiotext = Audio2Text(file)
#    bot.reply_to(message, "Audio2Text 结果: "+audiotext)

@bot.message_handler(func=lambda _: True)
def handle_message(message):
    if message.chat.type == "private":
        print(f"用户 {message.from_user.id} 使用了 Ask GPT 功能，问题是 {message.text}")
        try:
            try:
                bot.reply_to(message, "Thinking...", parse_mode='Markdown')
            except:
                bot.send_message(message.chat.id, "Thinking...", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
            replytxt = GPT(message.text)
        except:
            print("获取失败")
            print(Chat_MODEL)
            try:
                bot.reply_to(message, "连接服务器失败", parse_mode='Markdown')
                return 1
            except:
                bot.send_message(message.chat.id, "连接服务器失败", parse_mode='Markdown')
                print("为什么有人会删消息啊...")
                return 1
        try:
            bot.reply_to(message, replytxt, parse_mode='Markdown')
        except:
            bot.send_message(message.chat.id, replytxt, parse_mode='Markdown')
            print("为什么有人会删消息啊...")
    else:
        return 1

bot.polling()