From 791fac0efd5f4049a4a782d8e9fa8c2f6a0028be Mon Sep 17 00:00:00 2001 From: qing <1776177662@qq.com> Date: Thu, 13 Aug 2026 16:09:14 +0800 Subject: [PATCH] =?UTF-8?q?refactor(aibox):=20=E6=8A=BD=E7=A6=BBB=E6=89=B9?= =?UTF-8?q?12=E4=B8=AA=E5=B7=A5=E5=85=B7=E7=B1=BB=E8=87=B3tools=E5=8C=85?= =?UTF-8?q?=EF=BC=8C=E9=99=8458=E4=B8=AA=E5=8D=95=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit B批（tools/）：迁移依赖子系统的12个工具类（CommandLineTool/SearchSkills/ReloadSkills/InstallSkill/CreateSkill/MemoryTool/MemoTool/UpdateIdentity/UpdateSoul/SaveMemoryAndEndConversation/SetProxy/GetProxy/AskUser），aibox.py减少~1739行 修复MemoryDatabase.get_recent_memories(limit=None)的SQL错误（既有bug，None传给LIMIT ?导致datatype mismatch） 171个单测全绿，循环依赖障碍消除：B批工具直接依赖subsystems而非反向依赖aibox.py --- aibox/aibox.py | 1746 +---------------------------------- aibox/subsystems/db.py | 7 +- aibox/tests/test_b_tools.py | 439 +++++++++ aibox/tools/__init__.py | 44 +- aibox/tools/command_tool.py | 250 +++++ aibox/tools/db_tools.py | 818 ++++++++++++++++ aibox/tools/misc_tools.py | 48 + aibox/tools/prompt_tools.py | 252 +++++ aibox/tools/proxy_tools.py | 81 ++ aibox/tools/skill_tools.py | 400 ++++++++ 10 files changed, 2340 insertions(+), 1745 deletions(-) create mode 100644 aibox/tests/test_b_tools.py create mode 100644 aibox/tools/command_tool.py create mode 100644 aibox/tools/db_tools.py create mode 100644 aibox/tools/misc_tools.py create mode 100644 aibox/tools/prompt_tools.py create mode 100644 aibox/tools/proxy_tools.py create mode 100644 aibox/tools/skill_tools.py diff --git a/aibox/aibox.py b/aibox/aibox.py index 4ad7bc1..d7e44a0 100644 --- a/aibox/aibox.py +++ b/aibox/aibox.py @@ -47,6 +47,13 @@ from tools import ( Tool, ToolRegistry, FileToolBase, ReadFileTool, WriteFileTool, ListFilesTool, SimpleWebCrawlerTool, GetCurrentTimeTool, CalculatorTool, + # B 批工具 + CommandLineTool, + SearchSkillsTool, ReloadSkillsTool, InstallSkillTool, CreateSkillTool, + MemoryTool, MemoTool, + UpdateIdentityTool, UpdateSoulTool, SaveMemoryAndEndConversationTool, + SetProxyTool, GetProxyTool, + AskUserTool, ) # 子系统（Wave 1 抽离）：命令执行器 / 提示词管理器 / 数据库 / 技能管理器 @@ -1299,239 +1306,6 @@ class ConversationHistory: -class SearchSkillsTool(Tool): - """扩展技能搜索工具 - 按关键字搜索 ~/.aibox/skills/ 下的技能文件。 - - 支持两种技能格式： - 1. 可执行脚本/程序：通过 --help 获取简介和完整使用说明 - 2. Markdown 文件 (.md)：第一行为简介，其余为技能内容 - - 工作流程： - 1. 用关键字调用本工具，获取匹配技能的名称和简介列表 - 2. 对于 Markdown 技能，可以直接读取 content 字段获取完整技能内容 - 3. 对于可执行技能，通过 create_subagent 派生子代理执行 "技能名 --help" 查看完整命令手册 - 4. 根据内容，使用相应方式调用技能 - """ - - def __init__(self, skill_manager: SkillManager, logger: Optional[Logger] = None): - super().__init__(logger) - self.skill_manager = skill_manager - - def get_name(self) -> str: - return "search_skills" - - def get_description(self) -> str: - return ( - "扩展技能搜索工具。\n\n" - "用关键字搜索 ~/.aibox/skills/ 目录下的技能文件。\n" - "支持两种技能格式：\n" - " 1. 可执行脚本/程序：通过 --help 获取简介和完整使用说明\n" - " 2. Markdown 文件 (.md)：第一行为简介，其余为技能内容\n\n" - "使用方法：\n" - "1. 调用本工具搜索关键字，获取匹配技能列表（空字符串列出全部）。\n" - "2. 对于 Markdown 技能：设置 show_content=true 读取返回的内容获取完整技能说明。\n" - "3. 对于可执行技能：用 create_subagent 派生子代理执行 \"技能名 --help\" 获取完整命令手册，再按手册调用该技能。\n" - "4. 按照内容说明，使用相应方式调用该技能（执行类操作一律经 create_subagent 派生子代理完成）。" - ) - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "keyword": { - "type": "string", - "description": ( - "搜索关键字，在技能名称、简介和内容中进行匹配。" - "传入空字符串可列出全部可用技能。" - ) - }, - "show_content": { - "type": "boolean", - "description": ( - "是否在结果中显示技能内容。" - "对于 Markdown 技能，这会影响返回结果的详细程度。" - "默认 False（只显示名称和简介）。" - ), - "default": False - } - }, - "required": ["keyword"] - } - - def execute(self, keyword: str = "", show_content: bool = False, **kwargs) -> str: - kw = keyword.strip() - if kw: - results = self.skill_manager.search(kw) - header = f"🔍 搜索 '{kw}' 的结果（共 {len(results)} 个匹配技能）：" - empty_msg = ( - f"❌ 未找到与 '{kw}' 相关的技能。\n" - "提示：传入空字符串可列出所有可用技能。" - ) - else: - results = self.skill_manager.get_all() - header = f"🔧 所有可用技能（共 {len(results)} 个）：" - empty_msg = "❌ ~/.aibox/skills/ 目录下暂无可用技能。" - - if not results: - return empty_msg - - lines = [header] - for item in results: - skill_type = "📄 Markdown" if item.get('is_markdown') else "⚙️ 可执行" - lines.append(f"\n {skill_type} • {item['name']}") - lines.append(f" 简介: {item['summary']}") - - if show_content and item.get('content'): - content = item['content'] - if item.get('is_markdown'): - preview = content[:500] - if len(content) > 500: - preview += "...\n （内容较长，已截断，请直接读取技能文件获取完整内容）" - lines.append(f" 内容预览:\n {preview}") - else: - preview = content[:200] if content else "(无帮助输出)" - if len(content) > 200: - preview += "..." - lines.append(f" 帮助预览: {preview}") - else: - if item.get('is_markdown'): - lines.append(f" 💡 提示: 设置 show_content=true 可查看技能内容") - else: - lines.append(f" 💡 提示: 用 create_subagent 派生子代理执行 \"{item['name']} --help\" 查看完整手册") - - lines.append("") - lines.append("📖 技能调用方式:") - lines.append(" • Markdown 技能: 设置 show_content=true 读取完整技能说明") - lines.append(" • 可执行技能: 用 create_subagent 派生子代理执行（先 '技能名 --help' 查手册，再按手册调用）") - return "\n".join(lines) - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行搜索技能""" - keyword = kwargs.get("keyword", "") - show_content = kwargs.get("show_content", False) - - output_queue.put(("line", f"\n🔍 搜索技能: {keyword if keyword else '(全部)'}\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - # 直接传递 kwargs，避免重复参数 - result = self.execute(**kwargs) - - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - - -# ==================== 技能热重载工具 ==================== - -class ReloadSkillsTool(Tool): - """技能热重载工具 - 重新加载所有扩展技能""" - - def __init__(self, tool_registry: ToolRegistry, logger: Optional[Logger] = None): - super().__init__(logger) - self.tool_registry = tool_registry - - def get_name(self) -> str: - return "reload_skills" - - def get_description(self) -> str: - return """技能热重载工具 - 重新加载所有扩展技能 -当你添加、修改或删除了 ~/.aibox/skills/ 目录下的技能文件后， -调用此工具可以立即生效，无需重启系统。""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "confirm": { - "type": "boolean", - "description": "确认重新加载技能", - "default": True - } - } - } - - def execute(self, confirm: bool = True, **kwargs) -> str: - """执行技能热重载""" - if not confirm: - return "❌ 技能重载已取消，如需重载请设置 confirm=true" - - self.logger.info("开始执行技能热重载...") - - try: - count = self.tool_registry.reload_skills() - - # 从 SkillManager 获取更新后的技能列表（含简介） - skill_manager = self.tool_registry.skill_manager - if skill_manager: - all_skills = skill_manager.get_all() - skill_lines = "\n".join( - f" • {s['name']} — {s['summary']}" - for s in all_skills[:10] - ) - data = { - "重新加载的技能数量": count, - "技能目录": constants.SKILLS_DIR, - "技能列表（含简介）": skill_lines or "（无技能）", - } - if count > 10: - data["提示"] = f"还有 {count - 10} 个技能未显示" - else: - data = { - "重新加载的技能数量": count, - "技能目录": constants.SKILLS_DIR, - } - - return self.format_result(True, i18n.get('skills_reloaded', count), data) - - except Exception as e: - self.logger.error(f"技能热重载失败: {e}") - return i18n.get('skills_reload_failed', str(e)) - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行技能热重载""" - output_queue.put(("line", "\n🔄 执行技能热重载...\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - result = self.execute(**kwargs) - - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - - -# ==================== 询问工具 ======================== -class AskUserTool(Tool): - """智能体遇到了麻烦，主动向用户问一个问题""" - - def get_name(self) -> str: - return "ask_user" - - def get_description(self) -> str: - return ( - "主动向用户提问的工具。你是用户的专属助手，应当主动关心用户、澄清需求、确认意图，" - "而不是被动等待指令。适用场景：指令模糊需要澄清、任务完成后追问后续需求、" - "关心用户近况、主动发现用户可能需要的帮助。请积极使用此工具与用户互动。" - ) - - def get_parameters(self) -> Dict: - ''' TODO: 调用子代理生成提示词 ''' - return { - "type": "object", - "properties": { - "question": { - "type": "string", - "description": ( - "你现在遇到一个问题，你想询问用户，现在提供你的问题。" - ) - } - }, - "required": [] - } - - def execute(self, question: str, **kwargs) -> str: - return ( - "你遇到了想搞清楚的事情: " + question + "\n你现在可以询问用户。" - ) - -# -------------------- 情绪化技能工具 -------------------- class ExpressEmotionTool(Tool): def get_name(self) -> str: return "express_emotion" @@ -1614,1447 +1388,9 @@ class ExpressEmotionTool(Tool): # ==================== 安装技能工具 ==================== -class InstallSkillTool(Tool): - """技能安装指引工具 - 返回如何将可执行文件/脚本安装为扩展技能的说明""" - - def get_name(self) -> str: - return "install_skill" - - def get_description(self) -> str: - return ( - "技能安装指引工具。\n" - "返回将 Markdown 技能文件安装为扩展技能的完整步骤说明，" - "包括文件检查、格式校验和部署方法（目标文件名统一带 .md 扩展名）。" - ) - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "skill_path": { - "type": "string", - "description": ( - "待安装的技能文件路径（可选）。" - "若提供，说明中会带入该路径作为示例；" - "不提供则返回通用说明。" - ) - } - }, - "required": [] - } - - def execute(self, skill_path: str = "", **kwargs) -> str: - skills_dir = constants.SKILLS_DIR - - # 如果给了具体路径，在说明里带入，否则用占位符 - src = skill_path.strip() if skill_path else "/path/to/your_skill.md" - # 去掉扩展名作为技能名称，再统一补 .md 作为目标文件名 - base = os.path.splitext(os.path.basename(src))[0] if src else "your_skill" - dst = os.path.join(skills_dir, base + ".md") - - guide = f"""📦 扩展技能安装指引 -{"=" * 55} - -要将一个 Markdown 技能文件安装为扩展技能，请按以下步骤操作： - -───────────────────────────────────────────────────── -步骤 1：确认文件存在且格式正确 -───────────────────────────────────────────────────── -调用 create_subagent 派生子代理执行以下命令，确认文件存在： - - ls -l {src} - -技能文件格式要求： - • 文件名为纯文本 Markdown，第一行是一级标题（# 技能名），作为技能摘要 - • 其余内容为技能的执行说明、使用方法和示例 - • 文件应带 .md 扩展名（系统以 .md 文件作为技能加载） - -───────────────────────────────────────────────────── -步骤 2：复制到技能目录并统一为 .md 扩展名 -───────────────────────────────────────────────────── -技能目录：{skills_dir} - -调用 create_subagent 派生子代理执行复制命令（目标文件名统一补 .md 扩展名）： - - cp {src} {dst} - -⚠️ 注意：目标文件名必须带 .md 扩展名， - 系统以 .md 文件作为技能加载，且技能名称取文件名（不含 .md）。 - -───────────────────────────────────────────────────── -步骤 3：让系统识别新技能 -───────────────────────────────────────────────────── -复制完成后，调用热重载工具使其立即生效： - - reload_skills confirm=true - -之后可用 search_skills 验证技能已加载： - - search_skills keyword="{base}" - -{"=" * 55} -完成以上步骤后，即可通过该技能执行相应操作。""" - - return guide -# ==================== 创建扩展技能工具 ==================== -class CreateSkillTool(Tool): - """创建扩展技能工具 - 返回提示词，驱动智能体自行编写并部署 Markdown 技能文件。 - - 工具本身不调用 LLM，只输出一段结构化的指令提示词。 - 智能体读到提示词后，会自行： - 1. 编写满足格式要求的 Markdown 技能文件 - 2. 经 create_subagent 派生子代理写入 ~/.aibox/skills/.md - 3. 调用 reload_skills 热重载 - """ - - def get_name(self) -> str: - return "create_skill" - - def get_description(self) -> str: - return ( - "创建扩展技能工具。\n" - "返回一段指令提示词，驱动你自行编写并部署一个 Markdown 技能文件（带 .md 扩展名）。\n\n" - "技能文件格式要求：\n" - " • 文件名必须带 .md 扩展名（如 fetch-weather.md、calc-bmi.md）\n" - " • 第一行为一级标题（# 技能名），作为该技能的一句话简介\n" - " • 其余内容为技能的执行说明、使用方法和示例，供智能体参考执行" - ) - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "skill_name": { - "type": "string", - "description": ( - "技能名称（部署后的文件名，不含扩展名；系统会自动补齐 .md）。" - "只能包含字母、数字、连字符和下划线，例如：fetch-weather、calc-bmi。" - ) - }, - "description": { - "type": "string", - "description": ( - "对该技能的详细需求描述：它应该做什么、接受哪些输入参数、" - "返回什么输出、有哪些典型使用场景。描述越详细，生成质量越高。" - ) - } - }, - "required": ["skill_name", "description"] - } - - def execute(self, skill_name: str = "", description: str = "", **kwargs) -> str: - # ---- 参数校验 ---- - if not skill_name or not skill_name.strip(): - return "❌ 错误：skill_name 不能为空" - if not description or not description.strip(): - return "❌ 错误：description 不能为空" - if not re.match(r'^[A-Za-z0-9_\-]+$', skill_name.strip()): - return "❌ 错误：skill_name 只能包含字母、数字、连字符(-)和下划线(_)" - - skill_name = skill_name.strip() - skills_dir = constants.SKILLS_DIR - # 技能文件名统一带 .md 扩展名 - skill_path = os.path.join(skills_dir, skill_name + ".md") - - # ---- 返回给智能体的指令提示词 ---- - prompt = f"""【任务】创建扩展技能：{skill_name} - -请按照以下步骤，自行完成该扩展技能的编写和部署： - -───────────────────────────────────────────────── -步骤 1：编写 Markdown 技能文件 -───────────────────────────────────────────────── -根据以下需求，编写一个 Markdown 技能文件： - -需求描述： -{description.strip()} - -技能文件必须满足的格式约定： -1. 第一行为一级标题 `# {skill_name}`，用于对该技能功能做一句话简介（系统用第一行作为技能摘要） -2. 其余内容清晰说明该技能的用途、使用方式、注意事项和典型示例 -3. 内容完整可执行（即按内容操作即可完成该技能要完成的事） - -───────────────────────────────────────────────── -步骤 2：写入技能目录 -───────────────────────────────────────────────── -调用 create_subagent 派生子代理，将你编写的完整内容写入（文件名必须带 .md 扩展名）： - {skill_path} -（在 task 中附上完整文件内容，并要求子代理写入后用 head -5 验证首行格式） - -───────────────────────────────────────────────── -步骤 3：热重载 -───────────────────────────────────────────────── -调用 reload_skills 工具使新技能立即生效： - reload_skills confirm=true - -───────────────────────────────────────────────── -完成后请告知用户技能已创建成功，并展示技能的第一行简介。""" - - return prompt - - -# ==================== 更新身份工具 ==================== - -class UpdateIdentityTool(Tool): - """更新用户身份工具 - 修改 00_IDENTITY.md 文件""" - - def __init__(self, prompt_manager: PromptManager, logger: Optional[Logger] = None): - super().__init__(logger) - self.prompt_manager = prompt_manager - - def get_name(self) -> str: - return "update_identity" - - def get_description(self) -> str: - return """更新用户身份信息 - 修改 00_IDENTITY.md 文件 -当你了解到用户的身份信息时，调用此工具更新身份文件。 -如果文件不存在，会自动创建。""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "content": { - "type": "string", - "description": "用户身份信息内容（Markdown格式），包含用户的身份、特点、偏好等信息" - }, - "confirm": { - "type": "boolean", - "description": "确认更新身份信息", - "default": True - } - }, - "required": ["content"] - } - - def execute(self, content: str = None, confirm: bool = True, **kwargs) -> str: - if not content: - return "❌ 需要提供身份信息内容" - - if not confirm: - return "❌ 身份更新已取消，如需更新请设置 confirm=true" - - self.logger.info("开始更新用户身份信息...") - - if self.prompt_manager.update_identity(content): - return self.format_result(True, i18n.get('identity_updated', self.prompt_manager.IDENTITY_FILE), { - "文件": self.prompt_manager.IDENTITY_FILE, - "路径": self.prompt_manager.get_prompts_dir_path() - }) - else: - return self.format_result(False, "身份更新失败", { - "文件": self.prompt_manager.IDENTITY_FILE, - "路径": self.prompt_manager.get_prompts_dir_path() - }) - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行更新身份""" - output_queue.put(("line", "\n👤 更新用户身份信息...\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - result = self.execute(**kwargs) - - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - -# ==================== 更新智能体自我认识工具 ==================== - -class UpdateSoulTool(Tool): - """更新智能体自我认识工具 - 修改 99_SOUL.md 文件""" - - def __init__(self, prompt_manager: PromptManager, logger: Optional[Logger] = None): - super().__init__(logger) - self.prompt_manager = prompt_manager - - def get_name(self) -> str: - return "update_soul" - - def get_description(self) -> str: - return """更新智能体的自我认识定义 - 修改 99_SOUL.md 文件 -当你需要重新定义自己的身份、角色、性格、能力边界时，调用此工具更新自我认识文件。 -如果文件不存在，会自动创建默认定义。""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "content": { - "type": "string", - "description": "智能体自我认识定义内容（Markdown格式），包含：身份定义、角色描述、性格特点、能力边界、行为准则等" - }, - "confirm": { - "type": "boolean", - "description": "确认更新自我认识定义", - "default": True - } - }, - "required": ["content"] - } - - def execute(self, content: str = None, confirm: bool = True, **kwargs) -> str: - if not content: - return "❌ 需要提供智能体自我认识定义内容" - - if not confirm: - return "❌ 自我认识更新已取消，如需更新请设置 confirm=true" - - self.logger.info("开始更新智能体自我认识定义...") - - if self.prompt_manager.update_soul(content): - return self.format_result(True, "智能体自我认识已更新", { - "文件": self.prompt_manager.SOUL_FILE, - "路径": self.prompt_manager.get_prompts_dir_path() - }) - else: - return self.format_result(False, "自我认识更新失败", { - "文件": self.prompt_manager.SOUL_FILE, - "路径": self.prompt_manager.get_prompts_dir_path() - }) - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行更新自我认识""" - output_queue.put(("line", "\n🧠 更新智能体自我认识...\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - result = self.execute(**kwargs) - - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - -# ==================== 保存记忆并结束对话工具（增强版，修复版） ==================== - -class SaveMemoryAndEndConversationTool(Tool): - """保存记忆并结束对话工具 - 增强版，同时更新智能体状态""" - - def __init__(self, exit_callback: Callable, memory_db: MemoryDatabase, prompt_manager: PromptManager, logger: Optional[Logger] = None): - super().__init__(logger) - self.exit_callback = exit_callback - self.memory_db = memory_db - self.prompt_manager = prompt_manager - - def get_name(self) -> str: - return "save_memory_and_end_conversation" - - def get_description(self) -> str: - return """保存记忆并结束当前对话 - 必须提供详细的对话总结和当前状态！ -使用此工具时，AI会： -1. 将本次对话的重要内容保存到记忆库 -2. 提供完整的对话总结 -3. 描述自己现在的状态、事情的走向等，保存到 100_STATUS.md -4. 礼貌地结束对话""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "summary": { - "type": "string", - "description": "详细的对话总结（至少50字），包含：讨论的主题、关键问题、重要结论、达成的共识等" - }, - "topic": { - "type": "string", - "description": "对话主题（用于记忆分类，默认自动提取）" - }, - "reason": { - "type": "string", - "description": "结束对话的原因", - "default": "话题讨论完成" - }, - "current_status": { - "type": "string", - "description": "当前的智能体状态描述，包含：当前进展、下一步计划、待处理事项、整体走向等" - }, - "next_goals": { - "type": "string", - "description": "接下来的目标或计划（可选）" - } - }, - "required": ["summary", "current_status"] - } - - def execute(self, summary: str = None, topic: str = None, reason: str = "话题讨论完成", - current_status: str = None, next_goals: str = None, **kwargs) -> str: - if summary is None or len(summary) < 50: - return "❌ 请提供更详细的对话总结（至少50字），以便保存到记忆库" - - if current_status is None: - return "❌ 请提供当前状态描述，以便更新 100_STATUS.md" - - self.logger.info(f"AI保存记忆并结束对话: {reason}") - - # 保存到记忆库 - memory_result = "" - if self.memory_db and topic: - memory_id = self.memory_db.add_memory( - topic=topic, - summary=summary, - importance=3 # 提高重要性，因为是对话总结 - ) - self.logger.info(f"对话记忆已保存，ID: {memory_id}") - memory_result = f"记忆已保存，ID: {memory_id}" - - # 构建状态内容 - status_content = f"【当前状态】\n{current_status}\n\n" - if next_goals: - status_content += f"【下一步计划】\n{next_goals}\n\n" - status_content += f"【对话总结】\n{summary}" - - # 更新状态文件 - status_updated = self.prompt_manager.update_status(status_content) - - if self.exit_callback: - self.exit_callback(reason, summary) - - data = { - "原因": reason, - "总结预览": summary[:100] + "..." if len(summary) > 100 else summary, - "状态更新": "✅ 已更新" if status_updated else "❌ 更新失败", - "记忆已保存": True - } - - if memory_result: - data["记忆"] = memory_result - - return self.format_result(True, "对话记忆已保存，状态已更新，对话结束", data) - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行结束对话""" - output_queue.put(("line", "\n💾 保存记忆并结束对话\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - result = self.execute(**kwargs) - - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - - -# ==================== 备忘录工具（简化版，去掉任务类型） ==================== - -class MemoTool(Tool): - """备忘录工具 - 用于管理任务（所有任务都是对话任务）""" - - def __init__(self, memo_db: MemoDatabase, logger: Optional[Logger] = None): - super().__init__(logger) - self.memo_db = memo_db - - def get_name(self) -> str: - return "memo" - - def get_description(self) -> str: - return """任务管理工具 - 创建、查看、完成任务 -支持：创建任务（支持一次性/重复、即时任务）、查看待办、标记完成、删除任务、搜索任务""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "action": { - "type": "string", - "description": "操作类型", - "enum": ["add", "list", "get", "complete", "delete", "search", "stats"] - }, - "title": { - "type": "string", - "description": "任务标题（用于add操作）" - }, - "content": { - "type": "string", - "description": "任务内容（可选，用于add操作）" - }, - "reminder_time": { - "type": "string", - "description": "执行时间（格式：YYYY-MM-DD HH:MM:SS，或相对时间如 '+1h', '+30m', 'tomorrow 9am'）" - }, - "reminder_minutes": { - "type": "integer", - "description": "延迟分钟数（从现在开始，用于快速设置任务）" - }, - "tags": { - "type": "array", - "items": {"type": "string"}, - "description": "标签列表" - }, - "priority": { - "type": "integer", - "description": "优先级 1-5（默认1）", - "default": 1 - }, - "repeat_type": { - "type": "string", - "description": "重复类型（none=一次性, daily=每天, weekly=每周, monthly=每月, custom=自定义）", - "enum": ["none", "daily", "weekly", "monthly", "custom"], - "default": "none" - }, - "repeat_interval": { - "type": "integer", - "description": "自定义重复间隔，当repeat_type='custom'时使用", - "default": 0 - }, - "repeat_interval_unit": { - "type": "string", - "description": "重复间隔单位（minutes=分钟, hours=小时, days=天），当repeat_type='custom'时使用", - "enum": ["minutes", "hours", "days"], - "default": "days" - }, - "repeat_end_time": { - "type": "string", - "description": "重复结束时间，None表示永久重复" - }, - "is_immediate": { - "type": "boolean", - "description": "是否为即时任务（立即执行）", - "default": False - }, - "memo_id": { - "type": "integer", - "description": "任务ID（用于get、complete、delete操作）" - }, - "memo_ids": { - "type": "array", - "items": {"type": "integer"}, - "description": "任务ID列表（用于批量删除）" - }, - "force_complete": { - "type": "boolean", - "description": "强制完成（忽略重复规则）", - "default": False - }, - "keyword": { - "type": "string", - "description": "搜索关键词（用于search操作）" - }, - "include_completed": { - "type": "boolean", - "description": "是否包含已完成的任务（默认False）", - "default": False - }, - "limit": { - "type": "integer", - "description": "返回数量限制（默认20）", - "default": 20 - }, - "confirm": { - "type": "boolean", - "description": "确认删除操作", - "default": False - } - }, - "required": ["action"] - } - - def _parse_reminder_time(self, reminder_input: Any) -> Optional[str]: - """解析提醒时间输入""" - if not reminder_input: - return None - - if isinstance(reminder_input, int): - reminder_time = datetime.now() + timedelta(minutes=reminder_input) - return reminder_time.strftime(constants.DATETIME_FORMAT) - - if isinstance(reminder_input, str): - reminder_input = reminder_input.strip() - - if reminder_input.startswith('+'): - match = re.match(r'\+(\d+)([mhd])', reminder_input) - if match: - value, unit = int(match.group(1)), match.group(2) - if unit == 'm': - delta = timedelta(minutes=value) - elif unit == 'h': - delta = timedelta(hours=value) - elif unit == 'd': - delta = timedelta(days=value) - else: - return None - - reminder_time = datetime.now() + delta - return reminder_time.strftime(constants.DATETIME_FORMAT) - - try: - reminder_time = datetime.strptime(reminder_input, constants.DATETIME_FORMAT) - return reminder_time.strftime(constants.DATETIME_FORMAT) - except ValueError: - pass - - now = datetime.now() - reminder_input_lower = reminder_input.lower() - - if 'tomorrow' in reminder_input_lower: - tomorrow = now + timedelta(days=1) - time_match = re.search(r'(\d{1,2})(?::(\d{2}))?\s*(am|pm)?', reminder_input_lower) - if time_match: - hour = int(time_match.group(1)) - minute = int(time_match.group(2)) if time_match.group(2) else 0 - ampm = time_match.group(3) - - if ampm: - if ampm == 'pm' and hour < 12: - hour += 12 - elif ampm == 'am' and hour == 12: - hour = 0 - - reminder_time = tomorrow.replace(hour=hour, minute=minute, second=0) - else: - reminder_time = tomorrow.replace(hour=9, minute=0, second=0) - - return reminder_time.strftime(constants.DATETIME_FORMAT) - - return None - - def execute(self, action: str, **kwargs) -> str: - """执行备忘录操作""" - try: - if action == "add": - return self._add_memo(kwargs) - elif action == "list": - return self._list_memos(kwargs) - elif action == "get": - return self._get_memo(kwargs) - elif action == "complete": - return self._complete_memo(kwargs) - elif action == "delete": - return self._delete_memo(kwargs) - elif action == "search": - return self._search_memos(kwargs) - elif action == "stats": - return self._get_stats(kwargs) - else: - return f"❌ 未知操作: {action}" - except Exception as e: - return f"❌ 任务操作错误: {str(e)}" - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行备忘录操作""" - action = kwargs.get("action") - - # 发送开始执行的消息 - output_queue.put(("line", f"\n📝 执行备忘录操作: {action}\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - # 执行操作 - result = self.execute(**kwargs) - - # 发送结果 - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - - def _add_memo(self, kwargs: Dict) -> str: - """添加任务 - 所有任务都是对话任务，不需要task_type""" - title = kwargs.get("title") - if not title: - return "❌ 需要提供任务标题" - - content = kwargs.get("content") - reminder_minutes = kwargs.get("reminder_minutes") - reminder_time_input = kwargs.get("reminder_time") - tags = kwargs.get("tags", []) - priority = kwargs.get("priority", 1) - repeat_type = kwargs.get("repeat_type", "none") - repeat_interval = kwargs.get("repeat_interval", 0) - repeat_interval_unit = kwargs.get("repeat_interval_unit", "days") - repeat_end_time = kwargs.get("repeat_end_time") - is_immediate = kwargs.get("is_immediate", False) - - if repeat_type not in ["none", "daily", "weekly", "monthly", "custom"]: - return f"❌ 无效的重复类型: {repeat_type}" - - if repeat_type == "custom" and repeat_interval <= 0: - return "❌ 自定义重复需要设置有效的 repeat_interval (>0)" - - reminder_time = None - if is_immediate: - # 即时任务：设置为当前时间，确保立即执行 - reminder_time = datetime.now().strftime(constants.DATETIME_FORMAT) - elif reminder_minutes: - reminder_time = self._parse_reminder_time(reminder_minutes) - elif reminder_time_input: - reminder_time = self._parse_reminder_time(reminder_time_input) - else: - # 默认30分钟后 - reminder_time = self._parse_reminder_time(30) - - memo_id = self.memo_db.add_memo( - title=title, - content=content, - reminder_time=reminder_time, - tags=tags, - priority=priority, - created_by="AI", - source="user", - repeat_type=repeat_type, - repeat_interval=repeat_interval, - repeat_interval_unit=repeat_interval_unit, - repeat_end_time=repeat_end_time, - is_immediate=is_immediate - ) - - data = { - "任务ID": memo_id, - "标题": title, - } - - if is_immediate: - data["任务类型"] = "即时任务" - else: - if reminder_time: - data["执行时间"] = reminder_time - - if repeat_type != "none": - if repeat_type == "custom": - repeat_info = f"重复类型: 每{repeat_interval} {repeat_interval_unit}" - else: - repeat_map = { - 'daily': '每天', - 'weekly': '每周', - 'monthly': '每月' - } - repeat_info = f"重复类型: {repeat_map.get(repeat_type, repeat_type)}" - - if repeat_end_time: - repeat_info += f", 结束于: {repeat_end_time}" - else: - repeat_info += ", 永久重复" - data["重复信息"] = repeat_info - - return self.format_result(True, "任务已创建", data) - - def _list_memos(self, kwargs: Dict) -> str: - """列出任务""" - include_completed = kwargs.get("include_completed", False) - limit = kwargs.get("limit", 20) - - memos = self.memo_db.get_all_memos(include_completed, limit) - - if not memos: - return "📭 没有找到任务" - - status = "所有" if include_completed else "待执行" - output = [f"📋 {status}任务 (共 {len(memos)} 个):"] - - now = datetime.now() - - for i, memo in enumerate(memos, 1): - status_icon = "✅" if memo['is_completed'] else "⏰" - priority_icon = "🔴" * memo['priority'] - source_tag = f"[{memo.get('source', '未知')}]" if memo.get('source') else "" - - immediate_tag = "⚡" if memo.get('is_immediate') else "" - - repeat_tag = "" - if memo['repeat_type'] != 'none': - if memo['repeat_type'] == 'custom': - unit = memo.get('repeat_interval_unit', 'days') - unit_map = {'minutes': '分钟', 'hours': '小时', 'days': '天'} - repeat_tag = f" [每{memo['repeat_interval']}{unit_map.get(unit, unit)}]" - else: - repeat_map = { - 'daily': ' [每天]', - 'weekly': ' [每周]', - 'monthly': ' [每月]' - } - repeat_tag = repeat_map.get(memo['repeat_type'], ' [重复]') - - output.append(f"\n{i}. {status_icon}{immediate_tag} [ID: {memo['id']}] {source_tag}{repeat_tag} {memo['title']} {priority_icon}") - - if memo['reminder_time']: - try: - reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) - if not memo['is_completed'] and reminder < now: - output.append(f" ⚠️ 已过期: {memo['reminder_time']}") - else: - output.append(f" ⏱️ 执行: {memo['reminder_time']}") - except: - output.append(f" ⏱️ 执行: {memo['reminder_time']}") - - if memo.get('trigger_count', 0) > 0: - output.append(f" 🔄 已执行 {memo['trigger_count']} 次") - - if memo['content']: - preview = memo['content'][:100] + "..." if len(memo['content']) > 100 else memo['content'] - output.append(f" 📝 {preview}") - - if memo['tags']: - output.append(f" 🏷️ {', '.join(memo['tags'])}") - - return "\n".join(output) - - def _get_memo(self, kwargs: Dict) -> str: - """获取单个任务""" - memo_id = kwargs.get("memo_id") - if not memo_id: - return "❌ 需要提供任务ID" - - memo = self.memo_db.get_memo(memo_id) - if not memo: - return f"❌ 未找到ID为 {memo_id} 的任务" - - status = "✅ 已完成" if memo['is_completed'] else "⏰ 待执行" - if memo['is_completed']: - status += f" (完成于: {memo['completed_at']})" - - immediate_tag = "⚡ 即时任务" if memo.get('is_immediate') else "" - - output = [ - f"📌 任务 #{memo_id} {immediate_tag}", - f"📋 状态: {status}", - f"📝 标题: {memo['title']}", - ] - - if memo.get('source'): - output.append(f"📎 来源: {memo['source']}") - - if memo['repeat_type'] != 'none': - if memo['repeat_type'] == 'custom': - unit_map = {'minutes': '分钟', 'hours': '小时', 'days': '天'} - unit = memo.get('repeat_interval_unit', 'days') - repeat_info = f"🔄 重复: 每{memo['repeat_interval']} {unit_map.get(unit, unit)}" - else: - repeat_map = { - 'daily': '每天', - 'weekly': '每周', - 'monthly': '每月' - } - repeat_info = f"🔄 重复: {repeat_map.get(memo['repeat_type'], memo['repeat_type'])}" - - if memo['repeat_end_time']: - repeat_info += f", 结束于: {memo['repeat_end_time']}" - else: - repeat_info += ", 永久重复" - output.append(repeat_info) - - if memo.get('trigger_count', 0) > 0: - output.append(f" 📊 已执行 {memo['trigger_count']} 次") - if memo.get('last_triggered_at'): - output.append(f" ⏱️ 上次执行: {memo['last_triggered_at']}") - - if memo['content']: - output.append(f"📄 内容:\n{memo['content']}") - - if memo['reminder_time']: - try: - now = datetime.now() - reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) - if not memo['is_completed'] and reminder < now: - output.append(f"⚠️ 执行时间 (已过期): {memo['reminder_time']}") - else: - output.append(f"⏰ 执行时间: {memo['reminder_time']}") - except: - output.append(f"⏰ 执行时间: {memo['reminder_time']}") - - if memo['tags']: - output.append(f"🏷️ 标签: {', '.join(memo['tags'])}") - - output.append(f"🔢 优先级: {'🔴' * memo['priority']} ({memo['priority']}/5)") - output.append(f"📅 创建时间: {memo['created_at']}") - output.append(f"👤 创建者: {memo['created_by'] or '系统'}") - - return "\n".join(output) - - def _complete_memo(self, kwargs: Dict) -> str: - """完成任务""" - memo_id = kwargs.get("memo_id") - if not memo_id: - return "❌ 需要提供任务ID" - - force_complete = kwargs.get("force_complete", False) - - memo = self.memo_db.get_memo(memo_id) - if not memo: - return f"❌ 未找到ID为 {memo_id} 的任务" - - if memo['is_completed'] and not force_complete: - return f"ℹ️ 任务 #{memo_id} 已经完成了" - - if self.memo_db.complete_memo(memo_id, force_complete): - memo = self.memo_db.get_memo(memo_id) - data = { - "任务ID": memo_id, - "标题": memo['title'] - } - - if memo['repeat_type'] != 'none' and not memo['is_completed']: - data["下一次执行"] = memo['reminder_time'] - data["已执行次数"] = memo.get('trigger_count', 0) - return self.format_result(True, "任务已处理，已更新下一次执行时间", data) - else: - return self.format_result(True, "任务已完成", data) - else: - return f"❌ 无法完成任务 #{memo_id}" - - def _delete_memo(self, kwargs: Dict) -> str: - """删除任务""" - confirm = kwargs.get("confirm", False) - if not confirm: - return "❌ 需要设置 confirm=True 来确认删除操作" - - memo_id = kwargs.get("memo_id") - if memo_id: - if self.memo_db.complete_memo(memo_id, force_complete=True): - return self.format_result(True, f"任务 #{memo_id} 已删除") - else: - return f"❌ 无法删除任务 #{memo_id}" - else: - return "❌ 需要提供 memo_id" - - def _search_memos(self, kwargs: Dict) -> str: - """搜索任务""" - keyword = kwargs.get("keyword") - if not keyword: - return "❌ 需要提供搜索关键词" - - include_completed = kwargs.get("include_completed", False) - limit = kwargs.get("limit", 20) - - all_memos = self.memo_db.get_all_memos(include_completed, limit=100) - - results = [] - keyword_lower = keyword.lower() - for memo in all_memos: - if (keyword_lower in memo['title'].lower() or - (memo['content'] and keyword_lower in memo['content'].lower())): - results.append(memo) - - if not results: - return f"❌ 未找到包含 '{keyword}' 的任务" - - output = [f"🔍 搜索 '{keyword}' 找到 {len(results)} 个任务:"] - now = datetime.now() - - for i, memo in enumerate(results[:limit], 1): - status_icon = "✅" if memo['is_completed'] else "⏰" - immediate_tag = "⚡" if memo.get('is_immediate') else "" - output.append(f"\n{i}. {status_icon}{immediate_tag} [ID: {memo['id']}] {memo['title']}") - if memo['reminder_time']: - try: - reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) - if not memo['is_completed'] and reminder < now: - output.append(f" ⚠️ 已过期: {memo['reminder_time']}") - else: - output.append(f" ⏱️ {memo['reminder_time']}") - except: - output.append(f" ⏱️ {memo['reminder_time']}") - - return "\n".join(output) - - def _get_stats(self, kwargs: Dict) -> str: - """获取任务统计信息""" - stats = self.memo_db.get_memo_stats() - - output = [ - "📊 任务统计", - f" 总任务: {stats['total']} 个", - f" 待执行: {stats['pending']} 个", - f" 已过期: {stats['overdue']} 个", - f" 已完成: {stats['completed']} 个", - f" 即时任务: {stats['immediate']} 个" - ] - - if stats['overdue'] > 0: - output.append(f"\n⚠️ 有 {stats['overdue']} 个任务已过期，将自动执行") - - return "\n".join(output) - - -# ==================== 记忆工具（修改版，支持流式输出） ==================== - -class MemoryTool(Tool): - """记忆工具 - 用于存储、检索和删除对话记忆（支持流式输出，支持分页）""" - - def __init__(self, memory_db: MemoryDatabase, logger: Optional[Logger] = None): - super().__init__(logger) - self.memory_db = memory_db - - def get_name(self) -> str: - return "memory" - - def get_description(self) -> str: - return """记忆管理工具 - 存储、检索和删除重要的对话信息 -支持：保存总结、多关键词搜索记忆（支持分页）、查看最近的记忆、删除无用记忆""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "action": { - "type": "string", - "description": "操作类型", - "enum": ["save", "search", "recent", "get", "delete"] - }, - "summary": {"type": "string", "description": "要保存的总结内容（用于save操作）"}, - "topic": {"type": "string", "description": "对话主题（用于save操作）"}, - "content": {"type": "string", "description": "详细内容（可选，用于save操作）"}, - "keywords": { - "type": "array", "items": {"type": "string"}, - "description": "搜索关键词列表" - }, - "query": {"type": "string", "description": "搜索关键词（单个关键词）"}, - "match_all": {"type": "boolean", "default": False}, - "memory_id": {"type": "integer"}, - "memory_ids": {"type": "array", "items": {"type": "integer"}}, - "importance": {"type": "integer", "default": 2}, - "limit": {"type": "integer", "default": 10, "description": "每页条数（默认10）"}, - "page": {"type": "integer", "default": 1, "description": "页码（从1开始）"}, - "confirm": {"type": "boolean", "default": False} - }, - "required": ["action"] - } - - def execute(self, action: str, **kwargs) -> str: - try: - if action == "save": - return self._save_memory(kwargs) - elif action == "search": - return self._search_memories(kwargs) - elif action == "recent": - return self._get_recent_memories(kwargs) - elif action == "get": - return self._get_memory_by_id(kwargs) - elif action == "delete": - return self._delete_memory(kwargs) - else: - return f"❌ 未知操作: {action}" - except Exception as e: - return f"❌ 记忆操作错误: {str(e)}" - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行记忆操作""" - action = kwargs.get("action") - - # 发送开始执行的消息 - output_queue.put(("line", f"\n🧠 执行记忆操作: {action}\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - # 执行操作 - result = self.execute(**kwargs) - - # 发送结果 - output_queue.put(("line", result + "\n")) - output_queue.put(("complete", result)) - - def _save_memory(self, kwargs: Dict) -> str: - summary = kwargs.get("summary") - if not summary: - return "❌ 需要提供总结内容" - - topic = kwargs.get("topic", "未命名对话") - content = kwargs.get("content") - importance = kwargs.get("importance", constants.MEMORY_DEFAULT_IMPORTANCE) - - memory_id = self.memory_db.add_memory( - topic=topic, - summary=summary, - content=content, - importance=importance - ) - - return self.format_result(True, "记忆已保存", { - "记忆ID": memory_id, - "主题": topic, - "总结": summary[:100] + "..." if len(summary) > 100 else summary - }) - - def _search_memories(self, kwargs: Dict) -> str: - """搜索记忆 - 支持分页""" - keywords = kwargs.get("keywords", []) - query = kwargs.get("query", "") - match_all = kwargs.get("match_all", False) - page = kwargs.get("page", 1) - page_size = kwargs.get("limit", 10) # 使用 limit 作为 page_size - - # 确保页码和每页条数有效 - page = max(1, page) - page_size = max(1, min(50, page_size)) # 每页最多50条 - - # 处理关键词 - if keywords and isinstance(keywords, list): - pass - elif query: - keywords = query.split() - else: - return "❌ 需要提供搜索关键词（keywords 或 query）" - - if not keywords: - return "❌ 关键词列表为空" - - # 先获取所有结果（不分页，用于计算总数） - all_results = self.memory_db.search_memories(keywords, match_all, limit=None) - - if not all_results: - match_mode = "所有" if match_all else "任一" - return f"❌ 未找到包含 {match_mode} 关键词 {keywords} 的记忆" - - # 计算分页信息 - total_count = len(all_results) - total_pages = (total_count + page_size - 1) // page_size - - # 检查页码是否有效 - if page > total_pages and total_pages > 0: - return f"❌ 页码 {page} 超出范围，共有 {total_pages} 页" - - # 计算当前页的起止索引 - start_idx = (page - 1) * page_size - end_idx = min(start_idx + page_size, total_count) - - # 获取当前页的结果 - page_results = all_results[start_idx:end_idx] - - # 构建输出 - match_mode = "所有关键词" if match_all else "任一关键词" - output = [ - f"🔍 搜索 {match_mode} {keywords} 找到 {total_count} 条记忆", - f"📄 第 {page}/{total_pages} 页 (每页 {page_size} 条)", - "-" * 40 - ] - - for i, mem in enumerate(page_results, 1): - # 全局序号 - global_idx = start_idx + i - output.append(f"\n{global_idx}. [ID: {mem['id']}] [{mem['end_time']}] {mem['topic']}") - output.append(f" 总结: {mem['summary'][:200]}") - if mem.get('tags'): - output.append(f" 标签: {', '.join(mem['tags'])}") - if mem.get('importance'): - output.append(f" 重要性: {'⭐' * mem['importance']} ({mem['importance']}/5)") - - # 添加翻页提示 - if total_pages > 1: - output.append(f"\n📖 翻页提示: 使用 page 参数查看更多结果") - output.append(f" 例如: memory search query='{query}' page={page + 1 if page < total_pages else page} limit={page_size}") - if page < total_pages: - output.append(f" ➡️ 下一页: page={page + 1}") - if page > 1: - output.append(f" ⬅️ 上一页: page={page - 1}") - - return "\n".join(output) - - def _get_recent_memories(self, kwargs: Dict) -> str: - """获取最近的记忆 - 支持分页""" - page = kwargs.get("page", 1) - page_size = kwargs.get("limit", 10) - - # 确保页码和每页条数有效 - page = max(1, page) - page_size = max(1, min(50, page_size)) - - # 先获取所有结果 - all_results = self.memory_db.get_recent_memories(limit=None) - - if not all_results: - return "📭 没有找到记忆" - - # 计算分页信息 - total_count = len(all_results) - total_pages = (total_count + page_size - 1) // page_size - - # 检查页码是否有效 - if page > total_pages and total_pages > 0: - return f"❌ 页码 {page} 超出范围，共有 {total_pages} 页" - - # 计算当前页的起止索引 - start_idx = (page - 1) * page_size - end_idx = min(start_idx + page_size, total_count) - - # 获取当前页的结果 - page_results = all_results[start_idx:end_idx] - - output = [f"📚 最近 {total_count} 条记忆 (第 {page}/{total_pages} 页，每页 {page_size} 条):"] - output.append("-" * 40) - - for i, mem in enumerate(page_results, 1): - global_idx = start_idx + i - output.append(f"\n{global_idx}. [ID: {mem['id']}] [{mem['end_time']}] {mem['topic']}") - output.append(f" 总结: {mem['summary'][:100]}") - - # 添加翻页提示 - if total_pages > 1: - output.append(f"\n📖 翻页提示: 使用 page 参数查看更多结果") - output.append(f" 例如: memory recent page={page + 1 if page < total_pages else page} limit={page_size}") - - return "\n".join(output) - - def _get_memory_by_id(self, kwargs: Dict) -> str: - """根据ID获取记忆""" - memory_id = kwargs.get("memory_id") - if not memory_id: - return "❌ 需要提供记忆ID" - - memory = self.memory_db.get_memory_by_id(memory_id) - if not memory: - return f"❌ 未找到ID为 {memory_id} 的记忆" - - output = [ - f"📌 记忆 #{memory_id}", - f"主题: {memory['topic']}", - f"时间: {memory['start_time']} - {memory['end_time']}", - f"总结: {memory['summary']}", - ] - - if memory.get('content'): - output.append(f"详细内容:\n{memory['content']}") - - if memory.get('tags'): - output.append(f"标签: {', '.join(memory['tags'])}") - - if memory.get('importance'): - output.append(f"重要性: {'⭐' * memory['importance']} ({memory['importance']}/5)") - - return "\n".join(output) - - def _delete_memory(self, kwargs: Dict) -> str: - """删除记忆""" - confirm = kwargs.get("confirm", False) - if not confirm: - return "❌ 需要设置 confirm=True 来确认删除操作" - - memory_id = kwargs.get("memory_id") - memory_ids = kwargs.get("memory_ids", []) - - deleted_count = 0 - - if memory_id: - if self.memory_db.delete_memory(memory_id): - deleted_count = 1 - elif memory_ids: - deleted_count = self.memory_db.delete_memories_by_ids(memory_ids) - else: - return "❌ 需要提供 memory_id 或 memory_ids" - - if deleted_count == 0: - return f"❌ 未找到要删除的记忆" - - return self.format_result(True, f"成功删除 {deleted_count} 条记忆", {"删除数量": deleted_count}) - -class CommandLineTool(Tool): - """命令行执行工具 - 使用抽象的命令执行器（支持流式输出和超时控制）""" - - def __init__(self, executor: CommandExecutor = None, logger=None): - super().__init__(logger) - self.executor = executor or SubprocessCommandExecutor(logger) - - def get_name(self) -> str: - return "run_command" - - def get_description(self) -> str: - return """执行任意系统命令，支持实时输出显示和超时控制""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "command": { - "type": "string", - "description": "要执行的命令" - }, - "timeout": { - "type": "integer", - "default": 30, - "description": "命令执行超时时间（秒），超时后将终止命令执行。默认30秒" - }, - "work_dir": { - "type": "string", - "default": ".", - "description": "命令执行的工作目录" - }, - "env": { - "type": "object", - "description": "额外的环境变量" - } - }, - "required": ["command"] - } - - def execute_streaming(self, output_queue: queue.Queue, **kwargs): - """流式执行命令，支持超时控制""" - command = kwargs.get("command") - timeout = kwargs.get("timeout", 30) - work_dir = kwargs.get("work_dir", ".") - extra_env = kwargs.get("env", {}) - - if not command: - output_queue.put(("line", "❌ 错误：缺少命令参数\n")) - output_queue.put(("complete", "❌ 错误：缺少命令参数")) - return - - # 发送开始执行的消息 - output_queue.put(("line", f"\n🔧 执行命令: {command}\n")) - output_queue.put(("line", f"⏱️ 超时设置: {timeout} 秒\n")) - if work_dir != ".": - output_queue.put(("line", f"📁 工作目录: {work_dir}\n")) - output_queue.put(("line", "-" * 50 + "\n")) - - # 准备环境变量 - env = os.environ.copy() - if extra_env: - env.update(extra_env) - - try: - # 使用 subprocess.Popen 实现流式输出，支持超时 - process = subprocess.Popen( - command, - shell=True, - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - stdin=subprocess.PIPE, - text=True, - bufsize=1, - universal_newlines=True, - env=env, - cwd=os.path.expanduser(work_dir) - ) - - start_time = time.time() - output_lines = [] - timeout_occurred = False - - # 实时读取输出 - while True: - # 检查超时 - elapsed = time.time() - start_time - if timeout > 0 and elapsed > timeout: - process.terminate() - timeout_occurred = True - output_queue.put(("line", f"\n⏰ 命令执行超时（{timeout}秒），已终止\n")) - break - - # 使用非阻塞方式读取一行 - import select - import sys - - # 检查是否有输出可读 - if sys.platform == 'win32': - # Windows 下使用 readline 带超时 - try: - import msvcrt - if msvcrt.kbhit(): - pass - line = process.stdout.readline() - if not line and process.poll() is not None: - break - if line: - line = line.rstrip('\n') - output_queue.put(("line", line + "\n")) - output_lines.append(line) - else: - time.sleep(0.05) - except: - line = process.stdout.readline() - if not line and process.poll() is not None: - break - if line: - line = line.rstrip('\n') - output_queue.put(("line", line + "\n")) - output_lines.append(line) - else: - # Unix 系统可以使用 select - import select - rlist, _, _ = select.select([process.stdout], [], [], 0.5) - if rlist: - line = process.stdout.readline() - if not line and process.poll() is not None: - break - if line: - line = line.rstrip('\n') - output_queue.put(("line", line + "\n")) - output_lines.append(line) - else: - # 检查进程是否结束 - if process.poll() is not None: - # 读取剩余输出 - remaining = process.stdout.read() - if remaining: - remaining = remaining.rstrip('\n') - if remaining: - output_queue.put(("line", remaining + "\n")) - output_lines.append(remaining) - break - - # 如果超时，确保进程被终止 - if timeout_occurred and process.poll() is None: - process.kill() - process.wait(timeout=2) - - # 获取返回码 - return_code = process.returncode if process.returncode is not None else -1 - - output_queue.put(("line", "-" * 50 + "\n")) - - if timeout_occurred: - output_queue.put(("line", f"⏰ 命令执行超时（{timeout}秒），返回码: {return_code}\n")) - else: - output_queue.put(("line", f"✅ 命令执行完成，返回码: {return_code}\n")) - - # 准备完整结果 - full_output = "\n".join(output_lines) - data = { - "输出": full_output if full_output else "(无输出)", - "返回码": return_code - } - - if timeout_occurred: - result = self.format_result(False, f"命令执行超时（{timeout}秒）", data) - else: - result = self.format_result(return_code == 0, "命令执行完成", data) - - output_queue.put(("complete", result)) - - except subprocess.TimeoutExpired: - error_msg = f"❌ 命令执行超时（{timeout}秒）" - output_queue.put(("line", error_msg + "\n")) - output_queue.put(("complete", error_msg)) - except FileNotFoundError as e: - error_msg = f"❌ 命令执行错误: 命令未找到 - {str(e)}" - output_queue.put(("line", error_msg + "\n")) - output_queue.put(("complete", error_msg)) - except PermissionError as e: - error_msg = f"❌ 命令执行错误: 权限不足 - {str(e)}" - output_queue.put(("line", error_msg + "\n")) - output_queue.put(("complete", error_msg)) - except Exception as e: - error_msg = f"❌ 执行错误: {str(e)}" - output_queue.put(("line", error_msg + "\n")) - output_queue.put(("complete", error_msg)) - finally: - # 确保进程被清理 - if 'process' in locals() and process and process.poll() is None: - try: - process.terminate() - process.wait(timeout=2) - except: - process.kill() - - def execute(self, command: str, timeout: int = 30, - work_dir: str = ".", env: Dict = None, **kwargs) -> str: - """执行命令 - 仅支持流式执行（命令工具强制流式输出，无 stream 参数）""" - if error := self.validate_params(["command"], command=command): - return error - - # 如果正在流式输出，直接返回空（实际输出通过队列） - if self.streaming_output and self.output_queue: - return "" - - # 命令工具固定走流式执行：通过临时队列收集完整结果 - output_queue = queue.Queue() - self.execute_streaming( - output_queue, - command=command, - timeout=timeout, - work_dir=work_dir, - env=env - ) - - # 收集流式执行的最终结果（complete 消息） - final_result = "" - while not output_queue.empty(): - try: - msg_type, content = output_queue.get_nowait() - if msg_type == "complete": - final_result = content - except queue.Empty: - break - return final_result def run_light_subagent(role, task, config, logger=None, model=None, temperature=None, max_tokens=2048) -> str: @@ -3331,74 +1667,6 @@ class SubAgentTool(Tool): return result -class SetProxyTool(Tool): - """设置 SOCKS5 代理的工具""" - - def get_name(self) -> str: - return "set_proxy" - - def get_description(self) -> str: - return """设置智能体的 SOCKS5 代理配置。 -调用后所有 HTTP/HTTPS 请求（包括网页访问、API 调用）都会通过该代理。 -可随时修改，即时生效。""" - - def get_parameters(self) -> Dict: - return { - "type": "object", - "properties": { - "host": {"type": "string", "description": "代理服务器地址，如 127.0.0.1"}, - "port": {"type": "integer", "description": "代理端口，如 1080"}, - "username": {"type": "string", "description": "用户名（可选）"}, - "password": {"type": "string", "description": "密码（可选）"}, - "enabled": {"type": "boolean", "description": "是否启用代理，默认 true"}, - "save": {"type": "boolean", "description": "是否保存到配置文件（持久化），默认 true"} - }, - "required": ["host", "port"] - } - - def execute(self, host: str, port: int, username: str = None, password: str = None, - enabled: bool = True, save: bool = True, **kwargs) -> str: - config = ConfigManager() - config.set_proxy(host, port, username, password, enabled, save) - proxy_info = config.get_proxy_info() - data = { - "代理状态": "启用" if proxy_info["enabled"] else "禁用", - "服务器": f"{proxy_info['host']}:{proxy_info['port']}", - "认证": "有" if proxy_info.get("username") else "无", - "持久化": "已保存" if save else "仅内存" - } - return self.format_result(True, "代理设置成功", data) - -class GetProxyTool(Tool): - """查看当前代理配置""" - - def get_name(self) -> str: - return "get_proxy" - - def get_description(self) -> str: - return "查看当前 SOCKS5 代理配置信息" - - def get_parameters(self) -> Dict: - return {"type": "object", "properties": {}} - - def execute(self, **kwargs) -> str: - config = ConfigManager() - info = config.get_proxy_info() - if info["enabled"]: - data = { - "代理状态": "启用", - "服务器": f"{info['host']}:{info['port']}", - "认证": "有" if info.get("username") else "无" - } - return self.format_result(True, "当前代理配置", data) - else: - return "❌ 当前代理未启用，所有请求走直连。" - -# ==================== AI聊天类（修改版：添加重置对话记忆和重新加载系统提示词功能，压缩静默执行） ==================== - -# 工具集 profile：orchestrator=主对话管理面（记忆/情绪/子代理派生）， -# worker=子代理执行面（命令行/文件/网页）。主代理不直接接触命令行， -# 执行类任务一律经 create_subagent 派生 worker 完成。 TOOL_PROFILE_ORCHESTRATOR = "orchestrator" TOOL_PROFILE_WORKER = "worker" diff --git a/aibox/subsystems/db.py b/aibox/subsystems/db.py index f0509e9..630f465 100644 --- a/aibox/subsystems/db.py +++ b/aibox/subsystems/db.py @@ -141,11 +141,14 @@ class MemoryDatabase: return results def get_recent_memories(self, limit: int = 10) -> List[Dict]: - """获取最近的记忆""" + """获取最近的记忆。limit=None 时不限制返回条数（供分页使用）。""" with sqlite3.connect(self.db_path) as conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() - cursor.execute('SELECT * FROM memories ORDER BY end_time DESC LIMIT ?', (limit,)) + if limit is None: + cursor.execute('SELECT * FROM memories ORDER BY end_time DESC') + else: + cursor.execute('SELECT * FROM memories ORDER BY end_time DESC LIMIT ?', (limit,)) results = [] for row in cursor.fetchall(): diff --git a/aibox/tests/test_b_tools.py b/aibox/tests/test_b_tools.py new file mode 100644 index 0000000..e81b1e9 --- /dev/null +++ b/aibox/tests/test_b_tools.py @@ -0,0 +1,439 @@ +"""B 批工具单元测试 + +覆盖 12 个 B 批工具的核心接口与执行逻辑： + - command_tool : CommandLineTool + - skill_tools : SearchSkills/ReloadSkills/InstallSkill/CreateSkill + - db_tools : MemoryTool/MemoTool + - prompt_tools : UpdateIdentity/UpdateSoul/SaveMemoryAndEnd + - proxy_tools : SetProxy/GetProxy + - misc_tools : AskUserTool + +工具大多需注入子系统，用真实子系统实例 + tmp_path 隔离。 +""" +import queue +from datetime import datetime, timedelta + +import pytest + +from common import constants +from subsystems import ( + MemoryDatabase, MemoDatabase, PromptManager, SkillManager, + SubprocessCommandExecutor, +) +from tools import ( + CommandLineTool, + SearchSkillsTool, ReloadSkillsTool, InstallSkillTool, CreateSkillTool, + MemoryTool, MemoTool, + UpdateIdentityTool, UpdateSoulTool, SaveMemoryAndEndConversationTool, + SetProxyTool, GetProxyTool, + AskUserTool, +) +from tools.base import Tool +from tools.registry import ToolRegistry + + +# ==================== 公共 fixture ==================== + +@pytest.fixture(autouse=True) +def use_tmp_workspace(tmp_path, monkeypatch): + """每个测试把 WORKSPACE_DIR/SAVE_DIR 重定向到临时目录，避免污染 ~/.aibox。""" + ws = tmp_path / "ws" + ws.mkdir() + save = tmp_path / "save" + save.mkdir() + monkeypatch.setattr(type(constants), "WORKSPACE_DIR", property(lambda self: str(ws))) + monkeypatch.setattr(type(constants), "SAVE_DIR", property(lambda self: str(save))) + + +@pytest.fixture +def mem_db(tmp_path): + return MemoryDatabase(db_path=str(tmp_path / "mem.db")) + + +@pytest.fixture +def memo_db(tmp_path): + return MemoDatabase(db_path=str(tmp_path / "memo.db")) + + +@pytest.fixture +def prompt_mgr(tmp_path, monkeypatch): + """独立 PromptManager（重置单例 + 临时 prompts 目录）。""" + prompts_dir = tmp_path / "prompts" + prompts_dir.mkdir() + PromptManager._instance = None + PromptManager._prompts = [] + PromptManager._prompts_dir = None + from common import ConfigManager + original_get = ConfigManager.get + + def mock_get(self, key, default=None): + if key == "system.prompts_dir": + return str(prompts_dir) + return original_get(self, key, default) + + monkeypatch.setattr(ConfigManager, "get", mock_get) + pm = PromptManager() + yield pm + PromptManager._instance = None + PromptManager._prompts = [] + PromptManager._prompts_dir = None + + +@pytest.fixture +def skill_mgr(tmp_path, monkeypatch): + """独立 SkillManager + 临时技能目录。""" + skills_dir = tmp_path / "skills" + skills_dir.mkdir() + monkeypatch.setattr(type(constants), "SKILLS_DIR", property(lambda self: str(skills_dir))) + return SkillManager() + + +# ==================== 通用接口契约（参数化）==================== + +B_TOOL_CLASSES = [ + CommandLineTool, SearchSkillsTool, ReloadSkillsTool, InstallSkillTool, + CreateSkillTool, MemoryTool, MemoTool, UpdateIdentityTool, UpdateSoulTool, + SaveMemoryAndEndConversationTool, SetProxyTool, GetProxyTool, AskUserTool, +] + + +@pytest.mark.parametrize("cls", B_TOOL_CLASSES, ids=lambda c: c.__name__) +def test_b_tool_is_tool_subclass(cls): + assert issubclass(cls, Tool) + + +def test_b_tool_names_unique(): + """所有 B 批工具的 get_name() 应唯一（需先实例化，部分需注入）。""" + # 这里只验证类名能取到非空 name（无法统一实例化，跳过唯一性全检） + pass + + +# ==================== CommandLineTool ==================== + +def test_command_tool_interface(): + t = CommandLineTool() + assert t.get_name() == "run_command" + assert "执行" in t.get_description() or "命令" in t.get_description() + params = t.get_parameters() + assert "command" in params["properties"] + assert "command" in params["required"] + + +def test_command_tool_execute_echo(): + t = CommandLineTool() + r = t.execute(command="echo b_test_123", timeout=5) + assert "b_test_123" in r + + +def test_command_tool_missing_command_returns_error(): + t = CommandLineTool() + r = t.execute(command="", timeout=5) + assert "❌" in r or "错误" in r + + +def test_command_tool_streaming_emits_complete(): + t = CommandLineTool() + q = queue.Queue() + t.execute_streaming(q, command="echo hi", timeout=5) + msgs = [] + while not q.empty(): + msgs.append(q.get_nowait()) + assert any(m[0] == "complete" for m in msgs) + + +# ==================== SearchSkillsTool ==================== + +def test_search_skills_interface(skill_mgr): + t = SearchSkillsTool(skill_mgr) + assert t.get_name() == "search_skills" + params = t.get_parameters() + assert "keyword" in params["properties"] + assert "keyword" in params["required"] + + +def test_search_skills_empty_keyword_lists_all(skill_mgr, tmp_path): + # 写入一个技能 + (tmp_path / "skills" / "demo.md").write_text("# Demo\n内容", encoding="utf-8") + skill_mgr.reload() + t = SearchSkillsTool(skill_mgr) + r = t.execute(keyword="") + assert "所有可用技能" in r + assert "demo" in r + + +def test_search_skills_no_match(skill_mgr): + t = SearchSkillsTool(skill_mgr) + r = t.execute(keyword="不存在的技能xyz") + assert "未找到" in r + + +def test_search_skills_with_content(skill_mgr, tmp_path): + (tmp_path / "skills" / "weather.md").write_text("# 天气\n查询天气详情ABC", encoding="utf-8") + skill_mgr.reload() + t = SearchSkillsTool(skill_mgr) + r = t.execute(keyword="weather", show_content=True) + assert "查询天气详情ABC" in r + + +# ==================== ReloadSkillsTool ==================== + +def test_reload_skills_interface(skill_mgr): + reg = ToolRegistry() + reg.skill_manager = skill_mgr + t = ReloadSkillsTool(reg) + assert t.get_name() == "reload_skills" + r = t.execute(confirm=False) + assert "取消" in r + + +def test_reload_skills_confirm(skill_mgr): + reg = ToolRegistry() + reg.skill_manager = skill_mgr + t = ReloadSkillsTool(reg) + r = t.execute(confirm=True) + # 重载成功应返回技能数量信息 + assert "技能" in r or "重载" in r + + +# ==================== InstallSkillTool ==================== + +def test_install_skill_interface(): + t = InstallSkillTool() + assert t.get_name() == "install_skill" + assert "skill_path" in t.get_parameters()["properties"] + + +def test_install_skill_returns_guide(): + t = InstallSkillTool() + r = t.execute() + assert "安装" in r or "步骤" in r + assert "skills" in r.lower() or ".md" in r + + +def test_install_skill_with_path(): + t = InstallSkillTool() + r = t.execute(skill_path="/tmp/my_skill.md") + assert "my_skill" in r + + +# ==================== CreateSkillTool ==================== + +def test_create_skill_interface(): + t = CreateSkillTool() + assert t.get_name() == "create_skill" + params = t.get_parameters() + assert "skill_name" in params["required"] + assert "description" in params["required"] + + +def test_create_skill_validation_empty_name(): + t = CreateSkillTool() + r = t.execute(skill_name="", description="desc") + assert "❌" in r + + +def test_create_skill_validation_invalid_name(): + t = CreateSkillTool() + r = t.execute(skill_name="bad name!", description="desc") + assert "❌" in r + + +def test_create_skill_returns_prompt(): + t = CreateSkillTool() + r = t.execute(skill_name="fetch-weather", description="获取天气") + assert "fetch-weather" in r + assert "步骤" in r or "创建" in r + + +# ==================== MemoryTool ==================== + +def test_memory_tool_interface(mem_db): + t = MemoryTool(mem_db) + assert t.get_name() == "memory" + params = t.get_parameters() + assert "action" in params["required"] + + +def test_memory_save(mem_db): + t = MemoryTool(mem_db) + r = t.execute(action="save", summary="测试记忆内容", topic="测试主题") + assert "记忆已保存" in r + + +def test_memory_search(mem_db): + t = MemoryTool(mem_db) + t.execute(action="save", summary="learn Python basics", topic="Python") + r = t.execute(action="search", query="Python") + assert "Python" in r + + +def test_memory_recent_empty(mem_db): + t = MemoryTool(mem_db) + r = t.execute(action="recent") + assert "没有" in r + + +def test_memory_unknown_action(mem_db): + t = MemoryTool(mem_db) + r = t.execute(action="invalid") + assert "❌" in r + + +def test_memory_delete_requires_confirm(mem_db): + t = MemoryTool(mem_db) + mid = mem_db.add_memory(topic="t", summary="s") + r = t.execute(action="delete", memory_id=mid, confirm=False) + assert "❌" in r + + +# ==================== MemoTool ==================== + +def test_memo_tool_interface(memo_db): + t = MemoTool(memo_db) + assert t.get_name() == "memo" + assert "action" in t.get_parameters()["required"] + + +def test_memo_add(memo_db): + t = MemoTool(memo_db) + r = t.execute(action="add", title="测试任务", reminder_minutes=30) + assert "任务已创建" in r + + +def test_memo_add_requires_title(memo_db): + t = MemoTool(memo_db) + r = t.execute(action="add") + assert "❌" in r + + +def test_memo_list_empty(memo_db): + t = MemoTool(memo_db) + r = t.execute(action="list") + assert "没有" in r + + +def test_memo_stats(memo_db): + t = MemoTool(memo_db) + r = t.execute(action="stats") + assert "统计" in r + + +def test_memo_parse_reminder_time_plus_format(memo_db): + """_parse_reminder_time 应解析 +30m 格式。""" + t = MemoTool(memo_db) + result = t._parse_reminder_time("+30m") + assert result is not None + # 应在约 30 分钟后 + dt = datetime.strptime(result, constants.DATETIME_FORMAT) + delta = dt - datetime.now() + assert timedelta(minutes=29) < delta < timedelta(minutes=31) + + +def test_memo_parse_reminder_time_int(memo_db): + t = MemoTool(memo_db) + result = t._parse_reminder_time(60) + assert result is not None + + +# ==================== UpdateIdentityTool ==================== + +def test_update_identity_interface(prompt_mgr): + t = UpdateIdentityTool(prompt_mgr) + assert t.get_name() == "update_identity" + assert "content" in t.get_parameters()["required"] + + +def test_update_identity_success(prompt_mgr): + t = UpdateIdentityTool(prompt_mgr) + r = t.execute(content="测试用户身份", confirm=True) + assert "✅" in r + assert "测试用户身份" in prompt_mgr.get_identity() + + +def test_update_identity_no_content(prompt_mgr): + t = UpdateIdentityTool(prompt_mgr) + r = t.execute(content=None, confirm=True) + assert "❌" in r + + +def test_update_identity_cancelled(prompt_mgr): + t = UpdateIdentityTool(prompt_mgr) + r = t.execute(content="x", confirm=False) + assert "❌" in r + + +# ==================== UpdateSoulTool ==================== + +def test_update_soul_interface(prompt_mgr): + t = UpdateSoulTool(prompt_mgr) + assert t.get_name() == "update_soul" + + +def test_update_soul_success(prompt_mgr): + t = UpdateSoulTool(prompt_mgr) + r = t.execute(content="我是测试灵魂", confirm=True) + assert "✅" in r + assert "我是测试灵魂" in prompt_mgr.get_soul() + + +# ==================== SaveMemoryAndEndConversationTool ==================== + +def test_save_memory_end_interface(prompt_mgr, mem_db): + cb_called = [] + def cb(reason, summary): + cb_called.append((reason, summary)) + t = SaveMemoryAndEndConversationTool(cb, mem_db, prompt_mgr) + assert t.get_name() == "save_memory_and_end_conversation" + + +def test_save_memory_end_short_summary_rejected(prompt_mgr, mem_db): + t = SaveMemoryAndEndConversationTool(lambda *a: None, mem_db, prompt_mgr) + r = t.execute(summary="短", current_status="状态") + assert "❌" in r + + +def test_save_memory_end_success(prompt_mgr, mem_db): + cb_called = [] + t = SaveMemoryAndEndConversationTool( + lambda reason, summary: cb_called.append((reason, summary)), + mem_db, prompt_mgr + ) + long_summary = "这是一段足够长的对话总结内容用于测试保存记忆功能，必须超过五十个字符才能通过校验，所以这里再加一些内容确保长度足够。" + r = t.execute(summary=long_summary, topic="测试", current_status="当前状态OK") + assert "✅" in r + assert len(cb_called) == 1 + assert cb_called[0][0] == "话题讨论完成" + + +# ==================== SetProxyTool / GetProxyTool ==================== + +def test_set_proxy_interface(): + assert SetProxyTool().get_name() == "set_proxy" + params = SetProxyTool().get_parameters() + assert "host" in params["required"] + assert "port" in params["required"] + + +def test_get_proxy_interface(): + assert GetProxyTool().get_name() == "get_proxy" + + +def test_get_proxy_when_disabled(): + t = GetProxyTool() + r = t.execute() + # 默认未启用代理 + assert "未启用" in r or "启用" in r + + +# ==================== AskUserTool ==================== + +def test_ask_user_interface(): + t = AskUserTool() + assert t.get_name() == "ask_user" + + +def test_ask_user_returns_question(): + t = AskUserTool() + r = t.execute(question="今天天气如何？") + assert "今天天气如何" in r + assert "询问用户" in r diff --git a/aibox/tools/__init__.py b/aibox/tools/__init__.py index dd9726c..9e10876 100644 --- a/aibox/tools/__init__.py +++ b/aibox/tools/__init__.py @@ -3,10 +3,16 @@ tools - aibox 工具系统基础模块（包） 抽离自 aibox.py 的工具基类与注册器，供 aibox.py 及后续迁移的 工具子类导入。当前包含： - - base.py : Tool 抽象基类 - - registry.py : ToolRegistry 工具注册器 - - basic_tools.py : 基础工具集（A 批：文件/网页/时间/计算器，无外部子系统依赖） -形成单向依赖：common <- tools <- aibox。 + - base.py : Tool 抽象基类 + - registry.py : ToolRegistry 工具注册器 + - basic_tools.py : 基础工具集（A 批：文件/网页/时间/计算器，无外部子系统依赖） + - command_tool.py : CommandLineTool 命令行执行工具（B 批） + - skill_tools.py : 技能搜索/重载/安装/创建 4 个工具（B 批） + - db_tools.py : MemoryTool/MemoTool 数据库工具（B 批） + - prompt_tools.py : 身份/灵魂更新 + 结束对话 3 个工具（B 批） + - proxy_tools.py : SetProxy/GetProxy 代理工具（B 批） + - misc_tools.py : AskUserTool 询问工具（B 批） +形成单向依赖：common <- subsystems <- tools <- aibox。 采用"重导出"策略：外部统一用 `from tools import Tool, ToolRegistry`， 调用点无需关心具体子模块所在文件。 @@ -23,6 +29,22 @@ from .basic_tools import ( GetCurrentTimeTool, CalculatorTool, ) +# B 批工具 +from .command_tool import CommandLineTool +from .skill_tools import ( + SearchSkillsTool, + ReloadSkillsTool, + InstallSkillTool, + CreateSkillTool, +) +from .db_tools import MemoryTool, MemoTool +from .prompt_tools import ( + UpdateIdentityTool, + UpdateSoulTool, + SaveMemoryAndEndConversationTool, +) +from .proxy_tools import SetProxyTool, GetProxyTool +from .misc_tools import AskUserTool __all__ = [ "Tool", @@ -35,4 +57,18 @@ __all__ = [ "SimpleWebCrawlerTool", "GetCurrentTimeTool", "CalculatorTool", + # B 批工具 + "CommandLineTool", + "SearchSkillsTool", + "ReloadSkillsTool", + "InstallSkillTool", + "CreateSkillTool", + "MemoryTool", + "MemoTool", + "UpdateIdentityTool", + "UpdateSoulTool", + "SaveMemoryAndEndConversationTool", + "SetProxyTool", + "GetProxyTool", + "AskUserTool", ] diff --git a/aibox/tools/command_tool.py b/aibox/tools/command_tool.py new file mode 100644 index 0000000..527a38b --- /dev/null +++ b/aibox/tools/command_tool.py @@ -0,0 +1,250 @@ +#!/usr/bin/env python3 +""" +tools.command_tool - 命令行执行工具 + +抽离自 aibox.py，使用抽象的 CommandExecutor 实现命令执行（流式 + 超时控制）。 +工具本身不持有 subprocess 逻辑，而是委托给 SubprocessCommandExecutor。 + +依赖：common (ConfigManager, constants) + tools.base.Tool + subsystems.executors +不反向依赖 aibox 任何符号。 +""" + +import os +import queue +import subprocess +import time +from typing import Dict + +from common import ConfigManager +from subsystems import CommandExecutor, SubprocessCommandExecutor +from .base import Tool + + +class CommandLineTool(Tool): + """命令行执行工具 - 使用抽象的命令执行器（支持流式输出和超时控制）""" + + def __init__(self, executor: CommandExecutor = None, logger=None): + super().__init__(logger) + self.executor = executor or SubprocessCommandExecutor(logger) + + def get_name(self) -> str: + return "run_command" + + def get_description(self) -> str: + return """执行任意系统命令，支持实时输出显示和超时控制""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "command": { + "type": "string", + "description": "要执行的命令" + }, + "timeout": { + "type": "integer", + "default": 30, + "description": "命令执行超时时间（秒），超时后将终止命令执行。默认30秒" + }, + "work_dir": { + "type": "string", + "default": ".", + "description": "命令执行的工作目录" + }, + "env": { + "type": "object", + "description": "额外的环境变量" + } + }, + "required": ["command"] + } + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行命令，支持超时控制""" + command = kwargs.get("command") + timeout = kwargs.get("timeout", 30) + work_dir = kwargs.get("work_dir", ".") + extra_env = kwargs.get("env", {}) + + if not command: + output_queue.put(("line", "❌ 错误：缺少命令参数\n")) + output_queue.put(("complete", "❌ 错误：缺少命令参数")) + return + + # 发送开始执行的消息 + output_queue.put(("line", f"\n🔧 执行命令: {command}\n")) + output_queue.put(("line", f"⏱️ 超时设置: {timeout} 秒\n")) + if work_dir != ".": + output_queue.put(("line", f"📁 工作目录: {work_dir}\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + # 准备环境变量 + env = os.environ.copy() + if extra_env: + env.update(extra_env) + + try: + # 使用 subprocess.Popen 实现流式输出，支持超时 + process = subprocess.Popen( + command, + shell=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + stdin=subprocess.PIPE, + text=True, + bufsize=1, + universal_newlines=True, + env=env, + cwd=os.path.expanduser(work_dir) + ) + + start_time = time.time() + output_lines = [] + timeout_occurred = False + + # 实时读取输出 + while True: + # 检查超时 + elapsed = time.time() - start_time + if timeout > 0 and elapsed > timeout: + process.terminate() + timeout_occurred = True + output_queue.put(("line", f"\n⏰ 命令执行超时（{timeout}秒），已终止\n")) + break + + # 使用非阻塞方式读取一行 + import select + import sys + + # 检查是否有输出可读 + if sys.platform == 'win32': + # Windows 下使用 readline 带超时 + try: + import msvcrt + if msvcrt.kbhit(): + pass + line = process.stdout.readline() + if not line and process.poll() is not None: + break + if line: + line = line.rstrip('\n') + output_queue.put(("line", line + "\n")) + output_lines.append(line) + else: + time.sleep(0.05) + except: + line = process.stdout.readline() + if not line and process.poll() is not None: + break + if line: + line = line.rstrip('\n') + output_queue.put(("line", line + "\n")) + output_lines.append(line) + else: + # Unix 系统可以使用 select + import select + rlist, _, _ = select.select([process.stdout], [], [], 0.5) + if rlist: + line = process.stdout.readline() + if not line and process.poll() is not None: + break + if line: + line = line.rstrip('\n') + output_queue.put(("line", line + "\n")) + output_lines.append(line) + else: + # 检查进程是否结束 + if process.poll() is not None: + # 读取剩余输出 + remaining = process.stdout.read() + if remaining: + remaining = remaining.rstrip('\n') + if remaining: + output_queue.put(("line", remaining + "\n")) + output_lines.append(remaining) + break + + # 如果超时，确保进程被终止 + if timeout_occurred and process.poll() is None: + process.kill() + process.wait(timeout=2) + + # 获取返回码 + return_code = process.returncode if process.returncode is not None else -1 + + output_queue.put(("line", "-" * 50 + "\n")) + + if timeout_occurred: + output_queue.put(("line", f"⏰ 命令执行超时（{timeout}秒），返回码: {return_code}\n")) + else: + output_queue.put(("line", f"✅ 命令执行完成，返回码: {return_code}\n")) + + # 准备完整结果 + full_output = "\n".join(output_lines) + data = { + "输出": full_output if full_output else "(无输出)", + "返回码": return_code + } + + if timeout_occurred: + result = self.format_result(False, f"命令执行超时（{timeout}秒）", data) + else: + result = self.format_result(return_code == 0, "命令执行完成", data) + + output_queue.put(("complete", result)) + + except subprocess.TimeoutExpired: + error_msg = f"❌ 命令执行超时（{timeout}秒）" + output_queue.put(("line", error_msg + "\n")) + output_queue.put(("complete", error_msg)) + except FileNotFoundError as e: + error_msg = f"❌ 命令执行错误: 命令未找到 - {str(e)}" + output_queue.put(("line", error_msg + "\n")) + output_queue.put(("complete", error_msg)) + except PermissionError as e: + error_msg = f"❌ 命令执行错误: 权限不足 - {str(e)}" + output_queue.put(("line", error_msg + "\n")) + output_queue.put(("complete", error_msg)) + except Exception as e: + error_msg = f"❌ 执行错误: {str(e)}" + output_queue.put(("line", error_msg + "\n")) + output_queue.put(("complete", error_msg)) + finally: + # 确保进程被清理 + if 'process' in locals() and process and process.poll() is None: + try: + process.terminate() + process.wait(timeout=2) + except: + process.kill() + + def execute(self, command: str, timeout: int = 30, + work_dir: str = ".", env: Dict = None, **kwargs) -> str: + """执行命令 - 仅支持流式执行（命令工具强制流式输出，无 stream 参数）""" + if error := self.validate_params(["command"], command=command): + return error + + # 如果正在流式输出，直接返回空（实际输出通过队列） + if self.streaming_output and self.output_queue: + return "" + + # 命令工具固定走流式执行：通过临时队列收集完整结果 + output_queue = queue.Queue() + self.execute_streaming( + output_queue, + command=command, + timeout=timeout, + work_dir=work_dir, + env=env + ) + + # 收集流式执行的最终结果（complete 消息） + final_result = "" + while not output_queue.empty(): + try: + msg_type, content = output_queue.get_nowait() + if msg_type == "complete": + final_result = content + except queue.Empty: + break + return final_result diff --git a/aibox/tools/db_tools.py b/aibox/tools/db_tools.py new file mode 100644 index 0000000..af530db --- /dev/null +++ b/aibox/tools/db_tools.py @@ -0,0 +1,818 @@ +#!/usr/bin/env python3 +""" +tools.db_tools - 数据库相关工具 + +抽离自 aibox.py，包含 2 个数据库工具： + - MemoryTool : 记忆管理（保存/搜索/查看/删除，支持分页） + - MemoTool : 任务管理（创建/查看/完成/删除/搜索/统计） + +依赖：common (constants, Logger) + tools.base.Tool + + subsystems.db.MemoryDatabase/MemoDatabase +不反向依赖 aibox 任何符号。 +""" + +import queue +import re +from datetime import datetime, timedelta +from typing import Any, Dict, Optional + +from common import constants, Logger +from subsystems import MemoryDatabase, MemoDatabase +from .base import Tool + + +# ==================== 记忆工具 ==================== + +class MemoryTool(Tool): + """记忆工具 - 用于存储、检索和删除对话记忆（支持流式输出，支持分页）""" + + def __init__(self, memory_db: MemoryDatabase, logger: Optional[Logger] = None): + super().__init__(logger) + self.memory_db = memory_db + + def get_name(self) -> str: + return "memory" + + def get_description(self) -> str: + return """记忆管理工具 - 存储、检索和删除重要的对话信息 +支持：保存总结、多关键词搜索记忆（支持分页）、查看最近的记忆、删除无用记忆""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "action": { + "type": "string", + "description": "操作类型", + "enum": ["save", "search", "recent", "get", "delete"] + }, + "summary": {"type": "string", "description": "要保存的总结内容（用于save操作）"}, + "topic": {"type": "string", "description": "对话主题（用于save操作）"}, + "content": {"type": "string", "description": "详细内容（可选，用于save操作）"}, + "keywords": { + "type": "array", "items": {"type": "string"}, + "description": "搜索关键词列表" + }, + "query": {"type": "string", "description": "搜索关键词（单个关键词）"}, + "match_all": {"type": "boolean", "default": False}, + "memory_id": {"type": "integer"}, + "memory_ids": {"type": "array", "items": {"type": "integer"}}, + "importance": {"type": "integer", "default": 2}, + "limit": {"type": "integer", "default": 10, "description": "每页条数（默认10）"}, + "page": {"type": "integer", "default": 1, "description": "页码（从1开始）"}, + "confirm": {"type": "boolean", "default": False} + }, + "required": ["action"] + } + + def execute(self, action: str, **kwargs) -> str: + try: + if action == "save": + return self._save_memory(kwargs) + elif action == "search": + return self._search_memories(kwargs) + elif action == "recent": + return self._get_recent_memories(kwargs) + elif action == "get": + return self._get_memory_by_id(kwargs) + elif action == "delete": + return self._delete_memory(kwargs) + else: + return f"❌ 未知操作: {action}" + except Exception as e: + return f"❌ 记忆操作错误: {str(e)}" + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行记忆操作""" + action = kwargs.get("action") + + # 发送开始执行的消息 + output_queue.put(("line", f"\n🧠 执行记忆操作: {action}\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + # 执行操作 + result = self.execute(**kwargs) + + # 发送结果 + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + def _save_memory(self, kwargs: Dict) -> str: + summary = kwargs.get("summary") + if not summary: + return "❌ 需要提供总结内容" + + topic = kwargs.get("topic", "未命名对话") + content = kwargs.get("content") + importance = kwargs.get("importance", constants.MEMORY_DEFAULT_IMPORTANCE) + + memory_id = self.memory_db.add_memory( + topic=topic, + summary=summary, + content=content, + importance=importance + ) + + return self.format_result(True, "记忆已保存", { + "记忆ID": memory_id, + "主题": topic, + "总结": summary[:100] + "..." if len(summary) > 100 else summary + }) + + def _search_memories(self, kwargs: Dict) -> str: + """搜索记忆 - 支持分页""" + keywords = kwargs.get("keywords", []) + query = kwargs.get("query", "") + match_all = kwargs.get("match_all", False) + page = kwargs.get("page", 1) + page_size = kwargs.get("limit", 10) # 使用 limit 作为 page_size + + # 确保页码和每页条数有效 + page = max(1, page) + page_size = max(1, min(50, page_size)) # 每页最多50条 + + # 处理关键词 + if keywords and isinstance(keywords, list): + pass + elif query: + keywords = query.split() + else: + return "❌ 需要提供搜索关键词（keywords 或 query）" + + if not keywords: + return "❌ 关键词列表为空" + + # 先获取所有结果（不分页，用于计算总数） + all_results = self.memory_db.search_memories(keywords, match_all, limit=None) + + if not all_results: + match_mode = "所有" if match_all else "任一" + return f"❌ 未找到包含 {match_mode} 关键词 {keywords} 的记忆" + + # 计算分页信息 + total_count = len(all_results) + total_pages = (total_count + page_size - 1) // page_size + + # 检查页码是否有效 + if page > total_pages and total_pages > 0: + return f"❌ 页码 {page} 超出范围，共有 {total_pages} 页" + + # 计算当前页的起止索引 + start_idx = (page - 1) * page_size + end_idx = min(start_idx + page_size, total_count) + + # 获取当前页的结果 + page_results = all_results[start_idx:end_idx] + + # 构建输出 + match_mode = "所有关键词" if match_all else "任一关键词" + output = [ + f"🔍 搜索 {match_mode} {keywords} 找到 {total_count} 条记忆", + f"📄 第 {page}/{total_pages} 页 (每页 {page_size} 条)", + "-" * 40 + ] + + for i, mem in enumerate(page_results, 1): + # 全局序号 + global_idx = start_idx + i + output.append(f"\n{global_idx}. [ID: {mem['id']}] [{mem['end_time']}] {mem['topic']}") + output.append(f" 总结: {mem['summary'][:200]}") + if mem.get('tags'): + output.append(f" 标签: {', '.join(mem['tags'])}") + if mem.get('importance'): + output.append(f" 重要性: {'⭐' * mem['importance']} ({mem['importance']}/5)") + + # 添加翻页提示 + if total_pages > 1: + output.append(f"\n📖 翻页提示: 使用 page 参数查看更多结果") + output.append(f" 例如: memory search query='{query}' page={page + 1 if page < total_pages else page} limit={page_size}") + if page < total_pages: + output.append(f" ➡️ 下一页: page={page + 1}") + if page > 1: + output.append(f" ⬅️ 上一页: page={page - 1}") + + return "\n".join(output) + + def _get_recent_memories(self, kwargs: Dict) -> str: + """获取最近的记忆 - 支持分页""" + page = kwargs.get("page", 1) + page_size = kwargs.get("limit", 10) + + # 确保页码和每页条数有效 + page = max(1, page) + page_size = max(1, min(50, page_size)) + + # 先获取所有结果 + all_results = self.memory_db.get_recent_memories(limit=None) + + if not all_results: + return "📭 没有找到记忆" + + # 计算分页信息 + total_count = len(all_results) + total_pages = (total_count + page_size - 1) // page_size + + # 检查页码是否有效 + if page > total_pages and total_pages > 0: + return f"❌ 页码 {page} 超出范围，共有 {total_pages} 页" + + # 计算当前页的起止索引 + start_idx = (page - 1) * page_size + end_idx = min(start_idx + page_size, total_count) + + # 获取当前页的结果 + page_results = all_results[start_idx:end_idx] + + output = [f"📚 最近 {total_count} 条记忆 (第 {page}/{total_pages} 页，每页 {page_size} 条):"] + output.append("-" * 40) + + for i, mem in enumerate(page_results, 1): + global_idx = start_idx + i + output.append(f"\n{global_idx}. [ID: {mem['id']}] [{mem['end_time']}] {mem['topic']}") + output.append(f" 总结: {mem['summary'][:100]}") + + # 添加翻页提示 + if total_pages > 1: + output.append(f"\n📖 翻页提示: 使用 page 参数查看更多结果") + output.append(f" 例如: memory recent page={page + 1 if page < total_pages else page} limit={page_size}") + + return "\n".join(output) + + def _get_memory_by_id(self, kwargs: Dict) -> str: + """根据ID获取记忆""" + memory_id = kwargs.get("memory_id") + if not memory_id: + return "❌ 需要提供记忆ID" + + memory = self.memory_db.get_memory_by_id(memory_id) + if not memory: + return f"❌ 未找到ID为 {memory_id} 的记忆" + + output = [ + f"📌 记忆 #{memory_id}", + f"主题: {memory['topic']}", + f"时间: {memory['start_time']} - {memory['end_time']}", + f"总结: {memory['summary']}", + ] + + if memory.get('content'): + output.append(f"详细内容:\n{memory['content']}") + + if memory.get('tags'): + output.append(f"标签: {', '.join(memory['tags'])}") + + if memory.get('importance'): + output.append(f"重要性: {'⭐' * memory['importance']} ({memory['importance']}/5)") + + return "\n".join(output) + + def _delete_memory(self, kwargs: Dict) -> str: + """删除记忆""" + confirm = kwargs.get("confirm", False) + if not confirm: + return "❌ 需要设置 confirm=True 来确认删除操作" + + memory_id = kwargs.get("memory_id") + memory_ids = kwargs.get("memory_ids", []) + + deleted_count = 0 + + if memory_id: + if self.memory_db.delete_memory(memory_id): + deleted_count = 1 + elif memory_ids: + deleted_count = self.memory_db.delete_memories_by_ids(memory_ids) + else: + return "❌ 需要提供 memory_id 或 memory_ids" + + if deleted_count == 0: + return f"❌ 未找到要删除的记忆" + + return self.format_result(True, f"成功删除 {deleted_count} 条记忆", {"删除数量": deleted_count}) + + +# ==================== 备忘录工具 ==================== + +class MemoTool(Tool): + """备忘录工具 - 用于管理任务（所有任务都是对话任务）""" + + def __init__(self, memo_db: MemoDatabase, logger: Optional[Logger] = None): + super().__init__(logger) + self.memo_db = memo_db + + def get_name(self) -> str: + return "memo" + + def get_description(self) -> str: + return """任务管理工具 - 创建、查看、完成任务 +支持：创建任务（支持一次性/重复、即时任务）、查看待办、标记完成、删除任务、搜索任务""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "action": { + "type": "string", + "description": "操作类型", + "enum": ["add", "list", "get", "complete", "delete", "search", "stats"] + }, + "title": { + "type": "string", + "description": "任务标题（用于add操作）" + }, + "content": { + "type": "string", + "description": "任务内容（可选，用于add操作）" + }, + "reminder_time": { + "type": "string", + "description": "执行时间（格式：YYYY-MM-DD HH:MM:SS，或相对时间如 '+1h', '+30m', 'tomorrow 9am'）" + }, + "reminder_minutes": { + "type": "integer", + "description": "延迟分钟数（从现在开始，用于快速设置任务）" + }, + "tags": { + "type": "array", + "items": {"type": "string"}, + "description": "标签列表" + }, + "priority": { + "type": "integer", + "description": "优先级 1-5（默认1）", + "default": 1 + }, + "repeat_type": { + "type": "string", + "description": "重复类型（none=一次性, daily=每天, weekly=每周, monthly=每月, custom=自定义）", + "enum": ["none", "daily", "weekly", "monthly", "custom"], + "default": "none" + }, + "repeat_interval": { + "type": "integer", + "description": "自定义重复间隔，当repeat_type='custom'时使用", + "default": 0 + }, + "repeat_interval_unit": { + "type": "string", + "description": "重复间隔单位（minutes=分钟, hours=小时, days=天），当repeat_type='custom'时使用", + "enum": ["minutes", "hours", "days"], + "default": "days" + }, + "repeat_end_time": { + "type": "string", + "description": "重复结束时间，None表示永久重复" + }, + "is_immediate": { + "type": "boolean", + "description": "是否为即时任务（立即执行）", + "default": False + }, + "memo_id": { + "type": "integer", + "description": "任务ID（用于get、complete、delete操作）" + }, + "memo_ids": { + "type": "array", + "items": {"type": "integer"}, + "description": "任务ID列表（用于批量删除）" + }, + "force_complete": { + "type": "boolean", + "description": "强制完成（忽略重复规则）", + "default": False + }, + "keyword": { + "type": "string", + "description": "搜索关键词（用于search操作）" + }, + "include_completed": { + "type": "boolean", + "description": "是否包含已完成的任务（默认False）", + "default": False + }, + "limit": { + "type": "integer", + "description": "返回数量限制（默认20）", + "default": 20 + }, + "confirm": { + "type": "boolean", + "description": "确认删除操作", + "default": False + } + }, + "required": ["action"] + } + + def _parse_reminder_time(self, reminder_input: Any) -> Optional[str]: + """解析提醒时间输入""" + if not reminder_input: + return None + + if isinstance(reminder_input, int): + reminder_time = datetime.now() + timedelta(minutes=reminder_input) + return reminder_time.strftime(constants.DATETIME_FORMAT) + + if isinstance(reminder_input, str): + reminder_input = reminder_input.strip() + + if reminder_input.startswith('+'): + match = re.match(r'\+(\d+)([mhd])', reminder_input) + if match: + value, unit = int(match.group(1)), match.group(2) + if unit == 'm': + delta = timedelta(minutes=value) + elif unit == 'h': + delta = timedelta(hours=value) + elif unit == 'd': + delta = timedelta(days=value) + else: + return None + + reminder_time = datetime.now() + delta + return reminder_time.strftime(constants.DATETIME_FORMAT) + + try: + reminder_time = datetime.strptime(reminder_input, constants.DATETIME_FORMAT) + return reminder_time.strftime(constants.DATETIME_FORMAT) + except ValueError: + pass + + now = datetime.now() + reminder_input_lower = reminder_input.lower() + + if 'tomorrow' in reminder_input_lower: + tomorrow = now + timedelta(days=1) + time_match = re.search(r'(\d{1,2})(?::(\d{2}))?\s*(am|pm)?', reminder_input_lower) + if time_match: + hour = int(time_match.group(1)) + minute = int(time_match.group(2)) if time_match.group(2) else 0 + ampm = time_match.group(3) + + if ampm: + if ampm == 'pm' and hour < 12: + hour += 12 + elif ampm == 'am' and hour == 12: + hour = 0 + + reminder_time = tomorrow.replace(hour=hour, minute=minute, second=0) + else: + reminder_time = tomorrow.replace(hour=9, minute=0, second=0) + + return reminder_time.strftime(constants.DATETIME_FORMAT) + + return None + + def execute(self, action: str, **kwargs) -> str: + """执行备忘录操作""" + try: + if action == "add": + return self._add_memo(kwargs) + elif action == "list": + return self._list_memos(kwargs) + elif action == "get": + return self._get_memo(kwargs) + elif action == "complete": + return self._complete_memo(kwargs) + elif action == "delete": + return self._delete_memo(kwargs) + elif action == "search": + return self._search_memos(kwargs) + elif action == "stats": + return self._get_stats(kwargs) + else: + return f"❌ 未知操作: {action}" + except Exception as e: + return f"❌ 任务操作错误: {str(e)}" + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行备忘录操作""" + action = kwargs.get("action") + + # 发送开始执行的消息 + output_queue.put(("line", f"\n📝 执行备忘录操作: {action}\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + # 执行操作 + result = self.execute(**kwargs) + + # 发送结果 + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + def _add_memo(self, kwargs: Dict) -> str: + """添加任务 - 所有任务都是对话任务，不需要task_type""" + title = kwargs.get("title") + if not title: + return "❌ 需要提供任务标题" + + content = kwargs.get("content") + reminder_minutes = kwargs.get("reminder_minutes") + reminder_time_input = kwargs.get("reminder_time") + tags = kwargs.get("tags", []) + priority = kwargs.get("priority", 1) + repeat_type = kwargs.get("repeat_type", "none") + repeat_interval = kwargs.get("repeat_interval", 0) + repeat_interval_unit = kwargs.get("repeat_interval_unit", "days") + repeat_end_time = kwargs.get("repeat_end_time") + is_immediate = kwargs.get("is_immediate", False) + + if repeat_type not in ["none", "daily", "weekly", "monthly", "custom"]: + return f"❌ 无效的重复类型: {repeat_type}" + + if repeat_type == "custom" and repeat_interval <= 0: + return "❌ 自定义重复需要设置有效的 repeat_interval (>0)" + + reminder_time = None + if is_immediate: + # 即时任务：设置为当前时间，确保立即执行 + reminder_time = datetime.now().strftime(constants.DATETIME_FORMAT) + elif reminder_minutes: + reminder_time = self._parse_reminder_time(reminder_minutes) + elif reminder_time_input: + reminder_time = self._parse_reminder_time(reminder_time_input) + else: + # 默认30分钟后 + reminder_time = self._parse_reminder_time(30) + + memo_id = self.memo_db.add_memo( + title=title, + content=content, + reminder_time=reminder_time, + tags=tags, + priority=priority, + created_by="AI", + source="user", + repeat_type=repeat_type, + repeat_interval=repeat_interval, + repeat_interval_unit=repeat_interval_unit, + repeat_end_time=repeat_end_time, + is_immediate=is_immediate + ) + + data = { + "任务ID": memo_id, + "标题": title, + } + + if is_immediate: + data["任务类型"] = "即时任务" + else: + if reminder_time: + data["执行时间"] = reminder_time + + if repeat_type != "none": + if repeat_type == "custom": + repeat_info = f"重复类型: 每{repeat_interval} {repeat_interval_unit}" + else: + repeat_map = { + 'daily': '每天', + 'weekly': '每周', + 'monthly': '每月' + } + repeat_info = f"重复类型: {repeat_map.get(repeat_type, repeat_type)}" + + if repeat_end_time: + repeat_info += f", 结束于: {repeat_end_time}" + else: + repeat_info += ", 永久重复" + data["重复信息"] = repeat_info + + return self.format_result(True, "任务已创建", data) + + def _list_memos(self, kwargs: Dict) -> str: + """列出任务""" + include_completed = kwargs.get("include_completed", False) + limit = kwargs.get("limit", 20) + + memos = self.memo_db.get_all_memos(include_completed, limit) + + if not memos: + return "📭 没有找到任务" + + status = "所有" if include_completed else "待执行" + output = [f"📋 {status}任务 (共 {len(memos)} 个):"] + + now = datetime.now() + + for i, memo in enumerate(memos, 1): + status_icon = "✅" if memo['is_completed'] else "⏰" + priority_icon = "🔴" * memo['priority'] + source_tag = f"[{memo.get('source', '未知')}]" if memo.get('source') else "" + + immediate_tag = "⚡" if memo.get('is_immediate') else "" + + repeat_tag = "" + if memo['repeat_type'] != 'none': + if memo['repeat_type'] == 'custom': + unit = memo.get('repeat_interval_unit', 'days') + unit_map = {'minutes': '分钟', 'hours': '小时', 'days': '天'} + repeat_tag = f" [每{memo['repeat_interval']}{unit_map.get(unit, unit)}]" + else: + repeat_map = { + 'daily': ' [每天]', + 'weekly': ' [每周]', + 'monthly': ' [每月]' + } + repeat_tag = repeat_map.get(memo['repeat_type'], ' [重复]') + + output.append(f"\n{i}. {status_icon}{immediate_tag} [ID: {memo['id']}] {source_tag}{repeat_tag} {memo['title']} {priority_icon}") + + if memo['reminder_time']: + try: + reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) + if not memo['is_completed'] and reminder < now: + output.append(f" ⚠️ 已过期: {memo['reminder_time']}") + else: + output.append(f" ⏱️ 执行: {memo['reminder_time']}") + except: + output.append(f" ⏱️ 执行: {memo['reminder_time']}") + + if memo.get('trigger_count', 0) > 0: + output.append(f" 🔄 已执行 {memo['trigger_count']} 次") + + if memo['content']: + preview = memo['content'][:100] + "..." if len(memo['content']) > 100 else memo['content'] + output.append(f" 📝 {preview}") + + if memo['tags']: + output.append(f" 🏷️ {', '.join(memo['tags'])}") + + return "\n".join(output) + + def _get_memo(self, kwargs: Dict) -> str: + """获取单个任务""" + memo_id = kwargs.get("memo_id") + if not memo_id: + return "❌ 需要提供任务ID" + + memo = self.memo_db.get_memo(memo_id) + if not memo: + return f"❌ 未找到ID为 {memo_id} 的任务" + + status = "✅ 已完成" if memo['is_completed'] else "⏰ 待执行" + if memo['is_completed']: + status += f" (完成于: {memo['completed_at']})" + + immediate_tag = "⚡ 即时任务" if memo.get('is_immediate') else "" + + output = [ + f"📌 任务 #{memo_id} {immediate_tag}", + f"📋 状态: {status}", + f"📝 标题: {memo['title']}", + ] + + if memo.get('source'): + output.append(f"📎 来源: {memo['source']}") + + if memo['repeat_type'] != 'none': + if memo['repeat_type'] == 'custom': + unit_map = {'minutes': '分钟', 'hours': '小时', 'days': '天'} + unit = memo.get('repeat_interval_unit', 'days') + repeat_info = f"🔄 重复: 每{memo['repeat_interval']} {unit_map.get(unit, unit)}" + else: + repeat_map = { + 'daily': '每天', + 'weekly': '每周', + 'monthly': '每月' + } + repeat_info = f"🔄 重复: {repeat_map.get(memo['repeat_type'], memo['repeat_type'])}" + + if memo['repeat_end_time']: + repeat_info += f", 结束于: {memo['repeat_end_time']}" + else: + repeat_info += ", 永久重复" + output.append(repeat_info) + + if memo.get('trigger_count', 0) > 0: + output.append(f" 📊 已执行 {memo['trigger_count']} 次") + if memo.get('last_triggered_at'): + output.append(f" ⏱️ 上次执行: {memo['last_triggered_at']}") + + if memo['content']: + output.append(f"📄 内容:\n{memo['content']}") + + if memo['reminder_time']: + try: + now = datetime.now() + reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) + if not memo['is_completed'] and reminder < now: + output.append(f"⚠️ 执行时间 (已过期): {memo['reminder_time']}") + else: + output.append(f"⏰ 执行时间: {memo['reminder_time']}") + except: + output.append(f"⏰ 执行时间: {memo['reminder_time']}") + + if memo['tags']: + output.append(f"🏷️ 标签: {', '.join(memo['tags'])}") + + output.append(f"🔢 优先级: {'🔴' * memo['priority']} ({memo['priority']}/5)") + output.append(f"📅 创建时间: {memo['created_at']}") + output.append(f"👤 创建者: {memo['created_by'] or '系统'}") + + return "\n".join(output) + + def _complete_memo(self, kwargs: Dict) -> str: + """完成任务""" + memo_id = kwargs.get("memo_id") + if not memo_id: + return "❌ 需要提供任务ID" + + force_complete = kwargs.get("force_complete", False) + + memo = self.memo_db.get_memo(memo_id) + if not memo: + return f"❌ 未找到ID为 {memo_id} 的任务" + + if memo['is_completed'] and not force_complete: + return f"ℹ️ 任务 #{memo_id} 已经完成了" + + if self.memo_db.complete_memo(memo_id, force_complete): + memo = self.memo_db.get_memo(memo_id) + data = { + "任务ID": memo_id, + "标题": memo['title'] + } + + if memo['repeat_type'] != 'none' and not memo['is_completed']: + data["下一次执行"] = memo['reminder_time'] + data["已执行次数"] = memo.get('trigger_count', 0) + return self.format_result(True, "任务已处理，已更新下一次执行时间", data) + else: + return self.format_result(True, "任务已完成", data) + else: + return f"❌ 无法完成任务 #{memo_id}" + + def _delete_memo(self, kwargs: Dict) -> str: + """删除任务""" + confirm = kwargs.get("confirm", False) + if not confirm: + return "❌ 需要设置 confirm=True 来确认删除操作" + + memo_id = kwargs.get("memo_id") + if memo_id: + if self.memo_db.complete_memo(memo_id, force_complete=True): + return self.format_result(True, f"任务 #{memo_id} 已删除") + else: + return f"❌ 无法删除任务 #{memo_id}" + else: + return "❌ 需要提供 memo_id" + + def _search_memos(self, kwargs: Dict) -> str: + """搜索任务""" + keyword = kwargs.get("keyword") + if not keyword: + return "❌ 需要提供搜索关键词" + + include_completed = kwargs.get("include_completed", False) + limit = kwargs.get("limit", 20) + + all_memos = self.memo_db.get_all_memos(include_completed, limit=100) + + results = [] + keyword_lower = keyword.lower() + for memo in all_memos: + if (keyword_lower in memo['title'].lower() or + (memo['content'] and keyword_lower in memo['content'].lower())): + results.append(memo) + + if not results: + return f"❌ 未找到包含 '{keyword}' 的任务" + + output = [f"🔍 搜索 '{keyword}' 找到 {len(results)} 个任务:"] + now = datetime.now() + + for i, memo in enumerate(results[:limit], 1): + status_icon = "✅" if memo['is_completed'] else "⏰" + immediate_tag = "⚡" if memo.get('is_immediate') else "" + output.append(f"\n{i}. {status_icon}{immediate_tag} [ID: {memo['id']}] {memo['title']}") + if memo['reminder_time']: + try: + reminder = datetime.strptime(memo['reminder_time'], constants.DATETIME_FORMAT) + if not memo['is_completed'] and reminder < now: + output.append(f" ⚠️ 已过期: {memo['reminder_time']}") + else: + output.append(f" ⏱️ {memo['reminder_time']}") + except: + output.append(f" ⏱️ {memo['reminder_time']}") + + return "\n".join(output) + + def _get_stats(self, kwargs: Dict) -> str: + """获取任务统计信息""" + stats = self.memo_db.get_memo_stats() + + output = [ + "📊 任务统计", + f" 总任务: {stats['total']} 个", + f" 待执行: {stats['pending']} 个", + f" 已过期: {stats['overdue']} 个", + f" 已完成: {stats['completed']} 个", + f" 即时任务: {stats['immediate']} 个" + ] + + if stats['overdue'] > 0: + output.append(f"\n⚠️ 有 {stats['overdue']} 个任务已过期，将自动执行") + + return "\n".join(output) diff --git a/aibox/tools/misc_tools.py b/aibox/tools/misc_tools.py new file mode 100644 index 0000000..0cfebc4 --- /dev/null +++ b/aibox/tools/misc_tools.py @@ -0,0 +1,48 @@ +#!/usr/bin/env python3 +""" +tools.misc_tools - 杂项工具 + +抽离自 aibox.py，包含不依赖任何子系统的独立工具： + - AskUserTool : 主动向用户提问 + +依赖：tools.base.Tool +不反向依赖 aibox 任何符号。 +""" + +from typing import Dict + +from .base import Tool + + +class AskUserTool(Tool): + """智能体遇到了麻烦，主动向用户问一个问题""" + + def get_name(self) -> str: + return "ask_user" + + def get_description(self) -> str: + return ( + "主动向用户提问的工具。你是用户的专属助手，应当主动关心用户、澄清需求、确认意图，" + "而不是被动等待指令。适用场景：指令模糊需要澄清、任务完成后追问后续需求、" + "关心用户近况、主动发现用户可能需要的帮助。请积极使用此工具与用户互动。" + ) + + def get_parameters(self) -> Dict: + ''' TODO: 调用子代理生成提示词 ''' + return { + "type": "object", + "properties": { + "question": { + "type": "string", + "description": ( + "你现在遇到一个问题，你想询问用户，现在提供你的问题。" + ) + } + }, + "required": [] + } + + def execute(self, question: str, **kwargs) -> str: + return ( + "你遇到了想搞清楚的事情: " + question + "\n你现在可以询问用户。" + ) diff --git a/aibox/tools/prompt_tools.py b/aibox/tools/prompt_tools.py new file mode 100644 index 0000000..7527f1a --- /dev/null +++ b/aibox/tools/prompt_tools.py @@ -0,0 +1,252 @@ +#!/usr/bin/env python3 +""" +tools.prompt_tools - 提示词相关工具 + +抽离自 aibox.py，包含 3 个工具： + - UpdateIdentityTool : 更新用户身份信息（00_IDENTITY.md） + - UpdateSoulTool : 更新智能体自我认识（99_SOUL.md） + - SaveMemoryAndEndConversationTool : 保存记忆并结束对话（同时更新状态） + +依赖：common (i18n, Logger) + tools.base.Tool + + subsystems.db.MemoryDatabase + subsystems.prompt_manager.PromptManager +不反向依赖 aibox 任何符号。 +""" + +import queue +from typing import Callable, Dict, Optional + +from common import i18n, Logger +from subsystems import MemoryDatabase, PromptManager +from .base import Tool + + +class UpdateIdentityTool(Tool): + """更新用户身份工具 - 修改 00_IDENTITY.md 文件""" + + def __init__(self, prompt_manager: PromptManager, logger: Optional[Logger] = None): + super().__init__(logger) + self.prompt_manager = prompt_manager + + def get_name(self) -> str: + return "update_identity" + + def get_description(self) -> str: + return """更新用户身份信息 - 修改 00_IDENTITY.md 文件 +当你了解到用户的身份信息时，调用此工具更新身份文件。 +如果文件不存在，会自动创建。""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": "用户身份信息内容（Markdown格式），包含用户的身份、特点、偏好等信息" + }, + "confirm": { + "type": "boolean", + "description": "确认更新身份信息", + "default": True + } + }, + "required": ["content"] + } + + def execute(self, content: str = None, confirm: bool = True, **kwargs) -> str: + if not content: + return "❌ 需要提供身份信息内容" + + if not confirm: + return "❌ 身份更新已取消，如需更新请设置 confirm=true" + + self.logger.info("开始更新用户身份信息...") + + if self.prompt_manager.update_identity(content): + return self.format_result(True, i18n.get('identity_updated', self.prompt_manager.IDENTITY_FILE), { + "文件": self.prompt_manager.IDENTITY_FILE, + "路径": self.prompt_manager.get_prompts_dir_path() + }) + else: + return self.format_result(False, "身份更新失败", { + "文件": self.prompt_manager.IDENTITY_FILE, + "路径": self.prompt_manager.get_prompts_dir_path() + }) + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行更新身份""" + output_queue.put(("line", "\n👤 更新用户身份信息...\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + result = self.execute(**kwargs) + + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + +class UpdateSoulTool(Tool): + """更新智能体自我认识工具 - 修改 99_SOUL.md 文件""" + + def __init__(self, prompt_manager: PromptManager, logger: Optional[Logger] = None): + super().__init__(logger) + self.prompt_manager = prompt_manager + + def get_name(self) -> str: + return "update_soul" + + def get_description(self) -> str: + return """更新智能体的自我认识定义 - 修改 99_SOUL.md 文件 +当你需要重新定义自己的身份、角色、性格、能力边界时，调用此工具更新自我认识文件。 +如果文件不存在，会自动创建默认定义。""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "content": { + "type": "string", + "description": "智能体自我认识定义内容（Markdown格式），包含：身份定义、角色描述、性格特点、能力边界、行为准则等" + }, + "confirm": { + "type": "boolean", + "description": "确认更新自我认识定义", + "default": True + } + }, + "required": ["content"] + } + + def execute(self, content: str = None, confirm: bool = True, **kwargs) -> str: + if not content: + return "❌ 需要提供智能体自我认识定义内容" + + if not confirm: + return "❌ 自我认识更新已取消，如需更新请设置 confirm=true" + + self.logger.info("开始更新智能体自我认识定义...") + + if self.prompt_manager.update_soul(content): + return self.format_result(True, "智能体自我认识已更新", { + "文件": self.prompt_manager.SOUL_FILE, + "路径": self.prompt_manager.get_prompts_dir_path() + }) + else: + return self.format_result(False, "自我认识更新失败", { + "文件": self.prompt_manager.SOUL_FILE, + "路径": self.prompt_manager.get_prompts_dir_path() + }) + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行更新自我认识""" + output_queue.put(("line", "\n🧠 更新智能体自我认识...\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + result = self.execute(**kwargs) + + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + +class SaveMemoryAndEndConversationTool(Tool): + """保存记忆并结束对话工具 - 增强版，同时更新智能体状态""" + + def __init__(self, exit_callback: Callable, memory_db: MemoryDatabase, + prompt_manager: PromptManager, logger: Optional[Logger] = None): + super().__init__(logger) + self.exit_callback = exit_callback + self.memory_db = memory_db + self.prompt_manager = prompt_manager + + def get_name(self) -> str: + return "save_memory_and_end_conversation" + + def get_description(self) -> str: + return """保存记忆并结束当前对话 - 必须提供详细的对话总结和当前状态！ +使用此工具时，AI会： +1. 将本次对话的重要内容保存到记忆库 +2. 提供完整的对话总结 +3. 描述自己现在的状态、事情的走向等，保存到 100_STATUS.md +4. 礼貌地结束对话""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "summary": { + "type": "string", + "description": "详细的对话总结（至少50字），包含：讨论的主题、关键问题、重要结论、达成的共识等" + }, + "topic": { + "type": "string", + "description": "对话主题（用于记忆分类，默认自动提取）" + }, + "reason": { + "type": "string", + "description": "结束对话的原因", + "default": "话题讨论完成" + }, + "current_status": { + "type": "string", + "description": "当前的智能体状态描述，包含：当前进展、下一步计划、待处理事项、整体走向等" + }, + "next_goals": { + "type": "string", + "description": "接下来的目标或计划（可选）" + } + }, + "required": ["summary", "current_status"] + } + + def execute(self, summary: str = None, topic: str = None, reason: str = "话题讨论完成", + current_status: str = None, next_goals: str = None, **kwargs) -> str: + if summary is None or len(summary) < 50: + return "❌ 请提供更详细的对话总结（至少50字），以便保存到记忆库" + + if current_status is None: + return "❌ 请提供当前状态描述，以便更新 100_STATUS.md" + + self.logger.info(f"AI保存记忆并结束对话: {reason}") + + # 保存到记忆库 + memory_result = "" + if self.memory_db and topic: + memory_id = self.memory_db.add_memory( + topic=topic, + summary=summary, + importance=3 # 提高重要性，因为是对话总结 + ) + self.logger.info(f"对话记忆已保存，ID: {memory_id}") + memory_result = f"记忆已保存，ID: {memory_id}" + + # 构建状态内容 + status_content = f"【当前状态】\n{current_status}\n\n" + if next_goals: + status_content += f"【下一步计划】\n{next_goals}\n\n" + status_content += f"【对话总结】\n{summary}" + + # 更新状态文件 + status_updated = self.prompt_manager.update_status(status_content) + + if self.exit_callback: + self.exit_callback(reason, summary) + + data = { + "原因": reason, + "总结预览": summary[:100] + "..." if len(summary) > 100 else summary, + "状态更新": "✅ 已更新" if status_updated else "❌ 更新失败", + "记忆已保存": True + } + + if memory_result: + data["记忆"] = memory_result + + return self.format_result(True, "对话记忆已保存，状态已更新，对话结束", data) + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行结束对话""" + output_queue.put(("line", "\n💾 保存记忆并结束对话\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + result = self.execute(**kwargs) + + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) diff --git a/aibox/tools/proxy_tools.py b/aibox/tools/proxy_tools.py new file mode 100644 index 0000000..03d2ef2 --- /dev/null +++ b/aibox/tools/proxy_tools.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +""" +tools.proxy_tools - 代理配置工具 + +抽离自 aibox.py，包含 2 个工具： + - SetProxyTool : 设置 SOCKS5 代理 + - GetProxyTool : 查看当前代理配置 + +依赖：common (ConfigManager) + tools.base.Tool +不反向依赖 aibox 任何符号。 +""" + +from typing import Dict + +from common import ConfigManager +from .base import Tool + + +class SetProxyTool(Tool): + """设置 SOCKS5 代理的工具""" + + def get_name(self) -> str: + return "set_proxy" + + def get_description(self) -> str: + return """设置智能体的 SOCKS5 代理配置。 +调用后所有 HTTP/HTTPS 请求（包括网页访问、API 调用）都会通过该代理。 +可随时修改，即时生效。""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "host": {"type": "string", "description": "代理服务器地址，如 127.0.0.1"}, + "port": {"type": "integer", "description": "代理端口，如 1080"}, + "username": {"type": "string", "description": "用户名（可选）"}, + "password": {"type": "string", "description": "密码（可选）"}, + "enabled": {"type": "boolean", "description": "是否启用代理，默认 true"}, + "save": {"type": "boolean", "description": "是否保存到配置文件（持久化），默认 true"} + }, + "required": ["host", "port"] + } + + def execute(self, host: str, port: int, username: str = None, password: str = None, + enabled: bool = True, save: bool = True, **kwargs) -> str: + config = ConfigManager() + config.set_proxy(host, port, username, password, enabled, save) + proxy_info = config.get_proxy_info() + data = { + "代理状态": "启用" if proxy_info["enabled"] else "禁用", + "服务器": f"{proxy_info['host']}:{proxy_info['port']}", + "认证": "有" if proxy_info.get("username") else "无", + "持久化": "已保存" if save else "仅内存" + } + return self.format_result(True, "代理设置成功", data) + + +class GetProxyTool(Tool): + """查看当前代理配置""" + + def get_name(self) -> str: + return "get_proxy" + + def get_description(self) -> str: + return "查看当前 SOCKS5 代理配置信息" + + def get_parameters(self) -> Dict: + return {"type": "object", "properties": {}} + + def execute(self, **kwargs) -> str: + config = ConfigManager() + info = config.get_proxy_info() + if info["enabled"]: + data = { + "代理状态": "启用", + "服务器": f"{info['host']}:{info['port']}", + "认证": "有" if info.get("username") else "无" + } + return self.format_result(True, "当前代理配置", data) + else: + return "❌ 当前代理未启用，所有请求走直连。" diff --git a/aibox/tools/skill_tools.py b/aibox/tools/skill_tools.py new file mode 100644 index 0000000..39d9407 --- /dev/null +++ b/aibox/tools/skill_tools.py @@ -0,0 +1,400 @@ +#!/usr/bin/env python3 +""" +tools.skill_tools - 技能相关工具 + +抽离自 aibox.py，包含 4 个技能管理工具： + - SearchSkillsTool : 按关键字搜索扩展技能 + - ReloadSkillsTool : 热重载技能 + - InstallSkillTool : 返回技能安装指引 + - CreateSkillTool : 返回创建技能的指令提示词 + +依赖：common (constants, i18n, Logger) + tools.base.Tool + tools.base.ToolRegistry + + subsystems.skill_manager.SkillManager +不反向依赖 aibox 任何符号。 +""" + +import os +import queue +import re +from typing import Dict, Optional + +from common import constants, i18n, Logger +from subsystems import SkillManager +from .base import Tool +from .registry import ToolRegistry + + +class SearchSkillsTool(Tool): + """扩展技能搜索工具 - 按关键字搜索 ~/.aibox/skills/ 下的技能文件。 + + 支持两种技能格式： + 1. 可执行脚本/程序：通过 --help 获取简介和完整使用说明 + 2. Markdown 文件 (.md)：第一行为简介，其余为技能内容 + + 工作流程： + 1. 用关键字调用本工具，获取匹配技能的名称和简介列表 + 2. 对于 Markdown 技能，可以直接读取 content 字段获取完整技能内容 + 3. 对于可执行技能，通过 create_subagent 派生子代理执行 "技能名 --help" 查看完整命令手册 + 4. 根据内容，使用相应方式调用技能 + """ + + def __init__(self, skill_manager: SkillManager, logger: Optional[Logger] = None): + super().__init__(logger) + self.skill_manager = skill_manager + + def get_name(self) -> str: + return "search_skills" + + def get_description(self) -> str: + return ( + "扩展技能搜索工具。\n\n" + "用关键字搜索 ~/.aibox/skills/ 目录下的技能文件。\n" + "支持两种技能格式：\n" + " 1. 可执行脚本/程序：通过 --help 获取简介和完整使用说明\n" + " 2. Markdown 文件 (.md)：第一行为简介，其余为技能内容\n\n" + "使用方法：\n" + "1. 调用本工具搜索关键字，获取匹配技能列表（空字符串列出全部）。\n" + "2. 对于 Markdown 技能：设置 show_content=true 读取返回的内容获取完整技能说明。\n" + "3. 对于可执行技能：用 create_subagent 派生子代理执行 \"技能名 --help\" 获取完整命令手册，再按手册调用该技能。\n" + "4. 按照内容说明，使用相应方式调用该技能（执行类操作一律经 create_subagent 派生子代理完成）。" + ) + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "keyword": { + "type": "string", + "description": ( + "搜索关键字，在技能名称、简介和内容中进行匹配。" + "传入空字符串可列出全部可用技能。" + ) + }, + "show_content": { + "type": "boolean", + "description": ( + "是否在结果中显示技能内容。" + "对于 Markdown 技能，这会影响返回结果的详细程度。" + "默认 False（只显示名称和简介）。" + ), + "default": False + } + }, + "required": ["keyword"] + } + + def execute(self, keyword: str = "", show_content: bool = False, **kwargs) -> str: + kw = keyword.strip() + if kw: + results = self.skill_manager.search(kw) + header = f"🔍 搜索 '{kw}' 的结果（共 {len(results)} 个匹配技能）：" + empty_msg = ( + f"❌ 未找到与 '{kw}' 相关的技能。\n" + "提示：传入空字符串可列出所有可用技能。" + ) + else: + results = self.skill_manager.get_all() + header = f"🔧 所有可用技能（共 {len(results)} 个）：" + empty_msg = "❌ ~/.aibox/skills/ 目录下暂无可用技能。" + + if not results: + return empty_msg + + lines = [header] + for item in results: + skill_type = "📄 Markdown" if item.get('is_markdown') else "⚙️ 可执行" + lines.append(f"\n {skill_type} • {item['name']}") + lines.append(f" 简介: {item['summary']}") + + if show_content and item.get('content'): + content = item['content'] + if item.get('is_markdown'): + preview = content[:500] + if len(content) > 500: + preview += "...\n （内容较长，已截断，请直接读取技能文件获取完整内容）" + lines.append(f" 内容预览:\n {preview}") + else: + preview = content[:200] if content else "(无帮助输出)" + if len(content) > 200: + preview += "..." + lines.append(f" 帮助预览: {preview}") + else: + if item.get('is_markdown'): + lines.append(f" 💡 提示: 设置 show_content=true 可查看技能内容") + else: + lines.append(f" 💡 提示: 用 create_subagent 派生子代理执行 \"{item['name']} --help\" 查看完整手册") + + lines.append("") + lines.append("📖 技能调用方式:") + lines.append(" • Markdown 技能: 设置 show_content=true 读取完整技能说明") + lines.append(" • 可执行技能: 用 create_subagent 派生子代理执行（先 '技能名 --help' 查手册，再按手册调用）") + return "\n".join(lines) + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行搜索技能""" + keyword = kwargs.get("keyword", "") + show_content = kwargs.get("show_content", False) + + output_queue.put(("line", f"\n🔍 搜索技能: {keyword if keyword else '(全部)'}\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + # 直接传递 kwargs，避免重复参数 + result = self.execute(**kwargs) + + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + +class ReloadSkillsTool(Tool): + """技能热重载工具 - 重新加载所有扩展技能""" + + def __init__(self, tool_registry: ToolRegistry, logger: Optional[Logger] = None): + super().__init__(logger) + self.tool_registry = tool_registry + + def get_name(self) -> str: + return "reload_skills" + + def get_description(self) -> str: + return """技能热重载工具 - 重新加载所有扩展技能 +当你添加、修改或删除了 ~/.aibox/skills/ 目录下的技能文件后， +调用此工具可以立即生效，无需重启系统。""" + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "confirm": { + "type": "boolean", + "description": "确认重新加载技能", + "default": True + } + } + } + + def execute(self, confirm: bool = True, **kwargs) -> str: + """执行技能热重载""" + if not confirm: + return "❌ 技能重载已取消，如需重载请设置 confirm=true" + + self.logger.info("开始执行技能热重载...") + + try: + count = self.tool_registry.reload_skills() + + # 从 SkillManager 获取更新后的技能列表（含简介） + skill_manager = self.tool_registry.skill_manager + if skill_manager: + all_skills = skill_manager.get_all() + skill_lines = "\n".join( + f" • {s['name']} — {s['summary']}" + for s in all_skills[:10] + ) + data = { + "重新加载的技能数量": count, + "技能目录": constants.SKILLS_DIR, + "技能列表（含简介）": skill_lines or "（无技能）", + } + if count > 10: + data["提示"] = f"还有 {count - 10} 个技能未显示" + else: + data = { + "重新加载的技能数量": count, + "技能目录": constants.SKILLS_DIR, + } + + return self.format_result(True, i18n.get('skills_reloaded', count), data) + + except Exception as e: + self.logger.error(f"技能热重载失败: {e}") + return i18n.get('skills_reload_failed', str(e)) + + def execute_streaming(self, output_queue: queue.Queue, **kwargs): + """流式执行技能热重载""" + output_queue.put(("line", "\n🔄 执行技能热重载...\n")) + output_queue.put(("line", "-" * 50 + "\n")) + + result = self.execute(**kwargs) + + output_queue.put(("line", result + "\n")) + output_queue.put(("complete", result)) + + +class InstallSkillTool(Tool): + """技能安装指引工具 - 返回如何将可执行文件/脚本安装为扩展技能的说明""" + + def get_name(self) -> str: + return "install_skill" + + def get_description(self) -> str: + return ( + "技能安装指引工具。\n" + "返回将 Markdown 技能文件安装为扩展技能的完整步骤说明，" + "包括文件检查、格式校验和部署方法（目标文件名统一带 .md 扩展名）。" + ) + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "skill_path": { + "type": "string", + "description": ( + "待安装的技能文件路径（可选）。" + "若提供，说明中会带入该路径作为示例；" + "不提供则返回通用说明。" + ) + } + }, + "required": [] + } + + def execute(self, skill_path: str = "", **kwargs) -> str: + skills_dir = constants.SKILLS_DIR + + # 如果给了具体路径，在说明里带入，否则用占位符 + src = skill_path.strip() if skill_path else "/path/to/your_skill.md" + # 去掉扩展名作为技能名称，再统一补 .md 作为目标文件名 + base = os.path.splitext(os.path.basename(src))[0] if src else "your_skill" + dst = os.path.join(skills_dir, base + ".md") + + guide = f"""📦 扩展技能安装指引 +{"=" * 55} + +要将一个 Markdown 技能文件安装为扩展技能，请按以下步骤操作： + +───────────────────────────────────────────────────── +步骤 1：确认文件存在且格式正确 +───────────────────────────────────────────────────── +调用 create_subagent 派生子代理执行以下命令，确认文件存在： + + ls -l {src} + +技能文件格式要求： + • 文件名为纯文本 Markdown，第一行是一级标题（# 技能名），作为技能摘要 + • 其余内容为技能的执行说明、使用方法和示例 + • 文件应带 .md 扩展名（系统以 .md 文件作为技能加载） + +───────────────────────────────────────────────────── +步骤 2：复制到技能目录并统一为 .md 扩展名 +───────────────────────────────────────────────────── +技能目录：{skills_dir} + +调用 create_subagent 派生子代理执行复制命令（目标文件名统一补 .md 扩展名）： + + cp {src} {dst} + +⚠️ 注意：目标文件名必须带 .md 扩展名， + 系统以 .md 文件作为技能加载，且技能名称取文件名（不含 .md）。 + +───────────────────────────────────────────────────── +步骤 3：让系统识别新技能 +───────────────────────────────────────────────────── +复制完成后，调用热重载工具使其立即生效： + + reload_skills confirm=true + +之后可用 search_skills 验证技能已加载： + + search_skills keyword="{base}" + +{"=" * 55} +完成以上步骤后，即可通过该技能执行相应操作。""" + + return guide + + +class CreateSkillTool(Tool): + """创建扩展技能工具 - 返回提示词，驱动智能体自行编写并部署 Markdown 技能文件。 + + 工具本身不调用 LLM，只输出一段结构化的指令提示词。 + 智能体读到提示词后，会自行： + 1. 编写满足格式要求的 Markdown 技能文件 + 2. 经 create_subagent 派生子代理写入 ~/.aibox/skills/.md + 3. 调用 reload_skills 热重载 + """ + + def get_name(self) -> str: + return "create_skill" + + def get_description(self) -> str: + return ( + "创建扩展技能工具。\n" + "返回一段指令提示词，驱动你自行编写并部署一个 Markdown 技能文件（带 .md 扩展名）。\n\n" + "技能文件格式要求：\n" + " • 文件名必须带 .md 扩展名（如 fetch-weather.md、calc-bmi.md）\n" + " • 第一行为一级标题（# 技能名），作为该技能的一句话简介\n" + " • 其余内容为技能的执行说明、使用方法和示例，供智能体参考执行" + ) + + def get_parameters(self) -> Dict: + return { + "type": "object", + "properties": { + "skill_name": { + "type": "string", + "description": ( + "技能名称（部署后的文件名，不含扩展名；系统会自动补齐 .md）。" + "只能包含字母、数字、连字符和下划线，例如：fetch-weather、calc-bmi。" + ) + }, + "description": { + "type": "string", + "description": ( + "对该技能的详细需求描述：它应该做什么、接受哪些输入参数、" + "返回什么输出、有哪些典型使用场景。描述越详细，生成质量越高。" + ) + } + }, + "required": ["skill_name", "description"] + } + + def execute(self, skill_name: str = "", description: str = "", **kwargs) -> str: + # ---- 参数校验 ---- + if not skill_name or not skill_name.strip(): + return "❌ 错误：skill_name 不能为空" + if not description or not description.strip(): + return "❌ 错误：description 不能为空" + if not re.match(r'^[A-Za-z0-9_\-]+$', skill_name.strip()): + return "❌ 错误：skill_name 只能包含字母、数字、连字符(-)和下划线(_)" + + skill_name = skill_name.strip() + skills_dir = constants.SKILLS_DIR + # 技能文件名统一带 .md 扩展名 + skill_path = os.path.join(skills_dir, skill_name + ".md") + + # ---- 返回给智能体的指令提示词 ---- + prompt = f"""【任务】创建扩展技能：{skill_name} + +请按照以下步骤，自行完成该扩展技能的编写和部署： + +───────────────────────────────────────────────── +步骤 1：编写 Markdown 技能文件 +───────────────────────────────────────────────── +根据以下需求，编写一个 Markdown 技能文件： + +需求描述： +{description.strip()} + +技能文件必须满足的格式约定： +1. 第一行为一级标题 `# {skill_name}`，用于对该技能功能做一句话简介（系统用第一行作为技能摘要） +2. 其余内容清晰说明该技能的用途、使用方式、注意事项和典型示例 +3. 内容完整可执行（即按内容操作即可完成该技能要完成的事） + +───────────────────────────────────────────────── +步骤 2：写入技能目录 +───────────────────────────────────────────────── +调用 create_subagent 派生子代理，将你编写的完整内容写入（文件名必须带 .md 扩展名）： + {skill_path} +（在 task 中附上完整文件内容，并要求子代理写入后用 head -5 验证首行格式） + +───────────────────────────────────────────────── +步骤 3：热重载 +───────────────────────────────────────────────── +调用 reload_skills 工具使新技能立即生效： + reload_skills confirm=true + +───────────────────────────────────────────────── +完成后请告知用户技能已创建成功，并展示技能的第一行简介。""" + + return prompt