diff --git a/src/backend/hermes/client.py b/src/backend/hermes/client.py index 4ed4ad3254eac69c740534ebef4ee51d631e98d5..311e34f82b0bc5a291c29bda2b9c111b46441a01 100644 --- a/src/backend/hermes/client.py +++ b/src/backend/hermes/client.py @@ -5,6 +5,7 @@ from __future__ import annotations import json import re import time +from dataclasses import dataclass from typing import TYPE_CHECKING, Any from urllib.parse import urljoin, urlparse @@ -21,6 +22,10 @@ if TYPE_CHECKING: # HTTP 状态码常量 HTTP_OK = 200 +# 分页常量 +ITEMS_PER_PAGE = 16 # 每页最多16项 +MAX_PAGES = 100 # 最多请求100页 + class HermesAPIError(Exception): """Hermes API 错误异常""" @@ -41,6 +46,45 @@ def validate_url(url: str) -> bool: return re.match(r"^https?://", url) is not None +@dataclass +class HermesAgent: + """Hermes 智能体数据结构""" + + app_id: str + """应用ID""" + + name: str + """智能体名称""" + + author: str + """作者""" + + description: str + """描述""" + + icon: str + """图标""" + + favorited: bool + """是否已收藏""" + + published: bool | None = None + """是否已发布""" + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> HermesAgent: + """从字典创建智能体对象""" + return cls( + app_id=data.get("appId", ""), + name=data.get("name", ""), + author=data.get("author", ""), + description=data.get("description", ""), + icon=data.get("icon", ""), + favorited=data.get("favorited", False), + published=data.get("published"), + ) + + class HermesMessage: """Hermes 消息类""" @@ -345,6 +389,260 @@ class HermesChatClient(LLMClientBase): else: return models + async def get_available_agents(self) -> list[HermesAgent]: + """ + 获取当前用户可用的智能体列表 + + 通过调用 /api/app 接口获取当前用户可用的智能体列表。 + 支持分页获取所有智能体,每页最多16项,会自动请求所有页面。 + 这些智能体可以在聊天中使用,选择的智能体 ID 需要在 chat 接口中填入 appId 字段。 + 如果调用失败或没有返回,使用空列表。 + + Returns: + list[HermesAgent]: 可用的智能体列表(仅包含已发布的智能体) + + """ + start_time = time.time() + self.logger.info("开始请求 Hermes 智能体列表 API") + + all_agents = [] + current_page = 1 + total_apps = 0 + + try: + while True: + # 请求当前页 + page_agents, page_info = await self._get_agents_page(current_page) + + # 添加到总列表中 + all_agents.extend(page_agents) + + # 更新总数信息(第一次获取时) + if current_page == 1: + total_apps = page_info.get("total_apps", 0) + self.logger.info("总共有 %d 个应用需要获取", total_apps) + + current_page_from_response = page_info.get("current_page", current_page) + self.logger.info("获取第 %d 页完成,本页获得 %d 个智能体", current_page_from_response, len(page_agents)) + + # 检查是否还有更多页面 + # 每页最多16项,如果本页获取的数量少于16,说明是最后一页 + if len(page_agents) < ITEMS_PER_PAGE: + self.logger.info("已获取所有页面,共 %d 页", current_page) + break + + # 继续请求下一页 + current_page += 1 + + # 安全检查:避免无限循环 + if current_page > MAX_PAGES: + self.logger.warning("已达到最大页数限制(%d页),停止请求", MAX_PAGES) + break + + # 过滤已发布的智能体 + published_agents = [agent for agent in all_agents if agent.published is True] + + duration = time.time() - start_time + self.logger.info( + "获取智能体列表完成 - 总耗时: %.3fs, 总应用数: %d, 智能体数: %d, 已发布智能体: %d", + duration, + total_apps, + len(all_agents), + len(published_agents), + ) + + except (httpx.HTTPError, httpx.InvalidURL) as e: + # 网络请求异常 + duration = time.time() - start_time + log_exception(self.logger, "Hermes 智能体列表 API 请求异常", e) + log_api_request( + self.logger, + "GET", + f"{self.base_url}/api/app", + 500, + duration, + error=str(e), + ) + self.logger.warning("Hermes 智能体列表 API 请求异常,返回空列表") + return [] + else: + return published_agents + + async def _get_agents_page(self, page: int) -> tuple[list[HermesAgent], dict[str, Any]]: + """ + 获取指定页的智能体列表 + + Args: + page: 页码,从1开始 + + Returns: + tuple[list[HermesAgent], dict[str, Any]]: (智能体列表, 页面信息) + 页面信息包含: {"current_page": int, "total_apps": int} + + Raises: + httpx.HTTPError: 网络请求错误 + httpx.InvalidURL: URL错误 + + """ + client = await self._get_client() + app_url = urljoin(self.base_url, "/api/app") + + # 构建查询参数 + params = { + "appType": "agent", # 只获取智能体类型的应用 + "page": page, # 当前页码 + } + + headers = { + "Host": self._get_host_header(), + } + if self.auth_token: + headers["Authorization"] = f"Bearer {self.auth_token}" + + response = await client.get(app_url, headers=headers, params=params) + + # 处理HTTP错误状态 + if response.status_code != HTTP_OK: + error_msg = f"API 调用失败,状态码: {response.status_code}" + self.logger.warning("获取第 %d 页失败: %s", page, error_msg) + raise httpx.HTTPStatusError(error_msg, request=response.request, response=response) + + # 解析响应数据 + try: + data = response.json() + except json.JSONDecodeError as e: + error_msg = "响应 JSON 格式无效" + self.logger.warning("获取第 %d 页失败: %s", page, error_msg) + raise httpx.DecodingError(error_msg) from e + + # 验证响应结构 + if not self._validate_agent_response_structure_for_page(data, page): + return [], {} + + # 解析智能体列表和页面信息 + result = data["result"] + agents = self._parse_agent_list(result) + + page_info = { + "current_page": result.get("currentPage", page), + "total_apps": result.get("totalApps", 0), + } + + return agents, page_info + + def _validate_agent_response_structure_for_page(self, data: dict[str, Any], page: int) -> bool: + """验证单页智能体 API 响应结构""" + if not isinstance(data, dict): + self.logger.warning("第 %d 页响应格式无效:不是字典", page) + return False + + # 检查响应码 + code = data.get("code") + if code != 0: + message = data.get("message", "未知错误") + self.logger.warning("第 %d 页 API 返回错误: code=%s, message=%s", page, code, message) + return False + + # 检查result字段 + result = data.get("result") + if not isinstance(result, dict): + self.logger.warning("第 %d 页 result 字段不是对象", page) + return False + + # 检查applications字段 + applications = result.get("applications") + if not isinstance(applications, list): + self.logger.warning("第 %d 页 applications 字段不是数组", page) + return False + + return True + + def _handle_agent_api_error( + self, + url: str, + status_code: int, + duration: float, + error_msg: str, + ) -> list[HermesAgent]: + """处理智能体 API 错误,返回空列表""" + log_api_request( + self.logger, + "GET", + url, + status_code, + duration, + error=error_msg, + ) + self.logger.warning("Hermes 智能体列表 API %s,返回空列表", error_msg) + return [] + + def _validate_agent_response_structure( + self, + data: dict[str, Any], + url: str, + status_code: int, + duration: float, + ) -> bool: + """验证智能体 API 响应结构""" + if not isinstance(data, dict): + self._handle_agent_api_error(url, status_code, duration, "响应格式无效") + return False + + # 检查响应码 + code = data.get("code") + if code != 0: + message = data.get("message", "未知错误") + self._handle_agent_api_error(url, status_code, duration, f"API 返回错误: {message}") + return False + + # 检查result字段 + result = data.get("result") + if not isinstance(result, dict): + self._handle_agent_api_error(url, status_code, duration, "result 字段不是对象") + return False + + # 检查applications字段 + applications = result.get("applications") + if not isinstance(applications, list): + self._handle_agent_api_error(url, status_code, duration, "applications 字段不是数组") + return False + + return True + + def _parse_agent_list(self, result: dict[str, Any]) -> list[HermesAgent]: + """解析智能体列表数据(所有 Agent 类型应用,不过滤发布状态)""" + try: + agents = [] + applications = result.get("applications", []) + + if not isinstance(applications, list): + self.logger.warning("applications 字段不是列表类型") + return [] + + for app_data in applications: + if not isinstance(app_data, dict): + continue + + # 只处理Agent类型的应用 + app_type = app_data.get("appType") + if app_type != "agent": + continue + + try: + agent = HermesAgent.from_dict(app_data) + if agent.app_id and agent.name: # 确保必要字段存在 + agents.append(agent) + else: + self.logger.debug("跳过无效的智能体信息: %s", app_data) + except (KeyError, TypeError, ValueError) as e: + self.logger.warning("解析智能体信息失败: %s, 错误: %s", app_data, e) + + except (KeyError, TypeError, ValueError) as e: + self.logger.warning("解析智能体列表数据失败: %s, 错误: %s", result, e) + return [] + else: + return agents + async def close(self) -> None: """关闭 HTTP 客户端""" # 如果有未完成的会话,先停止它 @@ -359,7 +657,7 @@ class HermesChatClient(LLMClientBase): def _get_host_header(self) -> str: """ - 从base_url中提取主机名用于Host头部 + 从 base_url 中提取主机名用于 Host 请求头字段 Returns: str: 主机名,如 'www.eulercopilot.io' @@ -561,7 +859,7 @@ class HermesChatClient(LLMClientBase): return self._conversation_id def _build_chat_headers(self) -> dict[str, str]: - """构建聊天请求的HTTP头部""" + """构建聊天请求的 HTTP 头部""" headers = { "Host": self._get_host_header(), }