LlamaIndex DuckDBKVStore 深度解析基于 DuckDB 的键值存储集成实战指南【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_indexDuckDBKVStore 是 LlamaIndex 在 DuckDB 之上实现的一个键值存储KV Store集成它实现了llama_index.core.storage.kvstore.types.BaseKVStore抽象接口将 DuckDB 原生 JSON 类型、PyArrow 批量写入与复合主键约束封装为可供索引、文档存储复用的底层存储组件。本文将以docs/api_reference/api_reference/storage/kvstore/duckdb.md指向的DuckDBKVStore类为核心结合仓库源码、依赖配置与单元测试讲解其初始化参数、表结构、全部核心方法与上层集成方式帮助你掌握在 LlamaIndex 中使用 DuckDB 持久化元数据与文档数据的完整方案。一、模块定位KVStore 抽象体系中的 DuckDB 实现在 LlamaIndex 的存储分层中键值存储是最基础的抽象层之一。BaseKVStore 定义了统一的键值读写接口包括同步与异步两套方法put/aput、put_all/aput_all、get/aget、get_all/aget_all、delete/adelete。基类中约定DEFAULT_COLLECTION data即默认所有数据写入名为data的集合collection下。DuckDBKVStore 是这套抽象接口的 DuckDB 后端实现位于独立的集成包llama-index-storage-kvstore-duckdb核心源码为 duckdb/base.py。从包结构看模块的导入路径为llama_index.storage.kvstore.duckdb顶层仅导出DuckDBKVStore一个类见init.py。单元测试 test_storage_kvstore_duckdb.py 中的test_class验证了该类的继承关系def test_class(): names_of_base_classes [b.__name__ for b in DuckDBKVStore.__mro__] assert BaseKVStore.__name__ in names_of_base_classes这意味着任何接受BaseKVStore的 LlamaIndex 组件都可以无缝切换为 DuckDB 后端。二、安装与依赖约束该集成包的依赖声明在 pyproject.toml 中dependencies [ duckdb0.10.1,1.4.0, llama-index-core0.13.0,0.15, llama-index-vector-stores-duckdb0.4.2, pyarrow20.0.0, ]关键约束说明duckdb要求0.10.1,1.4.0这是底层数据库引擎负责连接、表管理与 SQL 执行pyarrow要求20.0.0用于put_all时将 Python 字典列表高效转换为 Arrow Table 后批量写入llama-index-core要求0.13.0,0.15提供BaseKVStore等核心抽象llama-index-vector-stores-duckdb要求0.4.2用于from_vector_store类方法与 DuckDB 向量存储共享连接。包名版本为0.3.0要求 Python3.10,4.0。安装方式与普通 LlamaIndex 集成包一致例如pip install llama-index-storage-kvstore-duckdb三、构造参数与连接管理DuckDBKVStore.__init__的完整签名如下源码 base.pydef __init__( self, database_name: str :memory:, table_name: str keyvalue, persist_dir: str ./storage, client: Optional[duckdb.DuckDBPyConnection] None, **kwargs: Any, ) - None:各参数含义参数默认值说明database_name:memory:DuckDB 数据库名称。传入:memory:表示纯内存数据库传入文件名如persisted.duckdb则持久化到磁盘table_namekeyvalue存储键值对的表名建表时若不存在会自动创建persist_dir./storage数据持久化目录。仅当database_name非:memory:时生效目录不存在会自动递归创建clientNone外部传入的 DuckDB 连接。传入时会在其基础上创建独立游标cursor用于与其他组件共享同一数据库连接关于persist_dir与database_name的配合从_connect类方法base.py可以看出classmethod def _connect(cls, database_name: str, persist_dir: str) - duckdb.DuckDBPyConnection: database_connection database_name if database_name ! :memory:: persist_path Path(persist_dir) if not persist_path.exists(): persist_path.mkdir(parentsTrue, exist_okTrue) database_connection str(persist_path / database_name) return duckdb.connect(database_connection)即磁盘模式下实际连接路径为persist_dir / database_name。测试中的磁盘用例test_storage_kvstore_duckdb.py与此一致def disk_store(): if os.path.exists(./storage/persisted.duckdb): os.remove(./storage/persisted.duckdb) return DuckDBKVStore(database_namepersisted.duckdb, persist_dir./storage)线程安全设计DuckDB 的连接对象并非线程安全。源码通过threading.local()为每个线程维护独立的游标base.pyproperty def client(self) - duckdb.DuckDBPyConnection: if self._shared_conn is None: self._shared_conn self._connect(self.database_name, self.persist_dir) if not hasattr(self._thread_local, conn) or self._thread_local.conn is None: self._thread_local.conn self._shared_conn.cursor() return self._thread_local.conn设计要点所有线程共享同一个底层连接_shared_conn但每个线程通过_shared_conn.cursor()获取自己的游标执行操作从而在多线程场景下保证安全。这从源码结构可以推断出该实现是面向多线程读取/写入场景设计的。四、底层表结构与初始化流程实例化时构造函数会调用_initialize_tablebase.py自动完成建表核心 SQL 如下CREATE TABLE IF NOT EXISTS keyvalue ( key VARCHAR, collection VARCHAR, value JSON, PRIMARY KEY (key, collection) ); CREATE INDEX IF NOT EXISTS collection_idx ON keyvalue (collection);表结构设计要点key VARCHAR键与collection组成复合主键collection VARCHAR集合名用于在同一张表中逻辑隔离不同业务数据value JSON值使用 DuckDB 原生 JSON 类型存储写入时通过json.dumps序列化读取时通过json.loads反序列化天然支持任意嵌套字典结构复合主键(key, collection)保证同一集合内键唯一为INSERT OR REPLACE的覆盖写语义提供支撑collection_idx索引加速按集合维度的过滤查询。此外初始化时还会安装 DuckDB 的json扩展conn.install_extension(json)与conn.load_extension(json)并将home_directory设置为用户主目录。建表完成后_initialize_table会校验表必须包含key与value两列否则抛出自定义异常DuckDBTableIncorrectColumnsError定义于 base.py防止误用结构不符的既有表。五、核心 API 全解DuckDBKVStore 完整实现了 BaseKVStore 的同步与异步接口。以下逐一说明其行为与底层实现。5.1 写入put / put_all / aput / aput_allput将单个键值对写入指定集合其实现直接委托给put_allbase.pydef put(self, key: str, val: dict, collection: str DEFAULT_COLLECTION) - None: self.put_all([(key, val)], collection)put_all是批量写入的核心实现利用 PyArrow 提升写入性能base.pydef put_all( self, kv_pairs: list[tuple[str, dict]], collection: str DEFAULT_COLLECTION, batch_size: int DEFAULT_BATCH_SIZE, ) - None: if len(kv_pairs) 0: return rows [ {key: key, collection: collection, value: json.dumps(value)} for key, value in kv_pairs ] arrow_table pyarrow.Table.from_pylist(rows) _ self.client.sql( queryf INSERT OR REPLACE INTO {self.table.alias} SELECT * from arrow_table; , )实现要点空列表直接返回不产生任何数据库操作数据先转为pyarrow.Table再通过 DuckDB 的 SQL 查询直接消费 Arrow 表完成写入避免逐行插入INSERT OR REPLACE结合复合主键(key, collection)实现存在即覆盖的语义测试test_put_twice验证了这一点同一 key 二次写入后get返回的是更新后的值DEFAULT_BATCH_SIZE 128定义于模块级base.py作为batch_size参数默认值保留。异步写入aput/aput_all通过asyncio.to_thread将同步操作提交到线程池执行base.py避免阻塞事件循环同时天然复用了线程安全的游标机制。5.2 读取get / get_all / aget / aget_allget使用 DuckDB 的表达式 API 构造过滤条件按(collection, key)精确定位base.pydef get(self, key: str, collection: str DEFAULT_COLLECTION) - Optional[dict]: expression: Expression ( ColumnExpression(collection) .__eq__(ConstantExpression(collection)) .__and__(ColumnExpression(key).__eq__(ConstantExpression(key))) ) row_result self.table.filter(filter_exprexpression).fetchone() if row_result is None: return None return json.loads(row_result[2])查询不到时返回None命中时对第三列value做json.loads还原为字典。get_all返回指定集合下的全部键值映射同样基于集合过滤并转为pyarrow.Table后以字典推导式聚合base.pydef get_all(self, collection: str DEFAULT_COLLECTION) - dict[str, dict]: filter_expr: Expression ColumnExpression(collection).__eq__( ConstantExpression(collection) ) table: pyarrow.Table self.table.filter( filter_exprfilter_expr ).fetch_arrow_table() as_list table.to_pylist() return {row[key]: json.loads(row[value]) for row in as_list}异步版本aget/aget_all同样通过asyncio.to_thread委托给同步实现。5.3 删除delete / adeletedelete构造 DELETE 语句按(collection, key)精确删除并返回Truebase.pydef delete(self, key: str, collection: str DEFAULT_COLLECTION) - bool: filter_expression ( ColumnExpression(collection) .__eq__(ConstantExpression(collection)) .__and__(ColumnExpression(key).__eq__(ConstantExpression(key))) ) command fDELETE FROM {self.table.alias} WHERE {filter_expression} _ self.client.execute(command) return Trueadelete直接委托给delete同步执行base.py。5.4 典型使用示例综合以上 API一个完整的读写删生命周期如下与测试用例 test_storage_kvstore_duckdb.py 行为一致from llama_index.storage.kvstore.duckdb import DuckDBKVStore # 内存模式 kv_store DuckDBKVStore() # 写入 kv_store.put(id_1, {name: John Doe, text: Hello, world!}) # 读取 assert kv_store.get(id_1) {name: John Doe, text: Hello, world!} # 批量写入与集合隔离 kv_store.put_all( [(id_1, {score: 95}), (id_2, {score: 88})], collectioncollection_1, ) # 删除 kv_store.delete(id_1) assert kv_store.get(id_1) is None # 磁盘持久化 disk_store DuckDBKVStore(database_namepersisted.duckdb, persist_dir./storage) disk_store.put(doc_1, {title: hello})异步场景对应测试test_asyncimport asyncio async def main(): kv_store DuckDBKVStore() await kv_store.aput(id_1, {name: John Doe}) assert await kv_store.aget(id_1) {name: John Doe} await kv_store.adelete(id_1) asyncio.run(main())六、与 DuckDB 向量存储共享连接from_vector_storefrom_vector_store类方法允许直接从已初始化的DuckDBVectorStore派生 KV Store复用其数据库名、持久化目录与连接对象base.pyclassmethod def from_vector_store( cls, duckdb_vector_store, table_name: str keyvalue ) - DuckDBKVStore: from llama_index.vector_stores.duckdb.base import DuckDBVectorStore assert isinstance(duckdb_vector_store, DuckDBVectorStore) return cls( database_nameduckdb_vector_store.database_name, table_nametable_name, persist_dirduckdb_vector_store.persist_dir, clientduckdb_vector_store.client, )该方法内部做了类型断言仅接受DuckDBVectorStore实例。测试test_from_vector_store验证了派生后属性继承关系与读写可用性def test_from_vector_store(): vector_store DuckDBVectorStore() kv_store DuckDBKVStore.from_vector_store(duckdb_vector_storevector_store) assert kv_store.database_name vector_store.database_name assert kv_store.table_name keyvalue assert kv_store.persist_dir vector_store.persist_dir kv_store.put(id_1, {name: John Doe, text: Hello, world!}) results kv_store.get_all() assert results[id_1] {name: John Doe, text: Hello, world!}这一设计使得向量索引与键值元数据可以共用一个 DuckDB 数据库文件便于统一管理与备份。七、上层应用DuckDBDocumentStore 与 DuckDBIndexStoreDuckDBKVStore 并非孤立组件它是 DuckDB 系文档存储与索引存储的底层基础。7.1 DuckDBDocumentStoreDuckDBDocumentStore 继承自KVDocumentStore将DuckDBKVStore作为底层存储class DuckDBDocumentStore(KVDocumentStore): def __init__( self, duckdb_kvstore: DuckDBKVStore, namespace: Optional[str] None, batch_size: int DEFAULT_BATCH_SIZE, ) - None: super().__init__(duckdb_kvstore, namespacenamespace, batch_sizebatch_size) # avoid conflicts with duckdb index store self._node_collection f{self._namespace}/doc这里可以看到 KVStore 的 collection 机制如何被上层复用文档存储通过namespace /doc作为集合名将节点数据与索引数据隔离在同一张keyvalue表中。7.2 DuckDBIndexStoreDuckDBIndexStore 同理继承自KVIndexStore并将集合名规范为namespace /index避免与文档存储冲突class DuckDBIndexStore(KVIndexStore): def __init__( self, duckdb_kvstore: DuckDBKVStore, namespace: Optional[str] None, collection_suffix: Optional[str] None, ) - None: super().__init__( duckdb_kvstore, namespacenamespace, collection_suffixcollection_suffix ) if self._collection.endswith(DEFAULT_COLLECTION_SUFFIX): self._collection f{self._namespace}/index因此一个DuckDBKVStore实例可以同时支撑文档存储与索引存储三者共享同一数据库文件配合DuckDBVectorStore即可构建完整的向量 文档 索引统一持久化方案。八、总结DuckDBKVStore 的价值在于以极少的代码量将 DuckDB 的高性能列式存储能力接入 LlamaIndex 的 KVStore 抽象层。其核心设计可以概括为四点JSON 原生存储value列使用 DuckDB JSON 类型json.dumps/json.loads完成序列化边界支持任意结构化字典PyArrow 批量写入put_all通过 Arrow Table 一次性灌入规避逐行插入的性能开销复合主键 覆盖写(key, collection)复合主键保证集合内键唯一INSERT OR REPLACE提供幂等写入语义线程安全与异步友好threading.local()每线程游标保障多线程安全asyncio.to_thread让异步 API 不阻塞事件循环。在需要将索引元数据、文档节点与向量数据统一落到单一 DuckDB 文件的场景下DuckDBKVStore与其衍生的DuckDBDocumentStore、DuckDBIndexStore构成了完整的落地方案。相关实现、依赖与测试均可在当前仓库中直接查看核心源码位于 llama-index-storage-kvstore-duckdb 包单元测试覆盖了初始化、读写、批量、覆盖、集合隔离、删除与异步等全部行为可作为自定义扩展时的参考蓝本。【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 LlamaIndex DuckDBKVStore 深度解析：基于 DuckDB 的键值存储集成实战指南
LlamaIndex DuckDBKVStore 深度解析基于 DuckDB 的键值存储集成实战指南【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_indexDuckDBKVStore 是 LlamaIndex 在 DuckDB 之上实现的一个键值存储KV Store集成它实现了llama_index.core.storage.kvstore.types.BaseKVStore抽象接口将 DuckDB 原生 JSON 类型、PyArrow 批量写入与复合主键约束封装为可供索引、文档存储复用的底层存储组件。本文将以docs/api_reference/api_reference/storage/kvstore/duckdb.md指向的DuckDBKVStore类为核心结合仓库源码、依赖配置与单元测试讲解其初始化参数、表结构、全部核心方法与上层集成方式帮助你掌握在 LlamaIndex 中使用 DuckDB 持久化元数据与文档数据的完整方案。一、模块定位KVStore 抽象体系中的 DuckDB 实现在 LlamaIndex 的存储分层中键值存储是最基础的抽象层之一。BaseKVStore 定义了统一的键值读写接口包括同步与异步两套方法put/aput、put_all/aput_all、get/aget、get_all/aget_all、delete/adelete。基类中约定DEFAULT_COLLECTION data即默认所有数据写入名为data的集合collection下。DuckDBKVStore 是这套抽象接口的 DuckDB 后端实现位于独立的集成包llama-index-storage-kvstore-duckdb核心源码为 duckdb/base.py。从包结构看模块的导入路径为llama_index.storage.kvstore.duckdb顶层仅导出DuckDBKVStore一个类见init.py。单元测试 test_storage_kvstore_duckdb.py 中的test_class验证了该类的继承关系def test_class(): names_of_base_classes [b.__name__ for b in DuckDBKVStore.__mro__] assert BaseKVStore.__name__ in names_of_base_classes这意味着任何接受BaseKVStore的 LlamaIndex 组件都可以无缝切换为 DuckDB 后端。二、安装与依赖约束该集成包的依赖声明在 pyproject.toml 中dependencies [ duckdb0.10.1,1.4.0, llama-index-core0.13.0,0.15, llama-index-vector-stores-duckdb0.4.2, pyarrow20.0.0, ]关键约束说明duckdb要求0.10.1,1.4.0这是底层数据库引擎负责连接、表管理与 SQL 执行pyarrow要求20.0.0用于put_all时将 Python 字典列表高效转换为 Arrow Table 后批量写入llama-index-core要求0.13.0,0.15提供BaseKVStore等核心抽象llama-index-vector-stores-duckdb要求0.4.2用于from_vector_store类方法与 DuckDB 向量存储共享连接。包名版本为0.3.0要求 Python3.10,4.0。安装方式与普通 LlamaIndex 集成包一致例如pip install llama-index-storage-kvstore-duckdb三、构造参数与连接管理DuckDBKVStore.__init__的完整签名如下源码 base.pydef __init__( self, database_name: str :memory:, table_name: str keyvalue, persist_dir: str ./storage, client: Optional[duckdb.DuckDBPyConnection] None, **kwargs: Any, ) - None:各参数含义参数默认值说明database_name:memory:DuckDB 数据库名称。传入:memory:表示纯内存数据库传入文件名如persisted.duckdb则持久化到磁盘table_namekeyvalue存储键值对的表名建表时若不存在会自动创建persist_dir./storage数据持久化目录。仅当database_name非:memory:时生效目录不存在会自动递归创建clientNone外部传入的 DuckDB 连接。传入时会在其基础上创建独立游标cursor用于与其他组件共享同一数据库连接关于persist_dir与database_name的配合从_connect类方法base.py可以看出classmethod def _connect(cls, database_name: str, persist_dir: str) - duckdb.DuckDBPyConnection: database_connection database_name if database_name ! :memory:: persist_path Path(persist_dir) if not persist_path.exists(): persist_path.mkdir(parentsTrue, exist_okTrue) database_connection str(persist_path / database_name) return duckdb.connect(database_connection)即磁盘模式下实际连接路径为persist_dir / database_name。测试中的磁盘用例test_storage_kvstore_duckdb.py与此一致def disk_store(): if os.path.exists(./storage/persisted.duckdb): os.remove(./storage/persisted.duckdb) return DuckDBKVStore(database_namepersisted.duckdb, persist_dir./storage)线程安全设计DuckDB 的连接对象并非线程安全。源码通过threading.local()为每个线程维护独立的游标base.pyproperty def client(self) - duckdb.DuckDBPyConnection: if self._shared_conn is None: self._shared_conn self._connect(self.database_name, self.persist_dir) if not hasattr(self._thread_local, conn) or self._thread_local.conn is None: self._thread_local.conn self._shared_conn.cursor() return self._thread_local.conn设计要点所有线程共享同一个底层连接_shared_conn但每个线程通过_shared_conn.cursor()获取自己的游标执行操作从而在多线程场景下保证安全。这从源码结构可以推断出该实现是面向多线程读取/写入场景设计的。四、底层表结构与初始化流程实例化时构造函数会调用_initialize_tablebase.py自动完成建表核心 SQL 如下CREATE TABLE IF NOT EXISTS keyvalue ( key VARCHAR, collection VARCHAR, value JSON, PRIMARY KEY (key, collection) ); CREATE INDEX IF NOT EXISTS collection_idx ON keyvalue (collection);表结构设计要点key VARCHAR键与collection组成复合主键collection VARCHAR集合名用于在同一张表中逻辑隔离不同业务数据value JSON值使用 DuckDB 原生 JSON 类型存储写入时通过json.dumps序列化读取时通过json.loads反序列化天然支持任意嵌套字典结构复合主键(key, collection)保证同一集合内键唯一为INSERT OR REPLACE的覆盖写语义提供支撑collection_idx索引加速按集合维度的过滤查询。此外初始化时还会安装 DuckDB 的json扩展conn.install_extension(json)与conn.load_extension(json)并将home_directory设置为用户主目录。建表完成后_initialize_table会校验表必须包含key与value两列否则抛出自定义异常DuckDBTableIncorrectColumnsError定义于 base.py防止误用结构不符的既有表。五、核心 API 全解DuckDBKVStore 完整实现了 BaseKVStore 的同步与异步接口。以下逐一说明其行为与底层实现。5.1 写入put / put_all / aput / aput_allput将单个键值对写入指定集合其实现直接委托给put_allbase.pydef put(self, key: str, val: dict, collection: str DEFAULT_COLLECTION) - None: self.put_all([(key, val)], collection)put_all是批量写入的核心实现利用 PyArrow 提升写入性能base.pydef put_all( self, kv_pairs: list[tuple[str, dict]], collection: str DEFAULT_COLLECTION, batch_size: int DEFAULT_BATCH_SIZE, ) - None: if len(kv_pairs) 0: return rows [ {key: key, collection: collection, value: json.dumps(value)} for key, value in kv_pairs ] arrow_table pyarrow.Table.from_pylist(rows) _ self.client.sql( queryf INSERT OR REPLACE INTO {self.table.alias} SELECT * from arrow_table; , )实现要点空列表直接返回不产生任何数据库操作数据先转为pyarrow.Table再通过 DuckDB 的 SQL 查询直接消费 Arrow 表完成写入避免逐行插入INSERT OR REPLACE结合复合主键(key, collection)实现存在即覆盖的语义测试test_put_twice验证了这一点同一 key 二次写入后get返回的是更新后的值DEFAULT_BATCH_SIZE 128定义于模块级base.py作为batch_size参数默认值保留。异步写入aput/aput_all通过asyncio.to_thread将同步操作提交到线程池执行base.py避免阻塞事件循环同时天然复用了线程安全的游标机制。5.2 读取get / get_all / aget / aget_allget使用 DuckDB 的表达式 API 构造过滤条件按(collection, key)精确定位base.pydef get(self, key: str, collection: str DEFAULT_COLLECTION) - Optional[dict]: expression: Expression ( ColumnExpression(collection) .__eq__(ConstantExpression(collection)) .__and__(ColumnExpression(key).__eq__(ConstantExpression(key))) ) row_result self.table.filter(filter_exprexpression).fetchone() if row_result is None: return None return json.loads(row_result[2])查询不到时返回None命中时对第三列value做json.loads还原为字典。get_all返回指定集合下的全部键值映射同样基于集合过滤并转为pyarrow.Table后以字典推导式聚合base.pydef get_all(self, collection: str DEFAULT_COLLECTION) - dict[str, dict]: filter_expr: Expression ColumnExpression(collection).__eq__( ConstantExpression(collection) ) table: pyarrow.Table self.table.filter( filter_exprfilter_expr ).fetch_arrow_table() as_list table.to_pylist() return {row[key]: json.loads(row[value]) for row in as_list}异步版本aget/aget_all同样通过asyncio.to_thread委托给同步实现。5.3 删除delete / adeletedelete构造 DELETE 语句按(collection, key)精确删除并返回Truebase.pydef delete(self, key: str, collection: str DEFAULT_COLLECTION) - bool: filter_expression ( ColumnExpression(collection) .__eq__(ConstantExpression(collection)) .__and__(ColumnExpression(key).__eq__(ConstantExpression(key))) ) command fDELETE FROM {self.table.alias} WHERE {filter_expression} _ self.client.execute(command) return Trueadelete直接委托给delete同步执行base.py。5.4 典型使用示例综合以上 API一个完整的读写删生命周期如下与测试用例 test_storage_kvstore_duckdb.py 行为一致from llama_index.storage.kvstore.duckdb import DuckDBKVStore # 内存模式 kv_store DuckDBKVStore() # 写入 kv_store.put(id_1, {name: John Doe, text: Hello, world!}) # 读取 assert kv_store.get(id_1) {name: John Doe, text: Hello, world!} # 批量写入与集合隔离 kv_store.put_all( [(id_1, {score: 95}), (id_2, {score: 88})], collectioncollection_1, ) # 删除 kv_store.delete(id_1) assert kv_store.get(id_1) is None # 磁盘持久化 disk_store DuckDBKVStore(database_namepersisted.duckdb, persist_dir./storage) disk_store.put(doc_1, {title: hello})异步场景对应测试test_asyncimport asyncio async def main(): kv_store DuckDBKVStore() await kv_store.aput(id_1, {name: John Doe}) assert await kv_store.aget(id_1) {name: John Doe} await kv_store.adelete(id_1) asyncio.run(main())六、与 DuckDB 向量存储共享连接from_vector_storefrom_vector_store类方法允许直接从已初始化的DuckDBVectorStore派生 KV Store复用其数据库名、持久化目录与连接对象base.pyclassmethod def from_vector_store( cls, duckdb_vector_store, table_name: str keyvalue ) - DuckDBKVStore: from llama_index.vector_stores.duckdb.base import DuckDBVectorStore assert isinstance(duckdb_vector_store, DuckDBVectorStore) return cls( database_nameduckdb_vector_store.database_name, table_nametable_name, persist_dirduckdb_vector_store.persist_dir, clientduckdb_vector_store.client, )该方法内部做了类型断言仅接受DuckDBVectorStore实例。测试test_from_vector_store验证了派生后属性继承关系与读写可用性def test_from_vector_store(): vector_store DuckDBVectorStore() kv_store DuckDBKVStore.from_vector_store(duckdb_vector_storevector_store) assert kv_store.database_name vector_store.database_name assert kv_store.table_name keyvalue assert kv_store.persist_dir vector_store.persist_dir kv_store.put(id_1, {name: John Doe, text: Hello, world!}) results kv_store.get_all() assert results[id_1] {name: John Doe, text: Hello, world!}这一设计使得向量索引与键值元数据可以共用一个 DuckDB 数据库文件便于统一管理与备份。七、上层应用DuckDBDocumentStore 与 DuckDBIndexStoreDuckDBKVStore 并非孤立组件它是 DuckDB 系文档存储与索引存储的底层基础。7.1 DuckDBDocumentStoreDuckDBDocumentStore 继承自KVDocumentStore将DuckDBKVStore作为底层存储class DuckDBDocumentStore(KVDocumentStore): def __init__( self, duckdb_kvstore: DuckDBKVStore, namespace: Optional[str] None, batch_size: int DEFAULT_BATCH_SIZE, ) - None: super().__init__(duckdb_kvstore, namespacenamespace, batch_sizebatch_size) # avoid conflicts with duckdb index store self._node_collection f{self._namespace}/doc这里可以看到 KVStore 的 collection 机制如何被上层复用文档存储通过namespace /doc作为集合名将节点数据与索引数据隔离在同一张keyvalue表中。7.2 DuckDBIndexStoreDuckDBIndexStore 同理继承自KVIndexStore并将集合名规范为namespace /index避免与文档存储冲突class DuckDBIndexStore(KVIndexStore): def __init__( self, duckdb_kvstore: DuckDBKVStore, namespace: Optional[str] None, collection_suffix: Optional[str] None, ) - None: super().__init__( duckdb_kvstore, namespacenamespace, collection_suffixcollection_suffix ) if self._collection.endswith(DEFAULT_COLLECTION_SUFFIX): self._collection f{self._namespace}/index因此一个DuckDBKVStore实例可以同时支撑文档存储与索引存储三者共享同一数据库文件配合DuckDBVectorStore即可构建完整的向量 文档 索引统一持久化方案。八、总结DuckDBKVStore 的价值在于以极少的代码量将 DuckDB 的高性能列式存储能力接入 LlamaIndex 的 KVStore 抽象层。其核心设计可以概括为四点JSON 原生存储value列使用 DuckDB JSON 类型json.dumps/json.loads完成序列化边界支持任意结构化字典PyArrow 批量写入put_all通过 Arrow Table 一次性灌入规避逐行插入的性能开销复合主键 覆盖写(key, collection)复合主键保证集合内键唯一INSERT OR REPLACE提供幂等写入语义线程安全与异步友好threading.local()每线程游标保障多线程安全asyncio.to_thread让异步 API 不阻塞事件循环。在需要将索引元数据、文档节点与向量数据统一落到单一 DuckDB 文件的场景下DuckDBKVStore与其衍生的DuckDBDocumentStore、DuckDBIndexStore构成了完整的落地方案。相关实现、依赖与测试均可在当前仓库中直接查看核心源码位于 llama-index-storage-kvstore-duckdb 包单元测试覆盖了初始化、读写、批量、覆盖、集合隔离、删除与异步等全部行为可作为自定义扩展时的参考蓝本。【免费下载链接】llama_indexLlamaIndex is the leading document agent and OCR platform项目地址: https://gitcode.com/GitHub_Trending/ll/llama_index创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考