摘要随着大语言模型LLM应用的快速发展向量数据库作为AI时代的关键基础设施正在成为RAG检索增强生成、语义搜索、智能推荐等场景的核心组件。本文将从向量嵌入的原理出发深入讲解向量相似度搜索与近似最近邻ANN算法横向对比Chroma、FAISS、Milvus、Pinecone、Qdrant、Weaviate等主流向量数据库的架构特点与适用场景并配以完整的Python代码示例帮助开发者快速掌握向量数据库的实战技能。关键词向量数据库、向量嵌入、ANN算法、相似度搜索、RAG、Chroma、FAISS、Milvus一、向量数据库基础1.1 什么是向量嵌入向量嵌入Vector Embedding是将高维稀疏数据如文本、图像、音频映射为低维稠密向量的技术手段。这些向量本质上是一串浮点数构成的数组每个维度编码了原始数据在语义空间中的某种特征。以文本为例苹果和香蕉在向量空间中距离较近因为它们同属水果而苹果和手机虽然字面有重叠但在语义空间中距离较远。通过这种映射方式AI系统能够理解数据之间的语义关系。# 使用 OpenAI 的 text-embedding-3-small 模型生成文本向量 from openai import OpenAI client OpenAI() def get_embedding(text: str) - list[float]: 生成文本的向量嵌入表示 response client.embeddings.create( modeltext-embedding-3-small, inputtext ) # 返回向量数组 return response.data[0].embedding # 示例生成三个水果名称的向量 texts [苹果, 香蕉, 手机] embeddings [get_embedding(t) for t in texts] print(f向量维度: {len(embeddings[0])}) print(f苹果向量前5维: {embeddings[0][:5]})1.2 向量相似度搜索原理向量相似度搜索的核心目标是从海量向量中快速找到与查询向量最相似的Top-K个结果。相似度的衡量通常有以下几种度量方式度量方式公式适用场景余弦相似度cos(θ) (A·B) / (|A||B|)文本语义相似度欧氏距离L2 √Σ(aᵢ - bᵢ)²图像特征、推荐系统点积A·B Σaᵢbᵢ排序打分、推荐系统import numpy as np def cosine_similarity(vec_a: list[float], vec_b: list[float]) - float: 计算两个向量的余弦相似度 a np.array(vec_a) b np.array(vec_b) # 余弦相似度 点积 / (模长乘积) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def euclidean_distance(vec_a: list[float], vec_b: list[float]) - float: 计算两个向量的欧氏距离 a np.array(vec_a) b np.array(vec_b) return float(np.linalg.norm(a - b)) # 示例 vec1 get_embedding(人工智能) vec2 get_embedding(机器学习) vec3 get_embedding(水果) print(f人工智能 vs 机器学习 余弦相似度: {cosine_similarity(vec1, vec2):.4f}) print(f人工智能 vs 水果 余弦相似度: {cosine_similarity(vec1, vec3):.4f})1.3 近似最近邻ANN算法精确的最近邻搜索KNN在数据量达到百万、千万级别时时间复杂度 O(N) 成为性能瓶颈。近似最近邻ANN算法通过牺牲少量精度换取数量级的性能提升是向量数据库的底层核心技术。主流ANN算法包括IVF倒排文件索引将向量空间聚类成多个桶查询时只搜索最近的几个桶HNSW分层导航小世界图构建多层图结构从上往下逐步逼近目标PQ乘积量化将高维向量压缩成短编码降低内存占用ANNOY随机投影树使用多棵随机投影树划分子空间二、主流向量数据库横向对比数据库开发语言部署方式索引算法适用场景特点ChromaPython本地/客户端HNSW开发测试轻量级、上手快FAISSC/Python本地IVF/PQ/HNSW研究、离线分析Facebook开源、性能强MilvusGo分布式集群HNSW/IVF/PQ生产环境分布式、云原生Pinecone-云服务HNSW云原生应用免运维、SaaSQdrantRust本地/容器HNSW高性能需求内存安全、高并发WeaviateGo分布式HNSW/BM25混合搜索图结构向量混合2.1 Chroma — 轻量级开发首选Chroma是目前最流行的本地向量数据库专为LLM应用原型开发设计。它将数据存储在本地SQLite文件中零配置即可使用非常适合快速验证RAG流程。# 安装pip install chromadb import chromadb # 初始化Chroma客户端本地持久化存储 client chromadb.PersistentClient(path./chroma_db) # 创建集合类似于表 collection client.create_collection( namearticles, metadata{description: 技术文章向量库} ) # 添加向量数据 collection.add( documents[ 大语言模型LLM是当前AI领域的核心技术, 向量数据库用于存储和检索高维向量数据, RAG技术结合了检索和生成两种能力 ], ids[doc1, doc2, doc3], metadatas[ {category: AI, author: 张三}, {category: 数据库, author: 李四}, {category: AI, author: 王五} ] ) # 相似度查询 results collection.query( query_texts[什么是大语言模型], n_results2 # 返回最相似的2条 ) print(查询结果:) for i, (doc, distance) in enumerate(zip(results[documents][0], results[distances][0])): print(f [{i1}] 距离{distance:.4f} | {doc}) # 删除文档 collection.delete(ids[doc3])2.2 FAISS — Facebook开源本地库FAISSFacebook AI Similarity Search是Facebook开源的高效相似度搜索库专注于在CPU/GPU上实现高速向量检索。它支持批量化索引构建和搜索是生产环境做离线分析或嵌入式部署的首选。# 安装pip install faiss-cpu或 faiss-gpu import numpy as np import faiss # 1. 准备数据生成10000条128维的随机向量模拟嵌入向量 dimension 128 # 向量维度 num_vectors 10000 # 向量数量 np.random.seed(42) # 生成随机向量并转为float32 vectors np.random.rand(num_vectors, dimension).astype(float32) # 2. 使用IVF索引加速查询 nlist 100 # 聚类中心数量 # 训练使用IVF索引需要先对数据进行聚类 quantizer faiss.IndexFlatIP(dimension) # IP 内积余弦相似度需先归一化 index faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT) index.train(vectors) # 训练索引 index.add(vectors) # 添加向量 # 设置搜索范围搜索的聚类中心数越大越精确但越慢 index.nprobe 10 # 3. 相似度搜索找最相似的5个 query_vector np.random.rand(1, dimension).astype(float32) k 5 # Top-K # 执行搜索 distances, indices index.search(query_vector, k) print(f查询向量维度: {query_vector.shape}) print(f找到的{k}个最近邻:) for i, (idx, dist) in enumerate(zip(indices[0], distances[0])): print(f 第{i1}个: 索引{idx}, 距离{dist:.4f}) # 4. 使用HNSW索引更精确但内存占用更大 print(\n--- 使用HNSW索引 ---) hnsw_index faiss.IndexHNSWFlat(dimension, 32) # 32为每层连接数 hnsw_index.hnsw.efConstruction 40 hnsw_index.add(vectors) hnsw_index.hnsw.efSearch 64 # 搜索时动态列表大小 distances_hnsw, indices_hnsw hnsw_index.search(query_vector, k) print(fHNSW搜索结果 (Top-{k}):) for i, (idx, dist) in enumerate(zip(indices_hnsw[0], distances_hnsw[0])): print(f 第{i1}个: 索引{idx}, 距离{dist:.4f})2.3 Milvus — 分布式生产级方案Milvus是面向生产环境设计的分布式向量数据库支持PB级数据存储提供丰富的索引类型和水平扩展能力适合企业级AI应用。# 安装pip install pymilvus from pymilvus import MilvusClient, DataType # 连接本地Milvus服务或远程地址 client MilvusClient(uri./milvus_demo.db) # 创建集合schema定义 schema MilvusClient.create_schema( auto_idTrue, enable_dynamic_fieldTrue, descriptionRAG知识库集合 ) # 添加字段 schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_namecontent, datatypeDataType.VARCHAR, max_length512) schema.add_field(field_nameembedding, datatypeDataType.FLOAT_VECTOR, dim1536) schema.add_field(field_namecategory, datatypeDataType.VARCHAR, max_length64) # 创建索引参数 index_params client.prepare_index_params() index_params.add_index( field_nameembedding, index_typeHNSW, # HNSW索引 metric_typeCOSINE, # 余弦相似度 params{M: 16, efConstruction: 200} ) # 创建集合 if client.has_collection(knowledge_base): client.drop_collection(knowledge_base) client.create_collection( collection_nameknowledge_base, schemaschema, index_paramsindex_params ) # 插入数据 documents [ Transformer是现代NLP的基石架构, 向量数据库解决大模型记忆问题, LangChain简化LLM应用开发流程 ] # 模拟嵌入向量实际使用时替换为真实embedding embeddings [np.random.rand(1536).tolist() for _ in documents] data [ {content: doc, embedding: emb, category: AI} for doc, emb in zip(documents, embeddings) ] result client.insert(collection_nameknowledge_base, datadata) print(f插入成功共 {result.insert_count} 条记录) # 搜索 query_embedding np.random.rand(1536).tolist() search_params {ef: 128} results client.search( collection_nameknowledge_base, data[query_embedding], limit3, search_paramssearch_params, output_fields[content, category] ) print(\n搜索结果:) for hits in results: for hit in hits: print(f 内容: {hit[entity][content]} | 距离: {hit[distance]:.4f}) # 删除集合 client.drop_collection(knowledge_base) client.close()2.4 Pinecone — 云原生SaaSPinecone是完全托管的云向量数据库用户无需关心基础设施运维支持快速弹性扩缩容适合快速上线的产品级应用。# 安装pip install pinecone-client from pinecone import Pinecone # 初始化需要API Key可在 pinecone.io 免费注册 pc Pinecone(api_keyyour-api-key) # 连接到索引 index pc.Index(llm-knowledge-base) # 插入向量Upsert vectors [ (vec1, [0.1] * 1536, {text: RAG是检索增强生成技术, source: AI论文}), (vec2, [0.2] * 1536, {text: 向量数据库支持高维向量检索, source: 技术文档}), (vec3, [0.3] * 1536, {text: 大模型需要外部知识库补充, source: 博客}), ] index.upsert(vectorsvectors) print(向量插入成功) # 相似度查询 query_result index.query( vector[0.15] * 1536, top_k2, include_metadataTrue ) print(\nTop-2 相似结果:) for match in query_result.matches: print(f ID: {match.id} | 分数: {match.score:.4f} | 文本: {match.metadata[text]}) # 删除向量 index.delete(ids[vec3])2.5 Qdrant — Rust实现高性能Qdrant由Rust语言实现以内存安全和高并发著称。其滤波搜索能力强大支持带条件过滤的向量查询在同等硬件条件下性能表现优异。# 安装pip install qdrant-client from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue # 连接Qdrant服务 client QdrantClient(urlhttp://localhost:6333) # 创建集合 collection_name products vector_size 128 if not client.collection_exists(collection_name): client.create_collection( collection_namecollection_name, vectors_configVectorParams(sizevector_size, distanceDistance.COSINE) ) # 批量插入向量 points [ PointStruct( id1, vectorlist(np.random.rand(vector_size)), payload{name: iPhone 15, category: 手机, price: 6999} ), PointStruct( id2, vectorlist(np.random.rand(vector_size)), payload{name: MacBook Pro, category: 电脑, price: 14999} ), PointStruct( id3, vectorlist(np.random.rand(vector_size)), payload{name: AirPods Pro, category: 耳机, price: 1899} ), ] client.upsert(collection_namecollection_name, pointspoints) print(插入成功) # 带过滤条件的搜索只查询手机类别 filter_condition Filter( must[FieldCondition(keycategory, matchMatchValue(value手机))] ) search_results client.search( collection_namecollection_name, query_vectorlist(np.random.rand(vector_size)), query_filterfilter_condition, limit1 ) print(\n过滤搜索结果category手机:) for result in search_results: print(f {result.payload[name]} | 分数: {result.score:.4f}) # 删除点 client.delete(collection_namecollection_name, points[1, 2, 3])2.6 Weaviate — 图结构向量混合Weaviate是原生支持GraphQL风格的向量数据库能够将向量搜索与传统BM25关键词搜索结合实现语义关键词的混合检索特别适合需要同时利用结构化数据过滤的场景。# 安装pip install weaviate-client import weaviate # 连接本地Weaviate服务 client weaviate.Client(http://localhost:8080) # 创建类Collection schema { class: Article, description: 技术文章, vectorizer: text2vec-transformers, # 使用本地transformer模型生成向量 moduleConfig: { text2vec-transformers: { vectorizeClassName: False # 类名不参与向量化 } }, properties: [ {name: title, dataType: [text]}, {name: content, dataType: [text]}, {name: category, dataType: [text]} ] } if client.schema.exists(Article): client.schema.delete_class(Article) client.schema.create_class(schema) print(Article 类创建成功) # 添加对象Weaviate自动生成向量 client.data_object.create( class_nameArticle, data_object{ title: 向量数据库详解, content: 本文介绍向量数据库的核心概念和使用方法, category: 数据库 } ) # 语义搜索 response client.query.get( Article, [title, content, category] ).with_near_text({ concepts: [AI检索技术] }).with_limit(2).do() print(\n语义搜索结果:) for obj in response[data][Get][Article]: print(f 标题: {obj[title]} | 类别: {obj[category]}) # 混合搜索向量 BM25关键词 hybrid_response client.query.get( Article, [title, content] ).with_hybrid( query数据库, alpha0.5 # 0.5表示向量搜索和关键词搜索各占50% ).with_limit(2).do() print(\n混合搜索结果:) for obj in hybrid_response[data][Get][Article]: print(f 标题: {obj[title]})三、索引算法详解3.1 IVF倒排文件索引IVFInverted File Index的核心思想是聚类倒排。先将所有向量通过K-Means聚类成N个桶Voroni单元查询时先找到查询向量所属的桶再在该桶内做精确搜索。这将搜索范围从N缩小到N/nlist。import faiss import numpy as np # 生成100000条256维向量 dimension 256 n_vectors 100_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # 构建IVF索引 nlist 1024 # 聚类中心数量 quantizer faiss.IndexFlatL2(dimension) index faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_L2) index.train(vectors) index.add(vectors) # 搜索性能对比 import time query np.random.rand(1, dimension).astype(float32) # nprobe1只搜索1个桶 index.nprobe 1 start time.time() _, _ index.search(query, 10) t1 time.time() - start # nprobe32搜索32个桶 index.nprobe 32 start time.time() _, _ index.search(query, 10) t2 time.time() - start print(fIVF nprobe1: {t1*1000:.2f}ms) print(fIVF nprobe32: {t2*1000:.2f}ms) print(f精度提升约 {(t2/t1):.1f}x 时间换取更高召回率)3.2 HNSW分层导航小世界图HNSWHierarchical Navigable Small World是一种基于图的近似最近邻算法。它构建一个多层图结构上层连接稀疏快速定位下层连接稠密精确结果。查询时从最上层开始通过贪心遍历逐步向下逼近。import faiss import numpy as np dimension 128 n_vectors 50_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # HNSW参数说明 # M: 每个节点的最大连接数越大越精确内存越高 # efConstruction: 构建时动态列表大小越大构建越慢索引质量越高 hnsw_index faiss.IndexHNSWFlat(dimension, 32) hnsw_index.hnsw.efConstruction 40 hnsw_index.add(vectors) # 查询时 efSearch 参数值越大搜索越精确但越慢 for ef in [16, 64, 256]: hnsw_index.hnsw.efSearch ef _, indices hnsw_index.search(np.random.rand(1, dimension).astype(float32), 10) print(fefSearch{ef}: Top-10结果索引前3个{indices[0][:3]})3.3 PQ乘积量化PQProduct Quantization将高维向量分割成多个子空间对每个子空间分别做K-Means量化从而将原始向量压缩成短编码。这使得在有限内存中存储数十亿向量成为可能。import faiss import numpy as np dimension 64 n_vectors 20_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # PQ参数m为子空间数量nbits为每个子空间的比特数 # 原始向量 64 dim * 4 bytes 256 bytes # 压缩后: m * nbits bits m 8 # 子空间数量 nbits 8 # 每个子空间用8bit编码256个聚类中心 pq_index faiss.IndexPQ(dimension, m, nbits) pq_index.train(vectors) pq_index.add(vectors) # 注意PQ索引的搜索结果是有损的 query np.random.rand(1, dimension).astype(float32) distances, indices pq_index.search(query, 5) print(fPQ压缩后每个向量仅需 {m * nbits / 8:.1f} bytes) print(f压缩比: {256 / (m * nbits / 8):.1f}x)3.4 ANNOY随机投影树ANNOYApproximate Nearest Neighbors Oh Yeah使用多棵随机投影二叉树来划分子空间查询时并行搜索所有树取最优结果。Google开发主要用于音乐推荐Audioscrobbler数据集。# 安装pip install annoy from annoy import AnnoyIndex import random dimension 40 n_vectors 10000 n_trees 50 # 构建索引 annoy_index AnnoyIndex(dimension, angular) # angular等价于余弦距离 for i in range(n_vectors): vec [random.gauss(0, 1) for _ in range(dimension)] annoy_index.add_item(i, vec) # 构建50棵树越多越精确越慢 annoy_index.build(n_trees) # 查询最近邻 query [random.gauss(0, 1) for _ in range(dimension)] results annoy_index.get_nns_by_vector(query, 5, search_k-1) print(fANNOY查询结果 (Top-5): {results}) # 指定搜索预算越大越精确 results_exact annoy_index.get_nns_by_vector(query, 5, search_k500) print(fANNOY精确查询 (search_k500): {results_exact})四、向量数据库核心操作4.1 插入向量import chromadb client chromadb.PersistentClient(path./demo_db) collection client.get_or_create_collection(demo) # 单条插入 collection.add( idssingle_doc, documents[这是一段技术文档], metadatas[{author: 工程师}] ) # 批量插入推荐性能更好 batch_data { ids: [fdoc_{i} for i in range(1000)], documents: [f文档内容 {i} for i in range(1000)], metadatas: [{index: i} for i in range(1000)] } collection.add(**batch_data) print(f批量插入完成共 {collection.count()} 条)4.2 相似度查询# 基于文本的语义搜索 results collection.query( query_texts[查找AI相关的技术文档], n_results5, where{author: 工程师}, # 元数据过滤条件 include[documents, distances, metadatas] # 指定返回字段 ) # 基于向量的搜索 import numpy as np query_vector np.random.rand(1536).tolist() results collection.query( query_embeddings[query_vector], n_results3 )4.3 范围查询Range Search范围查询返回与查询向量距离在指定阈值内的所有向量适用于附近类应用。from qdrant_client import QdrantClient client QdrantClient(urlhttp://localhost:6333) # 在Qdrant中执行范围查询搜索距离阈值内的所有向量 search_results client.search( collection_namedemo_collection, query_vectorlist(np.random.rand(128)), score_threshold0.7, # 相似度分数阈值0~1 limit100 # 最大返回数量 ) print(f范围内找到 {len(search_results)} 条记录)4.4 删除与更新# Chroma - 删除 collection.delete(ids[doc_1, doc_2]) # 删除指定ID collection.delete(where{category: 过时}) # 按条件删除 # Chroma - 更新先删后加 collection.update( idsdoc_1, documents[更新后的内容], metadatas[{status: updated}] ) # Milvus - 删除 client.delete( collection_nameknowledge_base, filterid in [1, 2, 3] ) # Milvus - 更新通过upsert实现 client.upsert( collection_nameknowledge_base, data[{id: 1, content: 更新内容, embedding: [0.1]*1536}] )五、使用场景5.1 RAG知识库RAGRetrieval-Augmented Generation是向量数据库最经典的应用场景。通过将企业知识库文档预先向量化并存储LLM在回答用户问题时先检索最相关的文档片段再结合上下文生成答案有效缓解大模型的幻觉问题。# 完整的RAG流程示例 import chromadb from openai import OpenAI client_chroma chromadb.PersistentClient(path./rag_db) collection client_chroma.get_or_create_collection(rag_knowledge_base) openai_client OpenAI() # 第一步文档预处理与入库 def process_document(text: str, doc_id: str): 将文档文本向量化并存储 embedding openai_client.embeddings.create( modeltext-embedding-3-small, inputtext ).data[0].embedding collection.add( idsdoc_id, documents[text], embeddings[embedding] ) # 第二步RAG检索 def rag_query(user_question: str, top_k: int 3) - str: 完整的RAG查询流程 1. 将用户问题向量化 2. 检索最相关的文档片段 3. 构建Prompt并调用LLM生成答案 # 1. 检索相关文档 query_embedding openai_client.embeddings.create( modeltext-embedding-3-small, inputuser_question ).data[0].embedding results collection.query( query_embeddings[query_embedding], n_resultstop_k ) # 2. 组合上下文 context \n\n.join(results[documents][0]) if results[documents] else 无相关背景信息 # 3. 构建PromptFew-shot prompt f基于以下背景信息回答用户问题。如果背景信息不足以回答请如实说明。 背景信息 {context} 用户问题{user_question} 回答 # 4. 调用LLM生成 response openai_client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.3 ) return response.choices[0].message.content # 示例查询 answer rag_query(向量数据库有哪些应用场景) print(fRAG回答{answer})5.2 以图搜图将图片通过视觉模型如CLIP、ResNet提取特征向量存入向量数据库实现基于内容的图像检索。import numpy as np import chromadb client chromadb.PersistentClient(path./image_search_db) collection client.get_or_create_collection(image_features) # 模拟图片向量实际使用CLIP等模型提取 def simulate_image_embedding(image_path: str) - list[float]: 模拟图片向量提取实际应使用clip.encode_image() np.random.seed(hash(image_path) % 2**32) return np.random.rand(512).tolist() # 添加图片到索引 images [ {id: img_001, path: /photos/cat.jpg, desc: 可爱的猫咪}, {id: img_002, path: /photos/dog.jpg, desc: 活泼的狗狗}, {id: img_003, path: /photos/car.jpg, desc: 红色跑车}, ] for img in images: emb simulate_image_embedding(img[path]) collection.add( idsimg[id], documents[img[desc]], embeddings[emb] ) # 搜索可爱的动物对应的图片 query_embedding simulate_image_embedding(cute animal) results collection.query( query_embeddings[query_embedding], n_results2 ) print(以图搜图结果:) for doc, dist in zip(results[documents][0], results[distances][0]): print(f {doc} | 距离: {dist:.4f})5.3 语义搜索超越关键词匹配在语义层面理解用户查询意图返回真正相关的结果。适用于搜索引擎、企业知识管理、内容推荐等场景。# 语义搜索示例使用Chroma import chromadb client chromadb.PersistentClient(path./semantic_search_db) collection client.get_or_create_collection(news_articles) # 批量添加新闻数据 articles [ 美联储宣布降息以应对经济下行压力, 人工智能芯片需求爆发式增长, 某地发生5.0级地震暂无人员伤亡, 新能源车销量突破历史新高, 科学家发现新的量子计算算法 ] collection.add( ids[fnews_{i} for i in range(len(articles))], documentsarticles ) # 语义查询金融政策应该匹配到美联储降息的新闻 query 货币政策和经济调控 results collection.query( query_texts[query], n_results2 ) print(f语义查询: {query}) print(返回结果:) for i, (doc, dist) in enumerate(zip(results[documents][0], results[distances][0])): print(f {i1}. {doc})5.4 推荐系统基于用户行为和物品特征构建向量表示通过向量相似度计算实现个性化推荐。import numpy as np import chromadb client chromadb.PersistentClient(path./recommendation_db) user_collection client.get_or_create_collection(user_profiles) item_collection client.get_or_create_collection(items) # 模拟用户兴趣向量和商品特征向量 def cosine_sim(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) # 用户画像向量由历史行为生成 user_vector np.random.rand(128).tolist() # 商品向量 item_vectors { item_101: np.random.rand(128).tolist(), item_102: np.random.rand(128).tolist(), item_103: np.random.rand(128).tolist(), item_104: np.random.rand(128).tolist(), } # 计算相似度并排序推荐 similarities [ (item_id, cosine_sim(user_vector, vec)) for item_id, vec in item_vectors.items() ] ranked sorted(similarities, keylambda x: x[1], reverseTrue) print(个性化推荐结果Top-3:) for item_id, score in ranked[:3]: print(f {item_id}: 相似度{score:.4f})六、综合代码示例构建本地RAG系统以下代码整合了本文的核心内容展示如何使用Chroma构建一个完整的本地RAG知识库问答系统 完整的本地RAG系统示例 功能从知识库检索相关片段 LLM生成回答 依赖chromadb, openai import chromadb from openai import OpenAI from typing import Optional class LocalRAGSystem: 本地RAG知识库系统 def __init__(self, db_path: str, openai_api_key: str): # 初始化Chroma持久化存储 self.client chromadb.PersistentClient(pathdb_path) self.collection self.client.get_or_create_collection( nameknowledge_base, metadata{description: 本地RAG知识库} ) # 初始化OpenAI客户端 self.llm OpenAI(api_keyopenai_api_key) self.embedding_model text-embedding-3-small def add_documents(self, documents: list[str], ids: Optional[list[str]] None): 批量添加文档到知识库 if ids is None: ids [fdoc_{i} for i in range(len(documents))] # 批量生成嵌入向量 embeddings [] batch_size 100 for i in range(0, len(documents), batch_size): batch documents[i:ibatch_size] response self.llm.embeddings.create( modelself.embedding_model, inputbatch ) embeddings.extend([r.embedding for r in response.data]) self.collection.add( idsids, documentsdocuments, embeddingsembeddings ) print(f✅ 成功添加 {len(documents)} 篇文档当前共 {self.collection.count()} 篇) def retrieve(self, query: str, top_k: int 3) - list[dict]: 从知识库检索最相关的文档 # 生成查询向量 response self.llm.embeddings.create( modelself.embedding_model, inputquery ) query_embedding response.data[0].embedding # 检索 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, distances, metadatas] ) # 格式化返回 retrieved [] for i in range(len(results[ids][0])): retrieved.append({ id: results[ids][0][i], document: results[documents][0][i], distance: results[distances][0][i], metadata: results[metadatas][0][i] }) return retrieved def generate(self, query: str, context: list[dict]) - str: 使用LLM结合上下文生成回答 # 组合上下文 context_text \n.join([ f[{i1}] {ctx[document]} for i, ctx in enumerate(context) ]) prompt f你是一个专业的问答助手。请基于以下参考信息回答用户问题。 如果参考信息不足以回答请如实说明不要编造答案。 参考信息 {context_text} 用户问题{query} 回答 response self.llm.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.3, max_tokens500 ) return response.choices[0].message.content def query(self, question: str, top_k: int 3) - dict: 完整的RAG查询检索 生成 # 1. 检索 retrieved self.retrieve(question, top_k) # 2. 生成 answer self.generate(question, retrieved) return { question: question, answer: answer, references: [ctx[document] for ctx in retrieved] } # 使用示例 if __name__ __main__: # 注意实际使用时替换为真实API Key # rag LocalRAGSystem(./my_rag_db, api_keysk-...) # 添加知识库文档 docs [ Transformer架构由Google在2017年提出是现代NLP的核心模型, BERT是基于Transformer的双向编码器表示在多项NLP任务上取得最优成绩, GPT系列是OpenAI开发的大型语言模型采用自回归生成方式, 向量数据库通过近似最近邻算法实现高速相似度检索, RAG检索增强生成结合了检索系统和生成模型的优势 ] # rag.add_documents(docs) # 结果 rag.query(什么是RAG技术) # print(f问题: {结果[question]}\n回答: {结果[answer]})七、总结与选型建议开发测试阶段推荐使用Chroma零配置、纯Python、SQLite后端5分钟即可跑通RAG流程。离线分析/研究阶段推荐FAISSFacebook出品性能极强支持GPU加速适合处理海量数据。生产环境根据规模选择中小规模百万级QdrantRust高性能或Milvus功能丰富大规模/云原生Milvus分布式集群或PineconeSaaS免运维需要混合检索Weaviate向量关键词融合选型核心考量因素数据规模本地/Chroma适合万级FAISS可达亿级Milvus/Pinecone支持十亿级以上查询延迟Qdrant和FAISS在同硬件下最快Pinecone延迟取决于云端配置运维成本Pinecone最低完全托管Chroma次之自建Milvus需要Kubernetes经验功能需求需要过滤查询优先Qdrant需要混合搜索优先Weaviate需要强一致性选Milvus
LLM应用开发之向量数据库详解
摘要随着大语言模型LLM应用的快速发展向量数据库作为AI时代的关键基础设施正在成为RAG检索增强生成、语义搜索、智能推荐等场景的核心组件。本文将从向量嵌入的原理出发深入讲解向量相似度搜索与近似最近邻ANN算法横向对比Chroma、FAISS、Milvus、Pinecone、Qdrant、Weaviate等主流向量数据库的架构特点与适用场景并配以完整的Python代码示例帮助开发者快速掌握向量数据库的实战技能。关键词向量数据库、向量嵌入、ANN算法、相似度搜索、RAG、Chroma、FAISS、Milvus一、向量数据库基础1.1 什么是向量嵌入向量嵌入Vector Embedding是将高维稀疏数据如文本、图像、音频映射为低维稠密向量的技术手段。这些向量本质上是一串浮点数构成的数组每个维度编码了原始数据在语义空间中的某种特征。以文本为例苹果和香蕉在向量空间中距离较近因为它们同属水果而苹果和手机虽然字面有重叠但在语义空间中距离较远。通过这种映射方式AI系统能够理解数据之间的语义关系。# 使用 OpenAI 的 text-embedding-3-small 模型生成文本向量 from openai import OpenAI client OpenAI() def get_embedding(text: str) - list[float]: 生成文本的向量嵌入表示 response client.embeddings.create( modeltext-embedding-3-small, inputtext ) # 返回向量数组 return response.data[0].embedding # 示例生成三个水果名称的向量 texts [苹果, 香蕉, 手机] embeddings [get_embedding(t) for t in texts] print(f向量维度: {len(embeddings[0])}) print(f苹果向量前5维: {embeddings[0][:5]})1.2 向量相似度搜索原理向量相似度搜索的核心目标是从海量向量中快速找到与查询向量最相似的Top-K个结果。相似度的衡量通常有以下几种度量方式度量方式公式适用场景余弦相似度cos(θ) (A·B) / (|A||B|)文本语义相似度欧氏距离L2 √Σ(aᵢ - bᵢ)²图像特征、推荐系统点积A·B Σaᵢbᵢ排序打分、推荐系统import numpy as np def cosine_similarity(vec_a: list[float], vec_b: list[float]) - float: 计算两个向量的余弦相似度 a np.array(vec_a) b np.array(vec_b) # 余弦相似度 点积 / (模长乘积) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def euclidean_distance(vec_a: list[float], vec_b: list[float]) - float: 计算两个向量的欧氏距离 a np.array(vec_a) b np.array(vec_b) return float(np.linalg.norm(a - b)) # 示例 vec1 get_embedding(人工智能) vec2 get_embedding(机器学习) vec3 get_embedding(水果) print(f人工智能 vs 机器学习 余弦相似度: {cosine_similarity(vec1, vec2):.4f}) print(f人工智能 vs 水果 余弦相似度: {cosine_similarity(vec1, vec3):.4f})1.3 近似最近邻ANN算法精确的最近邻搜索KNN在数据量达到百万、千万级别时时间复杂度 O(N) 成为性能瓶颈。近似最近邻ANN算法通过牺牲少量精度换取数量级的性能提升是向量数据库的底层核心技术。主流ANN算法包括IVF倒排文件索引将向量空间聚类成多个桶查询时只搜索最近的几个桶HNSW分层导航小世界图构建多层图结构从上往下逐步逼近目标PQ乘积量化将高维向量压缩成短编码降低内存占用ANNOY随机投影树使用多棵随机投影树划分子空间二、主流向量数据库横向对比数据库开发语言部署方式索引算法适用场景特点ChromaPython本地/客户端HNSW开发测试轻量级、上手快FAISSC/Python本地IVF/PQ/HNSW研究、离线分析Facebook开源、性能强MilvusGo分布式集群HNSW/IVF/PQ生产环境分布式、云原生Pinecone-云服务HNSW云原生应用免运维、SaaSQdrantRust本地/容器HNSW高性能需求内存安全、高并发WeaviateGo分布式HNSW/BM25混合搜索图结构向量混合2.1 Chroma — 轻量级开发首选Chroma是目前最流行的本地向量数据库专为LLM应用原型开发设计。它将数据存储在本地SQLite文件中零配置即可使用非常适合快速验证RAG流程。# 安装pip install chromadb import chromadb # 初始化Chroma客户端本地持久化存储 client chromadb.PersistentClient(path./chroma_db) # 创建集合类似于表 collection client.create_collection( namearticles, metadata{description: 技术文章向量库} ) # 添加向量数据 collection.add( documents[ 大语言模型LLM是当前AI领域的核心技术, 向量数据库用于存储和检索高维向量数据, RAG技术结合了检索和生成两种能力 ], ids[doc1, doc2, doc3], metadatas[ {category: AI, author: 张三}, {category: 数据库, author: 李四}, {category: AI, author: 王五} ] ) # 相似度查询 results collection.query( query_texts[什么是大语言模型], n_results2 # 返回最相似的2条 ) print(查询结果:) for i, (doc, distance) in enumerate(zip(results[documents][0], results[distances][0])): print(f [{i1}] 距离{distance:.4f} | {doc}) # 删除文档 collection.delete(ids[doc3])2.2 FAISS — Facebook开源本地库FAISSFacebook AI Similarity Search是Facebook开源的高效相似度搜索库专注于在CPU/GPU上实现高速向量检索。它支持批量化索引构建和搜索是生产环境做离线分析或嵌入式部署的首选。# 安装pip install faiss-cpu或 faiss-gpu import numpy as np import faiss # 1. 准备数据生成10000条128维的随机向量模拟嵌入向量 dimension 128 # 向量维度 num_vectors 10000 # 向量数量 np.random.seed(42) # 生成随机向量并转为float32 vectors np.random.rand(num_vectors, dimension).astype(float32) # 2. 使用IVF索引加速查询 nlist 100 # 聚类中心数量 # 训练使用IVF索引需要先对数据进行聚类 quantizer faiss.IndexFlatIP(dimension) # IP 内积余弦相似度需先归一化 index faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_INNER_PRODUCT) index.train(vectors) # 训练索引 index.add(vectors) # 添加向量 # 设置搜索范围搜索的聚类中心数越大越精确但越慢 index.nprobe 10 # 3. 相似度搜索找最相似的5个 query_vector np.random.rand(1, dimension).astype(float32) k 5 # Top-K # 执行搜索 distances, indices index.search(query_vector, k) print(f查询向量维度: {query_vector.shape}) print(f找到的{k}个最近邻:) for i, (idx, dist) in enumerate(zip(indices[0], distances[0])): print(f 第{i1}个: 索引{idx}, 距离{dist:.4f}) # 4. 使用HNSW索引更精确但内存占用更大 print(\n--- 使用HNSW索引 ---) hnsw_index faiss.IndexHNSWFlat(dimension, 32) # 32为每层连接数 hnsw_index.hnsw.efConstruction 40 hnsw_index.add(vectors) hnsw_index.hnsw.efSearch 64 # 搜索时动态列表大小 distances_hnsw, indices_hnsw hnsw_index.search(query_vector, k) print(fHNSW搜索结果 (Top-{k}):) for i, (idx, dist) in enumerate(zip(indices_hnsw[0], distances_hnsw[0])): print(f 第{i1}个: 索引{idx}, 距离{dist:.4f})2.3 Milvus — 分布式生产级方案Milvus是面向生产环境设计的分布式向量数据库支持PB级数据存储提供丰富的索引类型和水平扩展能力适合企业级AI应用。# 安装pip install pymilvus from pymilvus import MilvusClient, DataType # 连接本地Milvus服务或远程地址 client MilvusClient(uri./milvus_demo.db) # 创建集合schema定义 schema MilvusClient.create_schema( auto_idTrue, enable_dynamic_fieldTrue, descriptionRAG知识库集合 ) # 添加字段 schema.add_field(field_nameid, datatypeDataType.INT64, is_primaryTrue) schema.add_field(field_namecontent, datatypeDataType.VARCHAR, max_length512) schema.add_field(field_nameembedding, datatypeDataType.FLOAT_VECTOR, dim1536) schema.add_field(field_namecategory, datatypeDataType.VARCHAR, max_length64) # 创建索引参数 index_params client.prepare_index_params() index_params.add_index( field_nameembedding, index_typeHNSW, # HNSW索引 metric_typeCOSINE, # 余弦相似度 params{M: 16, efConstruction: 200} ) # 创建集合 if client.has_collection(knowledge_base): client.drop_collection(knowledge_base) client.create_collection( collection_nameknowledge_base, schemaschema, index_paramsindex_params ) # 插入数据 documents [ Transformer是现代NLP的基石架构, 向量数据库解决大模型记忆问题, LangChain简化LLM应用开发流程 ] # 模拟嵌入向量实际使用时替换为真实embedding embeddings [np.random.rand(1536).tolist() for _ in documents] data [ {content: doc, embedding: emb, category: AI} for doc, emb in zip(documents, embeddings) ] result client.insert(collection_nameknowledge_base, datadata) print(f插入成功共 {result.insert_count} 条记录) # 搜索 query_embedding np.random.rand(1536).tolist() search_params {ef: 128} results client.search( collection_nameknowledge_base, data[query_embedding], limit3, search_paramssearch_params, output_fields[content, category] ) print(\n搜索结果:) for hits in results: for hit in hits: print(f 内容: {hit[entity][content]} | 距离: {hit[distance]:.4f}) # 删除集合 client.drop_collection(knowledge_base) client.close()2.4 Pinecone — 云原生SaaSPinecone是完全托管的云向量数据库用户无需关心基础设施运维支持快速弹性扩缩容适合快速上线的产品级应用。# 安装pip install pinecone-client from pinecone import Pinecone # 初始化需要API Key可在 pinecone.io 免费注册 pc Pinecone(api_keyyour-api-key) # 连接到索引 index pc.Index(llm-knowledge-base) # 插入向量Upsert vectors [ (vec1, [0.1] * 1536, {text: RAG是检索增强生成技术, source: AI论文}), (vec2, [0.2] * 1536, {text: 向量数据库支持高维向量检索, source: 技术文档}), (vec3, [0.3] * 1536, {text: 大模型需要外部知识库补充, source: 博客}), ] index.upsert(vectorsvectors) print(向量插入成功) # 相似度查询 query_result index.query( vector[0.15] * 1536, top_k2, include_metadataTrue ) print(\nTop-2 相似结果:) for match in query_result.matches: print(f ID: {match.id} | 分数: {match.score:.4f} | 文本: {match.metadata[text]}) # 删除向量 index.delete(ids[vec3])2.5 Qdrant — Rust实现高性能Qdrant由Rust语言实现以内存安全和高并发著称。其滤波搜索能力强大支持带条件过滤的向量查询在同等硬件条件下性能表现优异。# 安装pip install qdrant-client from qdrant_client import QdrantClient from qdrant_client.models import Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue # 连接Qdrant服务 client QdrantClient(urlhttp://localhost:6333) # 创建集合 collection_name products vector_size 128 if not client.collection_exists(collection_name): client.create_collection( collection_namecollection_name, vectors_configVectorParams(sizevector_size, distanceDistance.COSINE) ) # 批量插入向量 points [ PointStruct( id1, vectorlist(np.random.rand(vector_size)), payload{name: iPhone 15, category: 手机, price: 6999} ), PointStruct( id2, vectorlist(np.random.rand(vector_size)), payload{name: MacBook Pro, category: 电脑, price: 14999} ), PointStruct( id3, vectorlist(np.random.rand(vector_size)), payload{name: AirPods Pro, category: 耳机, price: 1899} ), ] client.upsert(collection_namecollection_name, pointspoints) print(插入成功) # 带过滤条件的搜索只查询手机类别 filter_condition Filter( must[FieldCondition(keycategory, matchMatchValue(value手机))] ) search_results client.search( collection_namecollection_name, query_vectorlist(np.random.rand(vector_size)), query_filterfilter_condition, limit1 ) print(\n过滤搜索结果category手机:) for result in search_results: print(f {result.payload[name]} | 分数: {result.score:.4f}) # 删除点 client.delete(collection_namecollection_name, points[1, 2, 3])2.6 Weaviate — 图结构向量混合Weaviate是原生支持GraphQL风格的向量数据库能够将向量搜索与传统BM25关键词搜索结合实现语义关键词的混合检索特别适合需要同时利用结构化数据过滤的场景。# 安装pip install weaviate-client import weaviate # 连接本地Weaviate服务 client weaviate.Client(http://localhost:8080) # 创建类Collection schema { class: Article, description: 技术文章, vectorizer: text2vec-transformers, # 使用本地transformer模型生成向量 moduleConfig: { text2vec-transformers: { vectorizeClassName: False # 类名不参与向量化 } }, properties: [ {name: title, dataType: [text]}, {name: content, dataType: [text]}, {name: category, dataType: [text]} ] } if client.schema.exists(Article): client.schema.delete_class(Article) client.schema.create_class(schema) print(Article 类创建成功) # 添加对象Weaviate自动生成向量 client.data_object.create( class_nameArticle, data_object{ title: 向量数据库详解, content: 本文介绍向量数据库的核心概念和使用方法, category: 数据库 } ) # 语义搜索 response client.query.get( Article, [title, content, category] ).with_near_text({ concepts: [AI检索技术] }).with_limit(2).do() print(\n语义搜索结果:) for obj in response[data][Get][Article]: print(f 标题: {obj[title]} | 类别: {obj[category]}) # 混合搜索向量 BM25关键词 hybrid_response client.query.get( Article, [title, content] ).with_hybrid( query数据库, alpha0.5 # 0.5表示向量搜索和关键词搜索各占50% ).with_limit(2).do() print(\n混合搜索结果:) for obj in hybrid_response[data][Get][Article]: print(f 标题: {obj[title]})三、索引算法详解3.1 IVF倒排文件索引IVFInverted File Index的核心思想是聚类倒排。先将所有向量通过K-Means聚类成N个桶Voroni单元查询时先找到查询向量所属的桶再在该桶内做精确搜索。这将搜索范围从N缩小到N/nlist。import faiss import numpy as np # 生成100000条256维向量 dimension 256 n_vectors 100_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # 构建IVF索引 nlist 1024 # 聚类中心数量 quantizer faiss.IndexFlatL2(dimension) index faiss.IndexIVFFlat(quantizer, dimension, nlist, faiss.METRIC_L2) index.train(vectors) index.add(vectors) # 搜索性能对比 import time query np.random.rand(1, dimension).astype(float32) # nprobe1只搜索1个桶 index.nprobe 1 start time.time() _, _ index.search(query, 10) t1 time.time() - start # nprobe32搜索32个桶 index.nprobe 32 start time.time() _, _ index.search(query, 10) t2 time.time() - start print(fIVF nprobe1: {t1*1000:.2f}ms) print(fIVF nprobe32: {t2*1000:.2f}ms) print(f精度提升约 {(t2/t1):.1f}x 时间换取更高召回率)3.2 HNSW分层导航小世界图HNSWHierarchical Navigable Small World是一种基于图的近似最近邻算法。它构建一个多层图结构上层连接稀疏快速定位下层连接稠密精确结果。查询时从最上层开始通过贪心遍历逐步向下逼近。import faiss import numpy as np dimension 128 n_vectors 50_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # HNSW参数说明 # M: 每个节点的最大连接数越大越精确内存越高 # efConstruction: 构建时动态列表大小越大构建越慢索引质量越高 hnsw_index faiss.IndexHNSWFlat(dimension, 32) hnsw_index.hnsw.efConstruction 40 hnsw_index.add(vectors) # 查询时 efSearch 参数值越大搜索越精确但越慢 for ef in [16, 64, 256]: hnsw_index.hnsw.efSearch ef _, indices hnsw_index.search(np.random.rand(1, dimension).astype(float32), 10) print(fefSearch{ef}: Top-10结果索引前3个{indices[0][:3]})3.3 PQ乘积量化PQProduct Quantization将高维向量分割成多个子空间对每个子空间分别做K-Means量化从而将原始向量压缩成短编码。这使得在有限内存中存储数十亿向量成为可能。import faiss import numpy as np dimension 64 n_vectors 20_000 vectors np.random.rand(n_vectors, dimension).astype(float32) # PQ参数m为子空间数量nbits为每个子空间的比特数 # 原始向量 64 dim * 4 bytes 256 bytes # 压缩后: m * nbits bits m 8 # 子空间数量 nbits 8 # 每个子空间用8bit编码256个聚类中心 pq_index faiss.IndexPQ(dimension, m, nbits) pq_index.train(vectors) pq_index.add(vectors) # 注意PQ索引的搜索结果是有损的 query np.random.rand(1, dimension).astype(float32) distances, indices pq_index.search(query, 5) print(fPQ压缩后每个向量仅需 {m * nbits / 8:.1f} bytes) print(f压缩比: {256 / (m * nbits / 8):.1f}x)3.4 ANNOY随机投影树ANNOYApproximate Nearest Neighbors Oh Yeah使用多棵随机投影二叉树来划分子空间查询时并行搜索所有树取最优结果。Google开发主要用于音乐推荐Audioscrobbler数据集。# 安装pip install annoy from annoy import AnnoyIndex import random dimension 40 n_vectors 10000 n_trees 50 # 构建索引 annoy_index AnnoyIndex(dimension, angular) # angular等价于余弦距离 for i in range(n_vectors): vec [random.gauss(0, 1) for _ in range(dimension)] annoy_index.add_item(i, vec) # 构建50棵树越多越精确越慢 annoy_index.build(n_trees) # 查询最近邻 query [random.gauss(0, 1) for _ in range(dimension)] results annoy_index.get_nns_by_vector(query, 5, search_k-1) print(fANNOY查询结果 (Top-5): {results}) # 指定搜索预算越大越精确 results_exact annoy_index.get_nns_by_vector(query, 5, search_k500) print(fANNOY精确查询 (search_k500): {results_exact})四、向量数据库核心操作4.1 插入向量import chromadb client chromadb.PersistentClient(path./demo_db) collection client.get_or_create_collection(demo) # 单条插入 collection.add( idssingle_doc, documents[这是一段技术文档], metadatas[{author: 工程师}] ) # 批量插入推荐性能更好 batch_data { ids: [fdoc_{i} for i in range(1000)], documents: [f文档内容 {i} for i in range(1000)], metadatas: [{index: i} for i in range(1000)] } collection.add(**batch_data) print(f批量插入完成共 {collection.count()} 条)4.2 相似度查询# 基于文本的语义搜索 results collection.query( query_texts[查找AI相关的技术文档], n_results5, where{author: 工程师}, # 元数据过滤条件 include[documents, distances, metadatas] # 指定返回字段 ) # 基于向量的搜索 import numpy as np query_vector np.random.rand(1536).tolist() results collection.query( query_embeddings[query_vector], n_results3 )4.3 范围查询Range Search范围查询返回与查询向量距离在指定阈值内的所有向量适用于附近类应用。from qdrant_client import QdrantClient client QdrantClient(urlhttp://localhost:6333) # 在Qdrant中执行范围查询搜索距离阈值内的所有向量 search_results client.search( collection_namedemo_collection, query_vectorlist(np.random.rand(128)), score_threshold0.7, # 相似度分数阈值0~1 limit100 # 最大返回数量 ) print(f范围内找到 {len(search_results)} 条记录)4.4 删除与更新# Chroma - 删除 collection.delete(ids[doc_1, doc_2]) # 删除指定ID collection.delete(where{category: 过时}) # 按条件删除 # Chroma - 更新先删后加 collection.update( idsdoc_1, documents[更新后的内容], metadatas[{status: updated}] ) # Milvus - 删除 client.delete( collection_nameknowledge_base, filterid in [1, 2, 3] ) # Milvus - 更新通过upsert实现 client.upsert( collection_nameknowledge_base, data[{id: 1, content: 更新内容, embedding: [0.1]*1536}] )五、使用场景5.1 RAG知识库RAGRetrieval-Augmented Generation是向量数据库最经典的应用场景。通过将企业知识库文档预先向量化并存储LLM在回答用户问题时先检索最相关的文档片段再结合上下文生成答案有效缓解大模型的幻觉问题。# 完整的RAG流程示例 import chromadb from openai import OpenAI client_chroma chromadb.PersistentClient(path./rag_db) collection client_chroma.get_or_create_collection(rag_knowledge_base) openai_client OpenAI() # 第一步文档预处理与入库 def process_document(text: str, doc_id: str): 将文档文本向量化并存储 embedding openai_client.embeddings.create( modeltext-embedding-3-small, inputtext ).data[0].embedding collection.add( idsdoc_id, documents[text], embeddings[embedding] ) # 第二步RAG检索 def rag_query(user_question: str, top_k: int 3) - str: 完整的RAG查询流程 1. 将用户问题向量化 2. 检索最相关的文档片段 3. 构建Prompt并调用LLM生成答案 # 1. 检索相关文档 query_embedding openai_client.embeddings.create( modeltext-embedding-3-small, inputuser_question ).data[0].embedding results collection.query( query_embeddings[query_embedding], n_resultstop_k ) # 2. 组合上下文 context \n\n.join(results[documents][0]) if results[documents] else 无相关背景信息 # 3. 构建PromptFew-shot prompt f基于以下背景信息回答用户问题。如果背景信息不足以回答请如实说明。 背景信息 {context} 用户问题{user_question} 回答 # 4. 调用LLM生成 response openai_client.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.3 ) return response.choices[0].message.content # 示例查询 answer rag_query(向量数据库有哪些应用场景) print(fRAG回答{answer})5.2 以图搜图将图片通过视觉模型如CLIP、ResNet提取特征向量存入向量数据库实现基于内容的图像检索。import numpy as np import chromadb client chromadb.PersistentClient(path./image_search_db) collection client.get_or_create_collection(image_features) # 模拟图片向量实际使用CLIP等模型提取 def simulate_image_embedding(image_path: str) - list[float]: 模拟图片向量提取实际应使用clip.encode_image() np.random.seed(hash(image_path) % 2**32) return np.random.rand(512).tolist() # 添加图片到索引 images [ {id: img_001, path: /photos/cat.jpg, desc: 可爱的猫咪}, {id: img_002, path: /photos/dog.jpg, desc: 活泼的狗狗}, {id: img_003, path: /photos/car.jpg, desc: 红色跑车}, ] for img in images: emb simulate_image_embedding(img[path]) collection.add( idsimg[id], documents[img[desc]], embeddings[emb] ) # 搜索可爱的动物对应的图片 query_embedding simulate_image_embedding(cute animal) results collection.query( query_embeddings[query_embedding], n_results2 ) print(以图搜图结果:) for doc, dist in zip(results[documents][0], results[distances][0]): print(f {doc} | 距离: {dist:.4f})5.3 语义搜索超越关键词匹配在语义层面理解用户查询意图返回真正相关的结果。适用于搜索引擎、企业知识管理、内容推荐等场景。# 语义搜索示例使用Chroma import chromadb client chromadb.PersistentClient(path./semantic_search_db) collection client.get_or_create_collection(news_articles) # 批量添加新闻数据 articles [ 美联储宣布降息以应对经济下行压力, 人工智能芯片需求爆发式增长, 某地发生5.0级地震暂无人员伤亡, 新能源车销量突破历史新高, 科学家发现新的量子计算算法 ] collection.add( ids[fnews_{i} for i in range(len(articles))], documentsarticles ) # 语义查询金融政策应该匹配到美联储降息的新闻 query 货币政策和经济调控 results collection.query( query_texts[query], n_results2 ) print(f语义查询: {query}) print(返回结果:) for i, (doc, dist) in enumerate(zip(results[documents][0], results[distances][0])): print(f {i1}. {doc})5.4 推荐系统基于用户行为和物品特征构建向量表示通过向量相似度计算实现个性化推荐。import numpy as np import chromadb client chromadb.PersistentClient(path./recommendation_db) user_collection client.get_or_create_collection(user_profiles) item_collection client.get_or_create_collection(items) # 模拟用户兴趣向量和商品特征向量 def cosine_sim(a, b): return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) # 用户画像向量由历史行为生成 user_vector np.random.rand(128).tolist() # 商品向量 item_vectors { item_101: np.random.rand(128).tolist(), item_102: np.random.rand(128).tolist(), item_103: np.random.rand(128).tolist(), item_104: np.random.rand(128).tolist(), } # 计算相似度并排序推荐 similarities [ (item_id, cosine_sim(user_vector, vec)) for item_id, vec in item_vectors.items() ] ranked sorted(similarities, keylambda x: x[1], reverseTrue) print(个性化推荐结果Top-3:) for item_id, score in ranked[:3]: print(f {item_id}: 相似度{score:.4f})六、综合代码示例构建本地RAG系统以下代码整合了本文的核心内容展示如何使用Chroma构建一个完整的本地RAG知识库问答系统 完整的本地RAG系统示例 功能从知识库检索相关片段 LLM生成回答 依赖chromadb, openai import chromadb from openai import OpenAI from typing import Optional class LocalRAGSystem: 本地RAG知识库系统 def __init__(self, db_path: str, openai_api_key: str): # 初始化Chroma持久化存储 self.client chromadb.PersistentClient(pathdb_path) self.collection self.client.get_or_create_collection( nameknowledge_base, metadata{description: 本地RAG知识库} ) # 初始化OpenAI客户端 self.llm OpenAI(api_keyopenai_api_key) self.embedding_model text-embedding-3-small def add_documents(self, documents: list[str], ids: Optional[list[str]] None): 批量添加文档到知识库 if ids is None: ids [fdoc_{i} for i in range(len(documents))] # 批量生成嵌入向量 embeddings [] batch_size 100 for i in range(0, len(documents), batch_size): batch documents[i:ibatch_size] response self.llm.embeddings.create( modelself.embedding_model, inputbatch ) embeddings.extend([r.embedding for r in response.data]) self.collection.add( idsids, documentsdocuments, embeddingsembeddings ) print(f✅ 成功添加 {len(documents)} 篇文档当前共 {self.collection.count()} 篇) def retrieve(self, query: str, top_k: int 3) - list[dict]: 从知识库检索最相关的文档 # 生成查询向量 response self.llm.embeddings.create( modelself.embedding_model, inputquery ) query_embedding response.data[0].embedding # 检索 results self.collection.query( query_embeddings[query_embedding], n_resultstop_k, include[documents, distances, metadatas] ) # 格式化返回 retrieved [] for i in range(len(results[ids][0])): retrieved.append({ id: results[ids][0][i], document: results[documents][0][i], distance: results[distances][0][i], metadata: results[metadatas][0][i] }) return retrieved def generate(self, query: str, context: list[dict]) - str: 使用LLM结合上下文生成回答 # 组合上下文 context_text \n.join([ f[{i1}] {ctx[document]} for i, ctx in enumerate(context) ]) prompt f你是一个专业的问答助手。请基于以下参考信息回答用户问题。 如果参考信息不足以回答请如实说明不要编造答案。 参考信息 {context_text} 用户问题{query} 回答 response self.llm.chat.completions.create( modelgpt-4o, messages[{role: user, content: prompt}], temperature0.3, max_tokens500 ) return response.choices[0].message.content def query(self, question: str, top_k: int 3) - dict: 完整的RAG查询检索 生成 # 1. 检索 retrieved self.retrieve(question, top_k) # 2. 生成 answer self.generate(question, retrieved) return { question: question, answer: answer, references: [ctx[document] for ctx in retrieved] } # 使用示例 if __name__ __main__: # 注意实际使用时替换为真实API Key # rag LocalRAGSystem(./my_rag_db, api_keysk-...) # 添加知识库文档 docs [ Transformer架构由Google在2017年提出是现代NLP的核心模型, BERT是基于Transformer的双向编码器表示在多项NLP任务上取得最优成绩, GPT系列是OpenAI开发的大型语言模型采用自回归生成方式, 向量数据库通过近似最近邻算法实现高速相似度检索, RAG检索增强生成结合了检索系统和生成模型的优势 ] # rag.add_documents(docs) # 结果 rag.query(什么是RAG技术) # print(f问题: {结果[question]}\n回答: {结果[answer]})七、总结与选型建议开发测试阶段推荐使用Chroma零配置、纯Python、SQLite后端5分钟即可跑通RAG流程。离线分析/研究阶段推荐FAISSFacebook出品性能极强支持GPU加速适合处理海量数据。生产环境根据规模选择中小规模百万级QdrantRust高性能或Milvus功能丰富大规模/云原生Milvus分布式集群或PineconeSaaS免运维需要混合检索Weaviate向量关键词融合选型核心考量因素数据规模本地/Chroma适合万级FAISS可达亿级Milvus/Pinecone支持十亿级以上查询延迟Qdrant和FAISS在同硬件下最快Pinecone延迟取决于云端配置运维成本Pinecone最低完全托管Chroma次之自建Milvus需要Kubernetes经验功能需求需要过滤查询优先Qdrant需要混合搜索优先Weaviate需要强一致性选Milvus
