数据存储

插件经常需要保存和检索数据，例如用户偏好、会话状态、缓存信息或其他持久化内容。Nekro Agent 为每个插件提供了一个独立的、基于键值对 (Key-Value) 的存储系统，通过 plugin.store 对象进行访问。这个存储系统简化了数据持久化操作，并确保了不同插件数据之间的隔离。

plugin.store 概述

plugin.store 提供了一组异步方法来操作存储在数据库中的数据。其主要特点包括：

• 键值存储：简单直观的 KV 存储模型。
• 数据隔离：每个插件拥有其独立的命名空间，避免键名冲突。
• 作用域数据：支持存储不同作用域的数据：
  • 会话特定数据 (通过 chat_key): 数据与特定的聊天会话绑定。
  • 用户特定数据 (通过 user_key): 数据与特定的用户绑定（跨会话）。
  • 插件全局数据 (不指定 chat_key 或 user_key): 数据属于插件本身，不与任何特定会话或用户关联。
• 字符串存储：底层存储的值目前主要设计为字符串。对于复杂数据结构（如字典、列表、自定义对象），需要先序列化为字符串（通常是 JSON 字符串）再存入，取出后进行反序列化。

核心存储 API

以下是 plugin.store 提供的主要异步方法：

1. 设置数据 (set)

用于向存储中添加或更新一个键值对。

async def set(
    self,
    chat_key: str = "",    # 可选，会话标识
    user_key: str = "",    # 可选，用户标识
    store_key: str = "",   # 必需，存储的键名
    value: str = ""         # 必需，要存储的值 (字符串)
) -> Literal[0, 1]: # 返回 0 表示失败或未更改，1 表示成功
    # ... 实现 ...

参数说明：

• chat_key (str, 可选): 如果提供，数据将与此会话关联。
• user_key (str, 可选): 如果提供，数据将与此用户关联。
• store_key (str, 必需): 数据的唯一键名。
• value (str, 必需): 要存储的字符串数据。

示例：

from nekro_agent.api.schemas import AgentCtx
import json

# 存储会话特定数据
await plugin.store.set(chat_key=_ctx.from_chat_key, store_key="last_command", value="/weather London")

# 存储用户特定偏好
user_prefs = {"theme": "dark", "notifications": True}
await plugin.store.set(user_key=_ctx.from_user_id, store_key="preferences", value=json.dumps(user_prefs))

# 存储插件全局配置
await plugin.store.set(store_key="plugin_last_updated_timestamp", value=str(time.time()))

2. 获取数据 (get)

根据键名从存储中检索数据。

async def get(
    self,
    chat_key: str = "",    # 可选，会话标识
    user_key: str = "",    # 可选，用户标识
    store_key: str = ""   # 必需，存储的键名
) -> Optional[str]:      # 返回存储的字符串值，如果键不存在则返回 None
    # ... 实现 ...

参数说明：同 set 方法中的键参数。

示例：

# 获取会话特定数据
last_command = await plugin.store.get(chat_key=_ctx.from_chat_key, store_key="last_command")

# 获取用户偏好，如果不存在则使用默认值
prefs_str = await plugin.store.get(user_key=_ctx.from_user_id, store_key="preferences")
user_preferences = {} 
if prefs_str:
    user_preferences = json.loads(prefs_str)
else:
    user_preferences = {"theme": "light", "notifications": False} # 默认值

# 获取插件全局数据
timestamp_str = await plugin.store.get(store_key="plugin_last_updated_timestamp")
if timestamp_str:
    last_updated = float(timestamp_str)

3. 删除数据 (delete)

根据键名从存储中移除一个键值对。

async def delete(
    self,
    chat_key: str = "",    # 可选，会话标识
    user_key: str = "",    # 可选，用户标识
    store_key: str = ""   # 必需，存储的键名
) -> Literal[0, 1]:      # 返回 1 表示成功删除，0 表示键不存在或删除失败
    # ... 实现 ...

参数说明：同 set 方法中的键参数。

示例：

# 删除会话的特定缓存
await plugin.store.delete(chat_key=_ctx.from_chat_key, store_key="session_cache_data")

# 删除用户的某个设置
await plugin.store.delete(user_key=_ctx.from_user_id, store_key="old_setting")

4. 检查数据是否存在 (exists)

（注意：PluginStore 的代码中并未直接提供 exists 方法，但通常可以通过 get 方法的结果是否为 None 来判断。文档中提到过 exists，这里我们假设其可以通过 get 间接实现，或者未来可能添加。）

要检查一个键是否存在，你可以调用 get 并判断其返回值：

value = await plugin.store.get(chat_key=_ctx.from_chat_key, store_key="my_key")
if value is not None:
    # 键存在
    core.logger.info("'my_key' 存在于存储中。")
else:
    # 键不存在
    core.logger.info("'my_key' 不存在。")

存储结构化数据 (使用 Pydantic 和 JSON)

由于 plugin.store 直接存储的是字符串，当你需要存储更复杂的数据结构（如配置对象、笔记列表等）时，推荐使用 Pydantic 模型进行序列化和反序列化。

1. 定义 Pydantic 模型：描述你的数据结构。
2. 序列化：在调用 plugin.store.set() 之前，使用模型的 .model_dump_json() (Pydantic V2) 或 .json() (Pydantic V1) 方法将对象转换为 JSON 字符串。
3. 反序列化：从 plugin.store.get() 获取 JSON 字符串后，使用模型的 model_validate_json() (Pydantic V2) 或 parse_raw() (Pydantic V1) 方法将其转换回 Pydantic 模型实例。

示例：存储笔记数据

from pydantic import BaseModel
from typing import List, Dict, Optional
import time
import json # 仅用于对比，Pydantic模型有内建的json方法

class Note(BaseModel):
    id: str
    title: str
    content: str
    created_at: float
    tags: List[str] = []

class UserNotes(BaseModel):
    notes: Dict[str, Note] = {}

# --- 沙盒方法示例 --- 

@plugin.mount_sandbox_method(SandboxMethodType.BEHAVIOR, "add_user_note", "为当前用户添加一条笔记。")
async def add_note(_ctx: AgentCtx, note_id: str, title: str, content: str, tags_str: str = "") -> str:
    # 1. 获取现有笔记数据
    user_notes_json = await plugin.store.get(user_key=_ctx.from_user_id, store_key="all_notes")
    
    if user_notes_json:
        user_notes_data = UserNotes.model_validate_json(user_notes_json)
    else:
        user_notes_data = UserNotes()

    # 2. 创建新笔记并添加
    new_note = Note(
        id=note_id,
        title=title,
        content=content,
        created_at=time.time(),
        tags=tags_str.split(',') if tags_str else []
    )
    user_notes_data.notes[note_id] = new_note

    # 3. 序列化并存储
    await plugin.store.set(
        user_key=_ctx.from_user_id,
        store_key="all_notes",
        value=user_notes_data.model_dump_json()
    )
    return f"笔记 '{title}' 已添加。ID: {note_id}"

@plugin.mount_sandbox_method(SandboxMethodType.TOOL, "get_user_note_content", "获取用户指定ID的笔记内容。")
async def get_note_content(_ctx: AgentCtx, note_id: str) -> Optional[str]:
    user_notes_json = await plugin.store.get(user_key=_ctx.from_user_id, store_key="all_notes")
    if not user_notes_json:
        return "用户没有任何笔记。"
    
    user_notes_data = UserNotes.model_validate_json(user_notes_json)
    note = user_notes_data.notes.get(note_id)
    
    if note:
        return note.content
    return f"未找到 ID 为 '{note_id}' 的笔记。"

插件数据目录 (plugin.get_plugin_path())

除了键值存储外，如果插件需要存储较大的文件（如模型文件、大型数据集、用户上传的文件副本等），可以使用插件专属的文件目录。

plugin.get_plugin_path() 方法返回一个 pathlib.Path 对象，指向该插件在系统数据目录中的专属文件夹。路径通常是 data/plugins/<plugin_author>.<plugin_module_name>/。

from pathlib import Path

async def save_large_data_to_file(data_content: bytes, filename: str):
    plugin_data_dir: Path = plugin.get_plugin_path() # 获取插件数据根目录
    custom_files_dir = plugin_data_dir / "my_files"  # 创建一个子目录
    custom_files_dir.mkdir(parents=True, exist_ok=True) # 确保目录存在

    file_path = custom_files_dir / filename
    async with aiofiles.open(file_path, "wb") as f:
        await f.write(data_content)
    core.logger.info(f"大文件已保存到: {file_path}")

注意：直接操作文件系统时，请务必小心，并考虑路径转换问题（详见文件交互章节），尤其是当文件需要在沙盒和主服务之间共享或引用时。

最佳实践

1. 明确键名策略：使用清晰、有结构（例如，使用 . 或 : 分隔符）的 store_key，避免混淆。
   • 例如：user_prefs:theme, chat_state:topic, cache:external_api:last_fetch_time。
2. 数据序列化：对于非字符串数据，始终进行序列化（如转为 JSON）再存储，并在读取后反序列化。
3. 错误处理：妥善处理 get 时数据不存在 (None) 的情况，提供默认值或适当的逻辑。
4. 作用域选择：根据数据特性选择正确的存储作用域（会话、用户或全局）。
5. 数据清理：对于不再需要的临时数据或缓存，应及时使用 delete 方法清理，避免数据无限增长。
6. 版本兼容：如果插件升级导致存储的数据结构发生变化，需要考虑旧版本数据的兼容性迁移方案。
7. 大小限制：键值存储适合存储相对较小的数据。对于非常大的数据块或二进制文件，优先考虑使用插件的文件目录 (plugin.get_plugin_path()) 结合文件存储，然后在键值存储中保存文件的引用或元数据。