openJiuwen agent-core 存储层完全指南：BaseKVStore / BaseDbStore / BaseVectorStore 抽象与内置实现深度解析-平芜编程栈

人工智能
AI Agent
Agent 框架
大模型
工具调用
RAG
提示工程
强化学习

【免费下载链接】agent-core

openJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力

项目地址：https://gitcode.com/openJiuwen/agent-core

openjiuwen.core.foundation.store是 openJiuwen agent-core 框架的统一存储抽象层，为框架内的 memory（记忆）、session（会话）等模块提供 KV 存储、数据库存储与向量存储三类基础能力。读完本文，你将掌握三类存储抽象接口的完整契约、InMemoryKVStore/DbBasedKVStore/DefaultDbStore/GaussVectorStore四套内置实现的底层原理，并能通过create_vector_store()工厂与插件机制（register_vector_store+ entry_points）接入任意第三方向量后端，直接落地到 Agent 应用的记忆与持久化场景中。

模块总览：一个包，三类存储抽象

store.md 对应的源码位于 openjiuwen/core/foundation/store/。整个模块围绕三条抽象线展开：

抽象接口	用途	内置实现
`BaseKVStore`	键值存储（set/get/exists/delete/前缀查询/批量 mget 等）	`InMemoryKVStore`（纯内存）、`DbBasedKVStore`（SQLAlchemy 落库）
`BaseDbStore`	数据库存储（仅暴露异步 SQLAlchemy`AsyncEngine`）	`DefaultDbStore`
`BaseVectorStore`	向量存储（集合管理、文档写入、向量检索、文档删除、schema 迁移）	`GaussVectorStore`（GaussVector DB），另有`ChromaVectorStore`、`MilvusVectorStore`内置后端

从源码结构看，__init__.py的导出还包含BaseMessageStore、VectorSearchResult、CollectionSchema、FieldSchema、VectorDataType等配套类型，其中 KV/DB 相关的实现（BaseDbStore、DbBasedKVStore、DefaultDbStore）依赖 SQLAlchemy，因此模块采用了PEP 562 惰性加载——__getattr__只在真正访问这些符号时才执行 import，避免仅使用内存 KV 或向量存储的应用被拖入 SQLAlchemy 重依赖（见init.py）。

BaseKVStore：统一键值存储接口

BaseKVStore是所有 KV 后端的抽象基类，继承自ABC（base_kv_store.py）。从源码看，接口比文档列出的方法更完整，共 9 个抽象方法：

抽象方法	签名	说明
`set`	`async set(key: str, value: str \\| bytes)`	写入/覆盖键值对
`exclusive_set`	`async exclusive_set(key, value, expiry: int \\| None = None) -> bool`	原子化写入：仅当 key 不存在或已过期时写入成功，返回`True`；key 已存在且未过期则返回`False`（可用于分布式锁、一次性任务标记）
`get`	`async get(key) -> str \\| bytes \\| None`	读取，key 不存在返回`None`
`exists`	`async exists(key) -> bool`	判断 key 是否存在
`delete`	`async delete(key)`	删除，key 不存在时静默无操作
`get_by_prefix`	`async get_by_prefix(prefix) -> dict[str, str \\| bytes]`	按前缀批量读取
`delete_by_prefix`	`async delete_by_prefix(prefix, batch_size=None)`	按前缀批量删除，`batch_size`控制分批规模
`mget`	`async mget(keys: List[str]) -> List[str \\| bytes \\| None]`	批量读取，结果顺序与入参一致，缺失项为`None`
`batch_delete`	`async batch_delete(keys, batch_size=None) -> int`	批量删除，返回实际删除数量

注意文档中的value: str在源码中已扩展为str | bytes，两种实现也都支持 bytes 值。此外源码还提供了BasedKVStorePipeline流水线辅助类（base_kv_store.py），通过pipeline()收集set/get/exists操作后一次execute()批量执行，减少网络往返（内存实现中 pipelined 操作同样持锁）。

插件兼容性说明：BaseKVStore是稳定公共 API，第三方包可直接继承并导出；与向量存储不同，KV 存储没有create_kv_store工厂，调用方直接 import 并实例化类即可。

InMemoryKVStore：零依赖的内存实现

InMemoryKVStore无参构造，内部使用字典 +asyncio.Lock保证并发安全（in_memory_kv_store.py）：

内部字典_store: dict[str, tuple[str | bytes, Optional[float]]]同时保存值与过期时间戳；
所有读写在_lock保护下进行；
exclusive_set的expiry换算为绝对时间戳current_time + expiry；
过期 key 在get时按不存在处理（返回None），但不会被自动从字典删除——源码注释明确说明这是为了后续允许重新写入。

from openjiuwen.core.foundation.store import InMemoryKVStore kv = InMemoryKVStore() await kv.set("user:001:name", "Alice") await kv.get("user:001:name") # "Alice" await kv.exists("user:001:name") # True await kv.delete("user:001:name") await kv.get("user:001:name") # None

DbBasedKVStore：SQLAlchemy 落库实现

DbBasedKVStore(engine: AsyncEngine)基于 SQLAlchemy 异步引擎，使用kv_store表（key主键列 +value列），首次调用任何接口时自动建表（db_based_kv_store.py）：

内部通过async_sessionmaker(engine, expire_on_commit=False, class_=AsyncSession)创建会话工厂，建表动作由_create_table_if_not_exist()在双重检查锁（asyncio.Lock）下完成；
upsert 方言适配：_get_upsert_stmt()根据engine.dialect.name选择 SQLite 的on_conflict_do_update或 MySQL 的on_duplicate_key_update，因此同一套代码可同时跑在 SQLite/MySQL 上；
bytes 值编码：bytes 值以__BYTES__:前缀 + base64 形式存字符串，读取时自动解码还原；
exclusive_set的 expiry 实现：将{"exclusive_value": ..., "exclusive_expiry": ...}以 JSON 存入value列，检查时若 JSON 解析失败或旧 expiry 未过期则拒绝写入；
同样支持batch_size分批删除与pipeline()批量执行（pipeline 内部按操作类型分组，把 set 批量 upsert、get/exists 批量IN查询）。

from sqlalchemy.ext.asyncio import create_async_engine from openjiuwen.core.foundation.store import DbBasedKVStore async_engine = create_async_engine("sqlite+aiosqlite:///demo.db", echo=False) db_kv = DbBasedKVStore(async_engine) await db_kv.set("config:key1", "value1") await db_kv.get("config:key1") # "value1"

BaseDbStore：数据库访问的薄抽象

BaseDbStore只有一个抽象方法get_async_engine() -> AsyncEngine（base_db_store.py），返回异步 SQLAlchemy 引擎，供调用方执行建表、原生 SQL 等操作。它是 memory 等模块做"原始 SQL 能力"接线的标准通道。

DefaultDbStore(async_conn: AsyncEngine)是它的默认实现，构造时保存传入的async_conn，get_async_engine()直接原样返回（default_db_store.py）：

from sqlalchemy.ext.asyncio import create_async_engine from openjiuwen.core.foundation.store import DefaultDbStore async_engine = create_async_engine("sqlite+aiosqlite:///app.db") db_store = DefaultDbStore(async_engine) assert db_store.get_async_engine() is async_engine

BaseVectorStore：向量存储统一接口

BaseVectorStore定义了一套面向 RAG / 语义检索的完整向量存储契约，覆盖集合生命周期、文档 CRUD、向量检索、schema 迁移与元数据管理（源码见 base_vector_store.py，配套的FieldSchema/CollectionSchema数据模型与 Milvus FieldSchema 风格一致，支持auto_id、max_length、dim、element_type等字段属性）。

集合管理

create_collection(collection_name, schema, **kwargs)：按CollectionSchema或 schema 字典建集合，kwargs支持distance_metric（如"COSINE"、"L2"、"IP"）；
delete_collection(collection_name, **kwargs)/collection_exists(collection_name, **kwargs) -> bool：删除与存在性判断；
get_schema(collection_name, **kwargs) -> CollectionSchema：获取集合 schema；
list_collection_names() -> List[str]：列出全部集合名。

文档写入与检索

add_docs(collection_name, docs, **kwargs)：批量写文档，每篇文档含id（可选）、embedding: List[float]、text: str、metadata（可选），kwargs支持batch_size分批插入；
search(collection_name, query_vector, vector_field, top_k=5, filters=None, **kwargs) -> List[VectorSearchResult]：按向量相似度检索，filters仅支持标量字段等值过滤，kwargs可传metric_type、output_fields；每个结果含score: float（分数越高越相关）与fields: Dict[str, Any]（命中文档全部字段值）；
delete_docs_by_ids(collection_name, ids, **kwargs)：按 ID 删除；
delete_docs_by_filters(collection_name, filters, **kwargs)：按标量字段等值过滤删除。

schema 迁移与元数据

update_schema(collection_name, operations: List[BaseOperation])：为向量数据迁移服务，支持四类操作——AddScalarFieldOperation（新增标量字段）、RenameScalarFieldOperation（重命名字段）、UpdateScalarFieldTypeOperation（修改标量字段类型）、UpdateEmbeddingDimensionOperation（修改向量维度）；
get_collection_metadata(collection_name) -> Dict[str, Any]：返回distance_metric、vector_field、schema_version（未设置时为 0），实现应先查本地缓存，缓存未命中再从底层存储（如 Milvus collection properties）拉取；
update_collection_metadata(collection_name, metadata)：把元数据写入底层存储的 collection properties，所有值转字符串存储；校验schema_version必须为非负整数，并同步更新本地缓存。

GaussVectorStore：基于 GaussVector 数据库的实现

GaussVectorStore继承BaseVectorStore并实现全部抽象方法（vector/gauss_vector_store.py），构造参数：

GaussVectorStore( host="localhost", # GaussVector 服务器地址 port=5432, # 端口 database="postgres", # 数据库名 user="postgres", # 用户 password="", # 密码 **kwargs, # 如 connection_timeout、sslmode )

实现要点（按文档与源码注释）：

连接惰性加载：connection属性首次访问时才通过psycopg2建立连接，close()显式关闭；
支持集合（表）管理、文档增删改查与向量检索；
create_collection的kwargs较为丰富：distance_metric（默认"COSINE"，可选"L2"）、index_type（默认"diskann"，当前仅支持 DiskANN）、pg_nseg（PQ 分段数，默认 128）、pg_nclus（簇数，默认 16）、num_parallels（并行数，默认 32）；
add_docs支持batch_size分批插入（默认 128）；delete_docs_by_ids支持id_column参数（默认"id"）。

create_vector_store 工厂与三方插件机制

create_vector_store(store_type, **kwargs) -> BaseVectorStore | None是向量存储的工厂入口，解析顺序固定为三层（实现见init.py）：

内置后端（封闭集合，优先级最高）："chroma"、"milvus"、"gaussvector"三者恒定胜出，分别懒加载ChromaVectorStore、MilvusVectorStore、GaussVectorStore；
进程内显式注册：通过register_vector_store(name, factory)注册的第三方实现；
entry_points 插件：扫描openjiuwen.vector_stores组下安装的第三方插件。

插件加载失败（import 异常）或实例化失败时，工厂只打印 WARNING 日志并返回None，一个损坏的插件无法拖垮整个应用的工厂调用（_resolve_entry_point中对ep.load()与构造调用均做了 try/except，见init.py）。

# 内置后端：ChromaDB store = create_vector_store("chroma", persist_directory="./data/chroma") # 内置后端：Milvus store = create_vector_store("milvus", uri="http://localhost:19530") # 内置后端：GaussVector store = create_vector_store("gaussvector", host="localhost", port=5432)

register_vector_store：进程内注册私有后端

register_vector_store(name: str, factory: Callable[..., BaseVectorStore]) -> None用于不走 PyPI、无法用 entry_points 的私有后端（如仓库内私有实现），在应用初始化阶段、任何工作线程启动之前调用（非线程安全）：

注册后即可通过create_vector_store(name, ...)按名创建实例；
内置名称不可覆盖：对"chroma"/"milvus"/"gaussvector"调用注册会被保留在注册表中，但工厂解析时内置后端仍然胜出。

VECTOR_STORE_ENTRY_POINT_GROUP：插件发布契约

常量VECTOR_STORE_ENTRY_POINT_GROUP = "openjiuwen.vector_stores"是第三方插件必须声明的 entry_points 组名，属于稳定公共常量——改动该字符串会破坏所有已发布插件。插件作者在自己的pyproject.toml中声明：

[project.entry-points."openjiuwen.vector_stores"] my_backend = "my_package.my_vector_store:MyVectorStore"

用户安装后即可create_vector_store("my_backend", ...)获取实例。完整编写指南见 Store Plugin Development.md；插件框架的回归与故障隔离行为在 test_vector_store_plugin.py 中有系统化单元测试覆盖（内置后端解析回归、显式注册、entry_points 发现、名称冲突确定性解决、坏插件不崩工厂）。

端到端示例：从内存 KV 到向量检索

以下为文档提供的完整可运行示例（可直接保存执行，依赖sqlalchemy与对应的向量后端）：

import asyncio from sqlalchemy.ext.asyncio import create_async_engine, AsyncEngine from openjiuwen.core.foundation.store import ( BaseKVStore, BaseDbStore, InMemoryKVStore, DbBasedKVStore, DefaultDbStore, BaseVectorStore, CollectionSchema, FieldSchema, VectorDataType, create_vector_store, ) async def demo_kv_store(): # 1. In-memory KV storage (no external dependencies) kv = InMemoryKVStore() await kv.set("user:001:name", "Alice") value = await kv.get("user:001:name") print(value) # Alice exists = await kv.exists("user:001:name") print(exists) # True await kv.delete("user:001:name") print(await kv.get("user:001:name")) # None async def demo_db_store(): # 2. Database-based KV storage (requires SQLite/MySQL, etc.) async_engine = create_async_engine( "sqlite+aiosqlite:///demo.db", echo=False, ) db_kv = DbBasedKVStore(async_engine) await db_kv.set("config:key1", "value1") result = await db_kv.get("config:key1") print(result) # value1 async def demo_default_db_store(): # 3. DefaultDbStore only exposes Engine for modules like memory engine to create tables, execute SQL async_engine = create_async_engine("sqlite+aiosqlite:///app.db") db_store = DefaultDbStore(async_engine) engine = db_store.get_async_engine() assert engine is async_engine async def demo_vector_store(): # 4. Vector storage (using ChromaDB or Milvus) store = create_vector_store("chroma", persist_directory="./data/chroma") # Define collection schema schema = CollectionSchema( description="Document collection", enable_dynamic_field=False, ) schema.add_field(FieldSchema( name="id", dtype=VectorDataType.VARCHAR, max_length=256, is_primary=True, )) schema.add_field(FieldSchema( name="embedding", dtype=VectorDataType.FLOAT_VECTOR, dim=768, )) schema.add_field(FieldSchema( name="text", dtype=VectorDataType.VARCHAR, max_length=65535, )) schema.add_field(FieldSchema( name="metadata", dtype=VectorDataType.JSON, )) # Create collection await store.create_collection("documents", schema, distance_metric="cosine") # Add documents docs = [ { "id": "doc1", "embedding": [0.1] * 768, "text": "This is the first document", "metadata": {"category": "tech"}, }, { "id": "doc2", "embedding": [0.2] * 768, "text": "This is the second document", "metadata": {"category": "news"}, }, ] await store.add_docs("documents", docs) # Vector search query_vector = [0.15] * 768 results = await store.search( collection_name="documents", query_vector=query_vector, vector_field="embedding", top_k=10, filters={"category": "tech"}, ) for result in results: print(f"Score: {result.score:.4f}, Text: {result.fields.get('text')}") asyncio.run(demo_kv_store()) asyncio.run(demo_db_store()) asyncio.run(demo_default_db_store()) asyncio.run(demo_vector_store())

示例要点：FieldSchema构建时dim与FLOAT_VECTOR的约束校验（dim 缺失或非正数会触发STORE_VECTOR_SCHEMA_INVALID错误）由 pydantic validator 在构造阶段完成（base_vector_store.py）；检索时filters只做标量等值过滤，返回结果的fields包含命中文档的完整字段。

与 memory / session 模块的集成

这三类存储并非孤立能力，而是 memory 模块的底座。LongTermMemory.register_store(kv_store, vector_store=None, db_store=None, embedding_model=None, message_store=None)正是面向该抽象层的接入点（long_term_memory.py）：kv_store必填且必须为BaseKVStore实例，vector_store/db_store会做isinstance校验，不合法直接抛错。也就是说：

DbBasedKVStore+DefaultDbStore+ 向量后端组合，可支撑持久化、多进程共享的长期记忆与会话存储；
BaseDbStore通常配合LongTermMemory.register_store(db_store=...)使用，让 memory 引擎自行建表、执行 SQL；
InMemoryKVStore则适合单进程、无需持久化的开发调试与轻量缓存场景。

选型建议

场景	推荐实现	理由
单进程临时缓存、锁标记、开发调试	`InMemoryKVStore`	零外部依赖，`asyncio.Lock`保证并发安全
需持久化、多进程共享、与 memory/session 集成	`DbBasedKVStore`+`DefaultDbStore`	SQLite/MySQL 均可，自动建表，自动方言适配
RAG / 语义检索	`create_vector_store("chroma"/"milvus"/"gaussvector", ...)`	统一`BaseVectorStore`接口，可无缝切换后端
私有/自研向量后端	`register_vector_store(name, factory)`	进程内注册即可按名创建
对外发布的三方向量后端	entry_points 插件（`openjiuwen.vector_stores`组）	安装即用，工厂层故障隔离