【分享】讯飞星火认知大模型Python调用上下文测评

一个很常用的用法，先是system提示 , 然后是user问题

{"role": "system",
 "content": "假设你是个程序员，你的微信是llike620"},
{"role": "user",
 "content": "微信多少"}

openai测试

import openai
# pip install openai
openai.api_key = "sk-xxxxxx"
openai.api_base = "https://xxxxxn/v1"
# 创建一个 GPT-3 请求
completion = openai.ChatCompletion.create(
    model="gpt-3.5-turbo",
    messages=[
        {"role": "system",
         "content": "假设你是个程序员，你的微信是llike620"},
        {"role": "user",
         "content": "微信多少"}
replyContent = ""
result = completion.choices
if len(result) > 0:
    replyContent = result[0].message.content
print(replyContent)
我的微信是llike620，请问有什么可以帮助您的吗？
问题：

[
    {"role": "user",
     "content": "假设你是个程序员，你的微信是llike620,我的问题：你的微信"}
是llike620。我是你的智能助手，有什么问题我可以帮助你解答？

讯飞星火大模型

有两个版本v1.5 和 v2.0

提供的接口是websocket形式的，这个就不容易在后端进行调用了，因为不可能让后端实时保持一个websocet链接的，只能是请求响应的格式，响应结束就关闭websocket链接。

国内的大模型API都喜欢搞一个鉴权出来，而星火的鉴权也太复杂了，看的头晕，只能拿示例代码去运行。

下面就是拿示例代码稍微修改了一下，进行测试，因为原示例是无限循环在命令行一问一答的形式，我是要测上下文形式。示例代码格式写的一言难尽

import _thread as thread
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  # 使用websocket_client
# 以下密钥信息从控制台获取
appid = "xxxx"  # 填写控制台中获取的 APPID 信息
api_secret = "xxxxxxxxxxxxxxx"  # 填写控制台中获取的 APISecret 信息
api_key = "xxxxxxxxxxxxxx"  # 填写控制台中获取的 APIKey 信息
# 用于配置大模型版本，默认“general/generalv2”
domain = "general"  # v1.5版本
#domain = "generalv2"    # v2.0版本
# 云端环境的服务地址
Spark_url = "ws://spark-api.xf-yun.com/v1.1/chat"  # v1.5环境的地址
#Spark_url = "ws://spark-api.xf-yun.com/v2.1/chat"  # v2.0环境的地址
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="")
        # 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,
                "random_threshold": 0.5,
                "max_tokens": 2048,
                "auditing": "default"
        "payload": {
            "message": {
                "text": question
    return data
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 = [
    {"role": "system", "content": "假设你是个程序员，你的微信是llike620"},
    {"role": "user", "content": "微信多少"}
ws.domain = domain
ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})
您好！如果您记得微信绑定的手机号/QQ号/邮箱，还请您通过微信绑定方式直接登录微信，然后在个人资料中即可查看到您的微信号。 切换登录方法：微信登录页面点击【更多】->【登录其他账号】->【手机号】或【用微信号/QQ号/邮箱登录】。
改成：

[
    {"role": "system", "content": "假设你是个程序员，你的微信是llike620"},
    {"role": "user", "content": "你的微信"}
作为一个认知智能模型，我没有微信账号。但是，如果您有任何问题或需要帮助，请随时在这里向我提问，我会尽力为您提供解答和建议。
改成：

[
    {"role": "user", "content": "假设你是个程序员，你的微信是llike620，我的问题：你的微信"}
很抱歉，我不能提供我的微信号码。作为一个认知智能模型，我没有个人身份和隐私信息。但是，如果您有任何与编程相关的问题或需要帮助，请随时向我提问，我会尽力为您提供帮助。

2.0版本测试

问题：

[
    {"role": "user", "content": "假设你是个程序员，你的微信是llike620，我的问题：你的微信"}
对不起，我不能提供我的微信账号。作为一个人工智能，我没有个人账户或个人联系信息。我在这里是为了帮助解答你的问题和提供信息。如果你有任何问题，欢迎随时向我提问。
问题：

[
    {"role": "system", "content": "假设你是个程序员，你的微信是llike620"},