基于官方原版python API示例修改,使用需申请讯飞大模型API,注册申请地址:https://xinghuo.xfyun.cn/sparkapi
当前个人申请免费获得200万tokens,企业认证免费获得500万tokens,一年有效期,约等于700万字的文本交互,作为日常学习体验十分够用。
1、以下代码在官方的基础上进行多轮精简,全部代码仅110行,保留最基本对话功能,直到主动关闭或输入exit退出。
程序基于python,运行需先安装python3.8以上版本
不喜欢精简想要更复杂的?请看大神讯飞星火大模型 API 封装:https://aifeisheng.com/thread-170.htm
# coding: utf-8
import json
import ssl
import hmac
import hashlib
import base64
from datetime import datetime
from time import mktime
from wsgiref.handlers import format_date_time
from urllib.parse import urlencode, urlparse
import websocket
class WsParam:
def __init__(self, appid, api_key, api_secret, gpt_url):
self.appid = appid
self.api_key = api_key
self.api_secret = api_secret
self.host = urlparse(gpt_url).netloc
self.path = urlparse(gpt_url).path
self.gpt_url = gpt_url
def create_url(self):
now = datetime.now()
date = format_date_time(mktime(now.timetuple()))
signature_origin = f"host: {self.host}\ndate: {date}\nGET {self.path} HTTP/1.1"
signature_sha = hmac.new(self.api_secret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256).digest()
signature_sha_base64 = base64.b64encode(signature_sha).decode('utf-8')
authorization_origin = f'api_key="{self.api_key}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode('utf-8')
v = {
"authorization": authorization,
"date": date,
"host": self.host
}
return f"{self.gpt_url}?{urlencode(v)}"
def on_error(ws, error):
print("Error:", error)
def on_close(ws, close_status_code, close_msg):
print("回答完毕")
def on_open(ws):
data = json.dumps({
"header": {
"app_id": ws.appid,
"uid": "1234",
},
"parameter": {
"chat": {
"domain": ws.domain,
"temperature": 0.5,
"max_tokens": 4096,
"auditing": "default",
}
},
"payload": {
"message": {
"text": [{"role": "user", "content": ws.query}]
}
}
})
ws.send(data)
def on_message(ws, message):
data = json.loads(message)
code = data['header']['code']
if code != 0:
print(f'Error: {code}, {data}')
ws.close()
else:
choices = data["payload"]["choices"]
status = choices["status"]
content = choices["text"][0]["content"]
print(content, end='')
if status == 2:
#print("回答完毕")
ws.close()
def main(appid, api_secret, api_key, gpt_url, domain, query):
ws_param = WsParam(appid, api_key, api_secret, gpt_url)
ws_url = ws_param.create_url()
ws = websocket.WebSocketApp(ws_url, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
ws.appid = appid
ws.query = query
ws.domain = domain
ws.run_forever()
if __name__ == "__main__":
appid = "你"
api_secret = "的"
api_key = "api"
gpt_url = "ws://spark-api.xf-yun.com/v3.5/chat"
domain = "generalv3.5"
while True:
query = input("请输入你的问题(输入'exit'退出):")
if query == 'exit':
break
main(
appid=appid,
api_secret=api_secret,
api_key=api_key,
gpt_url=gpt_url,
domain=domain,
query=query
)
2、申请接口后,替换appid、api_secret、api_key参数
3、示例:写一篇3000字的论文(大概15秒)