""" main.py - MyAgent 主入口 ========================== 本地桌面端执行型AI助手。 支持: - 系统托盘后台运行 (pystray) - 交互式命令行 - 多聊天平台接入等 - Windows / macOS / Linux """ from __future__ import annotations import asyncio import os import sys import signal import threading from pathlib import Path # 确保项目根目录在 Python 路径中 # 兼容两种布局: (A) 扁平结构 skills/ 在脚本同目录 (B) 嵌套结构 skills/ 在 myagent/ 子目录 PROJECT_ROOT = Path(__file__).parent.resolve() if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) _MYAGENT_SUB = PROJECT_ROOT / 'myagent' if _MYAGENT_SUB.is_dir() and str(_MYAGENT_SUB) not in sys.path: sys.path.insert(0, str(_MYAGENT_SUB)) from config import get_config, ConfigManager from core.logger import setup_logger, get_logger from core.agent_storage import AgentStorage from core.llm import LLMClient, Message from core.task_queue import TaskQueue from memory.manager import MemoryManager from executor.engine import ExecutionEngine from aiskills.registry import SkillRegistry from agents.main_agent import MainAgent from agents.tool_agent import ToolAgent from agents.memory_agent import MemoryAgent from chatbot.base import ChatMessage, ChatResponse from chatbot.manager import ChatBotManager from core.utils import timestamp, detect_platform from core.config_broadcast import ConfigBroadcaster, ReloadType from core.update_manager import UpdateManager, UpdateType from core.version import get_version from core.permissions import PermissionManager from core.deps_checker import check_and_install_deps # 保留导入供其他模块使用 def _get_screen_resolution() -> tuple[int, int]: """获取当前屏幕分辨率，跨平台兼容 Windows / macOS / Linux。""" try: import tkinter as tk root = tk.Tk() root.withdraw() # 不显示窗口 w = root.winfo_screenwidth() h = root.winfo_screenheight() root.destroy() return w, h except Exception: pass # Linux 回退方案 try: import subprocess result = subprocess.run( ["xdpyinfo"], capture_output=True, text=True, timeout=5 ) for line in result.stdout.splitlines(): if "dimensions:" in line: parts = line.split() for p in parts: # Bug Fix: 使用更严格的正则匹配分辨率格式 (数字x数字) import re as _re m = _re.match(r"^(\d+)x(\d+)$", p) if m: return int(m.group(1)), int(m.group(2)) except Exception: pass return 1920, 1080 def _open_browser_kiosk(url: str): """打开浏览器窗口，使用系统已安装的 Chrome 或默认浏览器。 [v1.34.0] 优先使用系统已安装的 Chrome/Edge，而非 DrissionPage 控制的浏览器。 [v1.47.12] Termux 环境: 优先使用 termux-open-url（调用 Android 系统浏览器）。 [v1.48.0] 复用现有 Chrome 窗口打开新选项卡，避免每次都打开新窗口。 """ import shutil import subprocess import platform # [v1.47.12] Termux 环境: 优先使用 termux-open-url 打开 Android 系统浏览器 # 在 Termux proot 环境下，xdg-open 无法工作（Permission denied）， # 但 termux-open-url 可以跨 proot 边界调用 Android 系统浏览器 termux_open = shutil.which("termux-open-url") if termux_open: try: subprocess.Popen([termux_open, url], start_new_session=True) return except Exception: pass chrome_path = None system = platform.system() if system == "Windows": for p in ( os.path.join(os.environ.get("PROGRAMFILES", "C:\\Program Files"), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("PROGRAMFILES(X86)", "C:\\Program Files (x86)"), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("LOCALAPPDATA", ""), "Google", "Chrome", "Application", "chrome.exe"), os.path.join(os.environ.get("PROGRAMFILES", "C:\\Program Files"), "Microsoft", "Edge", "Application", "msedge.exe"), ): if p and os.path.isfile(p): chrome_path = p break elif system == "Darwin": for p in ( "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome", "/Applications/Chromium.app/Contents/MacOS/Chromium", "/Applications/Microsoft Edge.app/Contents/MacOS/Microsoft Edge", ): if os.path.isfile(p): chrome_path = p break else: # Linux # [v1.38.0] 优先使用 myagent-browser 包装器（自带 proot 兼容参数） for cmd in ("myagent-browser", "firefox", "google-chrome", "google-chrome-stable", "chromium", "epiphany", "microsoft-edge"): found = shutil.which(cmd) if found: # [v1.36.0] 跳过 snap 包装器（Ubuntu 的 chromium-browser 只是 snap 引导脚本） if "chromium-browser" in cmd or "chromium" in cmd: try: with open(found, "r") as f: first_kb = f.read(1024) if "snap" in first_kb.lower(): continue except Exception: pass chrome_path = found break # 环境变量覆盖 for key in ("CHROME_PATH", "BROWSER_PATH"): val = os.environ.get(key, "").strip() if val and os.path.isfile(val): chrome_path = val break if chrome_path: try: # [v1.48.0] 复用现有 Chrome 窗口打开新选项卡 # 不同平台使用不同的方式： # - Windows: 直接调用 chrome.exe，默认在现有窗口打开新选项卡 # - macOS: 使用 "open -a Google Chrome URL" 命令 # - Linux: 直接调用 chrome，默认在现有窗口打开新选项卡 if system == "Darwin": # macOS: 使用 open 命令，会复用现有 Chrome 窗口 subprocess.Popen(["open", "-a", "Google Chrome", url], start_new_session=False) elif system == "Windows": # Windows: 直接调用 chrome.exe，不使用 start_new_session # 这样会复用现有 Chrome 进程打开新选项卡 subprocess.Popen([chrome_path, url], start_new_session=False) else: # Linux: 直接调用 chrome，不使用 start_new_session subprocess.Popen([chrome_path, url], start_new_session=False) return except Exception: pass # 降级: 使用系统默认浏览器 import webbrowser webbrowser.open(url) # ============================================================================== # MyAgent 应用主类 # ============================================================================== class MyAgentApp: """ MyAgent 应用主类。 管理: - LLM 客户端 - 记忆系统 - 执行引擎 - 技能系统 - Agent 集群 - 聊天平台 - 系统托盘 - 任务队列 """ def __init__(self): self.config_mgr = get_config() self.config = self.config_mgr.config self.logger = None self._running = False self._restart_requested = False self._stop_requested = False self._start_requested = False # 核心组件 self.memory: MemoryManager | None = None self.executor: ExecutionEngine | None = None self.skill_registry: SkillRegistry | None = None self.llm: LLMClient | None = None self.task_queue: TaskQueue | None = None # Agent self.main_agent: MainAgent | None = None self.tool_agent: ToolAgent | None = None self.memory_agent: MemoryAgent | None = None self.agents: Dict[str, BaseAgent] = {} # Agent 注册表（供 api_server 等模块遍历） # 聊天平台 self.chat_manager: ChatBotManager | None = None # [v1.20.13] 会话级锁：防止聊天平台消息与 Web 请求并发覆盖 MainAgent 共享状态 self._session_locks: dict[str, asyncio.Lock] = {} self._session_locks_mutex = asyncio.Lock() # 配置热加载广播器 self.config_broadcaster: ConfigBroadcaster | None = None # 通信管理器 (Agent 间加密通信) self.communication_manager = None # 群聊管理器 self.group_manager = None # 自动更新管理器 self.update_manager: UpdateManager | None = None # 权限管理器 self.permission_manager: PermissionManager | None = None # Bug Fix: _running_service 必须在 __init__ 初始化，否则托盘菜单访问时 AttributeError self._running_service = False # Agent 存储（数据库） self.agent_storage: AgentStorage | None = None # 交互式会话 self._session_id = "cli_default" async def initialize(self): """初始化所有组件""" # 1. 日志 self.logger = setup_logger( "myagent", log_dir=str(self.config_mgr.logs_dir), level=self.config.log_level, ) self.logger.info("=" * 60) self.logger.info("MyAgent 正在启动...") self.logger.info(f"平台: {detect_platform()}") self.logger.info(f"数据目录: {self.config_mgr.data_dir}") self.logger.info("=" * 60) # 1.5 依赖检查已移至 start.js (install/reinstall 命令) # 启动时不再检查依赖，确保启动速度干净利落 # 2. LLM 客户端 llm_cfg = self.config.llm self.llm = LLMClient( provider=llm_cfg.provider, api_key=llm_cfg.api_key, base_url=llm_cfg.base_url, model=llm_cfg.model, temperature=llm_cfg.temperature, max_tokens=llm_cfg.max_tokens, timeout=llm_cfg.timeout, max_retries=llm_cfg.max_retries, reasoning=llm_cfg.reasoning, anthropic_api_key=llm_cfg.anthropic_api_key, context_window=llm_cfg.context_window, ) self.logger.info(f"LLM: {llm_cfg.provider}/{llm_cfg.model}") # 3. 记忆系统 mem_cfg = self.config.memory self.memory = MemoryManager(db_path=mem_cfg.db_path) self.memory.initialize() mem_stats = self.memory.get_stats() self.logger.info(f"记忆系统: {mem_stats['total_count']} 条记录") # 3.5 Agent 存储（数据库） self.agent_storage = AgentStorage() self.agent_storage._init_tables() self.logger.info("Agent 存储已初始化") # 4. 执行引擎 exe_cfg = self.config.executor # [v1.20.8] 设置工作目录为 workspace，Agent 生成的文件默认保存到这里 workspace_dir = str(self.config_mgr.data_dir / "workspace") os.makedirs(workspace_dir, exist_ok=True) self.executor = ExecutionEngine( timeout=exe_cfg.timeout, max_retries=exe_cfg.max_retries, auto_fix=exe_cfg.auto_fix, max_output_length=exe_cfg.max_output_length, execution_mode=exe_cfg.execution_mode, sandbox_image=exe_cfg.sandbox_image, sandbox_network=exe_cfg.sandbox_network, sandbox_memory=exe_cfg.sandbox_memory, work_dir=workspace_dir, ) self.logger.info(f"执行引擎: timeout={exe_cfg.timeout}s, auto_fix={exe_cfg.auto_fix}, work_dir={workspace_dir}") # 5. 技能系统 self.skill_registry = SkillRegistry() # 注册内置技能 self._register_builtin_skills() skills = self.skill_registry.list_skills() self.logger.info(f"技能系统: {len(skills)} 个技能已注册 - {skills}") # 6. 任务队列 self.task_queue = TaskQueue(max_workers=self.config.agent.max_parallel) await self.task_queue.start() self.logger.info(f"任务队列: workers={self.config.agent.max_parallel}") # 7. 配置广播器 self.config_broadcaster = ConfigBroadcaster() self.logger.info("配置广播器已初始化") # 8. 权限管理器 self.permission_manager = PermissionManager( data_dir=str(self.config_mgr.data_dir), ) self.permission_manager.load() # 设置全局权限管理器（供 BaseAgent.check_permission 使用） from agents.base import set_permission_manager set_permission_manager(self.permission_manager) perm_summary = self.permission_manager.get_config_summary() defaults = perm_summary.get("defaults", {}) self.logger.info( f"权限管理器: 已初始化 (默认: " f"执行={'✓' if defaults.get('execution') else '✗'} " f"文件读='{'✓' if defaults.get('file_read') else '✗'} " f"文件写='{'✓' if defaults.get('file_write') else '✗'} " f"联网='{'✓' if defaults.get('network') else '✗'} " f"本机通信='{'✓' if defaults.get('local_comm') else '✗'} " f"跨电脑通信='{'✓' if defaults.get('remote_comm') else '✗'})" ) # 9. 通信管理器 (Agent 间加密通信) try: from communication.manager import CommunicationManager comm_config = self.config.communication self.communication_manager = CommunicationManager( config=comm_config, data_dir=str(self.config_mgr.data_dir), ) self.communication_manager.initialize() if comm_config.enabled: # 注入权限检查器到通信管理器 self.communication_manager.set_permission_checker( self.permission_manager.check_permission ) await self.communication_manager.start() self.logger.info(f"通信管理器: agent_id={self.communication_manager.agent_id}") else: self.logger.info("通信管理器: 已初始化但未启用") except Exception as e: self.logger.warning(f"通信管理器初始化失败: {e}") self.communication_manager = None # 8.5 群聊管理器 try: from groups.manager import GroupManager self.group_manager = GroupManager( data_dir=str(self.config_mgr.data_dir), ) self.group_manager.initialize() self.logger.info(f"群聊管理器: {len(self.group_manager.list_groups())} 个群") except Exception as e: self.logger.warning(f"群聊管理器初始化失败: {e}") self.group_manager = None # 9. 创建 Agent self.memory_agent = MemoryAgent( llm=self.llm, memory_manager=self.memory, config=self.config, config_broadcaster=self.config_broadcaster, ) self.tool_agent = ToolAgent( llm=self.llm, memory_manager=self.memory, executor=self.executor, skill_registry=self.skill_registry, task_queue=self.task_queue, config=self.config, config_broadcaster=self.config_broadcaster, ) self.main_agent = MainAgent( llm=self.llm, memory_manager=self.memory, executor=self.executor, skill_registry=self.skill_registry, task_queue=self.task_queue, config=self.config, tool_agent=self.tool_agent, memory_agent=self.memory_agent, config_broadcaster=self.config_broadcaster, ) self.main_agent.init_context_builder( memory_manager=self.memory, skill_registry=self.skill_registry, knowledge_base_dir=str(self.config_mgr.data_dir / "organization" / "knowledge") ) # [v1.22.0] 初始化统一工具分发器 from core.tool_dispatcher import ToolDispatcher self.main_agent.dispatcher = ToolDispatcher( executor=self.executor, memory_agent=self.memory_agent, skill_registry=self.skill_registry, permission_checker=(self.main_agent.check_permission, self.main_agent.name), llm_client=self.llm, memory_manager=self.memory, knowledge_base_dir=str(self.config_mgr.data_dir / "organization" / "knowledge"), ) # 初始化 STT 引擎的 LLM API 配置（Whisper 兼容端点） try: from core.stt import set_llm_config cfg = self.config_mgr.config.llm set_llm_config(cfg.api_key or "", cfg.base_url or "") except Exception as e: self.logger.debug(f"STT LLM 配置初始化失败: {e}") self.logger.info("统一工具分发器已初始化") # [v1.22.0] 同步 SKILL.md 到知识库（按需加载，不再全量注入 system prompt） self._sync_skill_guides_to_knowledge() self.logger.info("Agent 集群已初始化 (主/工具/记忆)") # [v1.15.10] 注册到 agents 字典（供 api_server 等模块遍历） self.agents["main"] = self.main_agent self.agents["tool"] = self.tool_agent self.agents["memory"] = self.memory_agent # 10. 自动更新管理器 self.update_manager = UpdateManager( config_broadcaster=self.config_broadcaster, task_queue=self.task_queue, data_dir=str(self.config_mgr.data_dir), app_ref=self, ) await self.update_manager.initialize() self.logger.info(f"更新管理器: v{get_version()}, 定时检查=开启") # 11. 聊天平台 enabled_platforms = self.config_mgr.get_enabled_platforms() if enabled_platforms: self.chat_manager = ChatBotManager() self.chat_manager.setup_platforms( enabled_platforms, message_handler=self._handle_chat_message, ) self.logger.info(f"聊天平台: {[p.platform for p in enabled_platforms]}") else: self.logger.info("聊天平台: 未配置(仅CLI模式)") self._running = True self.logger.info("✅ MyAgent 启动完成!") # 检查是否是更新后重启 if os.environ.get("MYAGENT_UPDATED") == "true": from_ver = os.environ.get("MYAGENT_UPDATED_FROM", "?") to_ver = os.environ.get("MYAGENT_UPDATED_TO", "?") self.logger.info(f"🔄 检测到更新后重启: v{from_ver} → v{to_ver}") # 清理环境变量 os.environ.pop("MYAGENT_UPDATED", None) os.environ.pop("MYAGENT_UPDATED_FROM", None) os.environ.pop("MYAGENT_UPDATED_TO", None) def _register_builtin_skills(self): """注册内置技能""" from aiskills.search_skill import WebSearchSkill, WebReadSkill, URLReadSkill from aiskills.system_skill import ( SystemInfoSkill, ProcessListSkill, EnvironmentGetSkill, PathExpandSkill, ) from aiskills.chromedev_mcp import ( BrowserOpenSkill, BrowserClickSkill, BrowserFillSkill, BrowserScreenshotSkill, BrowserEvalSkill, BrowserNavigateSkill, BrowserCloseSkill, ) from aiskills.gui_skill import ( ScreenShotSkill, MouseClickSkill, MouseDragSkill, TypeTextSkill, HotkeySkill, WindowListSkill, WindowFocusSkill, ScreenElementSkill, ) # 搜索技能 for skill_cls in [WebSearchSkill, WebReadSkill, URLReadSkill]: self.skill_registry.register(skill_cls()) # 系统技能 for skill_cls in [ SystemInfoSkill, ProcessListSkill, EnvironmentGetSkill, PathExpandSkill, ]: self.skill_registry.register(skill_cls()) # 浏览器自动化技能 (ChromeDev MCP) for skill_cls in [ BrowserOpenSkill, BrowserClickSkill, BrowserFillSkill, BrowserScreenshotSkill, BrowserEvalSkill, BrowserNavigateSkill, BrowserCloseSkill, ]: self.skill_registry.register(skill_cls()) # 桌面 GUI 自动化技能 for skill_cls in [ ScreenShotSkill, MouseClickSkill, MouseDragSkill, TypeTextSkill, HotkeySkill, WindowListSkill, WindowFocusSkill, ScreenElementSkill, ]: self.skill_registry.register(skill_cls()) # ── 文档生成技能 (v1.17.3, 融合 MiniMax Skills 设计理念) ── from aiskills.pdf_skill import PDFCreateSkill, PDFReadSkill from aiskills.docx_skill import DOCXCreateSkill, DOCXReadSkill from aiskills.xlsx_skill import XLSXCreateSkill, XLSXReadSkill, XLSXEditSkill from aiskills.ppt_skill import PPTCreateSkill, PPTReadSkill for skill_cls in [ PDFCreateSkill, PDFReadSkill, DOCXCreateSkill, DOCXReadSkill, XLSXCreateSkill, XLSXReadSkill, XLSXEditSkill, PPTCreateSkill, PPTReadSkill, ]: self.skill_registry.register(skill_cls()) # ── 反检测浏览器技能 (DrissionPage — v1.30.0, v1.31.0) ── try: from aiskills.browser_stealth import ( StealthBrowserStartSkill, StealthBrowserNavigateSkill, StealthBrowserClickSkill, StealthBrowserFillSkill, StealthBrowserScreenshotSkill, StealthBrowserEvalSkill, StealthBrowserGetContentSkill, StealthBrowserCookieSkill, StealthBrowserCloseSkill, StealthBrowserWaitManualSkill, StealthBrowserWaitSkill, BrowserProfileListSkill, BrowserProfileCreateSkill, BrowserProfileDeleteSkill, # Site 管理 (v1.31.0) SiteManageListSkill, SiteManageShowSkill, SiteManageAddSkill, SiteManageRemoveSkill, SiteManageInitSkill, ) for skill_cls in [ StealthBrowserStartSkill, StealthBrowserNavigateSkill, StealthBrowserClickSkill, StealthBrowserFillSkill, StealthBrowserScreenshotSkill, StealthBrowserEvalSkill, StealthBrowserGetContentSkill, StealthBrowserCookieSkill, StealthBrowserCloseSkill, StealthBrowserWaitManualSkill, StealthBrowserWaitSkill, BrowserProfileListSkill, BrowserProfileCreateSkill, BrowserProfileDeleteSkill, SiteManageListSkill, SiteManageShowSkill, SiteManageAddSkill, SiteManageRemoveSkill, SiteManageInitSkill, ]: self.skill_registry.register(skill_cls()) except ImportError as e: self.logger.warning(f"反检测浏览器技能注册跳过（DrissionPage 未安装）: {e}") # ── [v1.35.2] 任务规划技能 ── from aiskills.task_plan_skill import TaskPlanSkill self.skill_registry.register(TaskPlanSkill()) # ── [v1.22.0] 内置平台工具不再注册存根 ── # command, recall_memory, file_send, playaudio, playvideo, web_control # 由 ToolDispatcher 直接处理，其元数据通过 BUILTIN_TOOLS 常量 # 在 registry.list_skills_by_category() 中自动合并展示。 def _sync_skill_guides_to_knowledge(self): """[v1.22.0] 将 SKILL.md 内容同步到知识库目录，供 get_knowledge 按需检索。 不再将所有 SKILL.md 全量注入 ，而是写入 {kb_dir}/_skill_guides/ 目录，由 RAG 索引，LLM 需要时 通过 按需检索。 """ import yaml from pathlib import Path kb_dir = self.config_mgr.data_dir / "organization" / "knowledge" skill_guides_dir = kb_dir / "_skill_guides" skill_guides_dir.mkdir(parents=True, exist_ok=True) # 扫描项目 aiskills/ 目录下的 SKILL.md project_skills_dir = Path(__file__).parent / "aiskills" if not project_skills_dir.exists(): self.logger.warning(f"aiskills 目录不存在: {project_skills_dir}") return synced = 0 for subdir in sorted(project_skills_dir.iterdir()): if not subdir.is_dir() or subdir.name.startswith(("_", ".")): continue skill_md = subdir / "SKILL.md" if not skill_md.exists(): continue try: content = skill_md.read_text(encoding="utf-8") if not content.startswith("---"): continue parts = content.split("---", 2) if len(parts) < 3: continue meta = yaml.safe_load(parts[1]) or {} name = meta.get("name", subdir.name) # 去重: 检查文件是否已存在且内容相同 target_file = skill_guides_dir / f"{name}.md" if target_file.exists(): existing = target_file.read_text(encoding="utf-8") if existing.strip() == content.strip(): continue # 内容相同，跳过 target_file.write_text(content, encoding="utf-8") synced += 1 except Exception as e: self.logger.warning(f"同步 SKILL.md 失败 ({subdir.name}): {e}") if synced > 0: self.logger.info(f"已同步 {synced} 个 SKILL.md 到知识库 ({skill_guides_dir})") else: self.logger.info("SKILL.md 知识库已是最新") async def process_message( self, user_message: str, session_id: str = "", stream_callback=None, ) -> str: """ 处理用户消息并返回回复。 Args: user_message: 用户消息 session_id: 会话 ID stream_callback: SSE 流式回调（群聊等场景需要，用于推送 v2_file 等事件） Returns: 助手回复文本 """ if not self.main_agent: return "⚠️ MyAgent 尚未初始化" session_id = session_id or self._session_id from agents.base import AgentContext context = AgentContext( session_id=session_id, user_message=user_message, ) try: result_context = await self.main_agent.process(context, stream_callback=stream_callback) response = result_context.working_memory.get( "final_response", "⚠️ 未能生成回复" ) return response except Exception as e: self.logger.error(f"处理消息异常: {e}", exc_info=True) return f"❌ 处理失败: {str(e)}" async def _get_session_lock(self, session_id: str) -> asyncio.Lock: """获取或创建会话级锁""" async with self._session_locks_mutex: if session_id not in self._session_locks: self._session_locks[session_id] = asyncio.Lock() return self._session_locks[session_id] async def _handle_chat_message(self, message: ChatMessage, bot): """ 处理来自聊天平台的消息。 [v1.20.8] 修复: 发送到平台前，将相对路径的文件链接替换为绝对 URL。 Args: message: 聊天消息 bot: 发送消息的机器人实例 """ # 生成会话 ID session_id = bot._generate_session_id(message) # 处理特殊命令 if message.text == "__cmd_clear__": if self.memory: self.memory.clear_conversation(session_id) await bot.send_message(ChatResponse( chat_id=message.chat_id, text="🗑️ 对话历史已清除。", )) return # 处理消息 # [v1.20.13] 使用会话级锁防止并发处理同一会话 # Bug Fix: 使用 try_acquire 模式避免 locked() + acquire 的 TOCTOU 竞争 session_lock = await self._get_session_lock(session_id) if not session_lock.acquire(blocking=False): return # 同一会话正在处理中，跳过（聊天平台消息有去重机制） try: response_text = await self.process_message(message.text, session_id) finally: session_lock.release() self._cleanup_session_locks() # [v1.20.8] 将回复中的相对路径文件链接替换为绝对 URL # (file_send 生成的 /api/file/xxx 链接在 Telegram/Discord 等外部平台无法访问) response_text = self._fix_file_urls_for_platform(response_text) # 发送回复 await bot.send_message(ChatResponse( chat_id=message.chat_id, user_id=message.user_id, text=response_text, )) def _fix_file_urls_for_platform(self, text: str) -> str: """[v1.20.8] 将文本中相对路径的文件下载链接替换为绝对 URL。 匹配模式: - /api/file/{file_id} → {public_url}/api/file/{file_id} - /api/file/{file_id}/download → {public_url}/api/file/{file_id}/download - (href="|src=")(/api/file/...) → 加上 public_url 前缀 """ import re from aiskills.file_send import get_public_base_url base_url = get_public_base_url() if not base_url: return text # 没有配置 public_url，不做替换 # 匹配 markdown 链接: [text](/api/file/...) text = re.sub( r'(\[([^\]]*)\]\()(/api/file/[^\)]+)(\))', lambda m: f'{m.group(1)}{base_url}{m.group(3)}{m.group(4)}', text, ) # 匹配 HTML 属性: href="/api/file/..." 或 src="/api/file/..." text = re.sub( r'((?:href|src)=["\'])(/api/file/[^"\']*)(["\'])', lambda m: f'{m.group(1)}{base_url}{m.group(2)}{m.group(3)}', text, ) # 匹配裸链接: /api/file/{12位ID}（可能后面有 /download） text = re.sub( r'(? ").strip() if not user_input: continue # 内置命令 if user_input.lower() in ("quit", "exit", "q"): print("再见!") break elif user_input.lower() == "help": self._print_help() continue elif user_input.lower() == "status": self._print_status() continue elif user_input.lower() == "skills": self._print_skills() continue elif user_input.lower() == "memory": self._print_memory_stats() continue elif user_input.lower() == "permissions": self._print_permissions() continue elif user_input.lower() == "sessions": self._print_sessions() continue elif user_input.lower().startswith("session "): new_session = user_input[8:].strip() if new_session: self._session_id = new_session print(f"📝 已切换到会话: {self._session_id}") continue elif user_input.lower() == "clear": if self.memory: self.memory.clear_conversation(self._session_id) print("🗑️ 对话历史已清除") continue # 处理消息 print("⏳ 思考中...", end="", flush=True) response = await self.process_message(user_input, self._session_id) print(f"\r🤖 助手: {response}") print() except KeyboardInterrupt: print("\n\n👋 再见!") break except EOFError: break except Exception as e: print(f"\n❌ 错误: {e}") print() def _print_help(self): print(""" 📋 可用命令: help - 显示帮助 status - 查看系统状态 skills - 列出所有技能 memory - 查看记忆统计 permissions - 查看权限配置 sessions - 查看会话列表 session - 切换会话 clear - 清除当前对话历史 quit/exit - 退出 💡 使用方式: 直接输入自然语言描述你的需求。 示例: - "帮我创建一个Python文件" - "搜索一下最新的AI新闻" - "查看系统信息" - "!system_info" (直接调用技能) """) def _print_status(self): print(f""" 📊 系统状态: 运行状态: {'✅ 运行中' if self._running else '❌ 已停止'} LLM: {self.config.llm.provider}/{self.config.llm.model} 当前会话: {self._session_id} 执行引擎: timeout={self.config.executor.timeout}s 任务队列: {self.task_queue.get_stats() if self.task_queue else 'N/A'} 记忆: {self.memory.get_stats() if self.memory else 'N/A'} 聊天平台: {self.chat_manager.get_stats() if self.chat_manager else 'N/A'} """) def _print_skills(self): print("\n🛠️ 已注册工具/技能:") if self.skill_registry: cats = self.skill_registry.list_skills_by_category() # 内置平台工具 bt = cats.get("builtin_tools", []) if bt: print(f"\n ── 内置平台工具 ({len(bt)}) ──") for s in bt: print(f" ⚙️ {s['name']}: {s.get('description', '')[:60]}") # Python 技能 ps = cats.get("python_skills", []) if ps: print(f"\n ── Python 可执行技能 ({len(ps)}) ──") for s in ps: disabled = " [禁用]" if s.get("disabled") else "" danger = " ⚠️" if s.get("dangerous") else "" print(f" 🔧 {s['name']}: {s.get('description', '')[:60]}{disabled}{danger}") # 技能指南 sg = cats.get("skill_guides", []) if sg: print(f"\n ── 技能指南 / RAG ({len(sg)}) ──") for s in sg: disabled = " [禁用]" if s.get("disabled") else "" print(f" 📝 {s['name']}: {s.get('description', '')[:60]}{disabled}") print() def _print_memory_stats(self): if self.memory: stats = self.memory.get_stats() print(f"\n🧠 记忆系统:") print(f" 会话记忆: {stats.get('session_count', 0)} 条") print(f" 全局记忆: {stats.get('global_count', 0)} 条") print(f" 总计: {stats.get('total_count', 0)} 条") print(f" 会话数: {stats.get('distinct_sessions', 0)}") print() def _print_permissions(self): if not self.permission_manager: print("\n🔑 权限管理器未初始化") print() return pm = self.permission_manager config = pm.get_config_summary() print("\n🔑 权限配置:") print() # 默认权限 defaults = config.get("defaults", {}) labels = config.get("permission_labels", {}) print(" 📋 全局默认权限:") for perm in config.get("all_permissions", []): label = labels.get(perm, perm) status = "✅ 开启" if defaults.get(perm, False) else "❌ 关闭" # Bug Fix: 使用 ASCII 空格而非 Unicode Ideographic Space (U+3000) 进行填充 print(f" {label:<8} ({perm}): {status}") print() # Per-agent 覆盖 agents = config.get("agents", {}) if agents: print(" 🤖 Agent 自定义权限:") for agent_name, perms in agents.items(): overrides = [] for perm in config.get("all_permissions", []): label = labels.get(perm, perm) val = perms.get(perm) if val is not None: status = "✅" if val else "❌" overrides.append(f"{label}={status}") if overrides: print(f" {agent_name}: {', '.join(overrides)}") else: print(f" {agent_name}: (使用默认)") else: print(" (无 Agent 自定义权限，所有 Agent 使用全局默认)") print() def _print_sessions(self): # TODO: 从记忆系统中获取活跃会话列表 print(f"\n📂 当前会话: {self._session_id}") print(" (提示: 使用 'session <名称>' 切换会话)") print() def _cleanup_session_locks(self): """清理过期的会话锁，防止内存泄漏。保留最近 100 个会话锁。""" if len(self._session_locks) > 100: # 按 FIFO 顺序移除最旧的锁 keys = list(self._session_locks.keys()) for k in keys[:len(keys) - 100]: del self._session_locks[k] async def shutdown(self): """关闭所有组件""" self.logger.info("MyAgent 正在关闭...") self._running = False # Bug Fix: 关闭所有反检测浏览器实例，避免 Chrome 进程残留 try: from aiskills.browser_stealth import close_stealth_browser close_stealth_browser(profile_name="") self.logger.info("已关闭所有反检测浏览器实例") except Exception as e: self.logger.debug(f"关闭反检测浏览器时出错: {e}") if self.communication_manager: await self.communication_manager.stop() if self.update_manager: await self.update_manager.stop_auto_check() if self.chat_manager: await self.chat_manager.stop_all() if self.task_queue: await self.task_queue.stop() if self.memory: self.memory.close() # 清理会话锁 self._session_locks.clear() self.logger.info("MyAgent 已关闭") # ============================================================================== # 系统托盘 # ============================================================================== def create_tray_icon(app: MyAgentApp, web_port: int = 8767): """ 创建系统托盘图标。 交互方式: - 左键单击 → 直接打开聊天界面 - 右键单击 → 弹出完整菜单 菜单项: - 打开聊天界面 - 打开管理后台 - 显示状态 (tray 通知) - 打开日志 / 工作目录 / 配置目录 - 设置开机自启 - 重启服务 - 退出 """ try: import pystray from PIL import Image, ImageDraw, ImageFont except Exception: # pystray 可能因缺少 GUI 环境 (X11/Wayland) 或未安装而失败 app.logger.info("pystray 不可用（未安装或缺少 GUI 环境），跳过系统托盘") return None # ── 图标颜色主题 ── _COLOR_NORMAL = "#4A90D9" # 蓝色 = 运行中 _COLOR_ERROR = "#E74C3C" # 红色 = 错误 _COLOR_BUSY = "#F39C12" # 橙色 = 忙碌 # ── 图标缓存 ── _icon_cache = {} def create_icon_image(color: str = _COLOR_NORMAL): """生成托盘图标 (圆角方形机器人头像)""" if color in _icon_cache: return _icon_cache[color] size = 64 img = Image.new("RGBA", (size, size), (0, 0, 0, 0)) draw = ImageDraw.Draw(img) # 圆角矩形背景 margin, radius = 2, 14 draw.rounded_rectangle( [margin, margin, size - margin, size - margin], radius=radius, fill=color, ) # 天线 draw.rounded_rectangle([28, 0, 36, 8], radius=2, fill=color) draw.ellipse([25, 0, 39, 6], fill=color) # 面板 (稍浅的内部) draw.rounded_rectangle( [6, 12, 58, 56], radius=10, fill=_lighten_color(color, 0.15), ) # 眼睛 (带高光) for cx in (22, 42): draw.ellipse([cx - 6, 22, cx + 6, 34], fill="white") draw.ellipse([cx - 3, 25, cx + 3, 31], fill="#333") draw.ellipse([cx - 1, 24, cx + 2, 27], fill="white") # 高光 # 嘴巴 draw.rounded_rectangle([24, 38, 40, 42], radius=2, fill="white") # 耳朵 for x in (0, 54): draw.rounded_rectangle([x, 24, x + 8, 36], radius=3, fill=color) _icon_cache[color] = img return img def _lighten_color(hex_color: str, factor: float) -> str: """将颜色变亮""" hex_color = hex_color.lstrip("#") r, g, b = int(hex_color[0:2], 16), int(hex_color[2:4], 16), int(hex_color[4:6], 16) r = min(255, int(r + (255 - r) * factor)) g = min(255, int(g + (255 - g) * factor)) b = min(255, int(b + (255 - b) * factor)) return f"#{r:02x}{g:02x}{b:02x}" def _open_path(path: str): """跨平台打开文件/目录""" if not os.path.exists(path): icon = _state.get("icon") if icon: try: icon.notify(f"路径不存在: {path}", "MyAgent") except Exception: pass app.logger.warning(f"无法打开不存在的路径: {path}") return import subprocess import shutil import platform as pf system = pf.system() try: if system == "Windows": os.startfile(path) # type: ignore elif system == "Darwin": subprocess.Popen(["open", path]) else: # [v1.47.12] Termux 环境: 优先使用 termux-open-url termux_open = shutil.which("termux-open-url") if termux_open: subprocess.Popen([termux_open, path], start_new_session=True) else: subprocess.Popen(["xdg-open", path]) except Exception as e: app.logger.warning(f"无法打开路径 {path}: {e}") # ── 状态追踪 (供动态菜单使用) ── _state = { "autostart_enabled": app.config.tray.auto_start if hasattr(app.config, 'tray') else False, "icon": None, # 由 run_with_tray 设置反向引用 } # ── 菜单动作 ── def open_web_ui(icon, item): _open_browser_kiosk(f"http://127.0.0.1:{web_port}/ui/") def open_chat_ui(icon, item): _open_browser_kiosk(f"http://127.0.0.1:{web_port}/ui/chat/chat_container.html") def open_logs(icon, item): _open_path(str(app.config_mgr.logs_dir)) def open_workdir(icon, item): wd = str(app.config_mgr.data_dir / "workspace") Path(wd).mkdir(parents=True, exist_ok=True) _open_path(wd) def open_config_dir(icon, item): _open_path(str(app.config_mgr.data_dir)) def show_status(icon, item): """以 tray 通知显示状态""" try: lines = [] if app.main_agent: stats = app.main_agent.get_stats() if isinstance(stats, dict): lines.append(f"对话: {stats.get('total_messages', 0)} 条") lines.append(f"会话: {stats.get('total_sessions', 0)} 个") if app.memory: mem = app.memory.get_stats() if isinstance(mem, dict): lines.append(f"记忆: {mem.get('total_count', 0)} 条") if app.task_queue: tq = app.task_queue.get_stats() if isinstance(tq, dict): pending = tq.get("pending", 0) running = tq.get("running", 0) lines.append(f"队列: {running} 运行, {pending} 等待") if not lines: lines.append("系统正常运行中") icon.notify("\n".join(lines), "MyAgent 状态") except Exception as e: icon.notify(f"获取状态失败: {e}", "MyAgent") def toggle_autostart(icon, item): new_state = not _state["autostart_enabled"] _state["autostart_enabled"] = new_state try: setup_auto_start(new_state) icon.notify( "已开启开机自启" if new_state else "已关闭开机自启", "MyAgent", ) except Exception as e: icon.notify(f"设置失败: {e}", "MyAgent") _state["autostart_enabled"] = not new_state # 更新菜单 (pystray 需要重建菜单来刷新 checked 状态) _refresh_menu(icon) def restart_service(icon, item): """重启 Web 服务""" try: icon.notify("正在重启并重新加载配置...", "MyAgent") app._restart_requested = True except Exception as e: icon.notify(f"重启失败: {e}", "MyAgent") def stop_service(icon, item): """停止 Web 服务""" try: app._stop_requested = True # 为了 UI 响应迅速，预设状态并刷新菜单 app._running_service = False _refresh_menu(icon) app.logger.info("用户请求停止服务") except Exception as e: icon.notify(f"停止失败: {e}", "MyAgent") def start_service(icon, item): """启动 Web 服务""" try: app._start_requested = True # 为了 UI 响应迅速，预设状态并刷新菜单 app._running_service = True _refresh_menu(icon) app.logger.info("用户请求启动服务") except Exception as e: icon.notify(f"启动失败: {e}", "MyAgent") def on_quit(icon, item): """退出应用""" try: app._running = False app.logger.info("从托盘菜单退出") except Exception: pass icon.stop() def _get_status_title(item): """动态生成状态标题 (pystray text callable: text(menu_item))""" if not app._running: return "MyAgent - 已停止" return "MyAgent - 运行中" def _autostart_checked(item): """返回当前自启状态的 checked 值""" return _state["autostart_enabled"] def _refresh_menu(icon): """刷新托盘菜单 (重建以更新动态状态)""" try: icon.menu = _build_menu() icon.update_menu() except Exception: pass # 部分平台不支持动态菜单刷新 def _build_menu(): """构建菜单 (Windows Win32 不支持 BMP 外 emoji, 使用纯文本)""" return pystray.Menu( pystray.MenuItem( _get_status_title, None, enabled=False, ), pystray.Menu.SEPARATOR, pystray.MenuItem("打开聊天界面", open_chat_ui, default=True), pystray.MenuItem("打开管理后台", open_web_ui), pystray.MenuItem("显示状态", show_status), pystray.Menu.SEPARATOR, pystray.MenuItem("打开工作目录", open_workdir), pystray.MenuItem("打开日志目录", open_logs), pystray.MenuItem("打开配置目录", open_config_dir), pystray.Menu.SEPARATOR, pystray.MenuItem("开机自启", toggle_autostart, checked=_autostart_checked), pystray.Menu.SEPARATOR, pystray.MenuItem("启动服务", start_service, visible=lambda item: not app._running_service), pystray.MenuItem("停止服务", stop_service, visible=lambda item: app._running_service), pystray.MenuItem("重启服务", restart_service), pystray.Menu.SEPARATOR, pystray.MenuItem("退出", on_quit), ) tray_icon = pystray.Icon( "myagent", create_icon_image(), f"MyAgent v{get_version()}", _build_menu(), ) # ── 单击直接打开聊天界面 ── # 重写 __call__: pystray 默认行为是单击弹出菜单，改为直接打开聊天界面。 # Windows 下右键仍由 _on_notify 独立处理弹出菜单，不受影响。 # X11/Linux 下菜单本身不原生支持，所以直接跳过菜单。 tray_icon.__call__ = lambda icon: open_chat_ui(icon, None) # 保存反向引用供外部使用 _state["icon"] = tray_icon app._tray_icon = tray_icon app._tray_state = _state app._tray_create_icon = create_icon_image # 供动态更新图标使用 return tray_icon def run_with_tray(app: MyAgentApp, web_port: int = 8767): """在系统托盘中运行 MyAgent""" tray = create_tray_icon(app, web_port) if tray is None: app.logger.warning("pystray 未安装，跳过系统托盘") return # 在后台线程运行托盘 (daemon=True 保证崩溃不影响主进程) def _tray_run(): try: tray.run() except Exception as e: app.logger.warning(f"系统托盘异常退出 (不影响主服务): {e}") tray_thread = threading.Thread(target=_tray_run, daemon=True) tray_thread.start() app.logger.info(f"系统托盘已启动 (聊天: http://127.0.0.1:{web_port}/ui/chat/chat_container.html)") # 启动完成通知 _tray_notify(app, "MyAgent 已启动", f"v{get_version()} | 端口: {web_port}") def _tray_notify(app: MyAgentApp, title: str, message: str): """发送系统托盘通知 (气球弹窗)""" if not hasattr(app.config, 'tray') or not app.config.tray.show_notifications: return tray_icon = getattr(app, '_tray_icon', None) if tray_icon: try: tray_icon.notify(message, title) except Exception: pass # 某些平台/桌面环境不支持通知 # ============================================================================== # 开机自启设置 # ============================================================================== def setup_auto_start(enable: bool = True): """ 设置开机自启。 Windows: 添加到启动文件夹 macOS: 使用 launchd (plist) Linux: 使用 systemd user service """ import platform system = platform.system() try: if system == "Windows": _setup_autostart_windows(enable) elif system == "Darwin": _setup_autostart_macos(enable) elif system == "Linux": _setup_autostart_linux(enable) except Exception as e: print(f"⚠️ 设置开机自启失败: {e}") def _setup_autostart_windows(enable: bool): """Windows: 启动文件夹快捷方式""" import winreg import os key_path = r"Software\Microsoft\Windows\CurrentVersion\Run" key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, key_path, 0, winreg.KEY_SET_VALUE) if enable: exe_path = os.path.abspath(sys.argv[0]) winreg.SetValueEx(key, "MyAgent", 0, winreg.REG_SZ, exe_path) print("✅ 已设置 Windows 开机自启") else: try: winreg.DeleteValue(key, "MyAgent") print("✅ 已取消 Windows 开机自启") except FileNotFoundError: print("ℹ️ 未找到开机自启项") winreg.CloseKey(key) def _setup_autostart_macos(enable: bool): """macOS: LaunchAgent plist""" import plistlib plist_dir = Path.home() / "Library" / "LaunchAgents" plist_dir.mkdir(parents=True, exist_ok=True) plist_path = plist_dir / "com.myagent.plist" if enable: plist_data = { "Label": "com.myagent", "ProgramArguments": [sys.executable, str(PROJECT_ROOT / "main.py"), "--tray"], "RunAtLoad": True, "KeepAlive": False, } with open(plist_path, "wb") as f: plistlib.dump(plist_data, f) print(f"✅ 已设置 macOS 开机自启: {plist_path}") else: if plist_path.exists(): plist_path.unlink() print("✅ 已取消 macOS 开机自启") def _setup_autostart_linux(enable: bool): """Linux: systemd user service""" service_dir = Path.home() / ".config" / "systemd" / "user" service_dir.mkdir(parents=True, exist_ok=True) service_path = service_dir / "myagent.service" if enable: service_content = f"""[Unit] Description=MyAgent AI Assistant After=network.target [Service] Type=simple ExecStart={sys.executable} {PROJECT_ROOT / 'main.py'} --tray WorkingDirectory={PROJECT_ROOT} Restart=on-failure [Install] WantedBy=default.target """ with open(service_path, "w") as f: f.write(service_content) print(f"✅ 已创建 systemd service: {service_path}") print(" 启用命令: systemctl --user enable myagent.service") else: if service_path.exists(): service_path.unlink() print("✅ 已取消 Linux 开机自启") # ============================================================================== # 首次运行引导 # ============================================================================== def _first_run_setup(config_mgr: ConfigManager): """ 首次运行引导：检测是否已配置 LLM API Key。 如果没有，进入交互式配置向导。 """ cfg = config_mgr.config # 判断是否需要配置: api_key 为空且 provider 不是 ollama（ollama 本地无需 key） has_api_key = bool(cfg.llm.api_key) or bool(cfg.llm.anthropic_api_key) is_ollama = cfg.llm.provider == "ollama" has_env_key = bool(os.environ.get("MYAGENT_LLM_API_KEY")) or \ bool(os.environ.get("MYAGENT_ANTHROPIC_API_KEY")) or \ bool(os.environ.get("OPENAI_API_KEY")) or \ bool(os.environ.get("ANTHROPIC_API_KEY")) if has_api_key or is_ollama or has_env_key: return # 已配置或通过环境变量提供，跳过 # 进入配置向导 print() print("=" * 55) print(" MyAgent - LLM API 配置向导") print(f" 版本: v{get_version()}") print("=" * 55) print() print(" 检测到尚未配置 LLM API Key，MyAgent 需要") print(" 一个大模型 API 才能工作。") print() print(" 支持的提供商:") print(" 1. OpenAI / 兼容 API (推荐)") print(" 2. Anthropic (Claude)") print(" 3. Ollama (本地模型，无需 API Key)") print(" 4. 自定义 / Custom (任意 OpenAI 兼容接口)") print() while True: try: choice = input(" 请选择 [1/2/3/4]: ").strip() except (EOFError, KeyboardInterrupt): print("\n\n 已取消配置。可通过 myagent-ai setup 重新配置。") sys.exit(0) if choice == "1": _setup_openai_compatible(config_mgr) break elif choice == "2": _setup_anthropic(config_mgr) break elif choice == "3": _setup_ollama(config_mgr) break elif choice == "4": _setup_custom(config_mgr) break else: print(" 无效选择，请输入 1/2/3/4") # 保存配置 config_mgr.save() print() print(" 配置已保存!") print(f" 配置文件: {config_mgr._config_file}") print() def _setup_openai_compatible(config_mgr: ConfigManager): """配置 OpenAI 或兼容 API""" print() print(" -- OpenAI / 兼容 API 配置 --") print() # 预设模板 print(" 预设模板:") print(" 1. OpenAI 官方 (https://api.openai.com/v1)") print(" 2. 智谱 GLM (https://open.bigmodel.cn/api/paas/v4)") print(" 3. DeepSeek (https://api.deepseek.com/v1)") print(" 4. 月之暗面 Moonshot (https://api.moonshot.cn/v1)") print(" 5. 自定义 Base URL") print() while True: try: choice = input(" 选择模板 [1/2/3/4/5]: ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) templates = { "1": ("https://api.openai.com/v1", "gpt-4o"), "2": ("https://open.bigmodel.cn/api/paas/v4", "glm-4-flash"), "3": ("https://api.deepseek.com/v1", "deepseek-chat"), "4": ("https://api.moonshot.cn/v1", "moonshot-v1-8k"), } if choice in templates: base_url, model = templates[choice] break elif choice == "5": try: base_url = input(" Base URL: ").strip().rstrip("/") model = input(" 模型名称 (如 gpt-4o): ").strip() or "gpt-4o" except (EOFError, KeyboardInterrupt): sys.exit(0) break else: print(" 无效选择") while True: try: api_key = input(" API Key: ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) if api_key: break print(" API Key 不能为空!") print() try: temp_input = input(f" Temperature (默认 0.1, 直接回车跳过): ").strip() temperature = float(temp_input) if temp_input else 0.1 except (EOFError, KeyboardInterrupt, ValueError): temperature = 0.1 # 写入配置 config_mgr.config.llm.provider = "openai" config_mgr.config.llm.api_key = api_key config_mgr.config.llm.base_url = base_url config_mgr.config.llm.model = model config_mgr.config.llm.temperature = temperature # 验证连接 print() print(" 正在验证连接...") try: import requests headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"} test_body = { "model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 5, } resp = requests.post( f"{base_url}/chat/completions", headers=headers, json=test_body, timeout=15, ) if resp.status_code == 200: print(" 连接成功!") else: print(f" API 返回 {resp.status_code}，请检查 Key 和 URL 是否正确") print(f" (已保存配置，启动后可在 Web 界面修改)") except Exception as e: print(f" 连接测试失败: {e}") print(f" (已保存配置，启动后可在 Web 界面修改)") def _setup_anthropic(config_mgr: ConfigManager): """配置 Anthropic Claude""" print() print(" -- Anthropic Claude 配置 --") print() while True: try: api_key = input(" Anthropic API Key: ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) if api_key: break print(" API Key 不能为空!") print() print(" 可选模型: claude-sonnet-4-20250514 / claude-opus-4-20250514") try: model = input(" 模型 (默认 claude-sonnet-4-20250514): ").strip() or "claude-sonnet-4-20250514" except (EOFError, KeyboardInterrupt): model = "claude-sonnet-4-20250514" config_mgr.config.llm.provider = "anthropic" config_mgr.config.llm.anthropic_api_key = api_key config_mgr.config.llm.model = model print() print(" 正在验证连接...") try: import requests headers = { "x-api-key": api_key, "anthropic-version": "2023-06-01", "Content-Type": "application/json", } test_body = { "model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 5, } resp = requests.post( "https://api.anthropic.com/v1/messages", headers=headers, json=test_body, timeout=15, ) if resp.status_code == 200: print(" 连接成功!") else: print(f" API 返回 {resp.status_code}，请检查 Key 是否正确") print(f" (已保存配置，启动后可在 Web 界面修改)") except Exception as e: print(f" 连接测试失败: {e}") print(f" (已保存配置，启动后可在 Web 界面修改)") def _setup_ollama(config_mgr: ConfigManager): """配置 Ollama 本地模型""" print() print(" -- Ollama 本地模型配置 --") print() try: base_url = input(" Ollama 地址 (默认 http://localhost:11434): ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) base_url = base_url or "http://localhost:11434" try: model = input(" 模型名称 (默认 llama3): ").strip() or "llama3" except (EOFError, KeyboardInterrupt): model = "llama3" config_mgr.config.llm.provider = "ollama" config_mgr.config.llm.ollama_base_url = base_url config_mgr.config.llm.ollama_model = model print() print(" 正在验证连接...") try: import requests resp = requests.get(f"{base_url}/api/tags", timeout=5) if resp.status_code == 200: models = [m.get("name", "") for m in resp.json().get("models", [])] print(f" 连接成功! 可用模型: {', '.join(models) if models else '(无)'}") else: print(f" 无法连接 Ollama，请确认 Ollama 已启动") print(f" (已保存配置，启动后可在 Web 界面修改)") except Exception as e: print(f" 连接测试失败: {e}") print(f" (已保存配置，启动后可在 Web 界面修改)") def _setup_custom(config_mgr: ConfigManager): """配置自定义 / Custom 模型 (OpenAI 兼容接口)""" print() print(" -- 自定义 / Custom 模型配置 --") print() print(" 支持任意 OpenAI API 兼容接口，包括:") print(" - 中转站 / 代理 API") print(" - 自部署的 vLLM / LocalAI") print(" - 企业内部 AI 网关") print() try: name = input(" 模型显示名称 (如: 我的 GPT 镜像): ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) while True: try: base_url = input(" Base URL (如 https://your-api.com/v1): ").strip().rstrip("/") except (EOFError, KeyboardInterrupt): sys.exit(0) if base_url: break print(" Base URL 不能为空!") while True: try: model = input(" 模型名称 (如 gpt-4o / deepseek-chat): ").strip() except (EOFError, KeyboardInterrupt): sys.exit(0) if model: break print(" 模型名称不能为空!") try: api_key = input(" API Key (选填，直接回车跳过): ").strip() except (EOFError, KeyboardInterrupt): api_key = "" # 写入配置 config_mgr.config.llm.provider = "custom" config_mgr.config.llm.api_key = api_key config_mgr.config.llm.base_url = base_url config_mgr.config.llm.model = model # 同时添加到模型库 model_id = f"custom-{model.replace('/', '-').lower()}" from config import ModelEntry entry = ModelEntry( id=model_id, name=name or f"Custom - {model}", provider="custom", model=model, base_url=base_url, api_key=api_key, max_tokens=4096, temperature=0.1, enabled=True, ) # 避免重复 existing_ids = [m.id for m in config_mgr.config.models_library] if model_id not in existing_ids: config_mgr.config.models_library.append(entry) print() print(" 正在验证连接...") try: import requests headers = {} if api_key: headers["Authorization"] = f"Bearer {api_key}" test_body = { "model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 5, } resp = requests.post( f"{base_url}/chat/completions", headers=headers, json=test_body, timeout=15, ) if resp.status_code == 200: print(" 连接成功!") else: print(f" 连接失败 (HTTP {resp.status_code})，请检查 Base URL 和模型名称") print(f" (已保存配置，启动后可在 Web 界面修改)") except Exception as e: print(f" 连接测试失败: {e}") print(f" (已保存配置，启动后可在 Web 界面修改)") # ============================================================================== # 主函数 # ============================================================================== def main(): """主入口""" import argparse parser = argparse.ArgumentParser(description="MyAgent - 本地桌面端执行型AI助手") parser.add_argument("--tray", action="store_true", help="以系统托盘模式运行") parser.add_argument("--web", type=int, nargs="?", const=8767, default=None, help="启动管理后台 Web UI (可选端口，默认8767)") parser.add_argument("--server", action="store_true", help="API 服务模式 (等同于 --web，绑定 0.0.0.0)") parser.add_argument("--host", type=str, default=None, help="Web UI 绑定地址 (默认 127.0.0.1，容器部署用 0.0.0.0，混合模式用 mixed)") parser.add_argument("--port", type=int, default=8767, help="Web UI 端口") parser.add_argument("--autostart", action="store_true", help="设置开机自启") parser.add_argument("--no-autostart", action="store_true", help="取消开机自启") parser.add_argument("--config", type=str, help="指定配置文件路径") parser.add_argument("--debug", action="store_true", help="调试模式") parser.add_argument("--setup", action="store_true", help="配置向导") args = parser.parse_args() # 配置 config_mgr = get_config() if args.debug: config_mgr.config.log_level = "DEBUG" if args.config: config_mgr._config_file = Path(args.config) config_mgr.load() # 开机自启 if args.autostart: setup_auto_start(True) return if args.no_autostart: setup_auto_start(False) return # 手动触发配置向导 if args.setup: _first_run_setup(config_mgr) return # ══════════════════════════════════════════════════════════ # 首次运行: 引导配置 LLM API Key # ══════════════════════════════════════════════════════════ # --server 等同于 --web，但默认绑定 0.0.0.0（用于容器/Docker 部署） if args.server and not args.web: args.web = args.port if args.server and not args.host: args.host = "0.0.0.0" web_port = args.web if args.web else args.port web_host = args.host or "mixed" if not web_port: # 仅 CLI 模式才用命令行配置向导 # Web 模式下由 chat.html 的 Web 向导处理 (检测 /api/setup/status) _first_run_setup(config_mgr) # 创建应用 app = MyAgentApp() app._restart_requested = False # 信号处理 — 仅设置标志, 不在信号处理器中调用 asyncio def signal_handler(sig, frame): app._running = False # 停止 tray (如果在运行) tray_icon = getattr(app, '_tray_icon', None) if tray_icon: try: tray_icon.stop() except Exception: pass signal.signal(signal.SIGINT, signal_handler) signal.signal(signal.SIGTERM, signal_handler) # 启动 async def run(): await app.initialize() # Web 管理后台 api_server = None if web_port: from web.api_server import ApiServer api_server = ApiServer(app) await api_server.start(port=web_port, host=web_host) app.logger.info(f"管理后台: http://127.0.0.1:{web_port}/ui/") # 非 tray 模式（纯 web 模式）自动打开浏览器 if not args.tray: try: _open_browser_kiosk(f"http://127.0.0.1:{web_port}/ui/chat/chat_container.html") except Exception: pass # 启动聊天平台(后台) if app.chat_manager: asyncio.create_task(app.chat_manager.start_all()) # 启动自动更新检查（后台） if app.update_manager: asyncio.create_task(app.update_manager.start_auto_check(interval=600)) app._running_service = web_port or app.chat_manager if args.tray: # 托盘模式: 后台等待 app.logger.info("后台运行中... (Ctrl+C 退出)") while app._running: await asyncio.sleep(1) # A. 处理停止请求 if getattr(app, '_stop_requested', False): app._stop_requested = False if app._running_service: app.logger.info("正在停止所有后台服务...") if api_server: try: await api_server.stop() except Exception: pass api_server = None if app.chat_manager: try: await app.chat_manager.stop_all() except Exception: pass app._running_service = False _tray_notify(app, "服务已停止", "管理后台与聊天平台已关闭") # B. 处理启动请求 if getattr(app, '_start_requested', False): app._start_requested = False if not app._running_service: app.logger.info("正在启动服务...") try: if web_port: from web.api_server import ApiServer api_server = ApiServer(app) await api_server.start(port=web_port, host=web_host) if app.chat_manager: asyncio.create_task(app.chat_manager.start_all()) app._running_service = True _tray_notify(app, "服务已启动", f"管理后台: http://127.0.0.1:{web_port}/ui/") except Exception as e: app.logger.error(f"启动失败: {e}") _tray_notify(app, "启动失败", str(e)) # C. 检查重启请求 if getattr(app, '_restart_requested', False): app._restart_requested = False app.logger.info("收到重启请求，正在重新加载所有组件...") # 1. 停止现有服务 if api_server: try: await api_server.stop() except Exception: pass if app.chat_manager: try: await app.chat_manager.stop_all() except Exception: pass # 2. 重新加载配置并初始化核心 try: app.config_mgr.reload() app.config = app.config_mgr.config await app.initialize() # 3. 启动 Web 服务 if web_port: from web.api_server import ApiServer api_server = ApiServer(app) await api_server.start(port=web_port, host=web_host) app.logger.info(f"Web 服务已重启: http://127.0.0.1:{web_port}/ui/") # 4. 启动聊天平台 if app.chat_manager: asyncio.create_task(app.chat_manager.start_all()) app._running_service = True _tray_notify(app, "MyAgent 已重启", f"核心配置已重新加载") except Exception as e: app.logger.error(f"重启过程中发生错误: {e}", exc_info=True) _tray_notify(app, "重启失败", str(e)) elif web_port: # Web 模式 (无托盘): 后台运行保持服务 app.logger.info("Web 服务运行中... (Ctrl+C 退出)") while app._running: await asyncio.sleep(1) else: # CLI 模式 await app.run_cli() await app.shutdown() # 停止 tray tray_icon = getattr(app, '_tray_icon', None) if tray_icon: try: tray_icon.stop() except Exception: pass if api_server: try: await api_server.stop() except Exception: pass try: asyncio.run(run()) except KeyboardInterrupt: pass finally: if app._running: try: asyncio.run(app.shutdown()) except RuntimeError: pass # event loop 可能已关闭 if __name__ == "__main__": main()