Langchain模块-DataCollection组件-文本嵌入与向量数据库embedding

 2025/12/08 

维度	Chroma	FAISS	Milvus
类型	嵌入式向量数据库	向量检索库（非数据库）	分布式向量数据库
部署方式	轻量级，Python集成或独立服务	库形式集成，无独立服务	独立的数据库服务，支持集群
持久化	✅ 支持（本地/云）	❌ 不支持（需额外实现）	✅ 支持（内置）
扩展性	中等，支持简单扩展	低，单机为主	高，分布式架构
生产就绪	适合中小规模	研究/中小应用	企业级大规模
查询功能	基础向量检索+元数据过滤	高性能向量检索	向量+标量混合查询、复杂过滤
社区生态	较新但增长快	成熟，Meta维护	活跃，CNCF项目
使用复杂度	最简单	中等	较高（需运维知识）
最佳场景	快速原型、中小RAG项目	学术研究、单机高性能检索	大规模生产系统、企业级应用
GPU支持	有限	完整GPU加速	支持（需配置）
近似搜索	✅ HNSW等	丰富算法（IVF、PQ等）	✅ 多种算法

持久化存储向量数据库

Chroma 0.4.x+ 后，db都是自动持久化的

我在做这个实验的时候HuggingFaceEmbeddings好像出了点问题，访问不了

# coding=utf-8
# 向量数据库的持久化
from langchain_deepseek import ChatDeepSeek
from langchain_huggingface import HuggingFaceEmbeddings
from langchain.text_splitter import CharacterTextSplitter
from langchain_community.vectorstores import Chroma
from langchain.chains.qa_with_sources.retrieval import RetrievalQAWithSourcesChain
from dotenv import load_dotenv
from langchain.schema import Document
from langchain.document_loaders import JSONLoader
import json
from huggingface_hub import snapshot_download

load_dotenv()

# 定义大模型
llm = ChatDeepSeek(model="deepseek-chat")

# 定义元数据
def metadata_fun(record: dict, metadata: dict) -> dict:
    # record现在是notable_films数组中的每个电影对象
    metadata["title"] = record.get("title")
    metadata["year"] = record.get("year")
    metadata["director"] = record.get("director") if record.get("director") else "昆汀·塔伦蒂诺"
    metadata["main_actors"] = ", ".join(record.get("main_actors", []))
    metadata["genres"] = ", ".join(record.get("genres", []))
    metadata["accolades"] = ", ".join(record.get("accolades", []))
    metadata["runtime_minutes"] = record.get("runtime_minutes")
    return metadata

# 加载json数据
loader = JSONLoader(
    file_path='/root/python-langchain/data_example.json',
    jq_schema='.notable_films[]',    # 提取数组中每个元素(每部电影)
    json_lines=False,                # 并非JSON Lines格式
    text_content=False,              # 允许非字符串内容
    metadata_func=metadata_fun
)
document = loader.load()

# 切割器定义
text_splitter = CharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=20,
    add_start_index=True
)
# 文档切割
docs = text_splitter.split_documents(document)

#  
embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={'device': 'cpu'}
)

ids = [str(i) for i in range(1,len(docs) + 1)]

# 创建db，并且持久化存储到/root/chroma.db
db_to_disk = Chroma.from_documents(docs,embeddings,persist_directory="/root/chroma_db",ids=ids)
db_to_disk.persist()

# 从本地数据创建db
db_to_disk = Chroma(persist_directory="/root/chroma_db",embedding_function=embeddings)

# 进行查询
query = "低俗小说"
docs = db_to_disk.similarity_search(query)

print(json.loads(docs[0].page_content))

自定义嵌入模型

HuggingFaceEmbeddings噶了，我找了其他的嵌入式模型，做个自定义嵌入模型

from langchain_core.embeddings import Embeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import CharacterTextSplitter
from typing import List, Dict, Any
import requests
import os
from langchain.document_loaders import JSONLoader
from dotenv import load_dotenv
from langchain.chains.qa_with_sources.retrieval import RetrievalQAWithSourcesChain
from langchain_deepseek import ChatDeepSeek

load_dotenv()

# 自定义一个类来调用
class SiliconFlowEmbeddings(Embeddings):
    def __init__(self, api_url: str, api_token: str):
        self.api_url = api_url
        self.api_token = api_token
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",
            "Content-Type": "application/json"
        })
    api_key = os.getenv("SILICONFLOW_API_KEY")  

    def _get_embedding(self, text: str) -> List[float]:
        payload = {
            "model": "BAAI/bge-large-zh-v1.5",
            "input": text,
            "encoding_format": "float",
            "dimensions": 1024
        }

        try:
            response = self.session.post(self.api_url, json=payload, timeout=10)
            response.raise_for_status()
            return response.json()["data"][0]["embedding"]
        except requests.exceptions.RequestException as e:
            raise RuntimeError(f"API request failed: {str(e)}") from e
    
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        return [self._get_embedding(text) for text in texts]
    
    def embed_query(self, text: str) -> List[float]:
        return self._get_embedding(text)

# 初始化嵌入式模型实例
embeddings = SiliconFlowEmbeddings(
    api_url="https://api.siliconflow.cn/v1/embeddings",
    api_token=os.environ["SILICONFLOW_API_KEY"]
)

# 自定义元数据，方便进行查询
def metadata_fun(record: dict,metadata: dict) ->dict:
    # 将json中的信息映射到文档的元数据中
    metadata["title"] = record.get("title")
    metadata["year"] = record.get("year")
    return metadata

# 文档加载
loader = JSONLoader(
    file_path='data_example.json',
    jq_schema=".notable_films[]",
    content_key="title",   # 哪个字段应该成为文档的文本内容
    text_content=False,    # 告诉JSONLoader不要期望字符串
    metadata_func=metadata_fun
)
data = loader.load()
print(data)

# 文档分割
text_splitter = CharacterTextSplitter(chunk_size=500, chunk_overlap=50)
split_documents = text_splitter.split_documents(data)

# 创建向量数据库并持久化
db = Chroma.from_documents(
    documents=split_documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)

# 查询最相关文档
query = "低俗小说"
docs = db.similarity_search(query)
print(docs)

FAISS向量数据库

FAISS相关性分数
FAISS库合并
metadata查询

# 使用FAISS向量数据库
from langchain.vectorstores import FAISS
from langchain.document_loaders import TextLoader
from langchain.text_splitter import CharacterTextSplitter
from typing import List
from dotenv import load_dotenv
from langchain.schema import Document
# 设置项目根路径
import sys
import os
sys.path.append('/root/python-langchain')
# 导入自定义的嵌入式模型
from embeddings.siliconflow_embeddings import SiliconFlowEmbeddings

load_dotenv()

# 文档加载
loader = TextLoader("data_example.txt")
document = loader.load()

# 文档切割
text_splitter = CharacterTextSplitter(
    chunk_size=20,
    chunk_overlap=5
)
docs = text_splitter.split_documents(document)

# 嵌入式模型初始化db
embeddings = SiliconFlowEmbeddings()
db = FAISS.from_documents(docs,embeddings)

# 测试搜索
query = "图灵测试是谁提出的？"
docs = db.similarity_search(query)
print(docs[0])

# 查询相似度，相似度越接近于0越相似
docs_and_sorces = db.similarity_search_with_score(query)
print(docs_and_sorces[0])

# np.float32(1.2514029)

# FAISS的合并库
# 准别另一个文档
texts = '''在医疗健康领域，AI辅助疾病诊断、药物研发和个性化治疗；金融行业利用AI进行风险评估、欺诈检测和算法交易；制造业通过AI实现质量控制、预测性维护和供应链优化；教育领域采用智能辅导系统和个性化学习路径；交通运输领域发展出自动驾驶技术和智能交通管理系统。
人工智能的发展也带来诸多伦理挑战，包括算法偏见、数据隐私、就业影响和责任归属等问题。各国政府和国际组织正在制定相关法规和伦理框架，如欧盟的人工智能法案和联合国的人工智能伦理建议。
当前人工智能研究的前沿方向包括多模态学习、具身智能、神经符号人工智能和可解释人工智能等。
产业界和学术界正在合作推动人工智能技术的创新，主要参与机构包括谷歌、微软、亚马逊、百度、腾讯等科技公司，以及麻省理工学院、斯坦福大学、清华大学等研究机构。
人工智能工具生态系统的成熟也为该领域的快速发展提供了支撑，包括TensorFlow、PyTorch等开发框架和各类云人工智能服务平台。
'''
documents2 = [Document(page_content=texts)]

# 创建新db
db2 = FAISS.from_documents(documents2, embeddings)
# 合并db与db2
db.merge_from(db2)

# 输出db
print(db.docstore.__dict__)

使用metadata进行查询

# 使用metadata进行查询
from langchain.schema import Document
from langchain.vectorstores import FAISS
from dotenv import load_dotenv
from langchain.schema import Document
# 设置项目根路径
import sys
import os
sys.path.append('/root/python-langchain')
# 导入自定义的嵌入式模型
from embeddings.siliconflow_embeddings import SiliconFlowEmbeddings

load_dotenv()

list_of_document = [
    Document(page_content="你好",metadata=dict(page=1)),
    Document(page_content="雷军",metadata=dict(page=1)),
    Document(page_content="你好",metadata=dict(page=2)),
    Document(page_content="周鸿祎 ",metadata=dict(page=2)),
    Document(page_content="你好",metadata=dict(page=3)),
    Document(page_content="李彦宏",metadata=dict(page=3)),
    Document(page_content="你好",metadata=dict(page=4)),
    Document(page_content="丁磊",metadata=dict(page=4)),
]

embeddings = SiliconFlowEmbeddings()
db = FAISS.from_documents(list_of_document,embeddings)

# response = db.similarity_search_with_score("你好")
# for doc,sorce in response:
#     print(f"Content: {doc.page_content},Metadata: {doc.metadata}, Score: {sorce}") 
    
response = db.similarity_search_with_score("你好",filter=dict(page=1))
for doc,sorce in response:
    print(f"Content: {doc.page_content},Metadata: {doc.metadata}, Score: {sorce}") 

# Content: 你好,Metadata: {'page': 1}, Score: 0.0
# Content: 雷军,Metadata: {'page': 1}, Score: 1.1747143268585205