LangChain Embedding与向量数据库

摘要

文档切好了得存起来，还得能根据问题找出来。Embedding把文字转成向量，向量数据库负责存和查。本文讲本地中文Embedding模型的使用、Chroma/Milvus的基本操作，以及怎么用元数据过滤缩小检索范围。

一、为什么需要Embedding

用户问"公司怎么处理迟到"，文档里写的是"员工应按时出勤，违者按制度处理"。关键词不一样，意思一样。

普通关键词搜不到，得靠语义检索。

Embedding的作用就是把文本转成一组数字（向量）。意思越接近的文本，向量距离越近。

python

# 伪代码示意
"人工智能" → [0.12, -0.35, 0.78, …]
"机器学习" → [0.11, -0.33, 0.80, …] # 距离近
"红烧肉" → [0.89, 0.45, -0.23, …] # 距离远

二、Embedding和聊天模型的区别

知识库项目中两者配合：Embedding找相关文档，聊天模型读文档后生成回答。

三、安装依赖

bash

pip install langchain-community chromadb pymilvus dashscope

使用本地中文Embedding模型（bge-small-zh），或者阿里云DashScope的text-embedding-v4。课堂用本地模型，不消耗API额度。

四、案例一：把文本转换成向量

python

from langchain_community.embeddings import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)

texts = ["人工智能正在改变世界", "机器学习是AI的一个分支"]

# 批量生成文档向量
vectors = embeddings.embed_documents(texts)
print(len(vectors))
print(len(vectors[0])) # 向量维度

# 生成查询向量
query_vec = embeddings.embed_query("什么是AI")
print(len(query_vec))

embed_documents用于文档列表，embed_query用于用户问题。两者使用的模型必须一致。

五、案例二：批量生成文本向量

在model_factory.py中封装Embedding方法：

python

from langchain_community.embeddings import HuggingFaceEmbeddings

def get_embeddings(model_name="BAAI/bge-small-zh-v1.5"):
return HuggingFaceEmbeddings(
model_name=model_name,
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)

使用：

python

from utils.model_factory import get_embeddings

embeddings = get_embeddings()

chunks = ["文档块1", "文档块2", "文档块3"]
vectors = embeddings.embed_documents(chunks)

query = "用户问题"
query_vector = embeddings.embed_query(query)

六、什么是向量数据库

文本转成向量后需要存起来，还要支持根据向量找相似内容。这就是向量数据库。

向量数据库保存三类数据：

文本内容（文档块正文）
向量（数字列表）
元数据（文件名、页码、分类等）

检索流程：

用户问题 → 转成查询向量 → 向量库搜索最相似的文档向量 → 返回对应的文本和元数据

七、向量数据库选型

课程用Chroma做演示，Milvus做案例。

八、Chroma向量数据库

python

from langchain_chroma import Chroma
from utils.model_factory import get_embeddings

embeddings = get_embeddings()

# 从文档创建向量库
vector_store = Chroma.from_documents(
documents=chunks, # Document列表
embedding=embeddings,
persist_directory="./chroma_db" # 持久化目录
)

# 检索
results = vector_store.similarity_search("员工迟到怎么处理", k=3)
for doc in results:
print(doc.page_content)
print(doc.metadata)

persist_directory指定持久化目录，下次启动时可以从磁盘加载，不用重新建库。

加载已有向量库：

python

vector_store = Chroma(
persist_directory="./chroma_db",
embedding_function=embeddings
)

九、相似度分数

similarity_search只返回文档，不返回分数。用similarity_search_with_score查看相似度：

python

results = vector_store.similarity_search_with_score("员工迟到怎么处理", k=3)

for doc, score in results:
print(f"分数：{score}")
print(f"内容：{doc.page_content[:100]}")
print("-" * 40)

不同向量库的分数含义不同，不要只凭固定数字做判断。结合Top K结果、文档内容、测试集效果综合评估。

十、Milvus向量数据库

Milvus是开源分布式向量数据库，适合大规模生产环境。

安装：

bash

pip install pymilvus langchain-milvus

确保Milvus服务已启动（Docker或本地部署）。

python

from langchain_milvus import Milvus
from utils.model_factory import get_embeddings

embeddings = get_embeddings()

# 连接Milvus并写入
vector_store = Milvus.from_documents(
documents=chunks,
embedding=embeddings,
connection_args={"host": "localhost", "port": "19530"},
collection_name="knowledge_base",
drop_old=True,
auto_id=True,
index_params={
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 128}
}
)

# 检索
results = vector_store.similarity_search("员工迟到怎么处理", k=3)

index_params参数说明：

index_type="HNSW"：分层导航小世界图，检索速度快，适合百万级以内向量。
metric_type="COSINE"：余弦相似度，中文Embedding首选。
M=16：每个节点最大连接邻居数，越大内存占用越高。
efConstruction=128：构建时候选邻居数，越大索引质量越好。

带分数的检索：

python

results = vector_store.similarity_search_with_score("员工迟到怎么处理", k=3)

返回的score是距离值，越小表示越相似。

十一、元数据过滤

知识库可能有多个分类，检索时可以按元数据过滤。

python

# 只检索人事制度
results = vector_store.similarity_search(
"迟到怎么处理",
k=3,
expr='file_name == "employee_handbook.txt"'
)

# 只检索PDF文件
results = vector_store.similarity_search(
"产品参数",
k=3,
expr='file_type == "pdf"'
)

注意Milvus使用expr表达式语法，不是filter参数。

元数据过滤适合：只检索某个部门的资料、只检索某个产品线的手册、只检索某个文件来源。

十二、企业案例：知识库检索索引

项目结构：

text

knowledge_base/
├── hr/
│ └── employee_handbook.txt
├── customer_service/
│ └── refund_policy.md
└── product/
└── product_manual.pdf

build_index.py
search_knowledge.py
document_loader.py
embedding_factory.py

embedding_factory.py：

python

from langchain_community.embeddings import HuggingFaceEmbeddings

def get_embeddings():
return HuggingFaceEmbeddings(
model_name="BAAI/bge-small-zh-v1.5",
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True}
)

document_loader.py：

python

from pathlib import Path
from langchain_community.document_loaders import TextLoader, PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

def load_and_split(directory_path, chunk_size=500, chunk_overlap=50):
docs = []
directory = Path(directory_path)

for file_path in sorted(directory.rglob("*")):
if not file_path.is_file():
continue
suffix = file_path.suffix.lower()
try:
if suffix == ".txt":
loader = TextLoader(str(file_path), encoding="utf-8")
elif suffix == ".md":
loader = TextLoader(str(file_path), encoding="utf-8")
elif suffix == ".pdf":
loader = PyPDFLoader(str(file_path))
else:
continue
loaded = loader.load()
for doc in loaded:
doc.metadata["file_name"] = file_path.name
doc.metadata["file_type"] = suffix[1:]
doc.metadata["category"] = file_path.parent.name
docs.extend(loaded)
except Exception as e:
print(f"加载失败：{file_path.name}，错误：{e}")

splitter = RecursiveCharacterTextSplitter(
chunk_size=chunk_size,
chunk_overlap=chunk_overlap
)
return splitter.split_documents(docs)

build_index.py：

python

from langchain_milvus import Milvus
from embedding_factory import get_embeddings
from document_loader import load_and_split

print("加载并切分文档…")
chunks = load_and_split("knowledge_base")
print(f"生成{len(chunks)}个文档块")

print("写入Milvus…")
embeddings = get_embeddings()

vector_store = Milvus.from_documents(
documents=chunks,
embedding=embeddings,
connection_args={"host": "localhost", "port": "19530"},
collection_name="knowledge_base",
drop_old=True,
index_params={
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 128}
}
)

print("索引构建完成")

search_knowledge.py：

python

from langchain_milvus import Milvus
from embedding_factory import get_embeddings

embeddings = get_embeddings()

vector_store = Milvus(
embedding_function=embeddings,
connection_args={"host": "localhost", "port": "19530"},
collection_name="knowledge_base"
)

while True:
question = input("\\n请输入问题（输入q退出）：")
if question.lower() == "q":
break

results = vector_store.similarity_search_with_score(question, k=3)

print(f"\\n相关文档块：")
for i, (doc, score) in enumerate(results, 1):
print(f"{i}. 来源：{doc.metadata.get('file_name')}")
print(f" 分类：{doc.metadata.get('category')}")
print(f" 相似度：{score:.4f}")
print(f" 内容：{doc.page_content[:150]}…")
print()

十三、检索效果调优

不准的话检查这些：

文档块太短或太长：调整chunk_size
上下文被切断：增大chunk_overlap
标题和正文分离：调整分隔符
查询和文档表达差异大：增加同义表达
Embedding模型不适合中文：换模型

准备一组测试问题，逐个观察Top 3结果是否相关。这是最早期的评估方式。

常见问题

DeepSeek能做Embedding吗

DeepSeek是聊天模型，不做Embedding。本课程用本地中文模型，不消耗API额度。

为什么要用同一个Embedding模型

建库和查询必须用同一个模型，否则向量空间不一致，检索效果差。

相似度分数阈值设多少合适

不要写死。不同模型、不同向量库的分数含义不同。先观察真实问题结果再定。

第一次运行慢

首次运行会下载模型文件到本地缓存，后续就快了。