1、获取api接口的ID和key
这里我们可以获取星火免费赠送的200万个token使用和测试，获取方法如下:
打卡网站讯飞星火认知大模型-AI大语言模型-星火大模型-科大讯飞 ，登录用户点击免费使用
1.1 创建应用
点击购买首次应该会让创建一个应用， 如下图，按要求内容随意填写，然后提交
1.2 购买token
创建完成应用，回去购买，我这里选择个人的(这些都是在完成认证及设置了支付密码的基础)
1.3 获取 ID和key
在工单中心这个大模型3.0,页面就是，appid这三个我们会用到
2、 Python调用api
我的Python版本是在3.8左右的
创建一个名为SparkApi.py的Python文件，将下面代码粘进去
下面官方的代码有个坑，就是answer = ""是个全局变量，这个会将所有的提问拼接在一起，不过这个影响不大，就是打印answer的结果不好看，只要我们输入时text列表清除历史输入，token还是不带历史
import _thread as threadimport 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 websocket # 使用websocket_clientanswer = ""class Ws_Param(object): # 初始化 def __init__(self, APPID, APIKey, APISecret, Spark_url): self.APPID = APPID self.APIKey = APIKey self.APISecret = APISecret self.host = urlparse(Spark_url).netloc self.path = urlparse(Spark_url).path self.Spark_url = Spark_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.Spark_url + '?' + urlencode(v) # 此处打印出建立连接时候的url,参考本demo的时候可取消上方打印的注释，比对相同参数时生成的url与自己代码生成的url是否一致 return url# 收到websocket错误的处理def on_error(ws, error): print("### error:", error)# 收到websocket关闭的处理def on_close(ws,one,two): print(" ")# 收到websocket连接建立的处理def on_open(ws): thread.start_new_thread(run, (ws,))def run(ws, *args): data = json.dumps(gen_params(appid=ws.appid, domain= ws.domain,question=ws.question)) 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 # print(1) if status == 2: ws.close()def gen_params(appid, domain,question): """ 通过appid和用户的提问来生成请参数 """ data = { "header": { "app_id": appid, "uid": "1234" }, "parameter": { "chat": { "domain": domain, "temperature": 0.5, "max_tokens": 2048 } }, "payload": { "message": { "text": question } } } return datadef main(appid, api_key, api_secret, Spark_url,domain, question): wsParam = Ws_Param(appid, api_key, api_secret, Spark_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.question = question ws.domain = domain ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})在随便建一个py文件，将下面代码粘进去，这个是调用客户端，上面文件是服务端
import SparkApi#以下密钥信息从控制台获取appid = "XXXXXXXX" #填写控制台中获取的 APPID 信息api_secret = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" #填写控制台中获取的 APISecret 信息api_key ="XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" #填写控制台中获取的 APIKey 信息domain = "generalv3" # v3版本#云端环境的服务地址Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat" # v3环境的地址（"wss://spark-api.xf-yun.com/v3.1/chat）text =[]# length = 0def getText(role,content): jsoncon = {} jsoncon["role"] = role jsoncon["content"] = content text.append(jsoncon) return textdef getlength(text): length = 0 for content in text: temp = content["content"] leng = len(temp) length += leng return lengthdef checklen(text): while (getlength(text) > 8000): del text[0] return text if __name__ == '__main__': text.clear while(1): Input = input("\n" +"我:") question = checklen(getText("user",Input)) SparkApi.answer ="" print("星火:",end = "") SparkApi.main(appid,api_key,api_secret,Spark_url,domain,question) getText("assistant",SparkApi.answer) # print(str(text))这里就需要把刚才控制台的信息替换掉那些xxxx
ps：这里就是一个python的简易问答，如果想掉用接口，在这个调用文件里加一个调用函数
def spark_api(question): """ :param question: :return: """ question = checklen(getText("user",question)) SparkApi.answer ="" SparkApi.main(appid,api_key,api_secret,Spark_url,domain,question) text.clear() return SparkApi.answer建议在使用接口函数调用是加上text.clear(),清除历史对话，否则在一个长的连接调用时历史的token会加越来越长，十分消耗token，不需要历史的建议clear
3 、应用分享
最近作者在做项目是使用到了api辅助数据处理，其中发现在在提示词中加入一个历史对话可能会让结果更标准。下面是测试样例：
def getText(role,content): jsoncon = {} history_put = """['工程','货物',]\n请从上面选项中选择一个属于下面文本的分类\n左侧边坡宣传标语 ,结果只输出1,2 ,如果都不属于输出0 """ text.append({'role':'user','content':history_put }) text.append({'role': 'assistant', 'content': '0'}) jsoncon["role"] = role jsoncon["content"] = content text.append(jsoncon) return text上面代码是作者在函数中加入一串历史对话，这种思路就是每次调用都是一次历史和本次调用，这样在输出的结果上大部分都是一致的。在我测试7000条数据中仅有10条是有解释性话术。
4、bug问题
bug：如下图：
这个是缺少包安装这个即可：websocket-client==1.4.2
如下图：
这个我认为是spark_url="ws(s)://spark-api.xf-yun.com/v3.1/chat" # 错误案例
正确的spark_url="wss://spark-api.xf-yun.com/v3.1/chat" 或者ws://spark-api.xf-yun.com/v3.1/chat
5、v3.5最新版接口新增功能补充
（1） 接口参数：对应的domain参数为generalv3.5 ，ws地址3.1改成3.5
（2）system：设置对话背景或者模型角色
如果想获取结合上下文的回答，需要开发者每次将历史问答信息一起传给服务端，如下示例
注意：text里面的所有content内容加一起的tokens需要控制在8192以内，开发者如有较长对话需求，需要适当裁剪历史信息。
使用方法--> 旧版本传入请求数据时列表中只有usr和assistant这两个字典数据，现在要是使用system，只需要在usr前加入提示语字典如下图。
也就是在上文应用分享中 getText函数，text.append( {"role":"system","content":"你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"} )后面每次调用接口都是自带system
# 参数构造示例如下{ "header": { "app_id": "12345", "uid": "12345" }, "parameter": { "chat": { "domain": "generalv3.5", "temperature": 0.5, "max_tokens": 1024, } }, "payload": { "message": { # 如果想获取结合上下文的回答，需要开发者每次将历史问答信息一起传给服务端，如下示例 # 注意：text里面的所有content内容加一起的tokens需要控制在8192以内，开发者如有较长对话需求，需要适当裁剪历史信息 "text": [ {"role":"system","content":"你现在扮演李白，你豪情万丈，狂放不羁；接下来请用李白的口吻和用户对话。"} #设置对话背景或者模型角色 {"role": "user", "content": "你是谁"} # 用户的历史问题 {"role": "assistant", "content": "....."} # AI的历史回答结果 # ....... 省略的历史对话 {"role": "user", "content": "你会做什么"} # 最新的一条问题，如无需上下文，可只传最新一条问题 ] } }}（3）了functions.text字段：用于方法的注册
# 参数构造示例如下,仅在原本生成的基础上，增加了functions.text字段，用于方法的注册{ "header": { "app_id": appid, "uid": "1234" }, "parameter": { "chat": { "domain": domain, "random_threshold": 0.5, "max_tokens": 2048, "auditing": "default" } }, "payload": { "message": { "text": [ {"role": "user", "content": ""} # 用户的提问 ] }, "functions": { "text": [ { "name": "天气查询", "description": "天气插件可以提供天气相关信息。你可以提供指定的地点信息、指定的时间点或者时间段信息，来检索诗词库，精准检索到天气信息。", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "地点，比如北京。" }, "date": { "type": "string", "description": "日期。" } }, "required": [ "location" ] } }, { "name": "税率查询", "description": "税率查询可以查询某个地方的个人所得税率情况。你可以提供指定的地点信息、指定的时间点，精准检索到所得税率。", "parameters": { "type": "object", "properties": { "location": { "type": "string", "description": "地点，比如北京。" }, "date": { "type": "string", "description": "日期。" } }, "required": [ "location" ] } } ] } } }6、参数说明
项目地址：
https://github.com/liukangjia666/spark_demo
小彩蛋：关注wx公众号--> AI疯人院
ps：我把讯飞3.5封装到了公众号聊天中,欢迎测试
