bge_embedding/tem.py

import requests
import time

def access_remote_api(url: str,headers:dict, data: dict, max_retries=1):
    """调用bge-m3，embedding"""
    # 类型检查
    time.sleep(0.08)  # 方式频繁调用接口
    for attempt in range(max_retries):
        try:
            response = requests.post(url=url, headers=headers, json=data)
            if response.status_code == 200:
                return response.json()
        except Exception as e:
            print('请求embedding模型失败', e)
            time.sleep(1)
            return None
    return None

if '__main__' == __name__:
    headers = {"Content-Type": "application/json",
               "JWT-TOKEN": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJJRCI6NywiVXNlcm5hbWUiOiJhZG1pbiIsIkRlcCI6IjEzNSIsImV4cCI6MTc3NjExOTExNCwiaXNzIjoiZ2luLWJsb2cifQ.0HTtzHZjyd2mHo8VCy8icYROxmntRMuQhyoZsAYRL_M"}
    data = [{"deviceId":"1","deviceItems":"ns=3;s=2#RO_CSDD_O","deviceName":"外供水PH","project_id":1420},{"deviceId":"1","deviceItems":"ns=3;s=2#RO_CSDD_O","deviceName":"外供水PH","project_id":1420}]
    resp = access_remote_api(url='http://120.55.44.4:8900/api/v1/jinke-cloud/device/current-data?time=1739859286292',
                      headers=headers,
                      data=data)
    print(resp)