LangChain RAG之向量

嵌入模型

通过嵌入模型转化为向量坐标，并存储到向量数据库中，再通过相似性搜索找到相关文档的过程。

• 将文本转换为数字向量表示
• 使用简单的数学运算来比较嵌入向量

• 嵌入模型是向量化的核心
• 市场上有海量的嵌入模型
• LangChain提供了标准接口
• 嵌入过程本质是一组高维坐标
• 常见的相似性算法有：
  • 余弦相似度：测量两个向量之间角度的余
  • 欧几里得距离：测量亮点之间的直线距离
  • 点积：测量一个向量到另一个向量的投影

LangChain的嵌入实现

LangChain封装了统一的嵌入模型接口，支持多文本和单问题嵌入，强调选择模型时需注意语言适配性、嵌入维度一致性及成本控制，推荐使用多语言模型，

主要的方法：

• embed_documents：用于嵌入多个文本（文档）
• embed_query：用于嵌入单个文本（查询）

主要合作模型

模型名	模型方	说明
OpenAIEmbeddings	openai	多个模型，一般嵌入维度1024
OllamaEmbeddings	多方	可以运行多种开源嵌入模型
JinaEmbeddings	Jina.ai	顶级多语言嵌入模型
ZhipuAIEmbeddings	zhipu	中文能力较强，嵌入维度1024

• 嵌入模型选择需要注意模型的适配语言
• 嵌入和检索的维度要相同
• 最好选择多语言嵌入模型
• 非开源嵌入模型一样有token消耗

嵌入模型的使用示例

示例一：embed_documents 向量转化和embed_query 向量查询

from langchain_openai import OpenAIEmbeddings

embeddings_model = OpenAIEmbeddings()
# 将下面内容转化为向量
embeddings = embeddings_model.embed_documents(
    [
        "Hi there!",
        "Oh, hello!",
        "What's your name?",
        "My friends call me World",
        "Hello World!"
    ]
)
len(embeddings), len(embeddings[0])

query_embedding = embeddings_model.embed_query("What is the meaning of life?")
print(query_embedding)

示例二：缓存嵌入结果（依托于向量数据库）

需要安装依赖

pip install --upgrade --quiet faiss-cpu

示例代码

from langchain.embeddings import CacheBackedEmbeddings
from langchain.storage import LocalFileStore
from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
from langchain_text_splitters import CharacterTextSplitter

underlying_embeddings = OpenAIEmbeddings()

# 从本地缓存加载
store = LocalFileStore("/tmp/langchain_cache")

cached_embedder = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings, store, namespace=underlying_embeddings.model
)

# 现在本地缓存什么都没有，则返回空
list(store.yield_keys())

# 创建向量存储
db = FAISS.from_documents(documents, cached_embedder)

# 再次创建将读取缓存，从而加快速度降低成本
db2 = FAISS.from_documents(documents, cached_embedder)

list(store.yield_keys())[:5]
  

示例三使用国产嵌入模型

选取硅基流动中的嵌入模型： https://cloud.siliconflow.cn/models?types=embedding

import os
from langchain_openai import OpenAIEmbeddings

embeddings_model = OpenAIEmbeddings(
    model="BAAI/bge-m3",
    api_key=os.environ.get("API_KEY"),
    base_url="https://api.siliconflow.cn/v1",
)

embeddings = embeddings_model.embed_documents(
    [
        "床前明月光",
        "疑是地上霜",
        "举头望明月",
        "低头思故乡",
        "李白《静夜思》"
    ]
)
len(embeddings), len(embeddings[0])

向量数据库

向量数据库的作用及与关系型数据库的区别，指出向量数据库用于存储高维向量数据并支持相似性搜索，适用于图像检索、语义搜索等场景，而关系型数据库适合结构化数据的精确查询。

• 问题被嵌入向量化后进入向量库
• 使用向量搜索找到相关文档片段

• 向量数据库是RAG的关键组件

• 向量数据库中存储了所有的外挂知识

• 向量库本身内置算法对于检索很关键

• 向量库与关系型数据库区别较大

  $$向量数据库 \neq 关系型数据库$$

• 向量数据库
  • 向量化存储
  • 相似度搜索
  • 高性能检索
• 关系型数据库
  • 结构化存储
  • 精确查询
  • 事务处理

LangChain的向量数据库操作

主要方法：

• add_documents：将文本列表添加到矢量存储中
• delete：从向量存储中删除文档列表
• similarity_search：搜索与给定查询类似的文档

常用向量数据库使用

名称	开源信息	官网	支持语言	云服务	特点介绍
Milvus	✔️ 21.5K star	https://github.com/milvus-io/milvus	Go、Python、C++	✔️	- 面向下一代的生产式AI向量数据库，支持云原生。 - 极高的检索性能，万亿向量数据的毫秒检索 - 非结构化数据的简单管理 - 高度可扩展与弹性，支持混合检索 - 统一的lambda架构，具有高可靠的故障转移 - 拥有超过1000个企业用户
Faiss	✔️ 3.2K star	https://github.com/facebookresearch/faiss	C++、Python	❌	- Meta团队开发 - 支持高效的相似性搜索和密集向量聚类 - 可以搜索任何大小的向量集 - 支持CPU计算和GPU计算
Pinecone	❌	https://www.pinecone.io/	❌	✔️	- 全托管的向量数据库，大幅减轻运维负担，用户可聚焦数据价值抽取；免费版支持500w向量存储，用法简单、价格低，能快速验证向量检索业务；特性上具备高速、准确、可扩展，还支持单元级元数据过滤、尖端稀疏-密集索引等高级功能。
Chroma	✔️ 7.4K star	https://github.com/chroma-core/chroma	Python	❌	- 简单：类型完整、测试全面、文档完整 - 整合：支持LangChain（Python和JS）、LlamaIndex等 - 开发流程：Python Notebook中运行的API可直接用于集群环境 - 功能丰富：查询、过滤、密度估计等
LanceDB	✔️ 1.6K star	https://github.com/lancedb/lancedb	Rust、Python、JavaScript	❌	- 支持存储、查询、过滤向量、元数据及多模态数据（文本、图像、视频、点云等） - 支持向量相似度搜索、全文检索和SQL语法 - 支持Python和JavaScript/Typescript - 零拷贝、自动版本控制，无需额外架构即可管理数据版本 - 支持基于GPU的向量索引 - 支持与LangChain、LlamaIndex、Pandas等生态集成

向量数据的操作

1.向量库的数据增加

import os
from langchain_openai import OpenAIEmbeddings

embeddings_model = OpenAIEmbeddings(
    model="BAAI/bge-m3",
    api_key=os.environ.get("API_KEY"),
    base_url="https://api.siliconflow.cn/v1",
)

# 引入一个内存向量数据库，它将向量暂存在内存中，并使用字典以及numpy计算搜索的余弦相似度。

from langchain_core.vectorstores import InMemoryVectorStore

vector_store = InMemoryVectorStore(embedding=embeddings_model)

from langchain_core.documents import Document

# 定义一组Document对象
document_1 = Document(
    page_content="今天在抖音学会了一个新菜：锅巴土豆泥！看起来简单，实际炸了厨房，连猫都嫌弃地走开了。",
    metadata={"source": "社交媒体"},
)

document_2 = Document(
    page_content="小区遛狗大爷今日播报：广场舞大妈占领健身区，遛狗群众纷纷撤退。现场气氛诡异，BGM已循环播放《最炫民族风》两小时。",
    metadata={"source": "社区新闻"},
)

documents = [document_1, document_2]

vector_store.add_documents(documents=documents)

# 可以为添加的文档增加ID索引，便于后面管理
vector_store.add_documents(documents=documents, ids=["doc1", "doc2"])

2.向量库的数据删除

根据1中的代码继续操作

vector_store.delete(ids=["doc1"])

3.向量库的相似性搜索

根据1中的代码继续操作

query = "遛狗"
docs = vector_store.similarity_search(query)
print(docs[0].page_content)

还可以使用“向量”查相似”向量“的方式来进行搜索

# 将"遛狗"转化为向量，这种方式会更精确
embedding_vector = embeddings_model.embed_query(query)
docs = vector_store.similarity_search_by_vector(embedding_vector)
print(docs[0].page_content)

注意：langchain只是在接口层面进行了封装，具体的搜索实现要依赖向量库本身的能力，比如Pinecone就可以进行元数据过滤，内存向量就不可以

以pinecone为例结合向量数据的能力进行相似性搜索

数据库地址：https://docs.pinecone.io/guides/get-started/quickstart

需要注册pinecone账号，获取key，注该数据库不是开源的

安装依赖

pip install -qU langchain-pinecone pinecone-notebooks

# 向量数据库配置
import os
from pinecone import Pinecone

pinecone_api_key = os.environ.get("API_KEY")

pc = Pinecone(api_key=pinecone_api_key)

# 数据库初始化
import time
from pinecone import ServerlessSpec

index_name = "langchain-test-index"  # change if desired
expected_dimension = 1024 # 你的 embedding 模型实际生成的维度

# 如果索引存在，先删除（注意：这会删除所有数据！）
if index_name in pc.list_indexes().names():
    pc.delete_index(index_name)

existing_indexes = [index_info["name"] for index_info in pc.list_indexes()]

if index_name not in existing_indexes:
    pc.create_index(
        name=index_name,
        dimension=expected_dimension, #注意维度要一致
        metric="cosine",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )
    while not pc.describe_index(index_name).status["ready"]:
        time.sleep(1)
# 定义索引
index = pc.Index(index_name)

# 嵌入模型设置
import os
from langchain_pinecone import PineconeVectorStore
from langchain_openai import OpenAIEmbeddings

embeddings_model = OpenAIEmbeddings(
    model="BAAI/bge-m3",
    api_key=os.environ.get("API_KEY"),
    base_url="https://api.siliconflow.cn/v1",
)

vector_store = PineconeVectorStore(index=index, embedding=embeddings_model)

# 向量数据库中添加数据
from uuid import uuid4
from langchain_core.documents import Document

document_1 = Document(
    page_content="今天早餐吃了老王家的生煎包，馅料实在得快从褶子里跳出来了！这才是真正的上海味道！",
    metadata={"source": "tweet"},
)

document_2 = Document(
    page_content="明日天气预报：北京地区将出现大范围雾霾，建议市民戴好口罩，看不见脸的时候请不要慌张。",
    metadata={"source": "news"},
)

document_3 = Document(
    page_content="终于搞定了AI聊天机器人！我问它'你是谁'，它回答'我是你爸爸'，看来还需要调教...",
    metadata={"source": "tweet"},
)

document_4 = Document(
    page_content="震惊！本市一男子在便利店抢劫，只因店员说'扫码支付才有优惠'，现已被警方抓获。",
    metadata={"source": "news"},
)

document_5 = Document(
    page_content="刚看完《流浪地球3》，特效简直炸裂！就是旁边大妈一直问'这是在哪拍的'有点影响观影体验。",
    metadata={"source": "tweet"},
)

document_6 = Document(
    page_content="新发布的小米14Ultra值不值得买？看完这篇测评你就知道为什么李老板笑得合不拢嘴了。",
    metadata={"source": "website"},
)

document_7 = Document(
    page_content="2025年中超联赛十大最佳球员榜单新鲜出炉，第一名居然是他？！",
    metadata={"source": "website"},
)

document_8 = Document(
    page_content="用LangChain开发的AI助手太神奇了！问它'人生的意义'，它给我推荐了一份外卖优惠券...",
    metadata={"source": "tweet"},
)

document_9 = Document(
    page_content="A股今日暴跌，分析师称原因是'大家都在抢着卖'，投资者表示很有道理。",
    metadata={"source": "news"},
)

document_10 = Document(
    page_content="感觉我马上要被删库跑路了，祝我好运 /(ㄒoㄒ)/~~",
    metadata={"source": "tweet"},
)

documents = [
    document_1,
    document_2,
    document_3,
    document_4,
    document_5,
    document_6,
    document_7,
    document_8,
    document_9,
    document_10,
]
uuids = [str(uuid4()) for _ in range(len(documents))]

vector_store.add_documents(documents=documents, ids=uuids)

# 删除最后一项数据操作
vector_store.delete(ids=[uuids[-1]])

• 相似性搜索支持元数据过滤

  results = vector_store.similarity_search(
      "看电影",
      k=1, # 召回数据条数，默认是4
      filter={"source": "tweet"}, # 支持元数据的过滤
  )
  for res in results:
      print(f"* {res.page_content} [{res.metadata}]")

• 通过source分数进行搜索

  results = vector_store.similarity_search_with_score(
      "明天热吗?", 
      k=1,  # 设置为1，则召回最相关的一条
      filter={"source": "news"}
  )
  for res, score in results:
      print(f"* [SIM={score:3f}] {res.page_content} [{res.metadata}]")

4. 高级使用：MMR（最大边际相关性）

注意：并非所有向量数据库都支持

该例子也是以向量数据库pinecone为示例，紧接3中的例子进行

vector_store.max_marginal_relevance_search(
    query="新手机",
    k=3,
    lambda_val=0.2, # 相关性和多样性之间的权衡
    filter={"source": "website"},
)

5.高级使用：混合搜索

先用关键字进行过滤，再使用向量机制进行搜索

安装依赖

pip install -qU pinecone-text

# 初始化pinecone数据库
from pinecone import ServerlessSpec

index_name = "langchain-pinecone-hybrid-search"  # change if desired
expected_dimension = 1024 # 你的 embedding 模型实际生成的维度

# 如果索引存在，先删除（注意：这会删除所有数据！）
if index_name in pc.list_indexes().names():
    pc.delete_index(index_name)

existing_indexes = [index_info["name"] for index_info in pc.list_indexes()]

if index_name not in existing_indexes:
    pc.create_index(
        name=index_name,
        dimension=expected_dimension, #注意维度要一致
        metric="dotproduct",
        spec=ServerlessSpec(cloud="aws", region="us-east-1"),
    )

index = pc.Index(index_name)

from pinecone_text.sparse import BM25Encoder

# or from pinecone_text.sparse import SpladeEncoder if you wish to work with SPLADE

# use default tf-idf values
bm25_encoder = BM25Encoder().default() # 混合搜索中的词法搜索，先进行过滤

# 关键字匹配，词法引擎初始化关键字
corpus = ["foo", "bar", "world", "hello"]

# fit tf-idf values on your corpus
bm25_encoder.fit(corpus)

# store the values to a json file
bm25_encoder.dump("bm25_values.json")

# load to your BM25Encoder object
bm25_encoder = BM25Encoder().load("bm25_values.json")

import os
from langchain_community.retrievers import (
    PineconeHybridSearchRetriever,
)
from langchain_openai import OpenAIEmbeddings

embeddings_model = OpenAIEmbeddings(
    model="BAAI/bge-m3",
    api_key=os.environ.get("API_KEY"),
    base_url="https://api.siliconflow.cn/v1",
)

# 词法引擎解析器
retriever = PineconeHybridSearchRetriever(
    embeddings=embeddings_model, sparse_encoder=bm25_encoder, index=index # 向量模型，词法引擎，索引
)

# 添加文本
retriever.add_texts(["foo", "bar", "world", "hello"])

# 检索
result = retriever.invoke("foo")
print(result[0])