app/controller.py - master - sc_p/EnergoAI

app/controller.py
110 строк3 KB
Павел Щёголев
Загрузить файлы в «»
12 мар 2026, 04:54
12 мар 2026, 04:54628f0d6
from langchain_core.runnables import Runnable
from typing import Any

class ChatController(Runnable):
    def __init__(
            self,
            agent,
            sandbox,
            output,
            llm,
            gigachat_config,
            llm_factory,
            agent_factory
    ):
        self.agent = agent
        self.sandbox = sandbox
        self.output = output
        self.llm = llm

        self.gigachat_config = gigachat_config
        self.llm_factory = llm_factory
        self.agent_factory = agent_factory

        self.chat_history = []
        self.tool_used = False

    def handle_text(self, user_text: str):
        if not self.agent:
            self.output.show_bot_message("Агент не инициализирован")
            return
        try:
            self.tool_used = False

            try:
                result = self.agent.invoke({"input": user_text})
            except Exception as e:
                self.output.show_bot_message(f"Ошибка агента: {e}")
                return

            if self.looks_like_action(user_text) and not self.tool_used:
                if "что ты умеешь" in user_text.lower():
                    pass
                else:
                    self.output.show_bot_message("В программе нет такой функции.")
                    return

        except Exception as e:
            self.output.show_bot_message(f"Ошибка агента: {e}")
            return

        if isinstance(result, dict):
            text = result.get("output") or str(result)
        else:
            text = str(result)

        self.output.show_bot_message(text)

    def looks_like_action(self, text: str) -> bool:
        verbs = [
            "сохрани", "сохранить", "экспорт",
            "открой", "закрой",
            "создай", "удали",
            "покажи", "очисти"
        ]
        t = text.lower()
        return any(v in t for v in verbs)

    def set_agent(self, agent):
        self.agent = agent

    def invoke(self, input: Any, config=None) -> Any:
        if isinstance(input, dict):
            text = input.get("input") or input.get("text")
        else:
            text = str(input)

        if not self.agent:
            raise RuntimeError("Агент не инициализирован")

        text = input.get("input") if isinstance(input, dict) else str(input)
        return self.agent.invoke({"input": text})

    def update_llm_setting(self, key: str, value):
        if not self.gigachat_config:
            return "Конфигурация GigaChat не загружена"

        if not hasattr(self.gigachat_config, key):
            return f"Неизвестный параметр: {key}"

        setattr(self.gigachat_config, key, value)

        self.llm = self.llm_factory(self.gigachat_config)

        # пересобираем агента
        self.agent = self.agent_factory(self.llm)

        return f"Параметр {key} установлен в {value}"

    def get_llm_settings(self) -> str:
        cfg = self.gigachat_config

        lines = [
            "Текущие настройки LLM:",
            f"model: {cfg.model}",
            f"temperature: {cfg.temperature}",
            f"top_p: {cfg.top_p}",
            f"max_tokens: {cfg.max_tokens}",
        ]

        return "\n".join(lines)