Python调用讯飞星火大模型v3.5接口

在国外，ChatGPT已经成为AI模型行业的大佬，但是国内如果需要使用，会有各种限制，本文介绍如何使用国内的模型。

在国内，讯飞星火大模型是一个非常优秀的中文预训练模型。本文将介绍如何使用Python调用讯飞星火大模型接口，实现文本生成等功能。

讯飞星火官网：讯飞星火认知大模型-AI大语言模型-星火大模型-科大讯飞

1、获取api接口的ID和key

可以获取星火免费赠送的200万个token使用和测试，个人学习使用完全够了

1.1 创建应用

       点击购买首次应该会让创建一个应用， 如下图，按要求内容随意填写，然后提交

1.2  购买token

        创建完成应用，回去购买，我这里选择个人的(这些都是在完成认证及设置了支付密码的基础)，我们选择免费的包

1.3 获取 ID和key

        在工单中心这个大模型3.5,页面就是，appid这三个我们会用到

1.4 接口文档的选择

官方提供很多版本的SDK开发包及文档，这里需要我们选择web，官方有介绍

1.5 获取python API接口文档

点击上图的WebAPI链接跳转到如下链接：

星火认知大模型Web API文档 | 讯飞开放平台文档中心

下拉文档到最后，下面会有一些不同语言的调用接口示例，我们选择python的：

点击后会自动下载示例程序包：

解压后如下：

打开查看如下内容所示：

1. # coding: utf-8
2. import _thread as thread
3. import os
4. import time
5. import base64
7. import base64
8. import datetime
9. import hashlib
10. import hmac
11. import json
12. from urllib.parse import urlparse
13. import ssl
14. from datetime import datetime
15. from time import mktime
16. from urllib.parse import urlencode
17. from wsgiref.handlers import format_date_time
19. import websocket
20. import openpyxl
21. from concurrent.futures import ThreadPoolExecutor, as_completed
22. import os
25. class Ws_Param(object):
26. # 初始化
27. def __init__(self, APPID, APIKey, APISecret, gpt_url):
28. self.APPID = APPID
29. self.APIKey = APIKey
30. self.APISecret = APISecret
31. self.host = urlparse(gpt_url).netloc
32. self.path = urlparse(gpt_url).path
33. self.gpt_url = gpt_url
35. # 生成url
36. def create_url(self):
37. # 生成RFC1123格式的时间戳
38. now = datetime.now()
39. date = format_date_time(mktime(now.timetuple()))
41. # 拼接字符串
42. signature_origin = "host: " + self.host + "\n"
43. signature_origin += "date: " + date + "\n"
44. signature_origin += "GET " + self.path + " HTTP/1.1"
46. # 进行hmac-sha256进行加密
47. signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
48. digestmod=hashlib.sha256).digest()
50. signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
52. authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
54. authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
56. # 将请求的鉴权参数组合为字典
57. v = {
58. "authorization": authorization,
59. "date": date,
60. "host": self.host
61. }
62. # 拼接鉴权参数，生成url
63. url = self.gpt_url + '?' + urlencode(v)
64. # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
65. return url
68. # 收到websocket错误的处理
69. def on_error(ws, error):
70. print("### error:", error)
73. # 收到websocket关闭的处理
74. def on_close(ws):
75. print("### closed ###")
78. # 收到websocket连接建立的处理
79. def on_open(ws):
80. thread.start_new_thread(run, (ws,))
83. def run(ws, *args):
84. data = json.dumps(gen_params(appid=ws.appid, query=ws.query, domain=ws.domain))
85. ws.send(data)
88. # 收到websocket消息的处理
89. def on_message(ws, message):
90. # print(message)
91. data = json.loads(message)
92. code = data['header']['code']
93. if code != 0:
94. print(f'请求错误: {code}, {data}')
95. ws.close()
96. else:
97. choices = data["payload"]["choices"]
98. status = choices["status"]
99. content = choices["text"][0]["content"]
100. print(content,end='')
101. if status == 2:
102. print("#### 关闭会话")
103. ws.close()
106. def gen_params(appid, query, domain):
107. """
108. 通过appid和用户的提问来生成请参数
109. """
111. data = {
112. "header": {
113. "app_id": appid,
114. "uid": "1234",
115. # "patch_id": [] #接入微调模型，对应服务发布后的resourceid
116. },
117. "parameter": {
118. "chat": {
119. "domain": domain,
120. "temperature": 0.5,
121. "max_tokens": 4096,
122. "auditing": "default",
123. }
124. },
125. "payload": {
126. "message": {
127. "text": [{"role": "user", "content": query}]
128. }
129. }
130. }
131. return data
134. def main(appid, api_secret, api_key, gpt_url, domain, query):
135. wsParam = Ws_Param(appid, api_key, api_secret, gpt_url)
136. websocket.enableTrace(False)
137. wsUrl = wsParam.create_url()
139. ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
140. ws.appid = appid
141. ws.query = query
142. ws.domain = domain
143. ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
146. if __name__ == "__main__":
147. main(
148. appid="",
149. api_secret="",
150. api_key="",
151. #appid、api_secret、api_key三个服务认证信息请前往开放平台控制台查看（https://console.xfyun.cn/services/bm35）
152. gpt_url="wss://spark-api.xf-yun.com/v3.5/chat",
153. # Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # v3.0环境的地址
154. # Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat" # v2.0环境的地址
155. # Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # v1.5环境的地址
156. domain="generalv3.5",
157. # domain = "generalv3" # v3.0版本
158. # domain = "generalv2" # v2.0版本
159. # domain = "general" # v2.0版本
160. query="给我写一篇100字的作文"
161. )

代码解释

这段代码定义了一个名为Ws_Param的类，用于处理WebSocket请求。以下是代码中各个方法的解释：

1. __init__(self, APPID, APIKey, APISecret, gpt_url):初始化方法，用于设置类的实例变量。其中，APPID、APIKey、APISecret分别表示讯飞开放平台的应用ID、API Key和API Secret;gpt_url表示讯飞语音合成服务的URL。
2. create_url(self):生成请求的URL。根据当前时间生成RFC1123格式的时间戳；然后，拼接签名字符串，包括host、date和GET请求行；接着，使用hmac-sha256算法对签名字符串进行加密；将加密后的签名字符串进行Base64编码，并将其添加到鉴权参数中，生成完整的URL。
3. on_error(ws, error):收到WebSocket错误的处理方法。当WebSocket连接发生错误时，会调用此方法。
4. on_close(ws):收到WebSocket关闭的处理方法。当WebSocket连接关闭时，会调用此方法。
5. on_open(ws):收到WebSocket连接建立的处理方法。当WebSocket连接建立时，会调用此方法。在此处，会启动一个新的线程来运行run函数。
6. run(ws, *args):运行函数，用于向讯飞语音合成服务发送请求。根据WebSocket实例的appid和question属性生成请求参数；然后，将请求参数转换为JSON字符串并通过WebSocket发送。
7. on_message(ws, message):收到WebSocket消息的处理方法。当从讯飞语音合成服务接收到消息时，会调用此方法。解析接收到的消息；然后，根据消息中的code判断请求是否成功；如果成功，则将返回的内容累加到全局变量result中，并打印出来；如果code不为0,表示请求失败，此时关闭WebSocket连接。

官方的代码有个坑，就是answer = ""是个全局变量，这个会将所有的提问拼接在一起，不过这个影响不大，就是打印answer的结果不好看，只要我们输入时text列表清除历史输入，token还是不带历史。

2、代码调试

直接运行发现报错

发现是on_close()方法少传两个参数，实际在传参时是有三个参数，这里我们给它随便补两个参数，然后调试发现不报错了

运行结果：

3、代码调整

虽然上面的代码可以直接运行了，但是没有交互，只能运行一次，并且不能获取用户输入，如果想实现这样的功能，需要调整代码：

参考官方请求参数：

这里我需要给原来的代码添加一个text的列表，将我们要问的问题全部写入到text列表中，然后传递给query参数

封装函数，添加如下代码：

1. text = []
4. # length = 0
6. def getText(role, content):
7. jsoncon = {}
9. history_put = """['工程','货物',]\n请从上面选项中选择一个属于下面文本的分类\n左侧边坡宣传标语
10. ,结果只输出1,2 ,如果都不属于输出0
11. """
12. text.append({'role': 'user', 'content': history_put})
13. text.append({'role': 'assistant', 'content': '0'})
14. # # 设置对话背景或者模型角色
15. # text.append({"role": "system", "content": "你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"})
16. jsoncon["role"] = role
17. jsoncon["content"] = content
18. text.append(jsoncon)
19. return text

发现接口调用示例对text长度有要求：

注意：text里面的所有content内容加一起的tokens需要控制在8192以内，开发者如有较长对话需求，需要适当裁剪历史信息

需要添加对text长度的检测和判断代码：

1. #获取长度
2. def getlength(text):
3. length = 0
4. for content in text:
5. temp = content["content"]
6. leng = len(temp)
7. length += leng
8. return length
10. #检测长度
11. def checklen(text):
12. while getlength(text) > 8000:
13. del text[0]
14. return text

接下来对主函数main进行修改，实现交互式及循环请求：

1. if __name__ == "__main__":
2. text.clear()
3. while 1:
4. Input = input("\n" + "我:")
5. query = checklen(getText("user", Input))
6. answer = ""
7. print("星火:", end="")
8. main(
9. appid="83e846c5", # 填写控制台中获取的 APPID 信息
10. api_secret="YTNkNzRiMDRkODBkMjFjOTBiNDM0ZDdl", # 填写控制台中获取的 APISecret 信息
11. api_key="678b91e71b311082d7bb915f50f44284", # 填写控制台中获取的 APIKey 信息
12. # appid、api_secret、api_key三个服务认证信息请前往开放平台控制台查看（https://console.xfyun.cn/services/bm35）
13. gpt_url="wss://spark-api.xf-yun.com/v3.5/chat",
14. # Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # v3.0环境的地址
15. # Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat" # v2.0环境的地址
16. # Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # v1.5环境的地址
17. domain="generalv3.5",
18. # domain = "generalv3" # v3.0版本
19. # domain = "generalv2" # v2.0版本
20. # domain = "general" # v2.0版本
21. query=query
22. )
23. # 这里是获取星火AI模型助手的回答
24. getText("assistant", answer)

在使用接口函数调用是加上text.clear(),清除历史对话，否则在一个长的连接调用时历史的token会加越来越长，十分消耗token，不需要历史的建议clear

请求参数这里需要调整，将之前的"text": [{"role": "user", "content": query}]改为如下，否则会报

json: cannot unmarshal array into Go struct field message.payload.message.text.content of type string'

调整后的代码如下：

1. def gen_params(appid, query, domain):
2. """
3. 通过appid和用户的提问来生成请参数
4. """
6. data = {
7. "header": {
8. "app_id": appid,
9. "uid": "1234",
10. # "patch_id": [] #接入微调模型，对应服务发布后的resourceid
11. },
12. "parameter": {
13. "chat": {
14. "domain": domain,
15. "temperature": 0.5,
16. "max_tokens": 4096,
17. "auditing": "default",
18. }
19. },
20. "payload": {
21. "message": {
22. "text": query
23. }
24. }
25. }
26. return data

4、代码测试

5、最终代码（亲测可用）

1. # coding: utf-8
2. import _thread as thread
3. import base64
4. import datetime
5. import hashlib
6. import hmac
7. import json
8. from urllib.parse import urlparse
9. import ssl
10. from datetime import datetime
11. from time import mktime
12. from urllib.parse import urlencode
13. from wsgiref.handlers import format_date_time
15. import websocket
17. text = []
19. # length = 0
21. class Ws_Param(object):
22. # 初始化
23. def __init__(self, APPID, APIKey, APISecret, gpt_url):
24. self.APPID = APPID
25. self.APIKey = APIKey
26. self.APISecret = APISecret
27. self.host = urlparse(gpt_url).netloc
28. self.path = urlparse(gpt_url).path
29. self.gpt_url = gpt_url
31. # 生成url
32. def create_url(self):
33. # 生成RFC1123格式的时间戳
34. now = datetime.now()
35. date = format_date_time(mktime(now.timetuple()))
37. # 拼接字符串
38. signature_origin = "host: " + self.host + "\n"
39. signature_origin += "date: " + date + "\n"
40. signature_origin += "GET " + self.path + " HTTP/1.1"
42. # 进行hmac-sha256进行加密
43. signature_sha = hmac.new(self.APISecret.encode('utf-8'), signature_origin.encode('utf-8'),
44. digestmod=hashlib.sha256).digest()
46. signature_sha_base64 = base64.b64encode(signature_sha).decode(encoding='utf-8')
48. authorization_origin = f'api_key="{self.APIKey}", algorithm="hmac-sha256", headers="host date request-line", signature="{signature_sha_base64}"'
50. authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
52. # 将请求的鉴权参数组合为字典
53. v = {
54. "authorization": authorization,
55. "date": date,
56. "host": self.host
57. }
58. # 拼接鉴权参数，生成url
59. url = self.gpt_url + '?' + urlencode(v)
60. # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致
61. return url
64. # 收到websocket错误的处理
65. def on_error(ws, error):
66. print("### error:", error)
69. # 收到websocket关闭的处理
70. def on_close(ws, one, two):
71. print("### closed ###")
74. # 收到websocket连接建立的处理
75. def on_open(ws):
76. thread.start_new_thread(run, (ws,))
79. def run(ws, *args):
80. data = json.dumps(gen_params(appid=ws.appid, query=ws.query, domain=ws.domain))
81. ws.send(data)
84. # 收到websocket消息的处理
85. def on_message(ws, message):
86. # print(message)
87. data = json.loads(message)
88. code = data['header']['code']
89. if code != 0:
90. print(f'请求错误: {code}, {data}')
91. ws.close()
92. else:
93. choices = data["payload"]["choices"]
94. status = choices["status"]
95. content = choices["text"][0]["content"]
96. print(content, end='')
97. global answer
98. answer += content
99. if status == 2:
100. print()
101. print("#### 关闭会话")
102. ws.close()
105. def gen_params(appid, query, domain):
106. """
107. 通过appid和用户的提问来生成请参数
108. """
110. data = {
111. "header": {
112. "app_id": appid,
113. "uid": "1234",
114. # "patch_id": [] #接入微调模型，对应服务发布后的resourceid
115. },
116. "parameter": {
117. "chat": {
118. "domain": domain,
119. "temperature": 0.5,
120. "max_tokens": 4096,
121. "auditing": "default",
122. }
123. },
124. "payload": {
125. "message": {
126. "text": query
127. }
128. }
129. }
130. return data
133. def getText(role, content):
134. jsoncon = {"role": role, "content": content}
135. # history_put = """['工程','货物',]\n请从上面选项中选择一个属于下面文本的分类\n左侧边坡宣传标语
136. # ,结果只输出1,2 ,如果都不属于输出0
137. # """
138. # text.append({'role': 'user', 'content': history_put})
139. # text.append({'role': 'assistant', 'content': '0'})
140. # # 设置对话背景或者模型角色
141. # text.append({"role": "system", "content": "你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"})
142. text.append(jsoncon)
143. return text
146. def getlength(text):
147. length = 0
148. for content in text:
149. temp = content["content"]
150. leng = len(temp)
151. length += leng
152. return length
155. def checklen(text):
156. while getlength(text) > 8000:
157. del text[0]
158. return text
161. def main(appid, api_secret, api_key, gpt_url, domain, query):
162. wsParam = Ws_Param(appid, api_key, api_secret, gpt_url)
163. websocket.enableTrace(False)
164. wsUrl = wsParam.create_url()
166. ws = websocket.WebSocketApp(wsUrl, on_message=on_message, on_error=on_error, on_close=on_close, on_open=on_open)
167. ws.appid = appid
168. ws.query = query
169. ws.domain = domain
170. ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
173. if __name__ == "__main__":
174. text.clear()
175. while 1:
176. Input = input("\n" + "我:")
177. query = checklen(getText("user", Input))
178. answer = ""
179. print("星火:", end="")
180. main(
181. appid="", # 填写控制台中获取的 APPID 信息
182. api_secret="", # 填写控制台中获取的 APISecret 信息
183. api_key="", # 填写控制台中获取的 APIKey 信息
184. # appid、api_secret、api_key三个服务认证信息请前往开放平台控制台查看（https://console.xfyun.cn/services/bm35）
185. gpt_url="wss://spark-api.xf-yun.com/v3.5/chat",
186. # Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # v3.0环境的地址
187. # Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat" # v2.0环境的地址
188. # Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat" # v1.5环境的地址
189. domain="generalv3.5",
190. # domain = "generalv3" # v3.0版本
191. # domain = "generalv2" # v2.0版本
192. # domain = "general" # v2.0版本
193. query=query
194. )
195. # 这里是获取星火AI模型助手的回答
196. getText("assistant", answer)

6、其他功能参数

system：设置对话背景或者模型角色

使用方法--> 旧版本传入请求数据时列表中只有usr和assistant这两个字典数据，现在要是使用system，只需要在usr前加入提示语字典如下图。

 也就是在上文应用分享中 getText函数，text.append(                   {"role":"system","content":"你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"} )后面每次调用接口都是自带system

1. # 参数构造示例如下
2. {
3. "header": {
4. "app_id": "12345",
5. "uid": "12345"
6. },
7. "parameter": {
8. "chat": {
9. "domain": "generalv3.5",
10. "temperature": 0.5,
11. "max_tokens": 1024,
12. }
13. },
14. "payload": {
15. "message": {
16. # 如果想获取结合上下文的回答，需要开发者每次将历史问答信息一起传给服务端，如下示例
17. # 注意：text里面的所有content内容加一起的tokens需要控制在8192以内，开发者如有较长对话需求，需要适当裁剪历史信息
18. "text": [
19. {"role":"system","content":"你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"} #设置对话背景或者模型角色
20. {"role": "user", "content": "你是谁"} # 用户的历史问题
21. {"role": "assistant", "content": "....."} # AI的历史回答结果
22. # ....... 省略的历史对话
23. {"role": "user", "content": "你会做什么"} # 最新的一条问题，如无需上下文，可只传最新一条问题
24. ]
25. }
26. }
27. }

测试如下：

一股子诗人的味道，哈哈！

7、问题解决

问题现象

在websocket同服务器进行连接时，出现没有enableTrace属性：

module 'websocket' has no attribute 'enableTrace'

问题原因

检查一下当前安装的websocket：

pip show websocket

检查这个库的相关发布信息：已经很久没维护了，早已被弃用：

pip_search websocket

解决方法

后续Python中websocket库改为使用websocket-client，需要重新安装：

卸载websocket，这个已弃用，websockets中没有enableTrace模块

 推荐安装如下库

pip install websocket-client -i https://pypi.tuna.tsinghua.edu.cn/simple/