python ChatGPT和国内AI的API接口
2024年4月17日大约 5 分钟约 1642 字
"""chatgpt
pip install qianfan, openai, zhipuai
"""
import requests
from typing import Literal, TypedDict, List, Union
from openai import OpenAI
from openai import APIStatusError, APITimeoutError, APIConnectionError, OpenAIError
from openai._constants import DEFAULT_TIMEOUT, DEFAULT_CONNECTION_LIMITS
import qianfan
from zhipuai import ZhipuAI
import os
import random
import httpx
class AIBase:
name: str = None # AI引擎名称
api_key_list = [] # api_key 列表
def __init__(self, api_key_list=None):
# 更新api_key
if isinstance(api_key_list, list):
self.api_key_list = api_key_list
@property
def api_key(self) -> str:
"""随机获取一个api-key"""
return random.choice(self.api_key_list)
def get_answer(self, question_str, prompt_str=None) -> str:
"""一问一答的简易方式"""
raise NotImplementedError("子类必须实现这个方法")
class Qianfan(AIBase):
name = '百度千帆'
access_key = os.environ.get("QIANFAN_ACCESS_KEY")
secret_key = os.environ.get('QIANFAN_SECRET_KEY')
def get_answer(self, question_str, prompt_str=None) -> str:
"""
模型广场：https://console.bce.baidu.com/qianfan/modelcenter/model/buildIn/list
SDK教程和模型查询：https://cloud.baidu.com/doc/WENXINWORKSHOP/s/xlmokikxe
SecretKey获取：https://console.bce.baidu.com/iam/#/iam/accesslist
"""
os.environ["QIANFAN_ACCESS_KEY"] = 'xxx'
os.environ['QIANFAN_SECRET_KEY'] = 'xxx'
chat_comp = qianfan.ChatCompletion()
# 调用默认模型，即 ERNIE-Bot-turbo
# messages列表必须为奇数，且奇数位角色为user
resp = chat_comp.do(
model='ERNIE-Bot-4',
system=prompt_str or '你是一个友好的中文聊天助手。', # 人设
messages=[{"role": "user", "content": question_str}],
)
return resp.get('result')
class Baichuan(AIBase):
name = '百川智能'
def get_answer(self, question_str, prompt_str=None) -> str:
"""
api 文档：https://platform.baichuan-ai.com/docs/api
api key管理：https://platform.baichuan-ai.com/console/apikey
"""
try:
url = "https://api.baichuan-ai.com/v1/chat/completions"
# 准备请求头
headers = {
'Content-Type': 'application/json',
'Authorization': 'Bearer xxx' # 确保替换 ${api_key} 为你的实际API密钥
}
# 准备请求的数据
data = {
"model": "Baichuan2-Turbo",
"messages": [
{"role": "user", "content": question_str}
],
"temperature": 0.3, # 多样性，越高，多样性越好
"stream": False
}
response = requests.post(url, json=data, headers=headers)
result = response.json()
return result['choices'][0]['message']['content']
except Exception as e:
return ''
class Zhipu(AIBase):
"""清华智普"""
name = '清华智普'
def get_answer(self, question_str, prompt_str=None) -> str:
"""
SDK教程：https://open.bigmodel.cn/dev/api#overview
API Key获取：https://open.bigmodel.cn/usercenter/apikeys
"""
try:
client = ZhipuAI(api_key="xxx") # 请填写您自己的APIKey
response = client.chat.completions.create(
model="glm-4", # 填写需要调用的模型名称
messages=[
{"role": "system", "content": prompt_str or '你是一个友好的中文聊天助手。'},
{"role": "user", "content": question_str}
],
)
return response.choices[0].message.content
except:
return ''
class QWen(AIBase):
"""通义千问
灵积模型服务（通义千问SDK）
KEY管理: https://dashscope.console.aliyun.com/apiKey
文本提问: https://help.aliyun.com/zh/dashscope/developer-reference/api-details
插件广场: https://dashscope.console.aliyun.com/plugin
插件使用: https://help.aliyun.com/zh/dashscope/developer-reference/invoke-model-plugins
"""
name = '通义千问'
api_key_list = ['xxx']
def _get_answer(self, messages: Union[str, list], plugin: Literal['ocr', 'pdf_extracter', 'all'] = None, format: Literal['text', 'message'] = None) -> Union[list, str]:
"""
提问完整接口
:param plugin: 提问时使用的插件， all 代表使用所有插件
:param format:
:return:
"""
headers = {'Authorization': f'Bearer {self.api_key}'}
# 使用插件
if plugin == 'pdf_extracter':
result_format = 'message' if not format else format
headers['X-DashScope-Plugin'] = '{"pdf_extracter":{}}'
elif plugin == 'ocr':
result_format = 'message' if not format else format
headers['X-DashScope-Plugin'] = '{"ocr":{}}'
elif plugin == 'all':
result_format = 'message' if not format else format
headers['X-DashScope-Plugin'] = '{"ocr":{}, "pdf_extracter": {}}'
else:
result_format = 'text' if not format else format
if isinstance(messages, str):
messages = [{'role': 'user', 'content': messages}]
form = {'model': 'qwen-max', 'input': {'messages': messages}, 'parameters': {'result_format': result_format}}
response = requests.post('https://dashscope.aliyuncs.com/api/v1/services/aigc/text-generation/generation', headers=headers, json=form)
if response.status_code == 200:
data = response.json()
if result_format == 'text':
return data['output']['text']
else:
if data['output']['choices'][0].get('messages'):
return data['output']['choices'][0]['messages']
else:
return_message = data['output']['choices'][0]['message']
return [return_message]
elif response.status_code <= 499:
return response.json()['message']
else:
return '服务器级别的错误。'
def get_answer(self, question_str, prompt_str=None) -> str:
"""直接使用一问一答的简易方式"""
messages = self._get_answer(question_str, 'all')
if messages:
answer = messages[-1]['content']
return answer
class Moonshot(AIBase):
name = '月之暗面'
def get_answer(self, question_str, prompt_str=None) -> str:
"""
API文档：https://platform.moonshot.cn/docs/api-reference
API Key管理：https://platform.moonshot.cn/console/api-keys
"""
try:
client = OpenAI(
api_key="xxx",
base_url="https://api.moonshot.cn/v1",
)
completion = client.chat.completions.create(
model="moonshot-v1-8k",
messages=[
{"role": "system", "content": prompt_str or '你是一个友好的中文聊天助手。'},
{"role": "user", "content": question_str}
],
temperature=0.3,
)
return completion.choices[0].message.content
except Exception as e:
return ''
class ChatGPT(AIBase):
"""
账号密码：xionglilong@vip.qq.com A6f53s123456
管理key：https://platform.openai.com/api-keys
文档：https://platform.openai.com/docs/api-reference/introduction
模型弃用列表：https://platform.openai.com/docs/deprecations
带图像理解的GPT4：https://platform.openai.com/docs/guides/vision
依赖：pip install openai
"""
name = 'ChatGPT'
api_key_list = ['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']
http_proxy = 'http://xxx:xxx@x.x.x.x:0000'
# 类型提示
class MessageDict(TypedDict):
role: Literal['user', 'assistant', 'system']
content: str
MessageList = List[MessageDict]
# 聊天模型
def _get_answer(self,
messages: Union[MessageList, str] = '',
prompt_str='',
response_format: Literal['text', 'json_object'] = 'text',
model: str = 'gpt-4-turbo-preview',
http_proxy: str = None,
) -> str:
"""
参数:
messages (str, list): 传递给chatgpt的提问
response_format ('text', 'json_object'): 默认情况下，返回的格式是否为json。目前仅 gpt-4-1106-preview、gpt-3.5-turbo-1106 支持json
model (str): 使用的模型名称
http_proxy (str): http代理
示例:
chatgpt.get_answer('你好')
"""
# 国内必须配置http代理，重写httpx客户端
if http_proxy or True:
custom_client = httpx.Client(
base_url='https://api.openai.com/v1',
timeout=DEFAULT_TIMEOUT,
limits=DEFAULT_CONNECTION_LIMITS,
transport=None,
proxies=http_proxy or self.http_proxy,
)
client = OpenAI(
api_key=self.api_key,
http_client=custom_client,
)
else:
client = OpenAI(api_key=self.api_key)
# 获取可用模型列表
# for item in client.models.list():
# print(item)
# 制作消息体
if not messages:
messages = [
# 通常，对话的格式首先为系统消息，然后是交替的用户和助手消息。
{'role': 'system', 'content': prompt_str or '你是一个友好的中文聊天助手。'}, # 系统
{'role': 'user', 'content': '你好'}, # 用户
# {'role': 'assistant', 'content': '你好'}, # 助手
]
elif isinstance(messages, str):
messages = [
{'role': 'system', 'content': prompt_str or '你是一个友好的中文聊天助手。'}, # 系统
{'role': 'user', 'content': messages}, # 用户
]
try:
response = client.chat.completions.create(
model=model,
response_format={"type": response_format},
messages=messages,
# stream=True,
# temperature=0.5,
# max_tokens=1024
)
content = response.choices[0].message.content
return content
except APIStatusError as e:
# API HTTP请求状态码错误：4xx 5xx
if e.status_code == 401:
return "[错误] 您的API密钥或请求组织无效。"
elif e.status_code == 429:
print(e)
return "[错误] 您的请求速率或积分已超过了当前的配额，请检查您的账单详细信息。"
elif e.status_code == 500:
return "[错误] OpenAI服务器已宕机，请等待恢复。"
elif e.status_code == 503:
return "[错误] OpenAI服务器负载已爆炸，请短暂等待后重试。"
except TimeoutError as e:
return "[错误] 连接超时，请检查您代理配置。"
except APIConnectionError as e:
return "[错误] 无法连接到服务，请检查您的网络设置、代理配置、SSL 证书或防火墙规则。"
except OpenAIError as e:
return f'[错误] {e}'
def get_answer(self, question_str, prompt_str=None) -> str:
"""一问一答简单模式"""
result = self._get_answer(question_str, prompt_str)
return result
class CharGPTProxy(AIBase):
"""ChatGPT 国内代理接口
微信号：wZ2022oc
"""
name = 'ChatGPT代理'
api_key_list = ['xxx']
def get_answer(self, question_str, prompt_str=None) -> str:
# 设置API请求参数
params = {
"model": "gpt-4-turbo-preview",
"messages": [
{
"role": "system",
"content": prompt_str or "你是一个友好的中文聊天助手。"
},
{
"role": "user",
"content": question_str
}
]
}
# 发送API请求
response = requests.post(
"https://nio.pp.ua/v1/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}"
},
json=params
)
if response.status_code == 504:
data = {'error': {'code': 504, 'message': 'CDN节点请求源服务器超时，请联系站长，检查源服务器连接是否通畅或CDN节点被拦截。'}}
print(self.name, ' :CDN节点请求源服务器超时，请联系站长，检查源服务器连接是否通畅或CDN节点被拦截。')
elif response.status_code > 400:
data = {'error': {'code': response.status_code, 'message': response.reason}}
print(self.name, ' :', response.reason)
else:
# 解析响应数据
data = response.json()
if 'error' in data:
if data['error'].get('code') == 'insufficient_user_quota':
return '[错误] 您的请求速率或积分已超过了当前的配额，请检查您的账单详细信息。'
else:
return f"[错误] {data['error']['message']}"
answer_str = data['choices'][0]['message']['content']
return answer_str
# 所有ai类的列表
ai_class_dict = {'通义千问': QWen, 'ChatGPT': ChatGPT, 'ChatGPT代理': CharGPTProxy, '百度千帆': Qianfan, '月之暗面': Moonshot, '清华智普': Zhipu, '百川智能': Baichuan}
if __name__ == '__main__':
for name, AiClass in ai_class_dict.items():
answer = AiClass().get_answer(question_str='你好，请简短的做一个自我介绍。')
print('------------------------------------------')
print(f'[{name}]', answer)