just test 3

src/get_answer_gigachat.py

@@ -1,151 +1,147 @@
-# from langchain_core.messages import HumanMessage, SystemMessage
-# from langchain_core.runnables.config import RunnableConfig
-# from langchain_gigachat.chat_models import GigaChat
-# import os
-# import re
-# from datetime import datetime
-# from pydantic import Field
-# from dotenv import load_dotenv
-# from typing import Any, Optional
-# import logging
-# from pathlib import Path
-# load_dotenv()
-
-# class AnswerGigaChat(GigaChat):
-#     THRESHOLD_INPUT_SYMBOLS: int = 40000
-#     THRESHOLD_COST: float = 5000.0
-#     ERROR_MESSAGE_EXCEEDED_COST: str = "\n\n\n\n\nALERT!!!!!!\nCOST NORM WAS EXCEEDED!!!!!!!!\n{} >= " + str(THRESHOLD_COST) + "\n\n\n\n"
-#     COST_PER_INPUT_TOKEN: float = 0.0
-#     COST_PER_OUTPUT_TOKEN: float = 1.95e-3
-#     LOG_FILE_PATH: str = os.path.join(os.path.dirname(__file__), "gigachat.log")
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.runnables.config import RunnableConfig
+from langchain_gigachat.chat_models import GigaChat
+import os
+import re
+from datetime import datetime
+from pydantic import Field
+from dotenv import load_dotenv
+from typing import Any, Optional
+import logging
+from pathlib import Path
+load_dotenv()
+
+class AnswerGigaChat(GigaChat):
+    THRESHOLD_INPUT_SYMBOLS: int = 40000
+    THRESHOLD_COST: float = 5000.0
+    ERROR_MESSAGE_EXCEEDED_COST: str = "\n\n\n\n\nALERT!!!!!!\nCOST NORM WAS EXCEEDED!!!!!!!!\n{} >= " + str(THRESHOLD_COST) + "\n\n\n\n"
+    COST_PER_INPUT_TOKEN: float = 0.0
+    COST_PER_OUTPUT_TOKEN: float = 1.95e-3
+    LOG_FILE_PATH: str = os.path.join(os.path.dirname(__file__), "gigachat.log")
     
-#     logger: Any = Field(default=None)
-
-#     def __init__(self):
-#         super().__init__(credentials=os.getenv("GIGACHAT_CREDENTIALS"),
-#                         verify_ssl_certs=False,
-#                         model="GigaChat-Max",
-#                         scope="GIGACHAT_API_CORP")
-#         self.logger = self._setup_logger(self.LOG_FILE_PATH)
-
-#     def _setup_logger(self, log_file: str) -> logging.Logger:
-#         logger = logging.getLogger(__name__)
-#         logger.setLevel(logging.INFO)
-
-#         if not logger.handlers:
-#             log_file = Path(log_file)
-#             log_file.parent.mkdir(parents=True, exist_ok=True)  # Создать папку для логов
-
-#             formatter = logging.Formatter(
-#                 '%(asctime)s - %(name)s - %(levelname)s - '
-#                 'input_tokens=%(input_tokens)d - output_tokens=%(output_tokens)d - '
-#                 'cost=%(cost).5f - execution_time=%(execution_time)s - '
-#                 'status=%(status)s'
-#             )
-
-#             file_handler = logging.FileHandler(log_file)
-#             file_handler.setFormatter(formatter)
-#             logger.addHandler(file_handler)
-
-#         return logger
-
-#     def invoke(self, input: Any, config: Optional[RunnableConfig] = None, **kwargs: Any) -> Any:
-#         # Извлекаем сообщения из входных данных
-#         if isinstance(input, list):
-#             # Если input - это список сообщений (например, [SystemMessage, HumanMessage])
-#             messages = input
-#         elif isinstance(input, dict) and "messages" in input:
-#             # Если input - это dict с ключом "messages" (формат LangGraph)
-#             messages = [
-#                 SystemMessage(content=msg["content"]) if msg["role"] == "system"
-#                 else HumanMessage(content=msg["content"])
-#                 for msg in input["messages"]
-#                 if msg["role"] in ["system", "user"]
-#             ]
-#         elif hasattr(input, "messages"):
-#             # Если input - это объект с атрибутом messages
-#             messages = input.messages
-#         else:
-#             # Попробуем интерпретировать input как одно сообщение
-#             messages = [HumanMessage(content=str(input))]
-
-#         # Проверяем длину ввода
-#         system_content = ""
-#         user_content = ""
-#         for msg in messages:
-#             if isinstance(msg, SystemMessage):
-#                 system_content += msg.content
-#             elif isinstance(msg, HumanMessage):
-#                 user_content += msg.content
-
-#         if not self._check_input_length(system_content, user_content):
-#             raise ValueError("Too long query")
-
-#         # Вызываем родительский метод invoke
-#         response = super().invoke(messages, config=config, **kwargs)
-
-#         # Логируем информацию о запросе
-#         num_input_tokens = response.usage_metadata["input_tokens"]
-#         num_output_tokens = response.usage_metadata["output_tokens"]
-#         cost = self._calculate_response_cost(num_input_tokens, num_output_tokens)
-#         self.logger.info(
-#             "got answer",
-#             extra={
-#                 "input_tokens": num_input_tokens,
-#                 "output_tokens": num_output_tokens,
-#                 "cost": cost,
-#                 "execution_time": str(datetime.now()),
-#                 "status": "success"
-#             }
-#         )
-
-#         # Проверяем общую стоимость
-#         total_cost = self._calculate_total_cost()
-#         if total_cost >= self.THRESHOLD_COST:
-#             error_message = self.ERROR_MESSAGE_EXCEEDED_COST.format(total_cost)
-#             print(error_message)
-#             response.content = error_message + response.content
-
-#         # Возвращаем ответ в соответствующем формате
-#         if isinstance(input, dict):
-#             return {**input, "messages": [{"role": "assistant", "content": response.content}]}
-#         return response
+    logger: Any = Field(default=None)
+
+    def __init__(self):
+        super().__init__(credentials=os.getenv("GIGACHAT_CREDENTIALS"),
+                        verify_ssl_certs=False,
+                        model="GigaChat-Max",
+                        scope="GIGACHAT_API_CORP")
+        self.logger = self._setup_logger(self.LOG_FILE_PATH)
+
+    def _setup_logger(self, log_file: str) -> logging.Logger:
+        logger = logging.getLogger(__name__)
+        logger.setLevel(logging.INFO)
+
+        if not logger.handlers:
+            log_file = Path(log_file)
+            log_file.parent.mkdir(parents=True, exist_ok=True)  # Создать папку для логов
+
+            formatter = logging.Formatter(
+                '%(asctime)s - %(name)s - %(levelname)s - '
+                'input_tokens=%(input_tokens)d - output_tokens=%(output_tokens)d - '
+                'cost=%(cost).5f - execution_time=%(execution_time)s - '
+                'status=%(status)s'
+            )
+
+            file_handler = logging.FileHandler(log_file)
+            file_handler.setFormatter(formatter)
+            logger.addHandler(file_handler)
+
+        return logger
+
+    def invoke(self, input: Any, config: Optional[RunnableConfig] = None, **kwargs: Any) -> Any:
+        # Извлекаем сообщения из входных данных
+        if isinstance(input, list):
+            # Если input - это список сообщений (например, [SystemMessage, HumanMessage])
+            messages = input
+        elif isinstance(input, dict) and "messages" in input:
+            # Если input - это dict с ключом "messages" (формат LangGraph)
+            messages = [
+                SystemMessage(content=msg["content"]) if msg["role"] == "system"
+                else HumanMessage(content=msg["content"])
+                for msg in input["messages"]
+                if msg["role"] in ["system", "user"]
+            ]
+        elif hasattr(input, "messages"):
+            # Если input - это объект с атрибутом messages
+            messages = input.messages
+        else:
+            # Попробуем интерпретировать input как одно сообщение
+            messages = [HumanMessage(content=str(input))]
+
+        # Проверяем длину ввода
+        system_content = ""
+        user_content = ""
+        for msg in messages:
+            if isinstance(msg, SystemMessage):
+                system_content += msg.content
+            elif isinstance(msg, HumanMessage):
+                user_content += msg.content
+
+        if not self._check_input_length(system_content, user_content):
+            raise ValueError("Too long query")
+
+        # Вызываем родительский метод invoke
+        response = super().invoke(messages, config=config, **kwargs)
+
+        # Логируем информацию о запросе
+        num_input_tokens = response.usage_metadata["input_tokens"]
+        num_output_tokens = response.usage_metadata["output_tokens"]
+        cost = self._calculate_response_cost(num_input_tokens, num_output_tokens)
+        self.logger.info(
+            "got answer",
+            extra={
+                "input_tokens": num_input_tokens,
+                "output_tokens": num_output_tokens,
+                "cost": cost,
+                "execution_time": str(datetime.now()),
+                "status": "success"
+            }
+        )
+
+        # Проверяем общую стоимость
+        total_cost = self._calculate_total_cost()
+        if total_cost >= self.THRESHOLD_COST:
+            error_message = self.ERROR_MESSAGE_EXCEEDED_COST.format(total_cost)
+            print(error_message)
+            response.content = error_message + response.content
+
+        # Возвращаем ответ в соответствующем формате
+        if isinstance(input, dict):
+            return {**input, "messages": [{"role": "assistant", "content": response.content}]}
+        return response
     
-#     def _check_input_length(self, system_message: str, user_message: str) -> bool:
-#         return len(system_message) + len(user_message) < self.THRESHOLD_INPUT_SYMBOLS
+    def _check_input_length(self, system_message: str, user_message: str) -> bool:
+        return len(system_message) + len(user_message) < self.THRESHOLD_INPUT_SYMBOLS
     
-#     def _calculate_response_cost(self, num_input_tokens: int, num_output_tokens: int) -> float:
-#         return num_input_tokens * self.COST_PER_INPUT_TOKEN + \
-#         num_output_tokens * self.COST_PER_OUTPUT_TOKEN
+    def _calculate_response_cost(self, num_input_tokens: int, num_output_tokens: int) -> float:
+        return num_input_tokens * self.COST_PER_INPUT_TOKEN + \
+        num_output_tokens * self.COST_PER_OUTPUT_TOKEN
         
-#     def _calculate_total_cost(self, 
-#                               start_date: str = '2025-06-01', 
-#                               end_date: str = str(datetime.now().date())):
-#         total_cost = 0.0
-#         start_date = datetime.strptime(start_date, '%Y-%m-%d').date()
-#         end_date = datetime.strptime(end_date, '%Y-%m-%d').date()
+    def _calculate_total_cost(self, 
+                              start_date: str = '2025-06-01', 
+                              end_date: str = str(datetime.now().date())):
+        total_cost = 0.0
+        start_date = datetime.strptime(start_date, '%Y-%m-%d').date()
+        end_date = datetime.strptime(end_date, '%Y-%m-%d').date()
         
-#         # Регулярное выражение для извлечения даты и cost из строки лога
-#         log_pattern = re.compile(
-#             r'^(?P<date>\d{4}-\d{2}-\d{2}) \d{2}:\d{2}:\d{2},\d{3} - .*? - .*? - '
-#             r'input_tokens=\d+ - output_tokens=\d+ - '
-#             r'cost=(?P<cost>\d+\.\d{2}) - '
-#             r'execution_time=.*? - status=.*$'
-#         )
+        # Регулярное выражение для извлечения даты и cost из строки лога
+        log_pattern = re.compile(
+            r'^(?P<date>\d{4}-\d{2}-\d{2}) \d{2}:\d{2}:\d{2},\d{3} - .*? - .*? - '
+            r'input_tokens=\d+ - output_tokens=\d+ - '
+            r'cost=(?P<cost>\d+\.\d{2}) - '
+            r'execution_time=.*? - status=.*$'
+        )
         
-#         with open(self.LOG_FILE_PATH, 'r', encoding='utf-8') as file:
-#             for line in file:
-#                 match = log_pattern.match(line)
-#                 if match:
-#                     log_date_str = match.group('date')
-#                     log_date = datetime.strptime(log_date_str, '%Y-%m-%d').date()
-#                     cost = float(match.group('cost'))
+        with open(self.LOG_FILE_PATH, 'r', encoding='utf-8') as file:
+            for line in file:
+                match = log_pattern.match(line)
+                if match:
+                    log_date_str = match.group('date')
+                    log_date = datetime.strptime(log_date_str, '%Y-%m-%d').date()
+                    cost = float(match.group('cost'))
                     
-#                     if start_date <= log_date <= end_date:
-#                         total_cost += cost
+                    if start_date <= log_date <= end_date:
+                        total_cost += cost
         
-#         return total_cost
-
-
-import streamlit as st
-st.write("Hello, World!")  
+        return total_cost

src/streamlit_app.py

@@ -1,105 +1,109 @@
-import streamlit as st
-from langchain_core.messages import HumanMessage, AIMessage
-from get_classification import get_graph_class
-from datetime import datetime
-
-
-def message_to_dict(messages):
-    result = []
-    for message in messages:
-        if isinstance(message, HumanMessage) or isinstance(message, AIMessage):
-            print("message:", message.content)
-        if message.content == "" or message.content is None:
-            continue
-        if isinstance(message, HumanMessage):
-            result.append({"role": "user", "content": message.content})
-        elif isinstance(message, AIMessage):
-            result.append({"role": "assistant", "content": message.content})
-    print("-" * 100)
-    return result
-
-
-def find_last_bot_message(messages):
-    """Находит последнее сообщение бота"""
-    for message in messages[::-1]:
-        if isinstance(message, AIMessage) and len(message.content) > 0:
-            return message.content
-    return None
-
-
-def display_chat_messages():
-    """Отображает историю сообщений в чате"""
-    for message in st.session_state.messages:
-        with st.chat_message(message["role"]):
-            st.markdown(message["content"])
-
-
-def save_broken_case():
-    messages_dict = st.session_state.messages
-    result_str = ""
-    for elem in messages_dict:
-        role = elem["role"]
-        content = elem["content"]
-        result_str += f"{role}: {content}\n"
+# import streamlit as st
+# from langchain_core.messages import HumanMessage, AIMessage
+# from get_classification import get_graph_class
+# from datetime import datetime
+
+
+# def message_to_dict(messages):
+#     result = []
+#     for message in messages:
+#         if isinstance(message, HumanMessage) or isinstance(message, AIMessage):
+#             print("message:", message.content)
+#         if message.content == "" or message.content is None:
+#             continue
+#         if isinstance(message, HumanMessage):
+#             result.append({"role": "user", "content": message.content})
+#         elif isinstance(message, AIMessage):
+#             result.append({"role": "assistant", "content": message.content})
+#     print("-" * 100)
+#     return result
+
+
+# def find_last_bot_message(messages):
+#     """Находит последнее сообщение бота"""
+#     for message in messages[::-1]:
+#         if isinstance(message, AIMessage) and len(message.content) > 0:
+#             return message.content
+#     return None
+
+
+# def display_chat_messages():
+#     """Отображает историю сообщений в чате"""
+#     for message in st.session_state.messages:
+#         with st.chat_message(message["role"]):
+#             st.markdown(message["content"])
+
+
+# def save_broken_case():
+#     messages_dict = st.session_state.messages
+#     result_str = ""
+#     for elem in messages_dict:
+#         role = elem["role"]
+#         content = elem["content"]
+#         result_str += f"{role}: {content}\n"
     
-    current_datetime = datetime.now()
-    formatted_datetime = current_datetime.strftime("%Y-%m-%d %H:%M:%S")
-
-    # with open("/Users/admin/my_documents/retrieval_part/services/broken_cases.txt", "a") as file:
-    #     file.write(f"{formatted_datetime}\n" + result_str + "\n" + "-" * 50 + "\n\n")
-
-
-def handle_user_input():
-    """Обрабатывает ввод пользователя и генерирует ответ бота"""
-    if prompt := st.chat_input("Введите ваш вопрос"):
-        st.session_state.messages.append({"role": "user", "content": prompt})
-        if prompt.lower().startswith("log"):
-            save_broken_case()
-            st.session_state.messages = []
-            display_chat_messages()
-        else:
-            with st.chat_message("user"):
-                st.markdown(prompt)
+#     current_datetime = datetime.now()
+#     formatted_datetime = current_datetime.strftime("%Y-%m-%d %H:%M:%S")
+
+#     # with open("/Users/admin/my_documents/retrieval_part/services/broken_cases.txt", "a") as file:
+#     #     file.write(f"{formatted_datetime}\n" + result_str + "\n" + "-" * 50 + "\n\n")
+
+
+# def handle_user_input():
+#     """Обрабатывает ввод пользователя и генерирует ответ бота"""
+#     if prompt := st.chat_input("Введите ваш вопрос"):
+#         st.session_state.messages.append({"role": "user", "content": prompt})
+#         if prompt.lower().startswith("log"):
+#             save_broken_case()
+#             st.session_state.messages = []
+#             display_chat_messages()
+#         else:
+#             with st.chat_message("user"):
+#                 st.markdown(prompt)
             
-            try:
-                if "bot" not in st.session_state:
-                    st.session_state.bot = get_graph_class(prompt)
-                st.session_state.bot.invoke(prompt)
-            except Exception as e:
-                st.error(f"Ошибка: {str(e)}")
+#             try:
+#                 if "bot" not in st.session_state:
+#                     st.session_state.bot = get_graph_class(prompt)
+#                 st.session_state.bot.invoke(prompt)
+#             except Exception as e:
+#                 st.error(f"Ошибка: {str(e)}")
             
-            # Извлекаем последнее сообщение бота
-            last_bot_message = find_last_bot_message(st.session_state.bot.messages)
-            st.session_state.messages.append(
-                {"role": "assistant", "content": last_bot_message}
-            )
-            with st.chat_message("assistant"):
-                st.markdown(last_bot_message)
-    raise ValueError
+#             # Извлекаем последнее сообщение бота
+#             last_bot_message = find_last_bot_message(st.session_state.bot.messages)
+#             st.session_state.messages.append(
+#                 {"role": "assistant", "content": last_bot_message}
+#             )
+#             with st.chat_message("assistant"):
+#                 st.markdown(last_bot_message)
+#     raise ValueError
+
 
+# def clear_chat():
+#     """Очищает чат и пересоздает бота"""
+#     st.session_state.messages = []
+#     del st.session_state.bot
 
-def clear_chat():
-    """Очищает чат и пересоздает бота"""
-    st.session_state.messages = []
-    del st.session_state.bot
 
+# def main():
+#     """Основная функция приложения"""
+#     # Заголовок приложения
+#     st.title("Чат-бот технической поддержки OpenVPN")
 
-def main():
-    """Основная функция приложения"""
-    # Заголовок приложения
-    st.title("Чат-бот технической поддержки OpenVPN")
+#     # Кнопка очистки чата
+#     if st.button("Clear"):
+#         clear_chat()
 
-    # Кнопка очистки чата
-    if st.button("Clear"):
-        clear_chat()
+#     if "messages" not in st.session_state:
+#         st.session_state.messages = []
 
-    if "messages" not in st.session_state:
-        st.session_state.messages = []
+#     # Отображение чата и обработка ввода
+#     display_chat_messages()
+#     handle_user_input()
 
-    # Отображение чата и обработка ввода
-    display_chat_messages()
-    handle_user_input()
 
+# if __name__ == "__main__":
+#     main()
 
-if __name__ == "__main__":
-    main()
+
+import streamlit as st
+st.write("Hello, World!")