自建 OpenAI 镜像站 Demo

testopenai.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
# coding: utf-8
 
import openai
 
# openai.log = "debug"
openai.api_key = "sk-"
openai.api_base = "https://api_gpt4.ai-gaochao.com/v1"
 
 
DEBUG = False
 
# 非流式响应
# completion = openai.ChatCompletion.create(model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello world!"}])
# print(completion.choices[0].message.content)
 
def gpt_api_stream(messages: list):
    """为提供的对话消息创建新的回答 (流式传输)
 
        gpt4长问题需要用流式传输，不然容易报错
    Args:
        messages (list): 完整的对话消息
        api_key (str): OpenAI API 密钥
 
    Returns:
        tuple: (results, error_desc)
    """
    try:
        response = openai.ChatCompletion.create(
            model='gpt-4',
            # model="gpt-3.5-turbo",
            messages=messages,
            stream=True,
            # temperature=0.5,
            # presence_penalty=0,
 
        )
        completion = {'role': '', 'content': ''}
        for event in response:
            if event['choices'][0]['finish_reason'] == 'stop':
                if DEBUG:
                    print(f'收到的完成数据: {completion}')
                break
            for delta_k, delta_v in event['choices'][0]['delta'].items():
                if DEBUG:
                    print(f'流响应数据: {delta_k} = {delta_v}')
                completion[delta_k] += delta_v
        messages.append(completion)  # 直接在传入参数 messages 中追加消息
        msg = completion['content']     # 解析返回的msg
        return (True, msg)
    except Exception as err:
        return (False, f'OpenAI API 异常: {err}')
 
def gpt_api_no_stream(messages: list):

    try:
        completion = openai.ChatCompletion.create(
            # model="gpt-3.5-turbo",
            # model="gpt-4",
            model="gpt-4-0613",
            messages=messages
        )
        # print(completion)
         
        msg = None
        choices = completion.get('choices', None)
        if choices:
            msg = choices[0]['message']['content']
        else:
            msg = completion['message']['content']
        return (True, msg)
    except Exception as err:
        return (False, f'OpenAI API 异常: {err}')
 
 
if __name__ == '__main__':
     
    # 一次性回答版
    prompt = 'There are 9 birds in the tree, the hunter shoots one, how many birds are left in the tree？'  # 短问题
    prompt = "鲁迅为什么暴打周树人"
    # prompt = "假如你是一个电力行业专家，请回答下面的问题：变压器有哪些部分组成"                                     # 长问题。可以把DEBUG变量设置为True来看流传输信息
    print("发送：", prompt)
    print("使用的apikey：", openai.api_key)
    messages = [{'role': 'user','content': prompt},]
    ret, msg = gpt_api_stream(messages)
    # ret, msg = gpt_api_no_stream(messages)
    print("返回：", msg)
    if DEBUG:
        print("messages：", messages)
 
 
     
    # # 问答版
    # messages = []
    # while True:
    #     prompt = input("Q: ")
    #     messages.append({'role': 'user','content': prompt})
    #     ret, msg = gpt_api_stream(messages)
    #     print("A：", msg)

selectmoney.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
import requests
from datetime import datetime, timedelta
 
def check_billing(api_key, api_url):
    # 计算起始日期和结束日期
    now = datetime.now()
    start_date = now - timedelta(days=90)
    end_date = now + timedelta(days=1)
    sub_date = now.replace(day=1)
 
    # 设置API请求URL和请求头
    url_subscription = f"{api_url}/v1/dashboard/billing/subscription"
    url_balance = f"{api_url}/dashboard/billing/credit_grants"
    url_usage = f"{api_url}/v1/dashboard/billing/usage?start_date={start_date.strftime('%Y-%m-%d')}&end_date={end_date.strftime('%Y-%m-%d')}"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
 
    try:
        # 获取API限额
        response = requests.get(url_subscription, headers=headers)
        if response.status_code != 200:
            print("查询返回异常。可能是：\n1. apikey没有余额了\n2. apikey无效\n3. apikey的base_url填写错了")
            return
 
        subscription_data = response.json()
        total_amount = subscription_data['hard_limit_usd']
 
        # 判断总用量是否大于20，若大于则更新start_date为sub_date
        if total_amount > 20:
            start_date = sub_date
 
        # 重新生成url_usage
        url_usage = f"{api_url}/v1/dashboard/billing/usage?start_date={start_date.strftime('%Y-%m-%d')}&end_date={end_date.strftime('%Y-%m-%d')}"
 
        # 获取已使用量
        response = requests.get(url_usage, headers=headers)
        usage_data = response.json()
        total_usage = usage_data['total_usage'] / 100
 
        # 计算剩余额度
        remaining = total_amount - total_usage
 
        # 输出总用量、总额及余额信息
        print(f"Total Amount: {total_amount:.2f}")
        print(f"Used: {total_usage:.2f}")
        print(f"Remaining: {remaining:.2f}")
 
        return [total_amount, total_usage, remaining]
    except Exception as error:
        print(error)
        return [None, None, None]
 
# 使用示例
api_key = "sk-"
api_base = "https://api_gpt4.ai-gaochao.com"
 
result = check_billing(api_key, api_base)