抱歉，您的浏览器无法访问本站
本页面需要浏览器支持（启用）JavaScript
了解详情 >

绕过双重封锁部署ChatGPT到zhenxun_bot

QQ机器人网络相关

发布于：2023年3月8日

字数：3.3k 字

时长：16 分钟
次浏览

作为一个从ChatGPT公测用到现在的用户，有些无奈很难言说。本来OpenAI就不对咱们这个区域开放，使用官方的API搭建应用可以不借助VPN访问，算是解除了区域限制。但是，从2023年3月2日傍晚开始，API接口就开始没有响应了，官网没有问题，四处查询发现可能是API的域名上了GFW名单（暂不确定，有可能重大会议过去后会恢复？）。

因此，现在摆在眼前的问题是如何绕过双重封锁调用OpenAI的API接口？最稳妥的方式当然是给服务器挂个全局代理，但是我的服务器本身就在作代理服务器，给服务器再上个代理会比较麻烦……这里记录下自己实现的方式，顺便记录下是如何部署ChatGPT到zhenxun_bot（这个bot真的超级好用！）上的。

本人在这方面是小白，只是记录实现过程。

此部分内容需要以部署zhenxun_bot为前提、有一个未上GFW名单的域名（国内需要实名）。

1. 部署ChatGPT到zhenxun_bot

时间过去太久，已经找不到写插件的原作者了…我是在原插件的基础上copy和修改了一部分代码，实际上就是在zhenxun_bot的AI插件基础上做的一点删改（可能有些没删干净，懒得查了）。如果没有修改路径的话，原文件路径是/zhenxun_bot/plugins/ai/data_source.py，下面代码替代原文件内容（原文件最好另存以防万一）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
import os
import random
import re
from utils.http_utils import AsyncHttpx
from configs.path_config import IMAGE_PATH, DATA_PATH
from services.log import logger
from utils.message_builder import image, face
from configs.config import Config, NICKNAME
from .utils import ai_message_manager
from copy import deepcopy
from transformers import GPT2TokenizerFast
import openai

openai.api_key = "xxxxxxxxxxxxxx"

try:
    import ujson as json
except ModuleNotFoundError:
    import json

session_config = {
    'preset': '你是一个大型语言模型，可以回答我的问题。如果我有任何问题，请随时告诉你，你会尽力为我解答。',
    'context': ''
}
sessions = {}
tokenizer = GPT2TokenizerFast.from_pretrained("gpt2-large")
check_url = "https://v2.alapi.cn/api/censor/text"
index = 0
anime_data = json.load(open(DATA_PATH / "anime.json", "r", encoding="utf8"))

# 获取对话session
def get_chat_session(sessionid):
    if sessionid not in sessions:
        config = deepcopy(session_config)
        config['id'] = sessionid
        sessions[sessionid] = config
    return sessions[sessionid]
    
def chat_with_gpt(prompt):
    try:
        resp = openai.Completion.create(
            model = "text-davinci-003", 
            temperature = 0.9,
            max_tokens=3000,
            top_p=1,
            presence_penalty=0,
            frequency_penalty=0,
            prompt=prompt)
        resp = resp['choices'][0]['text']
    except openai.OpenAIError as e:
        print('openai 接口报错: ' + str(e))
        resp = str(e)
    return resp

async def get_chat_result(text: str, img_url: str, user_id: int, nickname: str) -> str:
    """
    获取 AI 返回值，顺序： 特殊回复 -> GPT3 -> 青云客
    :param text: 问题
    :param img_url: 图片链接
    :param user_id: 用户id
    :param nickname: 用户昵称
    :return: 回答
    """
    global index
    ai_message_manager.add_message(user_id, text)
    special_rst = await ai_message_manager.get_result(user_id, nickname)
    if special_rst:
        ai_message_manager.add_result(user_id, special_rst)
        return special_rst
    if index == 5:
        index = 0
    if len(text) < 6 and random.random() < 0.6:
        keys = anime_data.keys()
        for key in keys:
            if text.find(key) != -1:
                return random.choice(anime_data[key]).replace("你", nickname)
    rst = await GPT_3(text, user_id)
    if not rst:
        rst = await xie_ai(text)
    if not rst:
        return no_result()
    if nickname:
        if len(nickname) < 5:
            if random.random() < 0.5:
                nickname = "~".join(nickname) + "~"
                if random.random() < 0.2:
                    if nickname.find("大人") == -1:
                        nickname += "大~人~"
        rst = str(rst).replace("小主人", nickname).replace("小朋友", nickname)
    ai_message_manager.add_result(user_id, rst)
    return rst


# GPT3接口
async def GPT_3(msg: str, sessionid: int) -> str:
    """
    获取GPT3接口的回复
    指令如下(群内需@机器人)：1.[重置会话] 请发送 重置会话2.[设置人格] 请发送 设置人格+人格描述3.[重置人格] 请发送 重置人格。
    注意：重置会话不会清空人格,重置人格会重置会话!设置人格后人格将一直存在，除非重置人格或重启逻辑端!
    """
    try:
        if msg.strip() == '':
            return '您好，我是人工智能助手，如果您有任何问题，请随时告诉我，我将尽力回答。\n如果您需要重置我们的会话，请回复`重置会话`'
        # 获得对话session
        session = get_chat_session(sessionid)
        if '重置会话' == msg.strip():
            session['context'] = ''
            return "会话已重置"
        if '重置人格' == msg.strip():
            session['context'] = ''
            session['preset'] = session_config['preset']
            return '人格已重置'
        if msg.strip().startswith('设置人格'):
            session['preset'] = msg.strip().replace('设置人格', '')
            session['context'] = ''
        # 处理上下文逻辑
        token_limit = 4096 - 3000 - len(tokenizer.encode(session['preset'])) - 3
        session['context'] = session['context'] + "\n\nQ:" + msg + "\nA:"
        ids = tokenizer.encode(session['context'])
        tokens = tokenizer.decode(ids[-token_limit:])
        # 计算可发送的字符数量
        char_limit = len(''.join(tokens))
        session['context'] = session['context'][-char_limit:]
        # 从最早的提问开始截取
        pos = session['context'].find('Q:')
        session['context'] = session['context'][pos:]
        # 设置预设
        msg = session['preset'] + '\n\n' + session['context']
        print(msg)
        # 与ChatGPT交互获得对话内容
        message = chat_with_gpt(msg)
        print("会话ID: " + str(sessionid))
        print("ChatGPT返回内容: ")
        print(message)
        return message 
    except Exception as error:
        traceback.print_exc()
        return str('异常: ' + str(error)) 

# 屑 AI
async def xie_ai(text: str) -> str:
    """
    获取青云客回复
    :param text: 问题
    :return: 青云可回复
    """
    res = await AsyncHttpx.get(f"http://api.qingyunke.com/api.php?key=free&appid=0&msg={text}")
    content = ""
    try:
        data = json.loads(res.text)
        if data["result"] == 0:
            content = data["content"]
            if "菲菲" in content:
                content = content.replace("菲菲", NICKNAME)
            if "艳儿" in content:
                content = content.replace("艳儿", NICKNAME)
            if "公众号" in content:
                content = ""
            if "{br}" in content:
                content = content.replace("{br}", "\n")
            if "提示" in content:
                content = content[: content.find("提示")]
            if "淘宝" in content or "taobao.com" in content:
                return ""
            while True:
                r = re.search("{face:(.*)}", content)
                if r:
                    id_ = r.group(1)
                    content = content.replace(
                        "{" + f"face:{id_}" + "}", str(face(int(id_)))
                    )
                else:
                    break
        return (
            content
            if not content and not Config.get_config("ai", "ALAPI_AI_CHECK")
            else await check_text(content)
        )
    except Exception as e:
        logger.error(f"Ai xie_ai 发生错误 {type(e)}：{e}")
        return ""


def hello() -> str:
    """
    一些打招呼的内容
    """
    result = random.choice(
        (
            "哦豁？！",
            "你好！Ov<",
            f"库库库，呼唤{NICKNAME}做什么呢",
            "我在呢！",
            "呼呼，叫俺干嘛",
        )
    )
    img = random.choice(os.listdir(IMAGE_PATH / "zai"))
    if img[-4:] == ".gif":
        result += image(img, "zai")
    else:
        result += image(img, "zai")
    return result

# 没有回答时回复内容
def no_result() -> str:
    """
    没有回答时的回复
    """
    return (
        random.choice(
            [
                "你在说啥子？",
                f"纯洁的{NICKNAME}没听懂",
                "下次再告诉你(下次一定)",
                "你觉得我听懂了吗？嗯？",
                "我！不！知！道！",
            ]
        )
        + image(random.choice(os.listdir(IMAGE_PATH / "noresult")), "noresult")
    )

async def check_text(text: str) -> str:
    """
    ALAPI文本检测，主要针对青云客API，检测为恶俗文本改为无回复的回答
    :param text: 回复
    """
    if not Config.get_config("alapi", "ALAPI_TOKEN"):
        return text
    params = {"token": Config.get_config("alapi", "ALAPI_TOKEN"), "text": text}
    try:
        data = (await AsyncHttpx.get(check_url, timeout=2, params=params)).json()
        if data["code"] == 200:
            if data["data"]["conclusion_type"] == 2:
                return ""
    except Exception as e:
        logger.error(f"检测违规文本错误...{type(e)}：{e}")
    return text

openai.api_key需要上官网获取后填入

实际上就是把原来的图灵接口替换成GPT3接口。引入openai库和transformers库，使用了前者的openai.Completion.create()方法和后者GPT2TokenizerFast.from_pretrained()预训练的GPT2模型和分词器。

关键在于前者，因为OpenAI的API网站已经上了GFW名单，所以我们现在就算有api_key也无法调用API接口（会显示超时）。以下是解决方法。

2. 托管域名到CLOUDFLARE

后面我们要用到CLOUDFLARE，没有账户的话注册一个：https://dash.cloudflare.com/

注册之后点击右边的Websites，按照操作流程添加主域名，修改两个DNS服务器名字。

比如我这里用阿里云买了一个域名，需要登录阿里云的域名控制台，点击管理，进入右边DNS修改页面

修改原来默认的DNS服务器为lorna.ns.cloudflare.com和ram.ns.cloudflare.com。

中间可能还需要你邮件确认，按照提示操作就可以。

3. 创建CLOUDFLARE Workers

该步骤参考来自github [noobnooc]，感谢大佬提供的解决方案！

回到CLOUDFLARE，点击右边的创建Workers——Create a Service，这里直接确认创建一个服务。

创建之后点击Quick edit修改workers代码如下（起到代理api.openai.com的作用）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
// Website you intended to retrieve for users.
const upstream = 'api.openai.com'

// Custom pathname for the upstream website.
const upstream_path = '/'

// Website you intended to retrieve for users using mobile devices.
const upstream_mobile = upstream

// Countries and regions where you wish to suspend your service.
const blocked_region = []

// IP addresses which you wish to block from using your service.
const blocked_ip_address = ['0.0.0.0', '127.0.0.1']

// Whether to use HTTPS protocol for upstream address.
const https = true

// Whether to disable cache.
const disable_cache = false

// Replace texts.
const replace_dict = {
    '$upstream': '$custom_domain',
}

addEventListener('fetch', event => {
    event.respondWith(fetchAndApply(event.request));
})

async function fetchAndApply(request) {
    const region = request.headers.get('cf-ipcountry').toUpperCase();
    const ip_address = request.headers.get('cf-connecting-ip');
    const user_agent = request.headers.get('user-agent');

    let response = null;
    let url = new URL(request.url);
    let url_hostname = url.hostname;

    if (https == true) {
        url.protocol = 'https:';
    } else {
        url.protocol = 'http:';
    }

    if (await device_status(user_agent)) {
        var upstream_domain = upstream;
    } else {
        var upstream_domain = upstream_mobile;
    }

    url.host = upstream_domain;
    if (url.pathname == '/') {
        url.pathname = upstream_path;
    } else {
        url.pathname = upstream_path + url.pathname;
    }

    if (blocked_region.includes(region)) {
        response = new Response('Access denied: WorkersProxy is not available in your region yet.', {
            status: 403
        });
    } else if (blocked_ip_address.includes(ip_address)) {
        response = new Response('Access denied: Your IP address is blocked by WorkersProxy.', {
            status: 403
        });
    } else {
        let method = request.method;
        let request_headers = request.headers;
        let new_request_headers = new Headers(request_headers);

        new_request_headers.set('Host', upstream_domain);
        new_request_headers.set('Referer', url.protocol + '//' + url_hostname);

        let original_response = await fetch(url.href, {
            method: method,
            headers: new_request_headers,
            body: request.body
        })

        connection_upgrade = new_request_headers.get("Upgrade");
        if (connection_upgrade && connection_upgrade.toLowerCase() == "websocket") {
            return original_response;
        }

        let original_response_clone = original_response.clone();
        let original_text = null;
        let response_headers = original_response.headers;
        let new_response_headers = new Headers(response_headers);
        let status = original_response.status;
		
		if (disable_cache) {
			new_response_headers.set('Cache-Control', 'no-store');
	    }

        new_response_headers.set('access-control-allow-origin', '*');
        new_response_headers.set('access-control-allow-credentials', true);
        new_response_headers.delete('content-security-policy');
        new_response_headers.delete('content-security-policy-report-only');
        new_response_headers.delete('clear-site-data');
		
		if (new_response_headers.get("x-pjax-url")) {
            new_response_headers.set("x-pjax-url", response_headers.get("x-pjax-url").replace("//" + upstream_domain, "//" + url_hostname));
        }
		
        const content_type = new_response_headers.get('content-type');
        if (content_type != null && content_type.includes('text/html') && content_type.includes('UTF-8')) {
            original_text = await replace_response_text(original_response_clone, upstream_domain, url_hostname);
        } else {
            original_text = original_response_clone.body
        }
		
        response = new Response(original_text, {
            status,
            headers: new_response_headers
        })
    }
    return response;
}

async function replace_response_text(response, upstream_domain, host_name) {
    let text = await response.text()

    var i, j;
    for (i in replace_dict) {
        j = replace_dict[i]
        if (i == '$upstream') {
            i = upstream_domain
        } else if (i == '$custom_domain') {
            i = host_name
        }

        if (j == '$upstream') {
            j = upstream_domain
        } else if (j == '$custom_domain') {
            j = host_name
        }

        let re = new RegExp(i, 'g')
        text = text.replace(re, j);
    }
    return text;
}


async function device_status(user_agent_info) {
    var agents = ["Android", "iPhone", "SymbianOS", "Windows Phone", "iPad", "iPod"];
    var flag = true;
    for (var v = 0; v < agents.length; v++) {
        if (user_agent_info.indexOf(agents[v]) > 0) {
            flag = false;
            break;
        }
    }
    return flag;
}

修改之后点击右下角Save and deploy，此时worker地址还不能直接代替openai的API地址，需要进一步绑定前面的域名（CLOUDFLARE Workers只能绑定托管到CLOUDFLARE的域名，所以有了前面一步）。

4. 绑定域名

点击Workers进入管理页面，点击Triggers——Add Custom Domain，将前面托管的域名填进去，可以用自己喜欢的二级域名：

大约过几分钟，custom domains显示Certificate 为 Activate即可。

这个时候就可以通过你绑定的域名来访问api.openai.com了，可以通过其他POST工具调试接口，就不多说了。

5. 修改openai库

前面做的一系列步骤是让你可以通过其他域名访问openai的API网站，但是前面第一步写的插件调用了openai.Completion.create()方法函数，此时仍然会直接访问api.openai.com，这个时候就是扒源代码修改了。

locate openai先找到服务器上openai下载的位置，在对应的路径修改，比如我的文件路径是/root/anaconda3/lib/python3.9/site-packages/openai，修改该路径下的__init__.py文件：

第34行api_base后面的网址改为刚刚绑定的网址（/v1的部分不要动）。

上述步骤完成后，重启zhenxun_bot就可以在不对自己服务器做任何代理的情况下正常调用OpenAI的API接口了。

顺便说一下，2022年12月申请的openAI账号每个账户有18美元的额度，现在（2023年3月）申请的账号就只有5美元额度了，emmmmmmm…

不过上面的那个插件用的是text-davinci-003模型，和ChatGPT用的模型稍有不同，就在前几天ChatGPT公开了API，所使用的模型为gpt-3.5-turbo。并且我前面用的方法是Create completion，和ChatGPT创建实例的方法Create chat completion是不同的，且收费也不一样，现在ChatGPT API收费标准是0.002美元/1000 tokens，token数和字数是不一样的，要看分词器怎么分，不过现在API输出上限均为4096 token。有空再更新一下模型方法，这里只是做个记录。

详细的API调用方法需要参考官网API Reference - OpenAI API

本文由Phantom创作，转载请注明出处

博客内容遵循 署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 协议

本文永久链接是：http://www.shelven.com/2023/03/08/a.html

更新于：2023年3月8日

ChatGPT

欢迎小伙伴们留言评论~