人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen 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纯内存、DbBasedKVStoreSQLAlchemy 落库BaseDbStore数据库存储仅暴露异步 SQLAlchemyAsyncEngineDefaultDbStoreBaseVectorStore向量存储集合管理、文档写入、向量检索、文档删除、schema 迁移GaussVectorStoreGaussVector 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 后端的抽象基类继承自ABCbase_kv_store.py。从源码看接口比文档列出的方法更完整共 9 个抽象方法抽象方法签名说明setasync set(key: str, value: str \| bytes)写入/覆盖键值对exclusive_setasync exclusive_set(key, value, expiry: int \| None None) - bool原子化写入仅当 key 不存在或已过期时写入成功返回Truekey 已存在且未过期则返回False可用于分布式锁、一次性任务标记getasync get(key) - str \| bytes \| None读取key 不存在返回Noneexistsasync exists(key) - bool判断 key 是否存在deleteasync delete(key)删除key 不存在时静默无操作get_by_prefixasync get_by_prefix(prefix) - dict[str, str \| bytes]按前缀批量读取delete_by_prefixasync delete_by_prefix(prefix, batch_sizeNone)按前缀批量删除batch_size控制分批规模mgetasync mget(keys: List[str]) - List[str \| bytes \| None]批量读取结果顺序与入参一致缺失项为Nonebatch_deleteasync batch_delete(keys, batch_sizeNone) - 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) # NoneDbBasedKVStoreSQLAlchemy 落库实现DbBasedKVStore(engine: AsyncEngine)基于 SQLAlchemy 异步引擎使用kv_store表key主键列 value列首次调用任何接口时自动建表db_based_kv_store.py内部通过async_sessionmaker(engine, expire_on_commitFalse, 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(sqliteaiosqlite:///demo.db, echoFalse) db_kv DbBasedKVStore(async_engine) await db_kv.set(config:key1, value1) await db_kv.get(config:key1) # value1BaseDbStore数据库访问的薄抽象BaseDbStore只有一个抽象方法get_async_engine() - AsyncEnginebase_db_store.py返回异步 SQLAlchemy 引擎供调用方执行建表、原生 SQL 等操作。它是 memory 等模块做原始 SQL 能力接线的标准通道。DefaultDbStore(async_conn: AsyncEngine)是它的默认实现构造时保存传入的async_connget_async_engine()直接原样返回default_db_store.pyfrom sqlalchemy.ext.asyncio import create_async_engine from openjiuwen.core.foundation.store import DefaultDbStore async_engine create_async_engine(sqliteaiosqlite:///app.db) db_store DefaultDbStore(async_engine) assert db_store.get_async_engine() is async_engineBaseVectorStore向量存储统一接口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、IPdelete_collection(collection_name, **kwargs)/collection_exists(collection_name, **kwargs) - bool删除与存在性判断get_schema(collection_name, **kwargs) - CollectionSchema获取集合 schemalist_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_k5, filtersNone, **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( hostlocalhost, # GaussVector 服务器地址 port5432, # 端口 databasepostgres, # 数据库名 userpostgres, # 用户 password, # 密码 **kwargs, # 如 connection_timeout、sslmode )实现要点按文档与源码注释连接惰性加载connection属性首次访问时才通过psycopg2建立连接close()显式关闭支持集合表管理、文档增删改查与向量检索create_collection的kwargs较为丰富distance_metric默认COSINE可选L2、index_type默认diskann当前仅支持 DiskANN、pg_nsegPQ 分段数默认 128、pg_nclus簇数默认 16、num_parallels并行数默认 32add_docs支持batch_size分批插入默认 128delete_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, urihttp://localhost:19530) # 内置后端GaussVector store create_vector_store(gaussvector, hostlocalhost, port5432)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( sqliteaiosqlite:///demo.db, echoFalse, ) 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(sqliteaiosqlite:///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( descriptionDocument collection, enable_dynamic_fieldFalse, ) schema.add_field(FieldSchema( nameid, dtypeVectorDataType.VARCHAR, max_length256, is_primaryTrue, )) schema.add_field(FieldSchema( nameembedding, dtypeVectorDataType.FLOAT_VECTOR, dim768, )) schema.add_field(FieldSchema( nametext, dtypeVectorDataType.VARCHAR, max_length65535, )) schema.add_field(FieldSchema( namemetadata, dtypeVectorDataType.JSON, )) # Create collection await store.create_collection(documents, schema, distance_metriccosine) # 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_namedocuments, query_vectorquery_vector, vector_fieldembedding, top_k10, filters{category: tech}, ) for result in results: print(fScore: {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_storeNone, db_storeNone, embedding_modelNone, message_storeNone)正是面向该抽象层的接入点long_term_memory.pykv_store必填且必须为BaseKVStore实例vector_store/db_store会做isinstance校验不合法直接抛错。也就是说DbBasedKVStoreDefaultDbStore 向量后端组合可支撑持久化、多进程共享的长期记忆与会话存储BaseDbStore通常配合LongTermMemory.register_store(db_store...)使用让 memory 引擎自行建表、执行 SQLInMemoryKVStore则适合单进程、无需持久化的开发调试与轻量缓存场景。选型建议场景推荐实现理由单进程临时缓存、锁标记、开发调试InMemoryKVStore零外部依赖asyncio.Lock保证并发安全需持久化、多进程共享、与 memory/session 集成DbBasedKVStoreDefaultDbStoreSQLite/MySQL 均可自动建表自动方言适配RAG / 语义检索create_vector_store(chroma/milvus/gaussvector, ...)统一BaseVectorStore接口可无缝切换后端私有/自研向量后端register_vector_store(name, factory)进程内注册即可按名创建对外发布的三方向量后端entry_points 插件openjiuwen.vector_stores组安装即用工厂层故障隔离相关单元测试可进一步印证实现行为test_in_memory_kv_store.py、test_db_based_kv_store.py 覆盖 KV 两类实现的接口契约、过期语义与批量操作test_vector_store_plugin.py 覆盖向量工厂与插件机制。赞分享人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习【免费下载链接】agent-coreopenJiuwen agent-core可提供AI Agent开发、运行、调优与演进相关的全套SDK能力项目地址https://gitcode.com/openJiuwen/agent-core点击查看免费下载相关推荐openJiuwen Agent-Core 对象存储抽象层实战指南深入解析 BaseObjectStorageClientopenJiuwen Agent Core 对象存储抽象层实战指南深入解析 BaseObjectStorageClient BaseObjectStorage人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen agent-core 存储抽象层foundation.store实战指南KV、数据库、向量存储与插件扩展openJiuwen agent core 存储抽象层foundation.store实战指南KV、数据库、向量存储与插件扩展 openjiuwen.co人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习openJiuwen agent-core 对象存储客户端BaseObjectStorageClient 抽象接口与 AioBotoClient 异步 S3 实现详解openJiuwen agent core 对象存储客户端BaseObjectStorageClient 抽象接口与 AioBotoClient 异步 S3人工智能AI AgentAgent 框架大模型工具调用RAG提示工程强化学习上一篇为什么选择Aurora探索这款地球系统基础模型的独特优势下一篇Boost.Beast中的DynamicBuffer概念解析创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考