PYTHON让大模型固定的返回JSON

在大模型技术流行的今天，各行各业的业务仿佛都可以让大模型介入进行降本增效，但是如何让大模型稳定的返回JSON格式却成为了面临的第一道难题，今天就通过PYTHON调用大模型让大模型返回JSON示例给大家做个演示。

实现的思路是：标准而清晰的大模型PROMPT

你的角色是：专业的JSON提取专家。
你的任务是：根据我提供的信息抽取信息，并严格按照此JSON返回：{"name":"xxx","age":xxx}
根据以上要求，我的输入是：我叫小明，今年16岁了。

这样请求后，大模型会返回
{
"name":"小明",
"age":16
}

# coding:utf8

import 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_time

import websocket

host_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_url


def 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}")