From 57702168e5ad1642c78468bf3a9c18c45aefc81b Mon Sep 17 00:00:00 2001 From: Hongyu Shi Date: Wed, 27 Aug 2025 16:29:19 +0800 Subject: [PATCH] =?UTF-8?q?feat(agent):=20=E6=B7=BB=E5=8A=A0=E6=99=BA?= =?UTF-8?q?=E8=83=BD=E4=BD=93=E9=80=89=E6=8B=A9=E5=8A=9F=E8=83=BD=E5=8F=8A?= =?UTF-8?q?=E7=9B=B8=E5=85=B3=E5=91=BD=E4=BB=A4=E8=A1=8C=E5=8F=82=E6=95=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Hongyu Shi --- src/main.py | 19 ++++- src/tool/oi_select_agent.py | 163 ++++++++++++++++++++++++++++++++++++ 2 files changed, 178 insertions(+), 4 deletions(-) create mode 100644 src/tool/oi_select_agent.py diff --git a/src/main.py b/src/main.py index d81a9db4..2572c776 100644 --- a/src/main.py +++ b/src/main.py @@ -1,6 +1,7 @@ """应用入口点""" import argparse +import asyncio import atexit import sys @@ -16,20 +17,26 @@ from log.manager import ( setup_logging, ) from tool.oi_backend_init import oi_backend_init +from tool.oi_select_agent import select_agent def parse_args() -> argparse.Namespace: """解析命令行参数""" parser = argparse.ArgumentParser(description="openEuler Intelligence") parser.add_argument( - "--logs", + "--init", action="store_true", - help="显示最新的日志内容（最多1000行）", + help="初始化 openEuler Intelligence 后端", ) parser.add_argument( - "--init", + "--agent", action="store_true", - help="初始化 openEuler Intelligence 后端", + help="选择默认智能体", + ) + parser.add_argument( + "--logs", + action="store_true", + help="显示最新的日志内容（最多1000行）", ) parser.add_argument( "--log-level", @@ -73,6 +80,10 @@ def main() -> None: oi_backend_init() return + if args.agent: + asyncio.run(select_agent()) + return + # 初始化配置和日志系统 config_manager = ConfigManager() diff --git a/src/tool/oi_select_agent.py b/src/tool/oi_select_agent.py new file mode 100644 index 00000000..6818bbae --- /dev/null +++ b/src/tool/oi_select_agent.py @@ -0,0 +1,163 @@ +"""智能体选择工具""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import TYPE_CHECKING, ClassVar + +from textual.app import App, ComposeResult +from textual.containers import Container + +from app.dialogs import AgentSelectionDialog +from backend.factory import BackendFactory +from config.manager import ConfigManager +from config.model import Backend +from log.manager import get_logger, log_exception, setup_logging + +if TYPE_CHECKING: + from logging import Logger + + +class AgentSelectorApp(App): + """智能体选择应用""" + + CSS_PATH = Path(__file__).parent.parent / "app" / "css" / "styles.tcss" + + BINDINGS: ClassVar = [ + ("escape", "quit", "退出"), + ] + + def __init__(self, agent_list: list[tuple[str, str]], current_agent: tuple[str, str]) -> None: + """初始化智能体选择应用""" + super().__init__() + self.agent_list = agent_list + self.current_agent = current_agent + self.selected_agent: tuple[str, str] | None = None + self.selection_made = False + + def compose(self) -> ComposeResult: + """构建应用界面""" + yield Container(id="main-container") + + def on_mount(self) -> None: + """应用挂载时显示选择对话框""" + + def on_agent_selected(selected_agent: tuple[str, str]) -> None: + """智能体选择回调""" + self.selected_agent = selected_agent + self.selection_made = True + self.exit() + + dialog = AgentSelectionDialog(self.agent_list, on_agent_selected, self.current_agent) + self.push_screen(dialog) + + async def action_quit(self) -> None: + """退出应用""" + self.exit() + + +async def get_agent_list(config_manager: ConfigManager, logger: Logger) -> list[tuple[str, str]]: + """获取智能体列表""" + # 创建 LLM 客户端 + llm_client = BackendFactory.create_client(config_manager) + + # 构建智能体列表 - 默认第一项为"智能问答"（无智能体） + agent_list = [("", "智能问答")] + + # 尝试获取可用智能体 + if not hasattr(llm_client, "get_available_agents"): + logger.warning("当前客户端不支持智能体功能，使用默认选项") + return agent_list + + try: + available_agents = await llm_client.get_available_agents() # type: ignore[attr-defined] + # 添加获取到的智能体 + agent_list.extend( + [ + (agent.app_id, agent.name) + for agent in available_agents + if hasattr(agent, "app_id") and hasattr(agent, "name") + ], + ) + logger.info("成功获取到 %d 个智能体", len(agent_list) - 1) + except (AttributeError, OSError, ValueError, RuntimeError) as e: + logger.warning("获取智能体列表失败，使用默认选项: %s", str(e)) + + return agent_list + + +def get_current_agent(config_manager: ConfigManager, agent_list: list[tuple[str, str]]) -> tuple[str, str]: + """获取当前默认智能体""" + current_app_id = config_manager.get_default_app() + current_agent = ("", "智能问答") + for agent in agent_list: + if agent[0] == current_app_id: + current_agent = agent + break + return current_agent + + +async def run_agent_selector( + agent_list: list[tuple[str, str]], + current_agent: tuple[str, str], +) -> tuple[str, str] | None: + """运行智能体选择器""" + app = AgentSelectorApp(agent_list, current_agent) + await app.run_async() + return app.selected_agent if app.selection_made else None + + +def handle_agent_selection( + config_manager: ConfigManager, + selected_agent: tuple[str, str] | None, +) -> None: + """处理智能体选择结果""" + if selected_agent: + selected_app_id, selected_name = selected_agent + + # 保存选择到配置 + config_manager.set_default_app(selected_app_id) + + sys.stdout.write(f"✓ 默认智能体已设置为: {selected_name}\n") + if selected_app_id: + sys.stdout.write(f" App ID: {selected_app_id}\n") + else: + sys.stdout.write(" 已设置为智能问答模式（无智能体）\n") + else: + sys.stdout.write("已取消选择\n") + + +async def select_agent() -> None: + """智能体选择功能主入口""" + # 初始化配置和日志系统 + config_manager = ConfigManager() + setup_logging(config_manager) + # 注意：这里不启用控制台输出，让日志只写入文件 + + logger = get_logger(__name__) + + # 检查是否使用 eulerintelli 后端 + if config_manager.get_backend() != Backend.EULERINTELLI: + sys.stderr.write("错误: 智能体功能需要使用 openEuler Intelligence 后端\n") + sys.stderr.write("请先运行以下命令切换后端：\n") + sys.stderr.write(" oi # 然后按下 Ctrl+S 进入设置界面切换到 openEuler Intelligence 后端\n") + sys.exit(1) + + try: + # 创建 LLM 客户端并获取智能体列表 + agent_list = await get_agent_list(config_manager, logger) + + # 获取当前默认智能体 + current_agent = get_current_agent(config_manager, agent_list) + + # 运行选择应用 + selected_agent = await run_agent_selector(agent_list, current_agent) + + # 处理选择结果 + handle_agent_selection(config_manager, selected_agent) + + except (OSError, ValueError, RuntimeError) as e: + log_exception(logger, "智能体选择功能发生错误", e) + sys.stderr.write(f"错误: {e}\n") + sys.exit(1) -- Gitee