Create gigiachat_requests.py

gigiachat_requests.py

@@ -0,0 +1,118 @@
+import os
+import base64
+import json
+import requests
+
+def get_access_token():
+    username = os.environ.get("GIGA_USERNAME")
+    password = os.environ.get("GIGA_SECRET")
+        
+    # Получаем строку с базовой авторизацией в формате Base64
+    auth_str = f'{username}:{password}'
+    auth_bytes = auth_str.encode('utf-8')
+    auth_base64 = base64.b64encode(auth_bytes).decode('utf-8')
+    url = os.getenv('GIGA_AUTH_URL')
+        
+    headers = {
+        'Authorization': f'Basic {auth_base64}',  # вставляем базовую авторизацию
+        'RqUID': os.getenv('GIGA_rquid'),
+        'Content-Type': 'application/x-www-form-urlencoded',
+        'Accept': 'application/json'
+    }
+        
+    data = {
+        'scope': os.getenv('GIGA_SCOPE')
+    }
+
+    response = requests.post(url, headers=headers, data=data, verify=False)
+    access_token = response.json()['access_token']
+    print('Got access token')
+    return access_token
+
+def get_number_of_tokens(prompt, access_token):
+    url_completion = os.getenv('GIGA_TOKENS_URL')
+        
+    data_copm = json.dumps({
+        "model": os.getenv('GIGA_MODEL'),
+        "input": [
+            prompt
+        ],
+    })
+        
+    headers_comp = {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json',
+        'Authorization': 'Bearer ' + access_token
+    }
+        
+    response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False)
+    response_data = response.json()
+
+    return response_data[0]['tokens']
+
+def get_completion_from_gigachat(prompt, max_tokens, access_token):
+    url_completion = os.getenv('GIGA_COMPLETION_URL')
+        
+    data_copm = json.dumps({
+        "model": os.getenv('GIGA_MODEL'),
+        "messages": [
+        {
+            "role": "user",
+            "content": prompt
+        }
+        ],
+        "stream": False,
+        "max_tokens": max_tokens,
+    })
+        
+    headers_comp = {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json',
+        'Authorization': 'Bearer ' + access_token
+    }
+        
+    response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False)
+    response_data = response.json()
+    answer_from_llm = response_data['choices'][0]['message']['content']
+
+    return answer_from_llm
+
+def process_transcribation_with_gigachat(prompt, transcript, max_tokens, access_token):
+    url_completion = os.getenv('GIGA_COMPLETION_URL')
+
+    data_copm = json.dumps({
+        "model": 'GigaChat-Plus',
+        "max_tokens": max_tokens,
+        "messages": [
+        {
+            "role": "user",
+            "content": prompt + transcript
+        }
+        ],
+        "stream": True,
+    })
+        
+    headers_comp = {
+        'Content-Type': 'application/json',
+        'Accept': 'application/json',
+        'Authorization': 'Bearer ' + access_token
+    }
+
+    output_text = ''
+    response = requests.post(url_completion, headers=headers_comp, data=data_copm, verify=False, stream=True)
+    for line in response.iter_lines():
+        if line:
+            decoded_line = line.decode('utf-8')
+            if decoded_line.startswith('data:'):
+                decoded_line = decoded_line[len('data:'):].strip()
+            try:
+                data = json.loads(decoded_line)
+                if "choices" in data:
+                    for choice in data["choices"]:
+                        if "delta" in choice and "content" in choice["delta"]:
+                            output_text += choice["delta"]["content"]
+            except json.JSONDecodeError:
+                continue
+
+
+    return output_text