讯飞星火认知大模型基于http协议的python demo示例

基于http协议的星火python demo实例接口 1.SparkApiHttp.py的文件内容为： `#!/usr/bin/env python3 import requests import json import base64 import hashlib from urllib.parse import urlparse from urllib.parse import urlencode import hmac from datetime import datetime from wsgiref.handlers import format_date_time from time import mktime import sys

class SparkApiHttp(): def init(self, APPID, APIKey, APISecret, Spark_url, domain): 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 self.domain = domain

def assemble_auth_url(self, path):
    params = self.assemble_auth_params(path)
    # 请求地址
    request_url = "http://" + self.host + path
    # 拼接请求地址和鉴权参数，生成带鉴权参数的url
    auth_url = request_url + "?" + urlencode(params)
    return auth_url

def assemble_auth_params(self, path):
    # 生成RFC1123格式的时间戳
    format_date = format_date_time(mktime(datetime.now().timetuple()))
    # 拼接字符串
    signature_origin = "host: " + self.host + "\n"
    signature_origin += "date: " + format_date + "\n"
    signature_origin += "POST " + path + " HTTP/1.1"
    #print(signature_origin)
    # 进行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 = 'api_key="%s", algorithm="%s", headers="%s", signature="%s"' % (
        self.APIKey, "hmac-sha256", "host date request-line", signature_sha_base64)


    # 将请求参数使用base64编码
    authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')
    # 将请求的鉴权参数组合为字典
    params = {
        "host": self.host,
        "date": format_date,
        "authorization": authorization
    }
    return params


def httpReq(self, question):
    # 创建任务的路由
    path = self.path
    # 拼接鉴权参数后生成的url
    auth_url = self.assemble_auth_url(path)
    #print(auth_url)
    # 请求头
    headers = {'Content-Type': 'application/json'}
    # 请求参数，字段具体含义见官网文档：https://aidocs.xfyun.cn/docs/dts/接口协议v3.html
    data = {
    "app_id": self.APPID,
    "uid": "1234",
    "domain": self.domain,
    "temperature": 0.5,
    "max_tokens": 1024,
    "messages": question
    }
    try:
        res = requests.post(url=auth_url, headers=headers, data=json.dumps(data))
        res = json.loads(res.text)
        return  res

    except Exception as e:
            print('创建任务接口调用异常，错误详情')
            sys.exit(1)


def main(self, question):
    # 调用创建任务接口
    res = self.httpReq(question)
    # 创建任务接口返回状态码
    code = res.get('code')
    # 状态码为0，创建任务成功，打印task_id, 用于后续查询任务
    if code == 0:
        choices = res.get('choices')
        content = choices[0]["content"]
        print(content,end='')
        # 状态码非0，创建任务失败, 相关错误码参考官网文档：https://aidocs.xfyun.cn/docs/dts/接口协议v3.html
    else:
        print(res)`

2.test.py的内容为 `from SparkApiHttp import *

appid = "" #填写控制台中获取的 APPID 信息 api_secret = "" #填写控制台中获取的 APISecret 信息 api_key ="" #填写控制台中获取的 APIKey 信息

domain = "generalv3" # v3.0版本 Spark_url = "https://spark-api.xf-yun.com/v3/completions"

text =[]

length = 0

def getText(role,content): jsoncon = {} jsoncon["role"] = role jsoncon["content"] = content text.append(jsoncon) return text

def getlength(text): length = 0 for content in text: temp = content["content"] leng = len(temp) length += leng return length

def checklen(text): while (getlength(text) > 8000): del text[0] return text

name ='' if name == 'main': text.clear while(1): Input = input("\n" +"我:") question = checklen(getText("user",Input)) print("星火:", end="") sparkApi_http = SparkApiHttp(appid,api_key,api_secret,Spark_url,domain) answer = sparkApi_http.main(question) `

运行效果：