Spaces:
Sleeping
Sleeping
File size: 2,091 Bytes
623a39b 014ba5e 623a39b 766dff5 cce3c9c 623a39b 6f8a3e7 cce3c9c 623a39b 766dff5 623a39b 0cea943 cce3c9c 623a39b 6f8a3e7 0cea943 623a39b e357d2f 623a39b 766dff5 623a39b e357d2f 623a39b 014ba5e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 |
import requests
import json
import os
from datetime import datetime, timedelta
import boto3
def gen_auth_token(auth_file):
# url = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"
url = "https://api.mlrnd.ru/api/v2/oauth"
# payload='scope=GIGACHAT_API_CORP'
payload='scope=API_v1'
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Accept': 'application/json',
'RqUID': '1b519047-0ee9-4b63-8599-e5ffc9c77e72',
'Authorization': os.getenv('GIGACHAT_API_TOKEN')
}
response = requests.request("POST", url, headers=headers, data=payload, verify=False)
with open(auth_file, 'w') as f:
json.dump(json.loads(response.text), f, ensure_ascii=False)
def get_text(content, auth_token=None):
# url = "https://gigachat.devices.sberbank.ru/api/v1/chat/completions"
url = "https://api.mlrnd.ru/api/v1/chat/completions"
payload = json.dumps({
"model": "Test_model",
"messages": content,
"temperature": 1,
"top_p": 0.1,
"n": 1,
"stream": False,
"max_tokens": 512,
"repetition_penalty": 1
})
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'Authorization': f'Bearer {auth_token}'
}
response = requests.request("POST", url, headers=headers, data=payload, verify=False)
return json.loads(response.text)
def generate(content=None, auth_file=None):
if auth_file is None or not os.path.isfile(auth_file):
gen_auth_token(auth_file)
with open(auth_file) as f:
auth_token = json.load(f)
if datetime.fromtimestamp(auth_token['expires_at']/1000) <= datetime.now() - timedelta(seconds=60):
gen_auth_token()
with open(auth_file) as f:
auth_token = json.load(f)
content = [{'role': x[0], 'content': x[1]} for x in content]
resp = get_text(content, auth_token['access_token'])
return resp["choices"][0]["message"]["content"]
def send_to_s3(data, name, session):
session.put_object(Bucket=os.getenv('S3_BUCKET'), Key=name, Body=json.dumps(data))
|