From fcd3e79cb7f672809619d9fc96118c0d5b01fd87 Mon Sep 17 00:00:00 2001 From: SMKRV Date: Sun, 24 Nov 2024 18:36:37 +0300 Subject: [PATCH] Release v2.0.0 --- custom_components/ha_text_ai/coordinator.py | 362 ++++++++++---------- custom_components/ha_text_ai/sensor.py | 10 +- 2 files changed, 190 insertions(+), 182 deletions(-) diff --git a/custom_components/ha_text_ai/coordinator.py b/custom_components/ha_text_ai/coordinator.py index 9773ce6..afb65a9 100644 --- a/custom_components/ha_text_ai/coordinator.py +++ b/custom_components/ha_text_ai/coordinator.py @@ -1,7 +1,6 @@ -"""Data coordinator for HA text AI.""" +"""DataUpdateCoordinator for HA Text AI.""" +from datetime import timedelta import logging -import time -from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from homeassistant.core import HomeAssistant @@ -9,10 +8,19 @@ from homeassistant.exceptions import HomeAssistantError from homeassistant.helpers.update_coordinator import DataUpdateCoordinator from homeassistant.util import dt as dt_util +from .const import ( + DOMAIN, + STATE_READY, + STATE_PROCESSING, + STATE_ERROR, + STATE_RATE_LIMITED, + STATE_MAINTENANCE, +) + _LOGGER = logging.getLogger(__name__) class HATextAICoordinator(DataUpdateCoordinator): - """Class to manage fetching data from API.""" + """Class to manage fetching data from the API.""" def __init__( self, @@ -28,6 +36,7 @@ class HATextAICoordinator(DataUpdateCoordinator): ) -> None: """Initialize coordinator.""" update_interval_td = timedelta(seconds=update_interval) + self.instance_name = instance_name super().__init__( hass, @@ -36,37 +45,38 @@ class HATextAICoordinator(DataUpdateCoordinator): update_interval=update_interval_td, ) - self.instance_name = instance_name + self.hass = hass self.client = client self.model = model - self.max_tokens = max_tokens self.temperature = temperature - self._max_history_size = max_history_size - self._is_anthropic = is_anthropic + self.max_tokens = max_tokens + self.max_history_size = max_history_size + self.is_anthropic = is_anthropic + + # Регистрируем instance в системе + self.hass.data.setdefault(DOMAIN, {}) + self.hass.data[DOMAIN][instance_name] = self + self._system_prompt = None + self._conversation_history = [] + self._performance_metrics = { + "total_tokens": 0, + "prompt_tokens": 0, + "completion_tokens": 0, + "successful_requests": 0, + "failed_requests": 0, + "total_errors": 0, + "average_latency": 0, + "max_latency": 0, + "min_latency": float('inf'), + } - # Статусы self._is_processing = False self._is_rate_limited = False self._is_maintenance = False + self.endpoint_status = "ready" - # История и метрики - self._history: List[Dict[str, Any]] = [] - self._request_count = 0 - self._error_count = 0 - self._last_error = None - self._performance_metrics = { - "total_requests": 0, - "total_errors": 0, - "avg_response_time": 0.0, - "last_request_time": None, - } - - # Версия API и последний ответ - self.api_version = "v1" - self.last_response = None - self.endpoint_status = "initialized" - + # Инициализация last_response self.last_response = { "timestamp": dt_util.utcnow().isoformat(), "question": "", @@ -74,24 +84,39 @@ class HATextAICoordinator(DataUpdateCoordinator): "model": model, "instance": instance_name, "error": None - } + } - @property - def is_processing(self) -> bool: - """Return True if currently processing a request.""" - return self._is_processing + self._start_time = dt_util.utcnow() + _LOGGER.info(f"Initialized HA Text AI coordinator with instance: {instance_name}") - @property - def is_rate_limited(self) -> bool: - """Return True if currently rate limited.""" - return self._is_rate_limited + async def _async_update_data(self) -> Dict[str, Any]: + """Update data via library.""" + return { + "state": self._get_current_state(), + "metrics": self._performance_metrics, + "last_response": self.last_response, + "is_processing": self._is_processing, + "is_rate_limited": self._is_rate_limited, + "is_maintenance": self._is_maintenance, + "endpoint_status": self.endpoint_status, + "uptime": (dt_util.utcnow() - self._start_time).total_seconds(), + "system_prompt": self._system_prompt, + "history_size": len(self._conversation_history), + } - @property - def is_maintenance(self) -> bool: - """Return True if in maintenance mode.""" - return self._is_maintenance + def _get_current_state(self) -> str: + """Get current state based on internal flags.""" + if self._is_processing: + return STATE_PROCESSING + elif self._is_rate_limited: + return STATE_RATE_LIMITED + elif self._is_maintenance: + return STATE_MAINTENANCE + elif self.last_response.get("error"): + return STATE_ERROR + return STATE_READY - async def async_ask_question( + async def async_process_question( self, question: str, model: Optional[str] = None, @@ -100,12 +125,22 @@ class HATextAICoordinator(DataUpdateCoordinator): system_prompt: Optional[str] = None, ) -> dict: """Process a question with optional parameters.""" + if not question: + raise ValueError("Question cannot be empty") + + _LOGGER.debug(f"Processing question for instance {self.instance_name}") + try: + self._is_processing = True + await self.async_update_ha_state() + temp_model = model or self.model temp_temperature = temperature or self.temperature temp_max_tokens = max_tokens or self.max_tokens temp_system_prompt = system_prompt or self._system_prompt + start_time = dt_util.utcnow() + messages = [] if temp_system_prompt: messages.append({"role": "system", "content": temp_system_prompt}) @@ -118,162 +153,127 @@ class HATextAICoordinator(DataUpdateCoordinator): "messages": messages, } - return await self.async_process_message(question, **kwargs) + response = await self.async_process_message(question, **kwargs) + + # Обновляем метрики + end_time = dt_util.utcnow() + latency = (end_time - start_time).total_seconds() + self._update_metrics(latency, response) + + # Обновляем историю + self._update_history(question, response) + + return response except Exception as err: - _LOGGER.error("Error processing question: %s", err) + self._handle_error(err) raise HomeAssistantError(f"Failed to process question: {err}") - async def _async_update_data(self) -> Dict[str, Any]: - """Update data from API.""" + finally: + self._is_processing = False + await self.async_update_ha_state() + + async def async_process_message(self, question: str, **kwargs) -> dict: + """Process message using the AI client.""" try: - await self.async_refresh_metrics() - return { - "status": self.endpoint_status, - "metrics": self._performance_metrics, - "last_update": dt_util.utcnow().isoformat(), - "last_response": self.last_response, - "is_processing": self._is_processing, - "is_rate_limited": self._is_rate_limited, - "is_maintenance": self._is_maintenance, - "instance": self.instance_name, - } - except Exception as e: - self._last_error = str(e) - self._error_count += 1 - _LOGGER.error("Error updating data: %s", str(e)) - raise HomeAssistantError(f"Error updating data: {str(e)}") - - async def async_refresh_metrics(self) -> None: - """Refresh performance metrics.""" - self._performance_metrics.update({ - "total_requests": self._request_count, - "total_errors": self._error_count, - "last_request_time": dt_util.utcnow().isoformat(), - "instance": self.instance_name, - }) - - async def async_process_message(self, message: str, **kwargs) -> dict: - """Process message and update last_response.""" - try: - self._is_processing = True - self.async_update_listeners() - - start_time = time.time() - - messages = kwargs.pop("messages", [{"role": "user", "content": message}]) - model = kwargs.pop("model", self.model) - temperature = kwargs.pop("temperature", self.temperature) - max_tokens = kwargs.pop("max_tokens", self.max_tokens) - - if self._is_anthropic: - response = await self.client.messages.create( - model=model, - max_tokens=max_tokens, - temperature=temperature, - messages=messages, - **kwargs - ) - response_text = response.content[0].text + if self.is_anthropic: + response = await self._process_anthropic_message(question, **kwargs) else: - response = await self.client.chat.completions.create( - model=model, - temperature=temperature, - max_tokens=max_tokens, - messages=messages, - **kwargs - ) - response_text = response.choices[0].message.content - - elapsed_time = time.time() - start_time - self._request_count += 1 - - if self._performance_metrics["avg_response_time"] == 0: - self._performance_metrics["avg_response_time"] = elapsed_time - else: - self._performance_metrics["avg_response_time"] = ( - (self._performance_metrics["avg_response_time"] * (self._request_count - 1) + elapsed_time) - / self._request_count - ) + response = await self._process_openai_message(question, **kwargs) self.last_response = { "timestamp": dt_util.utcnow().isoformat(), - "question": message, - "response": response_text, - "model": model, + "question": question, + "response": response["content"], + "model": kwargs.get("model", self.model), "instance": self.instance_name, "error": None } - self._history.append(self.last_response) - if len(self._history) > self._max_history_size: - self._history.pop(0) - - self.endpoint_status = "ready" - self._is_processing = False - self.async_update_listeners() - - return self.last_response + return response except Exception as err: - self._error_count += 1 - self._last_error = str(err) - self._performance_metrics["total_errors"] += 1 - - self.last_response = { - "timestamp": dt_util.utcnow().isoformat(), - "question": message, - "response": None, - "instance": self.instance_name, - "error": str(err) - } - - self.endpoint_status = "error" - self._is_processing = False - self.async_update_listeners() + self._handle_error(err) raise - async def async_set_system_prompt(self, prompt: str) -> None: - """Set the system prompt.""" - self._system_prompt = prompt - _LOGGER.debug("[%s] System prompt updated: %s", self.instance_name, prompt) - - async def async_clear_history(self) -> None: - """Clear conversation history.""" - self._history.clear() - _LOGGER.debug("[%s] Conversation history cleared", self.instance_name) - self.async_update_listeners() - - async def async_get_history( - self, - limit: Optional[int] = None, - filter_model: Optional[str] = None, - instance: Optional[str] = None - ) -> List[Dict[str, Any]]: - """Get conversation history with optional filtering.""" - history = self._history.copy() - - if instance and instance != self.instance_name: - return [] - - if filter_model: - history = [h for h in history if h.get("model") == filter_model] - - if limit and limit > 0: - history = history[-limit:] - - return history - - async def async_reset_metrics(self) -> None: - """Reset all metrics.""" - self._request_count = 0 - self._error_count = 0 - self._performance_metrics = { - "total_requests": 0, - "total_errors": 0, - "avg_response_time": 0.0, - "last_request_time": None, - "instance": self.instance_name, + async def _process_anthropic_message(self, question: str, **kwargs) -> dict: + """Process message using Anthropic API.""" + response = await self.client.messages.create( + model=kwargs["model"], + max_tokens=kwargs["max_tokens"], + messages=kwargs["messages"], + temperature=kwargs["temperature"], + ) + return { + "content": response.content[0].text, + "tokens": { + "prompt": response.usage.input_tokens, + "completion": response.usage.output_tokens, + "total": response.usage.input_tokens + response.usage.output_tokens + } } - _LOGGER.debug("[%s] Metrics reset", self.instance_name) - self.async_update_listeners() + + async def _process_openai_message(self, question: str, **kwargs) -> dict: + """Process message using OpenAI API.""" + response = await self.client.chat.completions.create( + model=kwargs["model"], + messages=kwargs["messages"], + temperature=kwargs["temperature"], + max_tokens=kwargs["max_tokens"], + ) + return { + "content": response.choices[0].message.content, + "tokens": { + "prompt": response.usage.prompt_tokens, + "completion": response.usage.completion_tokens, + "total": response.usage.total_tokens + } + } + + def _update_metrics(self, latency: float, response: dict) -> None: + """Update performance metrics.""" + metrics = self._performance_metrics + tokens = response.get("tokens", {}) + + metrics["total_tokens"] += tokens.get("total", 0) + metrics["prompt_tokens"] += tokens.get("prompt", 0) + metrics["completion_tokens"] += tokens.get("completion", 0) + metrics["successful_requests"] += 1 + + # Обновляем латентность + metrics["average_latency"] = ( + (metrics["average_latency"] * (metrics["successful_requests"] - 1) + latency) + / metrics["successful_requests"] + ) + metrics["max_latency"] = max(metrics["max_latency"], latency) + metrics["min_latency"] = min(metrics["min_latency"], latency) + + def _update_history(self, question: str, response: dict) -> None: + """Update conversation history.""" + self._conversation_history.append({ + "timestamp": dt_util.utcnow().isoformat(), + "question": question, + "response": response["content"] + }) + + # Ограничиваем размер истории + while len(self._conversation_history) > self.max_history_size: + self._conversation_history.pop(0) + + def _handle_error(self, error: Exception) -> None: + """Handle error and update metrics.""" + self._performance_metrics["total_errors"] += 1 + self._performance_metrics["failed_requests"] += 1 + + self.last_response = { + "timestamp": dt_util.utcnow().isoformat(), + "question": "", + "response": "", + "model": self.model, + "instance": self.instance_name, + "error": str(error) + } + + async def async_update_ha_state(self) -> None: + """Update Home Assistant state.""" + await self.async_request_refresh() diff --git a/custom_components/ha_text_ai/sensor.py b/custom_components/ha_text_ai/sensor.py index a1ef7ce..1b18241 100644 --- a/custom_components/ha_text_ai/sensor.py +++ b/custom_components/ha_text_ai/sensor.py @@ -66,6 +66,8 @@ async def async_setup_entry( ) -> None: """Set up the HA Text AI sensor.""" coordinator = hass.data[DOMAIN][entry.entry_id] + instance_name = coordinator.instance_name + _LOGGER.info(f"Setting up HA Text AI sensor with instance: {instance_name}") async_add_entities([HATextAISensor(coordinator, entry)], True) class HATextAISensor(CoordinatorEntity, SensorEntity): @@ -86,6 +88,9 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): self._attr_unique_id = f"{config_entry.entry_id}_{slugify(self._attr_name)}" self.entity_id = f"sensor.ha_text_ai_{slugify(self._attr_name)}" + # Сохраняем instance_name из координатора + self._instance_name = coordinator.instance_name + self._current_state = STATE_INITIALIZING self._error_count = 0 self._last_error = None @@ -103,13 +108,15 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): sw_version="1.0.0", ) + _LOGGER.info(f"Initialized sensor with instance: {self._instance_name}") + @property def native_value(self) -> StateType: """Return the native value of the sensor.""" if not self.coordinator.data: return STATE_DISCONNECTED - status = self.coordinator.data.get("status", STATE_READY) + status = self.coordinator.data.get("state", STATE_READY) self._current_state = status return status @@ -131,6 +138,7 @@ class HATextAISensor(CoordinatorEntity, SensorEntity): ATTR_API_STATUS: self._current_state, ATTR_TOTAL_ERRORS: self._error_count, ATTR_LAST_ERROR: self._last_error, + "instance_name": self._instance_name, # Добавляем instance_name в атрибуты } if not self.coordinator.data: