diff --git a/bots/rag_client.py b/bots/rag_client.py index 4991c4a..e7a8adb 100644 --- a/bots/rag_client.py +++ b/bots/rag_client.py @@ -8,6 +8,8 @@ HTTP-клиент для взаимодействия с RAG-сервером. Все методы повторяют сигнатуру RAGAPI для бесшовной замены. ИСТОРИЯ НЕ ПЕРЕДАЁТСЯ – сервер получает её из БД. + +Улучшена обработка сетевых ошибок с повторными попытками. """ import logging @@ -20,7 +22,7 @@ logger = logging.getLogger(__name__) class RAGClient: """ - HTTP-клиент для RAG-сервера. + HTTP-клиент для RAG-сервера с поддержкой повторных попыток при сетевых ошибках. """ def __init__( @@ -37,14 +39,14 @@ class RAGClient: server_url (str): базовый URL RAG-сервера (например, http://localhost:8080) timeout (int): таймаут на запрос в секундах max_retries (int): максимальное количество повторных попыток при ошибке - retry_delay (float): задержка между попытками (секунды) + retry_delay (float): базовая задержка между попытками (секунды) """ self.server_url = server_url.rstrip('/') self.timeout = timeout self.max_retries = max_retries self.retry_delay = retry_delay self._session: Optional[aiohttp.ClientSession] = None - logger.info(f"RAGClient инициализирован с сервером {server_url}") + logger.info(f"RAGClient инициализирован с сервером {server_url}, таймаут={timeout}с, макс. попыток={max_retries}") async def _get_session(self) -> aiohttp.ClientSession: """ @@ -52,9 +54,10 @@ class RAGClient: Сессия переиспользуется для всех запросов для эффективности. """ if self._session is None: - self._session = aiohttp.ClientSession( - timeout=aiohttp.ClientTimeout(total=self.timeout) - ) + # Создаём сессию с таймаутом, который применяется ко всем запросам + timeout_obj = aiohttp.ClientTimeout(total=self.timeout) + self._session = aiohttp.ClientSession(timeout=timeout_obj) + logger.debug("HTTP-сессия создана") return self._session async def close(self): @@ -77,33 +80,126 @@ class RAGClient: Исключения: aiohttp.ClientError: если все попытки не удались + asyncio.TimeoutError: если все попытки истекли по таймауту """ url = f"{self.server_url}{endpoint}" session = await self._get_session() + last_exception = None + for attempt in range(self.max_retries): try: + logger.debug(f"Попытка {attempt + 1}/{self.max_retries} к {endpoint}") async with session.post(url, json=payload) as resp: if resp.status == 200: return await resp.json() else: + # Обрабатываем HTTP ошибки (4xx, 5xx) error_text = await resp.text() - logger.error( - f"RAG-сервер вернул {resp.status}: {error_text[:200]}" - ) - raise aiohttp.ClientError( - f"HTTP {resp.status}: {error_text[:200]}" - ) - except (aiohttp.ClientError, asyncio.TimeoutError) as e: + error_msg = f"HTTP {resp.status}: {error_text[:200]}" + logger.error(error_msg) + # Для 5xx ошибок делаем повтор, для 4xx – нет (это ошибка клиента) + if 500 <= resp.status < 600: + raise aiohttp.ClientResponseError( + status=resp.status, + message=error_msg, + headers=resp.headers + ) + else: + # 4xx – не повторяем, сразу выбрасываем + raise aiohttp.ClientResponseError( + status=resp.status, + message=error_msg, + headers=resp.headers + ) + + except (aiohttp.ClientConnectorError, aiohttp.ServerDisconnectedError) as e: + # Ошибки подключения или разрыва соединения – повторяем с увеличенной задержкой logger.warning( - f"Попытка {attempt + 1}/{self.max_retries} к {endpoint} не удалась: {e}" + f"Сетевая ошибка при попытке {attempt + 1}/{self.max_retries} к {endpoint}: {e}" ) + last_exception = e if attempt < self.max_retries - 1: - # Экспоненциальная задержка: 1, 2, 4, ... секунд - await asyncio.sleep(self.retry_delay * (2 ** attempt)) + # Для сетевых ошибок задержка больше: 2, 4, 8 секунд + wait = (self.retry_delay * 2) * (2 ** attempt) + logger.info(f"Повтор через {wait:.1f} секунд") + await asyncio.sleep(wait) + else: + logger.error(f"Все попытки к {endpoint} не удались из-за сетевых проблем") + + except (aiohttp.ClientPayloadError, aiohttp.ClientResponseError) as e: + # Ошибки ответа (например, повреждённые данные) + logger.warning( + f"Ошибка ответа при попытке {attempt + 1}/{self.max_retries} к {endpoint}: {e}" + ) + last_exception = e + if attempt < self.max_retries - 1: + wait = self.retry_delay * (2 ** attempt) + logger.info(f"Повтор через {wait:.1f} секунд") + await asyncio.sleep(wait) + else: + logger.error(f"Все попытки к {endpoint} не удались из-за ошибок ответа") + + except asyncio.TimeoutError as e: + # Таймаут + logger.warning( + f"Таймаут при попытке {attempt + 1}/{self.max_retries} к {endpoint}: {e}" + ) + last_exception = e + if attempt < self.max_retries - 1: + wait = self.retry_delay * (2 ** attempt) + logger.info(f"Повтор через {wait:.1f} секунд") + await asyncio.sleep(wait) + else: + logger.error(f"Все попытки к {endpoint} не удались из-за таймаута") + + except aiohttp.ClientError as e: + # Все остальные ошибки клиента (общий класс) + logger.warning( + f"Ошибка клиента при попытке {attempt + 1}/{self.max_retries} к {endpoint}: {e}" + ) + last_exception = e + if attempt < self.max_retries - 1: + wait = self.retry_delay * (2 ** attempt) + logger.info(f"Повтор через {wait:.1f} секунд") + await asyncio.sleep(wait) else: logger.error(f"Все попытки к {endpoint} не удались") - raise + + except Exception as e: + # Неожиданная ошибка – не повторяем, логируем и выбрасываем + logger.error(f"Неожиданная ошибка при запросе к {endpoint}: {e}", exc_info=True) + raise + + # Если мы вышли из цикла, значит, все попытки провалились + error_msg = f"Не удалось выполнить запрос к {endpoint} после {self.max_retries} попыток" + if last_exception: + error_msg += f": {last_exception}" + logger.error(error_msg) + raise aiohttp.ClientError(error_msg) + + async def health_check(self) -> bool: + """ + Проверяет доступность RAG-сервера через эндпоинт /health. + Возвращает True, если сервер отвечает со статусом 200. + """ + try: + session = await self._get_session() + async with session.get(f"{self.server_url}/health", timeout=10) as resp: + if resp.status == 200: + data = await resp.json() + if data.get('status') == 'healthy': + logger.debug("RAG-сервер доступен (health check пройден)") + return True + else: + logger.warning(f"RAG-сервер вернул нездоровый статус: {data}") + return False + else: + logger.warning(f"Health check вернул HTTP {resp.status}") + return False + except Exception as e: + logger.error(f"Ошибка при health check: {e}") + return False async def query( self, @@ -139,6 +235,7 @@ class RAGClient: return result except Exception as e: logger.error(f"Ошибка в RAGClient.query: {e}", exc_info=True) + # Возвращаем структурированный ответ с ошибкой return { "answer": f"⚠️ Ошибка связи с RAG-сервером: {str(e)}", "intent": "ERROR", diff --git a/core/services/qdrant_service.py b/core/services/qdrant_service.py new file mode 100644 index 0000000..e69de29