【RAG】【vector_stores097】Timescale Vector Store 演示分析

1. 案例目标本案例演示如何使用Timescale Vector作为LlamaIndex的向量存储后端实现高效的向量相似性搜索和时间过滤功能。主要目标包括展示Timescale Vector与LlamaIndex的集成方法演示基础向量相似性搜索功能实现基于时间范围的向量过滤查询创建和管理不同类型的向量索引DiskANN、HNSW、IVFFLAT将Timescale Vector作为检索器和查询引擎使用Timescale Vector简介Timescale Vector是TimescaleDB的扩展专为AI应用优化的PostgreSQL向量数据库。它提供了增强的pgvector功能包括DiskANN索引算法、时间分区索引等特性特别适合处理带时间戳的向量数据。2. 技术栈与核心依赖LlamaIndex用于构建向量索引和查询引擎的核心框架Timescale Vector基于PostgreSQL的向量数据库扩展提供高效的向量存储和检索功能OpenAI Embeddings用于生成文本向量嵌入pandas用于数据处理和分析python-dotenv用于环境变量管理核心依赖安装pip install llama-index-embeddings-openai pip install llama-index-vector-stores-timescalevector3. 环境配置3.1 OpenAI API配置需要设置OpenAI API密钥可以通过环境变量或直接配置import os os.environ[OPENAI_API_KEY] sk-...3.2 Timescale数据库连接配置Timescale数据库服务URLimport os from dotenv import load_dotenv load_dotenv() TIMESCALE_SERVICE_URL os.environ.get(TIMESCALE_SERVICE_URL)获取Timescale服务URL可以通过Timescale控制台或环境变量获取服务URL格式通常为postgresql://username:passwordhost:port/database4. 案例实现4.1 数据准备首先下载Paul Graham的文章集作为示例数据!mkdir -p data/paul_graham/ !wget https://raw.githubusercontent.com/run-llama/llama_index/main/docs/examples/data/paul_graham/paul_graham_essay.txt -O data/paul_graham/paul_graham_essay.txt4.2 向量存储初始化创建TimescaleVectorStore实例并配置基本参数from llama_index.vector_stores.timescalevector import TimescaleVectorStore from llama_index.core import SimpleDirectoryReader, StorageContext, VectorStoreIndex # 加载文档 documents SimpleDirectoryReader(./data/paul_graham).load_data() # 初始化向量存储 ts_vector_store TimescaleVectorStore.from_params( service_urlTIMESCALE_SERVICE_URL, table_namepaul_graham_essay, ) # 创建存储上下文 storage_context StorageContext.from_defaults(vector_storets_vector_store) # 创建向量索引 index VectorStoreIndex.from_documents( documents, storage_contextstorage_context )4.3 基础向量相似性搜索实现基本的向量相似性搜索功能from llama_index.core.vector_stores import VectorStoreQuery # 创建查询引擎 query_engine index.as_query_engine() # 执行查询 response query_engine.query(Did the author work at YC?) print(str(response))4.4 索引管理Timescale Vector支持多种索引类型可以创建和管理不同的索引# 创建DiskANN索引默认 ts_vector_store.create_index() # 创建HNSW索引 ts_vector_store.create_index(hnsw, max_alpha1.0, num_neighbors30) # 创建IVFFLAT索引 ts_vector_store.create_index( ivfflat, num_lists100, m100 )4.5 时间过滤向量搜索Timescale Vector的一个强大功能是基于时间范围的向量过滤import datetime import uuid import pandas as pd # 加载Git日志数据 df pd.read_csv(commit_history.csv) df df.head(1000) # 取前1000条记录 # 创建辅助函数 def create_uuid(date_string): date_obj datetime.datetime.strptime(date_string, %Y-%m-%d %H:%M:%S%z) return uuid.uuid1(node0x0123456789ABCDEF, clock_seq_hi0x80, clock_seq_lo0x00, timeint(date_obj.timestamp() * 1e9)) def split_name(name): parts name.split() if len(parts) 1: return parts[0], parts[-1] return name, def create_date(date_string): date_obj datetime.datetime.strptime(date_string, %Y-%m-%d %H:%M:%S%z) return date_obj # 创建文本节点 nodes [] for _, row in df.iterrows(): first_name, last_name split_name(row[author]) date create_date(row[date]) uuid_obj create_uuid(row[date]) node TextNode( textrow[message], metadata{ commit: row[commit], author: row[author], date: date.strftime(%Y-%m-%d %H:%M:%S%z) }, id_str(uuid_obj) ) nodes.append(node) # 生成嵌入向量 from llama_index.embeddings.openai import OpenAIEmbedding embedding_model OpenAIEmbedding(modeltext-embedding-ada-002) for node in nodes: node_embedding embedding_model.get_text_embedding( node.get_content(metadata_modeall) ) node.embedding node_embedding # 添加节点到向量存储 ts_vector_store.add(nodes)4.6 时间范围过滤查询实现三种不同的时间范围过滤查询方法from datetime import datetime, timedelta # 方法1从开始日期和时间增量查询 start_dt datetime(2023, 8, 1) td timedelta(days7) query_embedding embedding_model.get_text_embedding( What changes were made to TimescaleDB functions? ) vector_store_query VectorStoreQuery( query_embeddingquery_embedding, similarity_top_k5 ) # 从开始日期和时间增量查询 query_result ts_vector_store.query( vector_store_query, start_datestart_dt, time_deltatd ) # 方法2从结束日期和时间增量查询 end_dt datetime(2023, 8, 31) query_result ts_vector_store.query( vector_store_query, end_dateend_dt, time_deltatd ) # 方法3在结束日期和时间增量之间查询 query_result ts_vector_store.query( vector_store_query, end_dateend_dt, time_deltatd )4.7 作为检索器和查询引擎使用将Timescale Vector作为LlamaIndex的检索器和查询引擎使用from llama_index.core import VectorStoreIndex # 创建索引 index VectorStoreIndex.from_vector_store(ts_vector_store) # 作为检索器使用 retriever index.as_retriever( vector_store_kwargs({start_date: start_dt, time_delta: td}) ) results retriever.retrieve(Whats new with TimescaleDB functions?) # 作为查询引擎使用 query_engine index.as_query_engine( vector_store_kwargs({start_date: start_dt, end_date: end_dt}) ) response query_engine.query( Whats new with TimescaleDB functions? When were these changes made and by whom? ) print(str(response))5. 案例效果高效向量存储成功将文档转换为向量并存储在Timescale Vector中精确相似性搜索能够根据查询向量返回最相似的文档内容灵活时间过滤支持多种时间范围过滤方式包括开始日期时间增量、结束日期时间增量等多种索引支持支持DiskANN、HNSW和IVFFLAT等不同类型的向量索引无缝集成与LlamaIndex框架无缝集成可作为检索器和查询引擎使用时间过滤优势Timescale Vector的时间过滤功能特别适合处理具有时间属性的向量数据如日志记录、新闻文章、社交媒体帖子等。通过时间分区查询效率显著提高只需搜索相关分区而非整个数据集。6. 案例实现思路本案例的实现思路如下环境准备安装必要的依赖库配置OpenAI API和Timescale数据库连接数据加载使用SimpleDirectoryReader加载文档数据或使用pandas加载结构化数据向量存储初始化创建TimescaleVectorStore实例配置表名和连接参数向量索引构建将文档转换为向量并存储在Timescale Vector中创建适当的索引查询实现实现基础向量查询和时间过滤查询功能高级功能将向量存储作为检索器和查询引擎集成到LlamaIndex工作流中时间过滤实现原理Timescale Vector通过将向量数据按时间分区存储实现高效的时间范围过滤。每个分区包含特定时间范围内的向量查询时只需搜索相关分区大幅提高查询效率。7. 扩展建议多模态数据支持扩展支持图像、音频等多模态数据的向量存储和检索自定义索引参数根据数据特性调整索引参数如DiskANN的max_alpha、HNSW的m和ef_construction等批量操作优化实现更高效的批量数据插入和更新操作分布式部署探索Timescale Vector在分布式环境下的部署和扩展方案实时数据流处理集成实时数据流处理框架实现动态向量更新和查询高级过滤功能结合元数据和时间过滤实现更复杂的查询条件组合性能监控添加查询性能监控和分析功能优化索引策略8. 总结本案例详细展示了如何使用Timescale Vector作为LlamaIndex的向量存储后端实现了高效的向量相似性搜索和时间过滤功能。Timescale Vector基于PostgreSQL构建提供了增强的向量存储和检索能力特别适合处理具有时间属性的向量数据。通过本案例我们学习了Timescale Vector的基本概念和特性如何初始化和配置TimescaleVectorStore基础向量相似性搜索的实现方法时间范围过滤查询的多种实现方式不同类型向量索引的创建和管理如何将Timescale Vector集成到LlamaIndex工作流中Timescale Vector的时间过滤功能是其独特优势特别适合处理日志记录、新闻文章、社交媒体帖子等具有时间属性的数据。通过时间分区查询效率显著提高只需搜索相关分区而非整个数据集。该案例为构建基于时间感知的向量搜索应用提供了完整的解决方案可作为开发类似应用的基础参考。