Python 如何调用讯飞星火大模型API

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接入”：

点击“在线调试”：
星火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，点击“立即购买”按钮：
购买token
选择应用，完成身份认证及设置支付密码的个人可以免费领取200万的token，有效期为1年。
免费token
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-8
import _thread as thread
import os
import time
import base64

import base64
import datetime
import hashlib
import hmac
import json
from urllib.parse import urlparse
import ssl
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_time

import websocket
import openpyxl
from concurrent.futures import ThreadPoolExecutor, as_completed
import os


class 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 data


def 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连接。

3.3 调试代码

注：项目仅支持 Python3.8+

安装PyPI上的包，在 Python 环境中执行命令：

pip install --upgrade spark_ai_python

安装 websocket 库：

pip install websocket-client

填入参数：
填入星火大模型的URL值和domain值（根据不同版本对应的值），以及接口认证信息（开放平台控制台查看）：
运行示例代码：

python [sparkAPI.py路径]

提示“ModuleNotFoundError: No module named ‘openpyxl’”：
缺少openpyxl
openpyxl 是用于读写Excel文件的库，这里没用到，可以删除导包的这行代码“import openpyxl”。
然后继续运行，又提示on_close()函数参数错误：

这里on_close()方法少传两个参数，实际在传参时需要有三个参数，这里补齐另外两个参数：

def on_close(ws, close_code, close_reason):
    print("### closed ###")

再次运行，成功！！！
运行成功