Embedding模型和向量数据库的使用

一、知识储备

Embedding、LLM、向量数据库、RAG的区别

Embedding 模型把文本转换成向量 一串数字，像： text-embedding-v4、BGE-M3

大语言模型 (LLM) ：理解并生成文本、人类可读的文字，像： GPT、Qwen、Claude

向量数据库：存储高维向量数据，适合非结构化数据。

• 向量存储：把 Embedding 模型生成的向量存起来。

• 创建向量索引：为了在海量数据中找得快，数据库会提前把向量建立成特殊的数学结构（比如树形结构、图结构等）。

• 相似性搜索（核心）

注意：向量数据库才是相似性搜索，Embedding模型只是把东西转成向量。

RAG = Embedding + 检索 + Text Splitter（切分器） +LLM生成

注意：Embedding不做切分，切分由RAG的Text Splitter（切分器）做。

二、Embedding计算

将一段文本（如 "我想知道迪士尼的退票政策"）通过 AI 模型转换成一组数字向量，这个向量可以用来衡量文本之间的语义相似度。

因为要调用连接阿里云百炼服务的 SDK（工具包），所以要openai。

pip install openai

import os from openai import OpenAI client = OpenAI( api_key="你的API", # 你访问阿里云服务的"钥匙" base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" # 阿里云服务的"地址" ) completion = client.embeddings.create( model="text-embedding-v4", #model ：选择阿里云提供的"翻译员"（Embedding 模型） input='我想知道迪士尼的退票政策', #input ：你要转换成数字的那句话 dimensions=1024, # 指定向量维度（数字越多，描述越精确） encoding_format="float" ) print(completion.model_dump_json())

输入："我想知道迪士尼的退票政策"
↓
这段代码调用阿里云服务
↓
输出：[0.123, -0.456, 0.789, ...] (1024个小数)

三、向量数据库FAISS的使用（存储向量和向量的相似检索）

此代码实现了一个"智能搜索"功能：把文档存入数据库，用户提问后自动找到最相关的文档并返回。

准备文档 → 转换成向量 → 存入数据库 → 用户提问 → 返回最相关的文档

第一步：准备钥匙

第二步：准备文档

第三步：把文档变成数字

第四步：把向量存入 FAISS 数据库

第五步：用户提问，搜索最相关的文档

import os import numpy as np import faiss from openai import OpenAI # Step1. 初始化 API 客户端 try: client = OpenAI( api_key="你的key", base_url="https://dashscope.aliyuncs.com/compatible-mode/v1" ) except Exception as e: print("初始化OpenAI客户端失败，请检查环境变量'DASHSCOPE_API_KEY'是否已设置。") print(f"错误信息: {e}") exit() # Step2. 准备示例文本和元数据 # 在实际应用中，这些数据可能来自数据库、文件等 documents = [ { "id": "doc1", "text": "迪士尼乐园的门票一经售出，原则上不予退换。但在特殊情况下，如恶劣天气导致园区关闭，可在官方指引下进行改期或退款。", "metadata": {"source": "official_faq_v1.pdf", "category": "退票政策", "author": "Admin"} }, { "id": "doc2", "text": "购买“奇妙年卡”的用户，可以享受一年内多次入园的特权，并且在餐饮和购物时有折扣。", "metadata": {"source": "annual_pass_rules.docx", "category": "会员权益", "author": "MarketingDept"} }, { "id": "doc3", "text": "对于在线购买的迪士尼门票，如果需要退票，必须在票面日期前48小时通过原购买渠道提交申请，并可能收取手续费。", "metadata": {"source": "online_policy.html", "category": "退票政策", "author": "E-commerceTeam"} }, { "id": "doc4", "text": "园区内的“加勒比海盗”项目因年度维护，将于下周暂停开放。", "metadata": {"source": "maintenance_notice.txt", "category": "园区公告", "author": "OpsDept"} } ] # Step3. 创建元数据存储和向量列表 # 我们使用一个简单的列表来存储元数据。列表的索引将作为FAISS的ID。 # 这种方式简单直接，适用于中小型数据集。 # 对于大型数据集，可以考虑使用字典或数据库（如Redis, SQLite） metadata_store = [] #保存原始文档内容和标签 vectors_list = [] #保存 Embedding 模型生成的向量 vector_ids = [] #建立向量和元数据之间的关联 print("正在为文档生成向量...") for i, doc in enumerate(documents): try: # 调用API生成向量 completion = client.embeddings.create( model="text-embedding-v4", input=doc["text"], dimensions=1024, encoding_format="float" ) # 获取向量 vector = completion.data[0].embedding vectors_list.append(vector) # 存储元数据，并使用列表索引作为唯一ID metadata_store.append(doc) vector_ids.append(i) # 自定义ID与列表索引一致 print(f" - 已处理文档 {i+1}/{len(documents)}") except Exception as e: print(f"处理文档 '{doc['id']}' 时出错: {e}") continue # 将向量列表转换为NumPy数组，FAISS需要这种格式 vectors_np = np.array(vectors_list).astype('float32') vector_ids_np = np.array(vector_ids) # Step4. 构建并填充 FAISS 索引 dimension = 1024 # 向量维度 k = 2 # 查找最近的3个邻居 # 创建一个基础的L2距离索引 index_flat_l2 = faiss.IndexFlatL2(dimension) # 使用IndexIDMap来包装基础索引，能够映射我们自定义的ID # 这就是关联向量和元数据的关键！ index = faiss.IndexIDMap(index_flat_l2) # 将向量和它们对应的ID添加到索引中 index.add_with_ids(vectors_np, vector_ids_np) print(f"\nFAISS 索引已成功创建，共包含 {index.ntotal} 个向量。") # Step5. 执行搜索并检索元数据 query_text = "我想了解一下迪士尼门票的退款流程" print(f"\n正在为查询文本生成向量: '{query_text}'") try: # 为查询文本生成向量 query_completion = client.embeddings.create( model="text-embedding-v4", input=query_text, dimensions=1024, encoding_format="float" ) query_vector = np.array([query_completion.data[0].embedding]).astype('float32') # 在FAISS索引中执行搜索 # search方法返回两个NumPy数组： # D: 距离 (distances) # I: 索引/ID (indices/IDs) distances, retrieved_ids = index.search(query_vector, k) # Step6. 展示结果 print("\n--- 搜索结果 ---") # `retrieved_ids[0]` 包含与查询最相似的k个向量的ID for i in range(k): doc_id = retrieved_ids[0][i] # 检查ID是否有效 if doc_id == -1: print(f"\n排名 {i+1}: 未找到更多结果。") continue # 使用ID从我们的元数据存储中检索信息 retrieved_doc = metadata_store[doc_id] print(f"\n--- 排名 {i+1} (相似度得分/距离: {distances[0][i]:.4f}) ---") print(f"ID: {doc_id}") print(f"原始文本: {retrieved_doc['text']}") print(f"元数据: {retrieved_doc['metadata']}") except Exception as e: print(f"执行搜索时发生错误: {e}")