133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
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() |