1 讯飞星火简介
讯飞星火是科大讯飞推出的一款先进的人工智能大模型，它具备强大的语言理解和知识问答能力，能够在多种场景中提供智能化服务。2024年6月27日，科大讯飞发布了讯飞星火大模型V4.0版本，全面对标GPT-4 Turbo。现有的模型类型如下：
Spark Lite是完全免费的，无限量Token，Spark Pro、Spark3.5 Max、Spark 4.0 Ultra目前是有免费额度的。
2 开发者快速指引
2.1 注册账号
打开讯飞星火大模型官方网站：
注册账号：
2.2 创建应用，获取接口认证信息
登录后，点击左下角的“API接入”：
点击“在线调试”：
右上角下拉菜单的「我的应用」进入控制台。若您的账户未曾创建过应用，系统会引导您创建您的第一个应用。
填写应用名称、应用分类、应用功能描述：
应用创建完成之后，就可以通过左侧的服务列表，选择您要使用的服务。在服务管理面板中，您将看到这个服务对应的可用量、历史用量、服务接口的认证信息（后面Python 调用API需要用到这3个参数）。
讯飞开放平台支持一个账户创建多个应用。当您需要返回应用列表页切换应用，可以点击页面左上角应用名称上方的返回按钮，或顶部右侧个人菜单中的「我的应用」。进入应用列表后，选择一个应用点击应用名称，即可进入这个应用对应的服务管理页。
同一个应用APPID可以用在多个业务上，这个没有限制，但考虑到多个业务共用一个APPID无法分业务统计用量，建议一个业务对应一个应用APPID。
2.3 获取Token
接口采用tokens方式计费。tokens与词表、分词方案相关，没有精确的计算方式，但是接口会返回本次计费的tokens数（详见接口文档响应参数描述)。接口计费会将请求text字段下所有的content内容均计费，开发者需要酌情考虑保留的历史对话信息数量，避免浪费tokens（最大的输入tokens见接口文档参数描述）。
以最新的Spark 4.0 Ultra为例，在左侧的列表中选中Spark 4.0 Ultra，点击“立即购买”按钮：
选择应用，完成身份认证及设置支付密码的个人可以免费领取200万的token，有效期为1年。
0元下单：
3 Python 调用 API
3.1 获取Web API 文档
点击链接：星火认知大模型Web API文档
详细阅读说明文档，重点看下请求参数：
role 参数的取值为[system,user,assistant]，system用于设置对话背景，user表示是用户的问题，assistant表示AI的回复。其中 system 只有Spark 4.0Ultra/Max支持，比如一些定制化的需求，使用Spark Lite 版本是无法实现的。
3.2 获取示例代码
下拉接口文档到最后，下面会有一些不同语言的调用示例以供参考，我们以 Python 为例：
第一个示例文件是单次对话，第二个则可以连续对话。这里我们先介绍第一个的使用方法。下载后解压，可以得到一个“sparkAPI.py”的文件，内容如下：
# coding: utf-8import _thread as threadimport osimport timeimport base64import base64import datetimeimport hashlibimport hmacimport jsonfrom urllib.parse import urlparseimport sslfrom datetime import datetimefrom time import mktimefrom urllib.parse import urlencodefrom wsgiref.handlers import format_date_timeimport websocketimport openpyxlfrom concurrent.futures import ThreadPoolExecutor, as_completedimport osclass Ws_Param(object): # 初始化 def __init__(self, APPID, APIKey, APISecret, gpt_url): self.APPID = APPID self.APIKey = APIKey self.APISecret = APISecret self.host = urlparse(gpt_url).netloc self.path = urlparse(gpt_url).path self.gpt_url = gpt_url # 生成url def create_url(self): # 生成RFC1123格式的时间戳 now = datetime.now() date = format_date_time(mktime(now.timetuple())) # 拼接字符串 signature_origin = "host: " + self.host + "\n" signature_origin += "date: " + date + "\n" signature_origin += "GET " + self.path + " HTTP/1.1" # 进行hmac-sha256进行加密 signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256).digest() signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8') authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"' authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') # 将请求的鉴权参数组合为字典 v = { "authorization": authorization, "date": date, "host": self.host } # 拼接鉴权参数，生成url url = self.gpt_url + '?' + urlencode(v) # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致 return url# 收到websocket错误的处理def on_error(ws, error): print("### error:", error)# 收到websocket关闭的处理def on_close(ws): print("### closed ###")# 收到websocket连接建立的处理def on_open(ws): thread.start_new_thread(run, (ws,))def run(ws, *args): data = json.dumps(gen_params(appid=ws.appid, query=ws.query, domain=ws.domain)) ws.send(data)# 收到websocket消息的处理def on_message(ws, message): # print(message) data = json.loads(message) code = data['header']['code'] if code != 0: print(f'请求错误: {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 gen_params(appid, query, domain): """ 通过appid和用户的提问来生成请参数 """ data = { "header": { "app_id": appid, "uid": "1234", # "patch_id": [] #接入微调模型，对应服务发布后的resourceid }, "parameter": { "chat": { "domain": domain, "temperature": 0.5, "max_tokens": 4096, "auditing": "default", } }, "payload": { "message": { "text": [{"role": "user", "content": query}] } } } return datadef main(appid, api_secret, api_key, gpt_url, domain, query): wsParam = Ws_Param(appid, api_key, api_secret, gpt_url) websocket.enableTrace(False) wsUrl = wsParam.create_url() ws = websocket.WebSocketApp(wsUrl, 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(sslopt={"cert_reqs": ssl.CERT_NONE})if __name__ == "__main__": main( appid="", api_secret="", api_key="", #appid、api_secret、api_key三个服务认证信息请前往开放平台控制台查看（https://console.xfyun.cn/services/bm35） gpt_url="wss://spark-api.xf-yun.com/v3.5/chat", # Max环境的地址 # Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # Pro环境的地址 # Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # Lite环境的地址 domain="generalv3.5", # Max版本 # domain = "generalv3" # Pro版本 # domain = "general" # Lite版本址 query="给我写一篇100字的作文" )这段代码定义了一个名为Ws_Param的类，用于处理WebSocket请求。以下是代码中各个方法的解释：
_init_(self, APPID, APIKey, APISecret, gpt_url)：初始化方法，用于设置类的实例变量。其中，APPID、APIKey、APISecret分别表示讯飞开放平台的应用ID、API Key和API Secret；gpt_url表示星火大模型的请求地址。create_url(self)：生成请求的URL。根据当前时间生成RFC1123格式的时间戳；然后，拼接签名字符串，包括host、date和GET请求行；接着，使用hmac-sha256算法对签名字符串进行加密；将加密后的签名字符串进行Base64编码，并将其添加到鉴权参数中，生成完整的URL。on_error(ws, error)：收到WebSocket错误的处理方法。当WebSocket连接发生错误时，会调用此方法。on_close(ws)：收到WebSocket关闭的处理方法。当WebSocket连接关闭时，会调用此方法。on_open(ws)：收到WebSocket连接建立的处理方法。当WebSocket连接建立时，会调用此方法。在此处，会启动一个新的线程来运行run函数。run(ws, *args)：运行函数，用于向讯飞星火大模型服务发送请求。根据WebSocket实例的appid和question属性生成请求参数，然后将请求参数转换为JSON字符串并通过WebSocket发送。on_message(ws, message)：收到WebSocket消息的处理方法。当从讯飞星火大模型服务接收到消息时，会调用此方法。解析接收到的消息，然后根据消息中的code判断请求是否成功；如果成功，则将返回的内容打印出来；如果code不为0，表示请求失败，此时关闭WebSocket连接。这里讯飞星火返回的内容是一段一段输出的，不是一次性全部输出，所以想要获得完整的回答，需要定义个全局变量将每一次输出的内容拼接起来。示例代码如下：# coding: utf-8import _thread as threadimport osimport timeimport base64import base64import datetimeimport hashlibimport hmacimport jsonfrom urllib.parse import urlparseimport sslfrom datetime import datetimefrom time import mktimefrom urllib.parse import urlencodefrom wsgiref.handlers import format_date_timeimport websocketimport openpyxlfrom concurrent.futures import ThreadPoolExecutor, as_completedimport osclass Ws_Param(object): # 初始化 def __init__(self, APPID, APIKey, APISecret, gpt_url): self.APPID = APPID self.APIKey = APIKey self.APISecret = APISecret self.host = urlparse(gpt_url).netloc self.path = urlparse(gpt_url).path self.gpt_url = gpt_url # 生成url def create_url(self): # 生成RFC1123格式的时间戳 now = datetime.now() date = format_date_time(mktime(now.timetuple())) # 拼接字符串 signature_origin = "host: " + self.host + "\n" signature_origin += "date: " + date + "\n" signature_origin += "GET " + self.path + " HTTP/1.1" # 进行hmac-sha256进行加密 signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'), digestmod=hashlib.sha256).digest() signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8') authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"' authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8') # 将请求的鉴权参数组合为字典 v = { "authorization": authorization, "date": date, "host": self.host } # 拼接鉴权参数，生成url url = self.gpt_url + '?' + urlencode(v) # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致 return url# 收到websocket错误的处理def on_error(ws, error): print("### error:", error)# 收到websocket关闭的处理def on_close(ws): print("### closed ###")# 收到websocket连接建立的处理def on_open(ws): thread.start_new_thread(run, (ws,))def run(ws, *args): data = json.dumps(gen_params(appid=ws.appid, query=ws.query, domain=ws.domain)) ws.send(data)# 收到websocket消息的处理def on_message(ws, message): # print(message) data = json.loads(message) code = data['header']['code'] if code != 0: print(f'请求错误: {code}, {data}') ws.close() else: choices = data["payload"]["choices"] status = choices["status"] content = choices["text"][0]["content"] print(content, end='') # 存放完整回答的全局变量 global answer answer += content if status == 2: print() print("#### 关闭会话") ws.close()def gen_params(appid, query, domain): """ 通过appid和用户的提问来生成请参数 """ data = { "header": { "app_id": appid, "uid": "1234", # "patch_id": [] #接入微调模型，对应服务发布后的resourceid }, "parameter": { "chat": { "domain": domain, "temperature": 0.5, "max_tokens": 4096, "auditing": "default", } }, "payload": { "message": { "text": [{"role": "user", "content": query}] } } } return datadef main(appid, api_secret, api_key, gpt_url, domain, query): wsParam = Ws_Param(appid, api_key, api_secret, gpt_url) websocket.enableTrace(False) wsUrl = wsParam.create_url() ws = websocket.WebSocketApp(wsUrl, 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(sslopt={"cert_reqs": ssl.CERT_NONE})if __name__ == "__main__":# 星火AI模型的回答answer = "" main( appid="", api_secret="", api_key="", #appid、api_secret、api_key三个服务认证信息请前往开放平台控制台查看（https://console.xfyun.cn/services/bm35） gpt_url="wss://spark-api.xf-yun.com/v3.5/chat", # Max环境的地址 # Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # Pro环境的地址 # Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # Lite环境的地址 domain="generalv3.5", # Max版本 # domain = "generalv3" # Pro版本 # domain = "general" # Lite版本址 query="给我写一篇100字的作文" )3.3 调试代码
注：项目仅支持 Python3.8+
安装PyPI上的包，在 Python 环境中执行命令：pip install --upgrade spark_ai_pythonpip install websocket-client填入参数：
填入星火大模型的URL值和domain值（根据不同版本对应的值），以及接口认证信息（开放平台控制台查看）：
运行示例代码：
python [sparkAPI.py路径]提示“ModuleNotFoundError: No module named ‘openpyxl’”：
openpyxl 是用于读写Excel文件的库，这里没用到，可以删除导包的这行代码“import openpyxl”。
然后继续运行，又提示on_close()函数参数错误：
这里on_close()方法少传两个参数，实际在传参时需要有三个参数，这里补齐另外两个参数：
def on_close(ws, close_code, close_reason): print("### closed ###")再次运行，成功！！！
