Spaces:
Sleeping
Sleeping
File size: 3,466 Bytes
1ec9b78 |
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 |
import os
import requests
import logging
import openai
from contextlib import contextmanager
from tenacity import retry, wait_random_exponential, stop_after_attempt
## 处理代理：
http_proxy = ""
https_proxy = ""
http_proxy = os.environ.get("HTTP_PROXY", http_proxy)
https_proxy = os.environ.get("HTTPS_PROXY", https_proxy)
# 重置系统变量，在不需要设置的时候不设置环境变量，以免引起全局代理报错
os.environ["HTTP_PROXY"] = ""
os.environ["HTTPS_PROXY"] = ""
# GPT_MODEL = "gpt-3.5-turbo-0613"
GPT_MODEL = "gpt-4"
@contextmanager
def retrieve_proxy(proxy=None):
"""
1, 如果proxy = NONE，设置环境变量，并返回最新设置的代理
2，如果proxy ！= NONE，更新当前的代理配置，但是不更新环境变量
"""
global http_proxy, https_proxy
if proxy is not None:
old_var = os.environ["HTTP_PROXY"], os.environ["HTTPS_PROXY"]
os.environ["HTTP_PROXY"] = proxy
os.environ["HTTPS_PROXY"] = proxy
yield proxy, proxy
# return old proxy
os.environ["HTTP_PROXY"], os.environ["HTTPS_PROXY"] = old_var
else:
old_var = os.environ["HTTP_PROXY"], os.environ["HTTPS_PROXY"]
os.environ["HTTP_PROXY"] = http_proxy
os.environ["HTTPS_PROXY"] = https_proxy
yield http_proxy, https_proxy # return new proxy
# return old proxy
os.environ["HTTP_PROXY"], os.environ["HTTPS_PROXY"] = old_var
@retry(wait=wait_random_exponential(multiplier=1, max=40), stop=stop_after_attempt(10))
def get_geoip(proxy=None):
try:
with retrieve_proxy(proxy):
response = requests.get("https://ipapi.co/json/", timeout=5)
data = response.json()
except Exception as e:
data = {"error": True, "reason": f"{e}"}
return e
country = data["country_name"]
if country == "China":
text = "**您的IP区域：中国。请立即检查代理设置，在不受支持的地区使用API可能导致账号被封禁。**"
logging.error(text)
return -1, text
else:
text = f"您的IP区域：{country}。"
logging.info(text)
return 1, text
@retry(wait=wait_random_exponential(multiplier=1, max=20), stop=stop_after_attempt(5))
def chat_completion_request(
messages,
functions=None,
function_call=None,
model=GPT_MODEL,
proxy=None,
**kwargs,
):
headers = {
"Content-Type": "application/json",
"Authorization": "Bearer " + openai.api_key,
}
json_data = {"model": model, "messages": messages}
if functions is not None:
json_data.update({"functions": functions})
if function_call is not None:
json_data.update({"function_call": function_call})
if kwargs is not None:
json_data.update(kwargs)
with retrieve_proxy(proxy):
try:
response = requests.post(
"https://api.openai.com/v1/chat/completions",
headers=headers,
json=json_data,
)
return response
except Exception as e:
print("Unable to generate ChatCompletion response")
print(f"Exception: {e}")
return e
def generate_messages(prompt):
messages = []
messages.append({"role": "system", "content": "You are an robot auto control code generator."})
messages.append({"role": "user", "content": f"{prompt}"})
return messages
|