在大模型技术流行的今天，各行各业的业务仿佛都可以让大模型介入进行降本增效，但是如何让大模型稳定的返回JSON格式却成为了面临的第一道难题，今天就通过PYTHON调用大模型让大模型返回JSON示例给大家做个演示。
实现的思路是：标准而清晰的大模型PROMPT
你的角色是：专业的JSON提取专家。
你的任务是：根据我提供的信息抽取信息，并严格按照此JSON返回：{"name":"xxx","age":xxx}
根据以上要求，我的输入是：我叫小明，今年16岁了。
这样请求后，大模型会返回
{
"name":"小明",
"age":16
}# coding:utf8import base64
import datetime
import hashlib
import hmac
import json
import ssl
from datetime import datetime
from time import mktime
from urllib.parse import urlencode
from wsgiref.handlers import format_date_timeimport websockethost_url = "wss://spark-api.xf-yun.com/v4.0/chat"
appid = "" # 控制台获取
api_secret = ""
api_key = ""def product_url(api_secret, api_key):now_time = datetime.now()now_date = format_date_time(mktime(now_time.timetuple()))# print(now_date)# 拼接鉴权原始餐宿# now_date = "Fri, 18 Oct 2024 07:39:19 GMT"origin_base = "host: " + "spark-api.xf-yun.com" + "\n"origin_base += "date: " + now_date + "\n"origin_base += "GET " + "/v4.0/chat " + "HTTP/1.1"# print(origin_base)# sha256加密signature_sha = hmac.new(api_secret.encode('utf-8'), origin_base.encode('utf-8'),digestmod=hashlib.sha256).digest()signature_sha = base64.b64encode(signature_sha).decode(encoding='utf-8')print(signature_sha)authorization_origin = "api_key=\"%s\", algorithm=\"%s\", headers=\"%s\", signature=\"%s\"" % (api_key, "hmac-sha256", "host date request-line", signature_sha)authorization = base64.b64encode(authorization_origin.encode('utf-8')).decode(encoding='utf-8')print(authorization)# 将请求的鉴权参数组合为字典dict_data = {"authorization": authorization,"date": now_date,"host": "spark-api.xf-yun.com"}ws_url = host_url + '?' + urlencode(dict_data)# print(ws_url)return ws_urldef on_message(ws, message):print(f"Received message: {message}")status = json.loads(message)["header"]["status"]text_list = json.loads(message)["payload"]["choices"]["text"]for text in text_list:print(text["content"], end="")# print(f"{res} ---标志：{mark}")# # print(status)if status == 2:ws.close()def on_error(ws, error):print(f"Error: {error},{ws}")def on_close(ws, reason, res):print(f"WebSocket connection closed,{ws}")def on_open(ws):print(f"WebSocket connection opened,{ws},ws连接建立成功...")# 这里可以发送初始消息给服务器，如果需要的话first_dict = {"payload": {"message": {"text": [{"role": "user","content": r"""你的角色是：专业的JSON提取专家。
你的任务是：根据我提供的信息抽取信息，并严格按照此JSON返回：{"name":"xxx","age":xxx}
根据以上要求，我的输入是：我叫小明，今年16岁了。"""}]}},"parameter": {"chat": {"max_tokens": 8192,"domain": "4.0Ultra","temperature": 0.01,"show_ref_label": True}},"header": {"uid": "20241023","app_id": appid}}# print(json.dumps(first_dict))ws.send(json.dumps(first_dict)) # 发送第一帧def close_connection(ws):print("Closing WebSocket connection...")ws.close()# 主函数入口
if __name__ == '__main__':start_time = datetime.now()websocket.enableTrace(False)ws_url = product_url(api_secret, api_key)ws_entity = websocket.WebSocketApp(ws_url, on_message=on_message, on_error=on_error, on_close=on_close,on_open=on_open)ws_entity.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE})end_time = datetime.now()print(f"\n大模型耗时： {end_time - start_time}")
