Parlant Qdrant 向量数据库适配器：用持久化向量存储替换默认内存存储的完整实践

Parlant Qdrant 向量数据库适配器用持久化向量存储替换默认内存存储的完整实践【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant本文基于 Parlant 仓库中的 Qdrant 适配器文档 与 Qdrant 适配器源码 展开讲解如何在 Parlant 中通过QdrantDatabase将术语表、固定回复、能力、旅程等四类向量存储从默认内存transient存储迁移到生产级持久化存储。读完本文你将掌握 Qdrant 适配器的完整接入代码、集合命名与自动同步机制、Where 过滤器到 Qdrant Filter 的映射规则以及 Windows 文件锁、嵌入器切换、数据持久化验证等常见问题的排查方法。为什么需要 Qdrant 适配器Parlant 默认的向量存储由 TransientVectorDatabase 提供底层基于nano_vectordb在内存中维护向量集合# src/parlant/adapters/vector_db/transient.py节选 self._databases[name] nano_vectordb.NanoVectorDB(embedder.dimensions)这种实现适合开发调试但服务重启后向量数据即丢失。Qdrant 适配器通过QdrantDatabase类实现了核心抽象 VectorDatabase 接口create_collection、get_collection、get_or_create_collection、delete_collection、upsert_metadata、remove_metadata、read_metadata将向量数据的持久化能力无缝接入 Parlant 的容器Container体系同时保持与现有向量存储接口的完全兼容——上层只需在configure_container中替换 Store 实现无需修改任何业务逻辑。前置条件安装 Qdrant 适配器以可选依赖方式安装pip install parlant[qdrant]从 pyproject.toml 可见qdrantextra 实际拉取的是qdrant-client1.7.0。选择存储方式本地文件系统或 Qdrant Cloud 远程集群。Python 版本文档中标注为 Python 3.8但从当前仓库的 pyproject.toml 看Parlant 实际要求requires-python 3.10,3.15该限制是为兼容 torch 2.8 与 triton因此请以 Python 3.10 为实际前提。本地存储模式需要一个可写目录Cloud 模式则需要一个 Qdrant Cloud 账户URL 与 API Key。快速上手在 configure_container 中接入 Qdrant核心思路是在 SDK 的configure_container钩子中创建QdrantDatabase实例然后把四个向量 StoreGlossaryStore、CannedResponseStore、CapabilityStore、JourneyStore全部替换为对应 VectorStore 实现。以下完整代码继承自 官方文档import parlant.sdk as p from pathlib import Path from contextlib import AsyncExitStack from parlant.adapters.vector_db.qdrant import QdrantDatabase from parlant.core.nlp.embedding import EmbedderFactory, EmbeddingCache, Embedder from parlant.core.loggers import Logger from parlant.core.nlp.service import NLPService from parlant.core.glossary import GlossaryVectorStore, GlossaryStore from parlant.core.canned_responses import CannedResponseVectorStore, CannedResponseStore from parlant.core.capabilities import CapabilityVectorStore, CapabilityStore from parlant.core.journeys import JourneyVectorStore, JourneyStore from parlant.adapters.db.transient import TransientDocumentDatabase async def configure_container(container: p.Container) - p.Container: embedder_factory EmbedderFactory(container) async def get_embedder_type() - type[Embedder]: return type(await container[NLPService].get_embedder()) exit_stack AsyncExitStack() qdrant_db await exit_stack.enter_async_context( QdrantDatabase( loggercontainer[Logger], pathPath(./qdrant_data), embedder_factoryEmbedderFactory(container), embedding_cache_providerlambda: container[EmbeddingCache], ) ) # For Qdrant Cloud, replace the above with: # qdrant_db await exit_stack.enter_async_context( # QdrantDatabase( # loggercontainer[Logger], # urlhttps://your-cluster-id.us-east4-0.gcp.cloud.qdrant.io, # api_keyyour-api-key-here, # embedder_factoryEmbedderFactory(container), # embedding_cache_providerlambda: container[EmbeddingCache], # ) # ) # Configure stores using vector database container[GlossaryStore] await exit_stack.enter_async_context( GlossaryVectorStore( id_generatorcontainer[p.IdGenerator], vector_dbqdrant_db, document_dbTransientDocumentDatabase(), embedder_factoryembedder_factory, embedder_type_providerget_embedder_type, ) # type: ignore ) container[CannedResponseStore] await exit_stack.enter_async_context( CannedResponseVectorStore( id_generatorcontainer[p.IdGenerator], vector_dbqdrant_db, document_dbTransientDocumentDatabase(), embedder_factoryembedder_factory, embedder_type_providerget_embedder_type, ) # type: ignore ) container[CapabilityStore] await exit_stack.enter_async_context( CapabilityVectorStore( id_generatorcontainer[p.IdGenerator], vector_dbqdrant_db, document_dbTransientDocumentDatabase(), embedder_factoryembedder_factory, embedder_type_providerget_embedder_type, ) # type: ignore ) container[JourneyStore] await exit_stack.enter_async_context( JourneyVectorStore( id_generatorcontainer[p.IdGenerator], vector_dbqdrant_db, document_dbTransientDocumentDatabase(), embedder_factoryembedder_factory, embedder_type_providerget_embedder_type, ) # type: ignore ) return container async def main(): async with p.Server(configure_containerconfigure_container) as server: agent await server.create_agent( nameMy Agent, descriptionAgent using Qdrant for persistent storage, ) # Test: Create a term to verify Qdrant is working term await agent.create_term( nameExample Term, descriptionThis is stored in Qdrant, ) print(fCreated term: {term.name}) # All vector operations now use Qdrant关键参数说明结合 QdrantDatabase 构造函数各参数含义如下参数说明logger从容器中取Logger用于记录超时重试等警告。tracer源码中还有tracer: Tracer参数用于链路追踪文档示例未显式传入时以容器默认行为为准。path本地 Qdrant 的数据目录如Path(./qdrant_data)启动后会在该目录下生成 Qdrant 数据库文件。与url互斥传path走本地模式。url/api_keyQdrant Cloud 集群地址与密钥。走远程模式时客户端会显式设置60 秒超时见下文源码分析以容忍大批量操作和较慢的网络。embedder_factoryEmbedderFactory用于按嵌入器类型创建 embedder 并决定向量维度。embedding_cache_provider嵌入缓存提供者通常注入container[EmbeddingCache]避免对相同文本重复调用嵌入模型。都不传从源码看path和url均未提供时QdrantClient会以:memory:初始化仅适合测试场景源码。四个 VectorStore 的构造参数完全一致其中id_generatorcontainer[p.IdGenerator]统一使用容器中的 ID 生成器vector_dbqdrant_db即上面的 Qdrant 数据库实例document_dbTransientDocumentDatabase()注意这里文档非向量存储仍可使用内存实现持久化的只是向量部分如需完整持久化可另行接入文档数据库适配器embedder_type_providerget_embedder_type一个异步回调返回当前 NLP 服务的嵌入器类型用于决定集合命名与维度。生命周期管理要点QdrantDatabase与四个 VectorStore 都是异步上下文管理器必须通过AsyncExitStack的enter_async_context注册确保服务关闭时按序退出、释放文件句柄这一点在 Windows 上尤其重要见下文。源码深潜Qdrant 适配器内部机制1. 双集合结构与命名规则每个逻辑集合在 Qdrant 中实际对应两个物理集合create_collection 源码嵌入集合{name}_{EmbedderTypeName}存放真实向量尺寸取embedder.dimensions距离度量固定为 COSINE未嵌入集合{name}_unembedded向量尺寸仅为 1占位向量[0]作为文档的SSOT单一事实来源存储完整 payload 与校验和checksum。例如嵌入器为OpenAITextEmbedding3Large时你会看到glossary_OpenAITextEmbedding3Large、glossary_unembedded这样的集合名——这正是文档“Check Collections”一节中列出集合名的来源。两个集合都会在id字段上创建 KEYWORD 类型的payload index_ensure_payload_index。文档 ID 是字符串而 Qdrant point ID 支持整数适配器用 SHA-256 哈希将其映射到 int64 安全范围内_string_id_to_inthash_value int(hashlib.sha256(doc_id.encode()).hexdigest()[:15], 16) return hash_value % (2**63 - 1)2. 写入与自动同步unembedded 到 embedded 的迁移get_collection/get_or_create_collection打开集合时会调用_load_collection_documents先用document_loader把 unembedded 集合里的旧文档逐一加载、按需迁移为新 schema再触发_index_collection完成向量重建按 payload 中的文档id建立两个集合的映射删除 unembedded 中已不存在的旧向量点只有当checksum变化时才重新调用嵌入模型否则直接复用旧向量避免不必要的模型调用加载失败的文档会被写入独立的failed_migrations集合以便调试源码通过metadata集合中的版本键{collection_name}_version判断是否需要重建索引。这就是文档中“Collection Sync集合在嵌入器或 schema 变化时自动同步大集合首次访问可能较慢”的底层实现。3. 超时重试与远程超时写入操作insert_one/update_one通过_retry_on_timeout_async包裹针对超时类错误做**最多 3 次、指数退避1s、2s、4s**的重试同时所有阻塞式 QdrantClient 调用都通过asyncio.to_thread放到线程池执行避免阻塞事件循环。远程模式下客户端初始化时设置 60 秒超时源码。4. Where 过滤器到 Qdrant Filter 的映射上层 Store 使用类 MongoDB 的Where字典过滤器适配器通过_convert_where_to_qdrant_filter将其翻译为 Qdrant 原生过滤条件Parlant Where 操作符Qdrant Filter 形式$eqFieldCondition(matchMatchValue(...))放入must$nemust_notMatchValue$gt/$gte/$lt/$lteFieldCondition(rangeRange(...))放入must$inFieldCondition(matchMatchAny(...))放入must$ninmust_notMatchAny$and递归转换为多个子条件放入must$or递归转换为多个子条件放入should另外执行find/find_one/update_one/delete_one/ 相似度检索前适配器会递归提取过滤器中出现的所有字段名并确保对应 payload index 存在源码若服务端过滤失败则回退为全量 scroll 内存过滤matches_filters保证正确性源码。5. 相似度检索与距离换算do_find_similar_documents先用嵌入器将查询文本向量化空向量直接返回空结果并打警告再调用query_points执行带过滤的 ANN 检索最后把 Qdrant 的余弦相似度得分换算为距离SimilarDocumentResult(document..., distance1.0 - result.score)读写并发由ReaderWriterLock控制查询取 reader 锁写操作取 writer 锁。元数据如各 Store 的 schema 版本VectorDocumentStoreMigrationHelper.get_store_version_key(...)则统一存放在名为metadata的集合中以单个固定 point__metadata__承载整个键值文档源码。6. Windows 文件锁处理本地模式下__aenter__在 Windowssys.platform win32上会把打开重试次数从 1 提升到 5遇到already accessed错误时按 0.05s 递增退避重试源码__aexit__则显式close()客户端、释放集合引用并在 Windows 上触发gc.collect()加 50ms 等待确保操作系统及时释放文件锁源码。这就是文档中“Windows File Locks使用async with上下文管理器适配器会自动处理文件锁重试”的具体实现。验证 Qdrant 集成是否生效检查集合Qdrant Cloud集合会出现在你的 Qdrant 控制台典型名称包括glossary_OpenAITextEmbedding3Largeglossary_unembeddedcapabilities_OpenAITextEmbedding3Largecanned_responses_OpenAITextEmbedding3Large本地 Qdrant指定路径下会生成包含 Qdrant 数据库文件的文件夹。确认没有回退到内存存储Qdrant 配置正确时parlant-data文件夹中不会出现向量文件Parlant 服务端的默认数据目录即parlant-data见 server.py 中DEFAULT_HOME_DIR的定义向量数据仅存储在 Qdrant本地或云端数据在服务重启后依然存在。测试向量检索与持久化term await agent.create_term( nameTest Term, descriptionThis should be stored in Qdrant, ) # Then chat with agent about test term - it should understand via vector search # Test persistence: close the server and run again # The term should still be available after restart仓库中的 tests/adapters/vector_db/test_qdrant.py 对同一机制做了完整覆盖可作为验证清单参考test_loading_collections验证关闭数据库后重新打开仍能取回文档持久化test_find_similar_documents验证 ANN 检索命中预期文档test_and_operator_with_multiple_conditions、test_or_operator_with_multiple_conditions、test_that_in_filter_works_with_list_of_strings分别验证$and/$or/$in过滤器test_that_documents_are_indexed_when_changing_embedder_type验证切换嵌入器后集合自动重建索引。常见问题与故障排查集成未生效仍在使用内存存储症状Qdrant 控制台中看不到集合parlant-data文件夹里出现了向量数据重启服务后数据丢失。解决确认configure_container中四个向量 Store 均已替换为 VectorStore 实现并注入qdrant_db确认使用AsyncExitStack正确管理QdrantDatabase与各 Store 的生命周期。Windows 文件锁在 Windows 上使用async with上下文管理器适配器会自动处理文件锁重试见上文源码分析。集合同步集合会在嵌入器或 schema 变化时自动同步大集合首次访问触发重建索引可能需要较长时间这是预期行为。更换嵌入器切换嵌入器类型后旧嵌入器命名的集合如glossary_OldEmbedder会一直保留直到手动删除——因为集合名中包含嵌入器类型名新集合会按新名字创建旧数据不会自动清理。性能生产环境建议使用 Qdrant Cloud 或 Qdrant 服务端本地模式不支持 payload 索引。使用本地 Qdrant 时你会看到相关警告这是预期行为可以忽略。其他建议使用嵌入缓存embedding_cache_provider减少对嵌入模型的重复调用使用 Qdrant Cloud/服务端以获得 payload 索引支持考虑拆分过大的集合。连接问题本地确认路径存在且可写远程核对 URL 与 API Key。数据未持久化检查文件路径是否正确、可写远程部署时核对连接配置通过“关闭服务再重启数据仍在”这一行为做最终验证。小结能力说明持久化存储以 Qdrant 替换基于nano_vectordb的默认内存存储重启不丢数据自动同步嵌入器或 schema 变化时自动重建嵌入集合checksum 未变的文档复用旧向量迁移容错加载失败的文档落入failed_migrations集合便于排查Windows 支持文件锁自动重试、显式关闭与 GC 处理过滤器支持$eq/$ne/$gt/$gte/$lt/$lte/$in/$nin/$and/$or全量映射含内存过滤回退依赖与前提pip install parlant[qdrant]qdrant-client1.7.0Python 3.10仓库requires-python 3.10,3.15本地可写目录或 Qdrant Cloud 账户实现与验证材料索引适配器实现 src/parlant/adapters/vector_db/qdrant.py、核心抽象 src/parlant/core/persistence/vector_database.py、默认内存实现 src/parlant/adapters/vector_db/transient.py、依赖声明 pyproject.toml、测试 tests/adapters/vector_db/test_qdrant.py、原始文档 docs/adapters/vector_db/qdrant.md。【免费下载链接】parlantBuild reliable customer-facing AI agents with Parlant: an interaction control harness optimized for controlled, consistent, and predictable LLM interactions.项目地址: https://gitcode.com/GitHub_Trending/pa/parlant创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考