GigaChat api: client.py

solutions/68ce8e2b06888ba023bf2212_GigaChat_api/client.py

@@ -0,0 +1,133 @@
+import json
+import time
+
+import requests
+
+# Конфигурация подключения к GigaChat API
+API_URL = "https://api.sberbank.ru/v1/gigachat/chat"
+API_KEY = "YOUR_GIGACHAT_API_KEY"  # Замените на ваш реальный ключ
+
+HEADERS = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+}
+
+# Пример локальной функции, которую может вызвать модель
+def calculate_sum(a: int, b: int) -> int:
+    """
+    Возвращает сумму двух целых чисел.
+    """
+    return a + b
+
+
+# Словарь доступных функций для вызова модели
+FUNCTIONS = {
+    "calculate_sum": calculate_sum,
+}
+
+
+def call_function(function_name: str, arguments: dict):
+    """
+    Вызов локальной функции по имени и аргументам.
+    Возвращает результат выполнения или сообщение об ошибке.
+    """
+    func = FUNCTIONS.get(function_name)
+    if not func:
+        return {"error": f"Функция {function_name} не найдена."}
+
+    try:
+        result = func(**arguments)
+        return {"result": result}
+    except Exception as exc:  # noqa: BLE001
+        return {"error": str(exc)}
+
+
+def send_message(message_text: str, conversation_id: str | None = None):
+    """
+    Отправка сообщения модели и обработка возможного вызова функции.
+    Возвращает финальный ответ от модели после выполнения всех функций.
+    """
+    payload = {
+        "messages": [
+            {"role": "user", "content": message_text},
+        ],
+    }
+    if conversation_id:
+        payload["conversationId"] = conversation_id
+
+    # Отправляем запрос к GigaChat
+    response = requests.post(API_URL, headers=HEADERS, json=payload)
+    response.raise_for_status()
+    data = response.json()
+
+    # Проверяем наличие вызова функции в ответе модели
+    function_call = None
+    for msg in data.get("messages", []):
+        if msg.get("role") == "assistant" and msg.get("functionCall"):
+            function_call = msg["functionCall"]
+            break
+
+    # Если функция не требуется, возвращаем обычный текстовый ответ
+    if not function_call:
+        assistant_msg = next(
+            (m for m in data["messages"] if m["role"] == "assistant"), None
+        )
+        return assistant_msg.get("content") if assistant_msg else ""
+
+    # Выполняем требуемую функцию
+    func_name = function_call["name"]
+    arguments = json.loads(function_call.get("arguments", "{}"))
+    function_result = call_function(func_name, arguments)
+
+    # Отправляем результат функции обратно модели для получения финального ответа
+    follow_up_payload = {
+        "messages": [
+            {"role": "assistant", "content": ""},  # пустое сообщение от ассистента
+            {
+                "role": "function",
+                "name": func_name,
+                "content": json.dumps(function_result),
+            },
+        ],
+        "conversationId": data.get("conversationId"),
+    }
+
+    follow_up_response = requests.post(API_URL, headers=HEADERS, json=follow_up_payload)
+    follow_up_response.raise_for_status()
+    follow_up_data = follow_up_response.json()
+
+    # Финальный ответ модели
+    final_msg = next(
+        (m for m in follow_up_data["messages"] if m["role"] == "assistant"), None
+    )
+    return final_msg.get("content") if final_msg else ""
+
+
+def main():
+    """
+    Основной цикл взаимодействия с пользователем.
+    """
+    conversation_id = None
+    print("Привет! Я подключён к GigaChat. Задавай вопросы (Ctrl+C для выхода).")
+    while True:
+        try:
+            user_input = input("\n> ")
+        except KeyboardInterrupt:
+            print("\nВыход.")
+            break
+
+        if not user_input.strip():
+            continue
+
+        # Отправляем сообщение и получаем ответ
+        answer = send_message(user_input, conversation_id)
+        print(f"\nОтвет модели: {answer}")
+
+        # Сохраняем идентификатор разговора для последующих сообщений
+        # (если API возвращает его в ответе)
+        if "conversationId" in locals():
+            conversation_id = locals()["conversationId"]
+
+
+if __name__ == "__main__":
+    main()