""" core/agent_storage.py - Agent 配置数据库管理 ============================================= 将 agent 的配置信息从 config.json 文件迁移到 SQLite 数据库。 文件类内容（soul.md, identity.md, user.md, workspace/, avatar.png）仍保留在本地目录。 数据库表结构: agents: 存储所有 agent 配置信息 迁移策略: 首次启动时自动检测并迁移已有 config.json 到数据库。 迁移完成后不再读写 config.json。 """ from __future__ import annotations import json import os import sqlite3 import time import uuid from dataclasses import dataclass, field, asdict from pathlib import Path from typing import Any, Dict, List, Optional from core.logger import get_logger from core.utils import next_agent_id, generate_id logger = get_logger("myagent.agent_storage") # ============================================================================== # 数据模型 # ============================================================================== @dataclass class AgentConfig: """Agent 配置（对应数据库一行）""" path: str = "" # Agent 路径 = 数字 aid，如 "1", "2", "3" id: str = "" # 数字 Agent ID，自增整数 name: str = "" # 显示名称 description: str = "" # 描述 avatar_color: str = "" # 头像颜色 avatar_emoji: str = "" # 头像 emoji avatar_image: str = "" # 头像图片 URL execution_mode: str = "sandbox" # 执行模式: local / sandbox enabled: bool = True # 是否启用 system: bool = False # 是否系统内置 Agent system_prompt: str = "" # 系统提示词 model: str = "" # 模型覆盖 parent: str = "" # 父 Agent 的数字 aid # 平台绑定 platform: str = "" # 平台标识 platform_token: str = "" # 平台 Token platform_app_id: str = "" # 平台 App ID platform_app_secret: str = "" # 平台 App Secret # 模型库引用 model_id: str = "" # 模型库 ID backup_model_ids: str = "" # 备选模型 ID（JSON 数组字符串） # 其他 work_dir: str = "" # 自定义工作目录 department: str = "" # 所属部门（已废弃，仅用于兼容旧API，实际数据在 agent_departments 表） # 时间戳 created_at: str = "" updated_at: str = "" # ---- 以下字段不存数据库，运行时从文件读取 ---- def to_dict(self) -> dict: """转为字典""" d = {} for k in self.__dataclass_fields__: v = getattr(self, k) if k == "enabled" or k == "system": d[k] = bool(v) else: d[k] = v if v is not None else "" return d def to_db_row(self) -> dict: """转为数据库行（只包含数据库字段）""" return self.to_dict() @classmethod def from_db_row(cls, row: sqlite3.Row) -> "AgentConfig": """从数据库行创建""" d = dict(row) return cls(**{k: d.get(k, "") for k in cls.__dataclass_fields__}) @classmethod def from_json(cls, path: str, data: dict) -> "AgentConfig": """从 config.json 字典创建""" return cls( path=path, id=data.get("id", ""), name=data.get("name", path), description=data.get("description", ""), avatar_color=data.get("avatar_color", ""), avatar_emoji=data.get("avatar_emoji", ""), avatar_image=data.get("avatar_image", ""), execution_mode=data.get("execution_mode", "sandbox"), enabled=bool(data.get("enabled", True)), system=bool(data.get("system", False)), system_prompt=data.get("system_prompt", ""), model=data.get("model", ""), parent=data.get("parent", ""), platform=data.get("platform", ""), platform_token=data.get("platform_token", ""), platform_app_id=data.get("platform_app_id", ""), platform_app_secret=data.get("platform_app_secret", ""), model_id=data.get("model_id", ""), backup_model_ids=json.dumps(data.get("backup_model_ids", []), ensure_ascii=False) if isinstance(data.get("backup_model_ids"), list) else data.get("backup_model_ids", ""), work_dir=data.get("work_dir", ""), created_at=data.get("created_at", ""), updated_at=data.get("updated_at", ""), ) # ============================================================================== # AgentStorage - 数据库管理类 # ============================================================================== class AgentStorage: """ Agent 配置数据库管理器。 职责: - 建表和版本管理 - Agent CRUD（增删改查） - 首次启动时自动从 config.json 迁移数据 - 提供兼容旧代码的接口 """ def __init__(self, db_path: str = ""): if not db_path: db_path = str(Path.home() / ".myagent" / "data" / "agents.db") self._db_path = db_path self._conn: Optional[sqlite3.Connection] = None self._lock = __import__("threading").Lock() def _get_conn(self) -> sqlite3.Connection: """获取数据库连接（线程安全）""" if self._conn is None: self._conn = sqlite3.connect(self._db_path, check_same_thread=False) self._conn.row_factory = sqlite3.Row self._conn.execute("PRAGMA journal_mode=WAL") self._conn.execute("PRAGMA foreign_keys=ON") self._init_tables() return self._conn def _init_tables(self): """初始化数据库表""" conn = self._get_conn() conn.executescript(""" -- Agent 配置表 CREATE TABLE IF NOT EXISTS agents ( path TEXT PRIMARY KEY, id TEXT NOT NULL DEFAULT '', name TEXT NOT NULL DEFAULT '', description TEXT NOT NULL DEFAULT '', avatar_color TEXT NOT NULL DEFAULT '', avatar_emoji TEXT NOT NULL DEFAULT '', avatar_image TEXT NOT NULL DEFAULT '', execution_mode TEXT NOT NULL DEFAULT 'sandbox', enabled INTEGER NOT NULL DEFAULT 1, system INTEGER NOT NULL DEFAULT 0, system_prompt TEXT NOT NULL DEFAULT '', model TEXT NOT NULL DEFAULT '', parent TEXT NOT NULL DEFAULT '', platform TEXT NOT NULL DEFAULT '', platform_token TEXT NOT NULL DEFAULT '', platform_app_id TEXT NOT NULL DEFAULT '', platform_app_secret TEXT NOT NULL DEFAULT '', model_id TEXT NOT NULL DEFAULT '', backup_model_ids TEXT NOT NULL DEFAULT '[]', work_dir TEXT NOT NULL DEFAULT '', created_at TEXT NOT NULL DEFAULT '', updated_at TEXT NOT NULL DEFAULT '' ); -- Agent-部门多对多关联表（支持一个 agent 属于多个部门） CREATE TABLE IF NOT EXISTS agent_departments ( agent_path TEXT NOT NULL, dept_path TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT '', PRIMARY KEY (agent_path, dept_path) ); CREATE INDEX IF NOT EXISTS idx_agents_id ON agents(id); CREATE INDEX IF NOT EXISTS idx_agents_parent ON agents(parent); CREATE INDEX IF NOT EXISTS idx_agents_enabled ON agents(enabled); CREATE INDEX IF NOT EXISTS idx_agent_depts_agent ON agent_departments(agent_path); CREATE INDEX IF NOT EXISTS idx_agent_depts_dept ON agent_departments(dept_path); """) conn.commit() # ========================================================================== # CRUD 操作 # ========================================================================== def get(self, path: str) -> Optional[AgentConfig]: """根据 path 获取 agent 配置""" conn = self._get_conn() with self._lock: row = conn.execute("SELECT * FROM agents WHERE path = ?", (path,)).fetchone() if row: return AgentConfig.from_db_row(row) return None def get_by_id(self, agent_id: str) -> Optional[AgentConfig]: """根据唯一 ID 获取 agent 配置""" conn = self._get_conn() with self._lock: row = conn.execute("SELECT * FROM agents WHERE id = ?", (agent_id,)).fetchone() if row: return AgentConfig.from_db_row(row) return None def list_all(self, enabled_only: bool = False) -> List[AgentConfig]: """列出所有 agent""" conn = self._get_conn() with self._lock: if enabled_only: rows = conn.execute("SELECT * FROM agents WHERE enabled = 1 ORDER BY path").fetchall() else: rows = conn.execute("SELECT * FROM agents ORDER BY path").fetchall() return [AgentConfig.from_db_row(r) for r in rows] def list_children(self, parent_path: str) -> List[AgentConfig]: """列出子 agent""" conn = self._get_conn() with self._lock: rows = conn.execute("SELECT * FROM agents WHERE parent = ? ORDER BY path", (parent_path,)).fetchall() return [AgentConfig.from_db_row(r) for r in rows] def exists(self, path: str) -> bool: """检查 agent 是否存在""" conn = self._get_conn() with self._lock: row = conn.execute("SELECT 1 FROM agents WHERE path = ?", (path,)).fetchone() return row is not None def exists_id(self, agent_id: str) -> bool: """检查 ID 是否存在""" conn = self._get_conn() with self._lock: row = conn.execute("SELECT 1 FROM agents WHERE id = ?", (agent_id,)).fetchone() return row is not None def create(self, cfg: AgentConfig) -> AgentConfig: """创建 agent，返回创建后的配置（含自动填充字段）""" conn = self._get_conn() now = _now_iso() if not cfg.id: cfg.id = next_agent_id() if not cfg.created_at: cfg.created_at = now if not cfg.updated_at: cfg.updated_at = now if not cfg.name: cfg.name = cfg.path with self._lock: conn.execute(""" INSERT INTO agents (path, id, name, description, avatar_color, avatar_emoji, avatar_image, execution_mode, enabled, system, system_prompt, model, parent, platform, platform_token, platform_app_id, platform_app_secret, model_id, backup_model_ids, work_dir, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cfg.path, cfg.id, cfg.name, cfg.description, cfg.avatar_color, cfg.avatar_emoji, cfg.avatar_image, cfg.execution_mode, int(cfg.enabled), int(cfg.system), cfg.system_prompt, cfg.model, cfg.parent, cfg.platform, cfg.platform_token, cfg.platform_app_id, cfg.platform_app_secret, cfg.model_id, cfg.backup_model_ids, cfg.work_dir, cfg.created_at, cfg.updated_at, )) conn.commit() return cfg def update(self, path: str, updates: dict) -> Optional[AgentConfig]: """ 更新 agent 配置。 updates: 要更新的字段字典，只允许更新已知字段。 返回更新后的配置，如果不存在返回 None。 """ # 允许更新的字段白名单 ALLOWED = { "name", "description", "avatar_color", "avatar_emoji", "avatar_image", "execution_mode", "enabled", "system_prompt", "model", "platform", "platform_token", "platform_app_id", "platform_app_secret", "model_id", "backup_model_ids", "work_dir", } filtered = {k: v for k, v in updates.items() if k in ALLOWED} if not filtered: return self.get(path) conn = self._get_conn() now = _now_iso() # backup_model_ids 需要 JSON 序列化 if "backup_model_ids" in filtered and isinstance(filtered["backup_model_ids"], list): filtered["backup_model_ids"] = json.dumps(filtered["backup_model_ids"], ensure_ascii=False) # enabled 需要转 int if "enabled" in filtered: filtered["enabled"] = int(bool(filtered["enabled"])) filtered["updated_at"] = now set_parts = ", ".join(f"{k} = ?" for k in filtered) values = list(filtered.values()) + [path] with self._lock: cursor = conn.execute(f"UPDATE agents SET {set_parts} WHERE path = ?", values) conn.commit() if cursor.rowcount == 0: return None return self.get(path) def rename(self, old_path: str, new_path: str) -> bool: """ 重命名 agent（更新 path 及所有子 agent 的 path）。 同时需要更新 parent 字段引用。 """ conn = self._get_conn() with self._lock: # 检查是否存在 existing = conn.execute("SELECT 1 FROM agents WHERE path = ?", (old_path,)).fetchone() if not existing: return False # 检查目标是否已存在 target = conn.execute("SELECT 1 FROM agents WHERE path = ?", (new_path,)).fetchone() if target: return False now = _now_iso() # 1. 更新自身 conn.execute("UPDATE agents SET path = ?, updated_at = ? WHERE path = ?", (new_path, now, old_path)) # 2. 更新所有子 agent 的 path（前缀替换）和 parent old_prefix = old_path + "/" new_prefix = new_path + "/" children = conn.execute("SELECT path FROM agents WHERE path LIKE ?", (old_prefix + "%",)).fetchall() for child in children: child_old = child["path"] child_new = new_prefix + child_old[len(old_prefix):] conn.execute("UPDATE agents SET path = ?, parent = ?, updated_at = ? WHERE path = ?", (child_new, new_path, now, child_old)) # 3. 更新其他 agent 中 parent 引用 conn.execute("UPDATE agents SET parent = ?, updated_at = ? WHERE parent = ?", (new_path, now, old_path)) conn.commit() return True def delete(self, path: str) -> bool: """ 删除 agent 及其所有子 agent。 返回删除的行数。 """ conn = self._get_conn() with self._lock: # 先删除关联记录（包括自身和所有子 agent） # 删除自身关联 conn.execute("DELETE FROM agent_departments WHERE agent_path = ?", (path,)) # 删除子 agent 关联（递归删除所有以 path/ 开头的） conn.execute("DELETE FROM agent_departments WHERE agent_path LIKE ?", (path + "/%",)) # 先删子 agent conn.execute("DELETE FROM agents WHERE path LIKE ?", (path + "/%",)) # 再删自身 cursor = conn.execute("DELETE FROM agents WHERE path = ?", (path,)) conn.commit() return cursor.rowcount > 0 def count(self) -> int: """统计 agent 总数""" conn = self._get_conn() with self._lock: row = conn.execute("SELECT COUNT(*) as cnt FROM agents").fetchone() return row["cnt"] if row else 0 def search(self, keyword: str, limit: int = 20) -> List[AgentConfig]: """按名称或描述搜索 agent""" conn = self._get_conn() with self._lock: rows = conn.execute( "SELECT * FROM agents WHERE name LIKE ? OR description LIKE ? ORDER BY path LIMIT ?", (f"%{keyword}%", f"%{keyword}%", limit), ).fetchall() return [AgentConfig.from_db_row(r) for r in rows] def get_all_ids(self) -> List[str]: """获取所有 agent ID 列表（用于调试）""" conn = self._get_conn() with self._lock: rows = conn.execute("SELECT id FROM agents WHERE id != ''").fetchall() return [r["id"] for r in rows] # ========================================================================== # Agent-部门关联管理（多对多） # ========================================================================== def get_agent_departments(self, agent_path: str) -> List[str]: """ 获取 agent 所属的所有部门路径列表。 Args: agent_path: Agent 路径 Returns: 部门路径列表（按添加顺序） """ conn = self._get_conn() with self._lock: rows = conn.execute( "SELECT dept_path FROM agent_departments WHERE agent_path = ? ORDER BY created_at", (agent_path,) ).fetchall() return [row["dept_path"] for row in rows] def get_department_agents(self, dept_path: str) -> List[str]: """ 获取部门下的所有 agent 路径列表。 Args: dept_path: 部门路径 Returns: agent 路径列表 """ conn = self._get_conn() with self._lock: rows = conn.execute( "SELECT agent_path FROM agent_departments WHERE dept_path = ? ORDER BY created_at", (dept_path,) ).fetchall() return [row["agent_path"] for row in rows] def add_agent_to_department(self, agent_path: str, dept_path: str) -> bool: """ 将 agent 添加到部门（多对多关联）。 Args: agent_path: Agent 路径 dept_path: 部门路径 Returns: True 表示添加成功，False 表示已存在或失败 """ conn = self._get_conn() now = _now_iso() with self._lock: try: conn.execute( "INSERT OR IGNORE INTO agent_departments (agent_path, dept_path, created_at) VALUES (?, ?, ?)", (agent_path, dept_path, now) ) conn.commit() return True except Exception as e: logger.warning(f"添加 agent {agent_path} 到部门 {dept_path} 失败: {e}") return False def remove_agent_from_department(self, agent_path: str, dept_path: str) -> bool: """ 从部门移除 agent。 Args: agent_path: Agent 路径 dept_path: 部门路径 Returns: True 表示移除成功 """ conn = self._get_conn() with self._lock: try: conn.execute( "DELETE FROM agent_departments WHERE agent_path = ? AND dept_path = ?", (agent_path, dept_path) ) conn.commit() return True except Exception as e: logger.warning(f"从部门 {dept_path} 移除 agent {agent_path} 失败: {e}") return False def set_agent_departments(self, agent_path: str, dept_paths: List[str]) -> None: """ 设置 agent 的部门列表（替换模式）。 Args: agent_path: Agent 路径 dept_paths: 部门路径列表 """ conn = self._get_conn() now = _now_iso() with self._lock: # 删除旧关联 conn.execute("DELETE FROM agent_departments WHERE agent_path = ?", (agent_path,)) # 插入新关联 for dept_path in dept_paths: conn.execute( "INSERT OR IGNORE INTO agent_departments (agent_path, dept_path, created_at) VALUES (?, ?, ?)", (agent_path, dept_path, now) ) conn.commit() # ========================================================================== # Agent-部门关联管理（多对多） # ========================================================================== def get_agent_departments(self, agent_path: str) -> List[str]: """ 获取 agent 所属的所有部门路径列表。 Args: agent_path: Agent 路径 Returns: 部门路径列表（按添加顺序） """ conn = self._get_conn() with self._lock: rows = conn.execute( "SELECT dept_path FROM agent_departments WHERE agent_path = ? ORDER BY created_at", (agent_path,) ).fetchall() return [row["dept_path"] for row in rows] def get_department_agents(self, dept_path: str) -> List[str]: """ 获取部门下的所有 agent 路径列表。 Args: dept_path: 部门路径 Returns: agent 路径列表 """ conn = self._get_conn() with self._lock: rows = conn.execute( "SELECT agent_path FROM agent_departments WHERE dept_path = ? ORDER BY created_at", (dept_path,) ).fetchall() return [row["agent_path"] for row in rows] def add_agent_to_department(self, agent_path: str, dept_path: str) -> bool: """ 将 agent 添加到部门（多对多关联）。 Args: agent_path: Agent 路径 dept_path: 部门路径 Returns: True 表示添加成功，False 表示已存在或失败 """ conn = self._get_conn() now = _now_iso() with self._lock: try: conn.execute( "INSERT OR IGNORE INTO agent_departments (agent_path, dept_path, created_at) VALUES (?, ?, ?)", (agent_path, dept_path, now) ) conn.commit() return True except Exception as e: logger.warning(f"添加 agent {agent_path} 到部门 {dept_path} 失败: {e}") return False def remove_agent_from_department(self, agent_path: str, dept_path: str) -> bool: """ 从部门移除 agent。 Args: agent_path: Agent 路径 dept_path: 部门路径 Returns: True 表示移除成功 """ conn = self._get_conn() with self._lock: try: conn.execute( "DELETE FROM agent_departments WHERE agent_path = ? AND dept_path = ?", (agent_path, dept_path) ) conn.commit() return True except Exception as e: logger.warning(f"从部门 {dept_path} 移除 agent {agent_path} 失败: {e}") return False def set_agent_departments(self, agent_path: str, dept_paths: List[str]) -> None: """ 设置 agent 的部门列表（替换模式）。 Args: agent_path: Agent 路径 dept_paths: 部门路径列表 """ conn = self._get_conn() now = _now_iso() with self._lock: # 删除旧关联 conn.execute("DELETE FROM agent_departments WHERE agent_path = ?", (agent_path,)) # 插入新关联 for dept_path in dept_paths: conn.execute( "INSERT OR IGNORE INTO agent_departments (agent_path, dept_path, created_at) VALUES (?, ?, ?)", (agent_path, dept_path, now) ) conn.commit() def set_department_agents(self, dept_path: str, agent_paths: List[str]) -> None: """ 设置部门下的 agent 列表（替换模式）。 Args: dept_path: 部门路径 agent_paths: agent 路径列表 """ conn = self._get_conn() now = _now_iso() with self._lock: # 删除旧关联 conn.execute("DELETE FROM agent_departments WHERE dept_path = ?", (dept_path,)) # 插入新关联 for agent_path in agent_paths: conn.execute( "INSERT OR IGNORE INTO agent_departments (agent_path, dept_path, created_at) VALUES (?, ?, ?)", (agent_path, dept_path, now) ) conn.commit() def close(self): """关闭数据库连接""" if self._conn: self._conn.close() self._conn = None # ============================================================================== # 工具函数 # ============================================================================== def _now_iso() -> str: """返回 ISO 格式时间戳""" from datetime import datetime, timezone try: from core.utils import get_config_tz tz = get_config_tz() return datetime.now(tz).strftime("%Y-%m-%dT%H:%M:%S") except Exception: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S")