Embeddings 与 VectorStore:把文字变成坐标,把"意思相近"变成"距离很近"
Day09 我们有了 Document,Day10 把它切成了小块。今天回答 RAG 最核心的一问:怎么按"意思"而不是"关键词"找到相关的块?答案是两层抽象:Embeddings(把文本变成向量)+ VectorStore(存向量、按相似度搜)。我们直接读 core 里的接口定义,再用官方自带的 InMemoryVectorStore(一个纯 Python dict 实现)把"入库→检索"的全过程看得清清楚楚。
add_documents);来了问题,先算出问题的坐标,再在地图上找离它最近的 k 个点(similarity_search)。今天所有源码都在讲这两件事。痛点:关键词搜索搜不到"意思相近"的内容
LIKE '%请假%' 或倒排索引,直接搜空。而且就算搜到了,几万字的文档也不能全塞给模型(又贵又超上下文)。怎么办?Embeddings 定义"怎么把文本变向量",VectorStore 定义"怎么存与搜",几十家厂商(OpenAI/FAISS/Chroma/Milvus…)都实现同一套接口,你的业务代码一行不用改就能换后端。Embeddings 抽象:一共就 4 个方法
先看"变坐标"这一半。整个抽象在 libs/core/langchain_core/embeddings/embeddings.py:8,出奇地小:
# libs/core/langchain_core/embeddings/embeddings.py:8
class Embeddings(ABC):
"""Interface for embedding models. ..."""
@abstractmethod
def embed_documents(self, texts: list[str]) -> list[list[float]]: # :37
"""Embed search docs.""" # 一批文档 → 一批向量
@abstractmethod
def embed_query(self, text: str) -> list[float]: # :48
"""Embed query text.""" # 一个问题 → 一个向量
async def aembed_documents(self, texts: list[str]) -> list[list[float]]: # :58
return await run_in_executor(None, self.embed_documents, texts) # 默认:丢线程池跑同步版
async def aembed_query(self, text: str) -> list[float]: # :69
return await run_in_executor(None, self.embed_query, text)
embed_documents(texts)入库用:一次给一批文本算坐标,返回 list[list[float]]。批量是刻意的——调 embedding API 按条算太慢太贵,批着发划算。embed_query(text)查询用:给一个问题算坐标,返回单个向量。类文档(:25 附近)专门解释:多数模型两者算法相同,但抽象上分开——有些模型对"查询"和"文档"用不同的编码方式(如加不同前缀),所以留了两个口子。aembed_*(默认实现)异步版不是抽象方法:默认用 run_in_executor 把同步版丢进线程池。子类只须实现 2 个同步方法就能跑,有原生异步 SDK 的再覆写异步版提速——和 Day05 ChatModel 的套路一模一样。DeterministicFakeEmbedding(embeddings/fake.py:70)——用文本的 hash 做随机种子生成"假向量",同一段文本永远得到同一个向量,写测试超好用。VectorStore 抽象:向量库的统一接口
"档案馆"这一半在 libs/core/langchain_core/vectorstores/base.py:43。类很大(一千多行),但骨干方法就这几个:
| 方法 | 位置 | 干什么 |
|---|---|---|
add_texts / add_documents | base.py:46 / :234 | 入库:把文本(算好向量后)存进去,返回 ids |
similarity_search(query, k=4) | base.py:361 | 检索:返回与 query 最相似的 k 个 Document |
similarity_search_with_score | base.py:417 | 同上,但带相似度分数 |
max_marginal_relevance_search | base.py:659 | MMR 检索:相关性 + 多样性兼顾(去重相似结果) |
from_texts(texts, embedding) | base.py:848 | 类方法:一步"建馆 + 入库" |
as_retriever(**kwargs) | base.py:905 | 把自己包装成 Retriever(明天 Day12 的主角) |
其中 add_documents 的默认实现(base.py:234)体现了"Document 只是 text+metadata 的壳":它拆出 page_content 和 metadata 转调 add_texts。而 as_retriever 的实现只有两行(base.py:960-961):
# libs/core/langchain_core/vectorstores/base.py:960
tags = kwargs.pop("tags", None) or [*self._get_retriever_tags()]
return VectorStoreRetriever(vectorstore=self, tags=tags, **kwargs) # 明天 Day12 拆它
VectorStore 自己不做任何向量计算——它只是一份"档案馆服务公约":会入库、会按相似度搜、能升级成 Retriever。FAISS、Chroma、PGVector、Milvus……全都签了这份公约(继承它),所以你把 InMemoryVectorStore 换成 Chroma,上层代码零改动。这和 Day05 里"几十家聊天模型共用 BaseChatModel"是同一个设计哲学:抽象在 core,实现在 partners。InMemoryVectorStore:入库就是"算向量 + 存 dict"
core 自带一个"教学级但可用于生产小数据"的实现:InMemoryVectorStore(libs/core/langchain_core/vectorstores/in_memory.py:34)。它的"仓库"就是一个普通字典(in_memory.py:169:self.store: dict[str, dict[str, Any]] = {})。看入库 add_documents(in_memory.py:188):
# libs/core/langchain_core/vectorstores/in_memory.py:188
def add_documents(self, documents, ids=None, **kwargs) -> list[str]:
texts = [doc.page_content for doc in documents] # ① 取出纯文本
vectors = self.embedding.embed_documents(texts) # ② ★批量算向量(调 Embeddings 接口)
if ids and len(ids) != len(texts):
raise ValueError(...) # ids 数量必须对得上
id_iterator = iter(ids) if ids else iter(doc.id for doc in documents)
ids_ = []
for doc, vector in zip(documents, vectors, strict=False):
doc_id = next(id_iterator)
doc_id_ = doc_id or str(uuid.uuid4()) # ③ 没给 id 就发一个 uuid
ids_.append(doc_id_)
self.store[doc_id_] = { # ④ ★存进字典:一条记录四个字段
"id": doc_id_,
"vector": vector, # 坐标
"text": doc.page_content, # 原文
"metadata": doc.metadata, # 元数据(来源、页码…)
}
return ids_
self.embedding.embed_documents★入库的唯一"智能"环节。InMemoryVectorStore 构造时要传入一个 Embeddings 实例——L02 的接口在这里被消费。向量库自己不懂语义,语义全靠 embedding 模型。self.store[doc_id_] = {...}所谓"向量库",最朴素的形态就是每条记录 = 原文 + 坐标 + 元数据。存原文是因为检索完要把文字(不是数字)还给你拼 prompt。uuid4 兜底没显式给 id、Document 也没带 id,就发 uuid。有 id 则支持"同 id 覆盖"——天然当 upsert 用。aadd_documents(in_memory.py:226):唯一区别是 ② 换成 await self.embedding.aembed_documents(texts)——存 dict 本来就是内存操作,不需要异步。相似度检索:余弦相似度 + argsort 取 top-k
检索的心脏是 _similarity_search_with_score_by_vector(in_memory.py:291)——所有 similarity_search* 变体最后都汇到这里:
# libs/core/langchain_core/vectorstores/in_memory.py:291
def _similarity_search_with_score_by_vector(self, embedding, k=4, filter=None):
docs = list(self.store.values()) # ① 取出全部记录
if filter is not None: # ② 可选:先按元数据过滤
docs = [doc for doc in docs
if filter(Document(id=doc["id"], page_content=doc["text"],
metadata=doc["metadata"]))]
if not docs:
return []
similarity = cosine_similarity( # ③ ★问题向量 vs 所有文档向量
[embedding], [doc["vector"] for doc in docs])[0] # 一次矩阵运算全算完
top_k_idx = similarity.argsort()[::-1][:k] # ④ ★按相似度降序,取前 k 个下标
return [(Document(id=doc_dict["id"], page_content=doc_dict["text"],
metadata=doc_dict["metadata"]),
float(similarity[idx].item()), # ⑤ 附上分数
doc_dict["vector"])
for idx in top_k_idx if (doc_dict := docs[idx])]
cosine_similarity★真正的数学在 vectorstores/utils.py:39 的 _cosine_similarity:numpy 算"两个向量夹角的余弦",值越接近 1 越相似。"意思相近"到这一步彻底变成了纯几何。argsort()[::-1][:k]numpy 排序取前 k。注意这是 O(N) 暴力扫全库——每次检索都和所有向量比一遍。几千条毫无压力;上百万条就得换 FAISS/Milvus 这类带近似最近邻(ANN)索引的实现,而接口不变。filter=可调用对象过滤条件是一个"收 Document 返回 bool"的函数,比如 lambda d: d.metadata["source"] == "hr.md"。先过滤再算相似度,实现"只在人事制度里搜"。外层入口similarity_search(query)(in_memory.py:404)就是先 embed_query(query) 算问题坐标,再调本函数、剥掉分数只留 Document。串起来 + 今日小结
store = InMemoryVectorStore(embedding=OpenAIEmbeddings(model="text-embedding-3-small")),然后 store.add_documents([Document(page_content="员工休假需至少提前三个工作日提交 OA 流程", metadata={"source": "hr.md"}), Document(page_content="午餐补贴每日 25 元", metadata={"source": "welfare.md"})]) → 内部调 embed_documents 得到 2 个 1536 维向量,self.store 里多出 2 条记录(key 是 uuid)。接着 store.similarity_search_with_score("请假要提前几天申请?", k=1) → embed_query 算问题向量 → 余弦相似度约 [0.71, 0.18] → argsort 取第 0 条 → 返回 [(Document(page_content="员工休假需至少提前三个工作日提交 OA 流程", metadata={"source": "hr.md"}), 0.71)]。零关键词重叠,照样命中。👶 小白:为什么要分 embed_documents 和 embed_query 两个方法?都是"文本变向量",一个不就够了?
👨🏫 老师:多数时候确实一样(很多实现里 embed_query 就是 embed_documents([text])[0],比如 embeddings/fake.py:66 的 FakeEmbeddings 就这么干)。但有些模型是"非对称检索"设计——文档和问题要加不同的指令前缀编码,效果才最好。抽象层把两个口子都留出来,实现者才有发挥空间。这是接口设计的常见考量:按"角色"建模,而不是按"当前实现碰巧相同"建模。
🧠 今天你应该能回答
- Embedding 解决什么问题?(把"语义相近"变成"向量距离近",让语义搜索可计算)
Embeddings抽象有哪几个方法?(embed_documents / embed_query + 两个默认线程池实现的异步版)VectorStore的三大能力?(add 入库 / similarity_search 检索 / as_retriever 升级成检索器)InMemoryVectorStore的存储结构?(一个 dict:id → {id, vector, text, metadata})- 它的检索是怎么算的?(embed_query → 与全库逐条余弦相似度 → argsort 取 top-k,O(N) 暴力)
- 为什么换 embedding 模型要全量重建库?(不同模型的向量空间不互通)
✋ 10 分钟动手
cd /Users/bitmart/work/codes/github/AI_WORK/langchain
# 1. Embeddings 抽象:就这么小
sed -n '8,80p' libs/core/langchain_core/embeddings/embeddings.py
# 2. VectorStore 骨干方法
grep -n "def add_texts\|def similarity_search\|def as_retriever\|def from_texts" \
libs/core/langchain_core/vectorstores/base.py
# 3. InMemory 实现:入库 + 检索
sed -n '188,222p' libs/core/langchain_core/vectorstores/in_memory.py # add_documents
sed -n '291,333p' libs/core/langchain_core/vectorstores/in_memory.py # 余弦 top-k
sed -n '39,60p' libs/core/langchain_core/vectorstores/utils.py # cosine_similarity 本尊
as_retriever() 埋了个钩子——把向量库包装成 Retriever 之后,它就成了一个标准 Runnable,能直接用 | 接进 LCEL 链。明天读 BaseRetriever 的模板方法设计,并把 检索 + Prompt + 模型 组装成一条完整可跑的 RAG 问答链。