当前位置：首页 > news >正文

LLM模型-讯飞星火与百度文心api调用

news 2025/8/25 21:40:50

spark-wenxin

- - - 1-讯飞星火
    - - 1_1-SparkApi.py
      - 1_2- Chat_spark.py
      - 1_3-调用api
    - 2-百度文心
    - - 2_1.code
    - 3-两者之间比较与openai

1-讯飞星火

进入讯飞官网进行创建应用，获取相关密钥APPID，APISecret，APIKey，选择最新版本
首次调用讯飞官方api可能有两问题：

1-No module named 'websocket'#安装 pip install websocket_client==1.4.2
2-spark_url="ws(s)://spark-api.xf-yun.com/v3.1/chat" #改为"ws://spark-api.xf-yun.com/v3.1/chat"

现搞了个完整的代码如下：
第一个相关请求的代码：SparkApi.py 相关参数可自行修改

1_1-SparkApi.py

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_timeimport websocket  # 
answer = ""class Ws_Param(object):# 初始化def __init__(self, APPID, APIKey, APISecret, Spark_url):self.APPID = APPIDself.APIKey = APIKeyself.APISecret = APISecretself.host = urlparse(Spark_url).netlocself.path = urlparse(Spark_url).pathself.Spark_url = Spark_url# 生成urldef 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}# 拼接鉴权参数，生成urlurl = 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 answeranswer += 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,"random_threshold": 0.5,"max_tokens": 2048,#"auditing": "default"}},"payload": {"message": {"text": question}}}return datadef main(appid, api_key, api_secret, Spark_url,domain, question):# print("星火:")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 = appidws.question = questionws.domain = domainws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})

1_2- Chat_spark.py

通过创建应用获取密钥，选择版本创建，这里用的是3.0版本

import SparkApiclass SparkProcessor:#以下密钥信息从控制台获取3.0appid = "xxx"     #填写控制台中获取的 APPID 信息api_secret = "xxxx"   #填写控制台中获取的 APISecret 信息api_key ="xxx"    #填写控制台中获取的 APIKey 信息# # #2.0# appid = "xxx"     #填写控制台中获取的 APPID 信息# api_secret = "xxx"   #填写控制台中获取的 APISecret 信息# api_key ="xxx"    #填写控制台中获取的 APIKey 信息#用于配置大模型版本，默认“general/generalv2”#domain = "general"   # v1.5版本# domain = "generalv2"    # v2.0版本domain = "generalv3"    # v3.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环境的地址Spark_url = "ws://spark-api.xf-yun.com/v3.1/chat"import SparkApiclass SparkProcessor:@staticmethoddef _get_text(role, content):#jsoncon = {}jsoncon["role"] = rolejsoncon["content"] = contentreturn jsoncon#实例方法#def get_text(self, role, content):#    jsoncon = {}#   jsoncon["role"] = role#   jsoncon["content"] = content#    return jsoncon@staticmethoddef _get_length(text):length = 0for content in text:temp = content["content"]leng = len(temp)length += lengreturn length@staticmethoddef _check_len(text):#私有方法while SparkProcessor._get_length(text) > 8000:del text[0]return text#类方法@classmethoddef spark_api(cls, question):text = []question = cls._check_len([cls._get_text("user", question)])SparkApi.answer = ""SparkApi.main(cls._get_appid(),cls._get_api_key(),cls._get_api_secret(),cls._get_spark_url(),cls._get_domain(),question,)text.clear()return SparkApi.answer# 私有方法获取敏感信息@staticmethoddef _get_appid():return "xxx"@staticmethoddef _get_api_secret():return "xxx"@staticmethoddef _get_api_key():return "xxx"@staticmethoddef _get_domain():return "xxx"@staticmethoddef _get_spark_url():return "xxx"

1_3-调用api

prompt的基本写法规则：
任务：明确而简洁地陈述Prompt 要求模型生成的内容
指令：模型在生成文本时应遵循的指令。
角色：模型在生成文本时应扮演的角色。

prompt1="""将文本分类为中性、负面或正面。文本：我认为这次假期还可以。情感："""
from Chat_spark import SparkProcessor
result = SparkProcessor.spark_api(prompt1)
print(result)#中性

2-百度文心

人设+任务目标及解决方式+附加条件完整陈述

2_1.code

创建应用获取相关密钥

# -*- coding: utf-8 -*-
# @Author  : Elvis
# @Time    : 2023/11/13 10:56
# @File    : get_chat.py
# @describe: ""
api_key = 'xxx'
api_id = 'xxx'
secret_key = 'xxx'
name = 'CHAT'import requests
import jsondef get_access_token(api_k, secret_k):"""使用 API Key，Secret Key 获取access_token，替换下列示例中的应用API Key、应用Secret Key"""url = "https://aip.baidubce.com/oauth/2.0/token?grant_type=client_credentials&client_id=%s&client_secret=%s" % (api_k, secret_k)
#     print('url: ', url)payload = json.dumps("")headers = {'Content-Type': 'application/json','Accept': 'application/json'}response = requests.request("POST", url, headers=headers, data=payload)return response.json().get("access_token")def get_main(prompt,sys=None):res = get_access_token(api_key, secret_key)
#     print('res: ', res)#     url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions?access_token=" + get_access_token()url = "https://aip.baidubce.com/rpc/2.0/ai_custom/v1/wenxinworkshop/chat/completions_pro?access_token=" + respayload = json.dumps({"messages": [{"role": "user","content": prompt}],
#          "stream": True,"temperature":0.1,#(0, 1.0]"system":sys})headers = {'Content-Type': 'application/json'}response = requests.request("POST", url, headers=headers, data=payload)res = response.textprint(res)res1 = json.loads(res)
#     print(res1)return res1['result']