test.py - master - Vitmgn/mytest

test.py
109 строк5 KB
Vitmgn
update test.py
02 фев 2025, 21:57
02 фев 2025, 21:57b16412e
from langchain_gigachat import GigaChat
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_community.cache import InMemoryCache
from langchain.prompts import ChatPromptTemplate
from langchain.globals import set_llm_cache
from langchain_core.runnables.history import RunnableWithMessageHistory
import logging
import os
import requests
import truststore

# Настройка SSLContext для использования системного хранилища сертификатов
truststore.inject_into_ssl()

url = "https://ngw.devices.sberbank.ru:9443/api/v2/oauth"

payload={
    'scope': 'GIGACHAT_API_PERS'
}
headers = {
    'Content-Type': 'application/x-www-form-urlencoded',
    'Accept': 'application/json',
    'RqUID': '2dc61fec-6050-4323-bb07-6e0220a1a2ed',
    'Authorization': 'Basic N2I1ZWRmZGItOTYyYi00MTY2LThjZWItZjY0ZGE3MDQyMWQ3OjNjNDMzNDAxLWJmNTQtNDg3Yi1iMDcwLTFiYmIzNDhmY2FjMg=='
}

#response = requests.request("POST", url, headers=headers, data=payload)

#print(response.text)

# Вставьте ваш API-ключ здесь
API_KEY = 'eyJjdHkiOiJqd3QiLCJlbmMiOiJBMjU2Q0JDLUhTNTEyIiwiYWxnIjoiUlNBLU9BRVAtMjU2In0.XCk7dSB3w28cC20hy6bEQhp5QeCgL-za7JsxEuigtKEvyOLIrgOAOxZmu86UxasBBKvKvOZpH8JQ72vLUdaQ_wDaCNAE_kDklm8QCOMjfs04WgqxI41jZPfBUcctoCHGR_SADnwTRcSS1ydmQdAiYP95H8rU_WVjel1S0pMiK7oU78abxtSdoLUaZS1z4jI3xd5LucZ_HuKuFRAJX8Dxil3rR7eFQgVxLdChZRv3erAu8uK1utsNT0kNSDjLu1zfvNZZ5jWrFDQH6RGe2wPW4r6m9V4nYBI4EJHtfR544GNP4kFkMIdSlBO8S6C8p8hxwLE3spRChKirbCaygWTcWQ.HB_bOmW6pl_9a9FrtiRzpg.W6gMdl0WyR-XOWCFGJb6VypC6QuBOA1rp9vvWo7cKNtPPxO-2x_qgHcfbGKMfgQluzz_ROPh7cVHwMNlJzDikbQWmLubVi5XoU2Grx3ggShDKE0OpGt4UwhN0-UfC-okyRe1GsR3fNK19ssmKlOz2wt3Zf2l4A-eF_1_FYv2GX06hMBrvsNBw63u8M0H1qY33-Ddb1_5qrAYo4JRCvWk27CCnF23JkQto3c73cSSRQ8gXCScBz3YvXG8YSOKdUpxfB4-SiBmFsW5DheImRn0Jk0ORBD3Z4VRLfW25F63OetctP_ytZxyUcFEPVrcbOkFvg3d-mfP65aUNJaY3UUD1fjkWfMmdewuUBlf7jN2YcucMEpVWUzNs8yuBDNJUHb5ggbh5oIIUFJvuOkHUudgo7beDDB3COhFlMveAB3zamw3_v6TtdS4cR3A9m9kY9GgEWAK9kQ66FSkrJ1gss8_NvzxGRgVgsjpu4zhrqNGpblApXjBCZ66OmsdelPN2cjmn1T76oFsB2lO__ukbaeGz1RVfKDwQTUUrCm6Gidj_cwfTztau1lH6n9jmV49s3Fch3PKj4v5MW6QZYTUWIvKucxCU8jjgvEKyX-tihv-ts2-2euxRjJ_pcZyLN8nsnO2YBjGRcKcOhsxjwKRGN9CkdYuUkMOR2eXAG-jE5sAs0MSIOVWfTEeZQ-GNmbwTrpEYTcHB1bgGnpXKWif6H8SGgzSiQxqs_svX68-u1MFx4w.LsXlpq0lp7n2RsQOAJhKmRhHvIUNcFp9rN6JEZfFyFM'

# Инициализация клиента GigaChat с отключенной проверкой SSL
giga = GigaChat(
    api_key=API_KEY,
    model="GigaChat",
    credentials='N2I1ZWRmZGItOTYyYi00MTY2LThjZWItZjY0ZGE3MDQyMWQ3OjNjNDMzNDAxLWJmNTQtNDg3Yi1iMDcwLTFiYmIzNDhmY2FjMg==',
    verify_ssl=False,  # Отключение проверки SSL (используйте только для тестов)
    scope="GIGACHAT_API_PERS"  # Укажите область действия токена
)

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Включение кэширования
set_llm_cache(InMemoryCache())

# Авторизация в GigaChat
#giga = GigaChat(credentials="YOUR_GIGACHAT_TOKEN")

# Системный промпт
SYSTEM_PROMPT = (
    "Ты профессиональный шеф-повар. Отвечай кратко и по делу."
)

# Шаблон промпта
prompt_template = ChatPromptTemplate.from_messages([
    ("system", SYSTEM_PROMPT),
    ("placeholder", "{chat_history}"),
    ("human", "{input}"),
    ("ai", ""),
])

# Цепочка
runnable = prompt_template | giga

# Память с использованием Redis
def get_memory(session_id):
    return RedisChatMessageHistory(session_id=session_id, url="redis://localhost:6379/0")

# RunnableWithMessageHistory
chain_with_memory = RunnableWithMessageHistory(
    runnable,
    get_memory,
    input_messages_key="input",
    history_messages_key="chat_history",
)

# Функция для нормализации входных данных
def normalize_input(user_input):
    return user_input.strip().lower()

# Пример использования
def main():
    print("Добро пожаловать в кулинарного помощника! Напишите 'выход' или 'exit', чтобы завершить.")
    session_id = "user_session_1"  # Уникальный ID для каждой сессии
    while True:
        user_input = input("Вы: ")
        if user_input.lower() in ["выход", "exit"]:
            print("Шеф-повар: До свидания! Приятного приготовления!")
            break
        normalized_input = normalize_input(user_input)
        logger.info(f"Получен запрос: {normalized_input}")
        try:
            response = chain_with_memory.invoke(
                {"input": normalized_input},
                config={"configurable": {"session_id": session_id}}
            )
            response_text = response.content.strip()
            logger.info(f"Ответ модели: {response_text}")
            print(f"Шеф-повар: {response_text}")
        except Exception as e:
            logger.error(f"Ошибка при обработке запроса: {e}")
            print("Шеф-повар: Произошла ошибка. Попробуйте еще раз.")

if __name__ == "__main__":
    main()