Day 11 / 共 20 天 · 阶段3 数据与 RAG

Embeddings 与 VectorStore:把文字变成坐标,把"意思相近"变成"距离很近"

Day09 我们有了 Document,Day10 把它切成了小块。今天回答 RAG 最核心的一问:怎么按"意思"而不是"关键词"找到相关的块?答案是两层抽象:Embeddings(把文本变成向量)+ VectorStore(存向量、按相似度搜)。我们直接读 core 里的接口定义,再用官方自带的 InMemoryVectorStore(一个纯 Python dict 实现)把"入库→检索"的全过程看得清清楚楚。

📍 你在 20 天里的位置(阶段3:数据与 RAG · D09-12)
S1 LCEL S2 模型·消息·提示 D09 Document D10 文本切分 D11 Embeddings/向量库 D12 Retriever/RAG S4 工具/Agent S5 进阶收官
💡 先用两个类比兜住今天 类比一:Embedding 像给每本书标一个"地图坐标"。图书馆几万本书,你不可能每本翻一遍。Embedding 模型读完一段文字,输出一串数字(向量),相当于把它钉在一张巨大的"语义地图"上——讲相近内容的文字,坐标就挨得近。"猫粮怎么选"和"如何给猫咪挑食物"关键词几乎不重叠,但在地图上是邻居。类比二:VectorStore 像挂这张地图的"档案馆"——入库时给每段文字算好坐标存起来(add_documents);来了问题,先算出问题的坐标,再在地图上找离它最近的 k 个点similarity_search)。今天所有源码都在讲这两件事。
L01

痛点:关键词搜索搜不到"意思相近"的内容

🤔 痛点你想做一个"公司制度问答机器人"。用户问"请假要提前几天申请?",而制度文档里写的是"员工休假需至少提前三个工作日提交 OA 流程"。一个共同的关键词都没有——"请假"vs"休假"、"申请"vs"提交流程"。用传统的 LIKE '%请假%' 或倒排索引,直接搜空。而且就算搜到了,几万字的文档也不能全塞给模型(又贵又超上下文)。怎么办?
💡 本质:把"语义匹配"降维成"几何最近邻"Embedding 模型是在海量语料上训出来的"读心器":它把任意文本映射成一个 N 维向量(比如 1536 个 float),且保证语义相近 → 向量夹角小。于是"找意思最相关的段落"这个 AI 难题,被降维成一个纯数学问题——算余弦相似度、排个序、取前 k 个。LangChain 在这里做的事是定标准:Embeddings 定义"怎么把文本变向量",VectorStore 定义"怎么存与搜",几十家厂商(OpenAI/FAISS/Chroma/Milvus…)都实现同一套接口,你的业务代码一行不用改就能换后端。
L02

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 的套路一模一样。
大白话这个接口翻译过来就是两句话:"给我一摞文件,我给每份盖个坐标章"(embed_documents);"给我一个问题,我告诉你它站在地图哪里"(embed_query)。任何厂商只要会这两下,就能接进 LangChain 的 RAG 体系。core 里还自带 DeterministicFakeEmbeddingembeddings/fake.py:70)——用文本的 hash 做随机种子生成"假向量",同一段文本永远得到同一个向量,写测试超好用。
L03

VectorStore 抽象:向量库的统一接口

"档案馆"这一半在 libs/core/langchain_core/vectorstores/base.py:43。类很大(一千多行),但骨干方法就这几个:

方法位置干什么
add_texts / add_documentsbase.py:46 / :234入库:把文本(算好向量后)存进去,返回 ids
similarity_search(query, k=4)base.py:361检索:返回与 query 最相似的 k 个 Document
similarity_search_with_scorebase.py:417同上,但带相似度分数
max_marginal_relevance_searchbase.py:659MMR 检索:相关性 + 多样性兼顾(去重相似结果)
from_texts(texts, embedding)base.py:848类方法:一步"建馆 + 入库"
as_retriever(**kwargs)base.py:905把自己包装成 Retriever(明天 Day12 的主角)

其中 add_documents 的默认实现(base.py:234)体现了"Document 只是 text+metadata 的壳":它拆出 page_contentmetadata 转调 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
L04

InMemoryVectorStore:入库就是"算向量 + 存 dict"

core 自带一个"教学级但可用于生产小数据"的实现:InMemoryVectorStorelibs/core/langchain_core/vectorstores/in_memory.py:34)。它的"仓库"就是一个普通字典(in_memory.py:169self.store: dict[str, dict[str, Any]] = {})。看入库 add_documentsin_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_documentsin_memory.py:226):唯一区别是 ② 换成 await self.embedding.aembed_documents(texts)——存 dict 本来就是内存操作,不需要异步。
L05

相似度检索:余弦相似度 + argsort 取 top-k

检索的心脏是 _similarity_search_with_score_by_vectorin_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。
语义地图:入库钉坐标,检索找最近邻 向量空间(语义地图) 休假需提前三日… OA 提交流程… 食堂菜单 报销制度 ❓请假提前几天 ① 入库:embed_documents → store ② 提问:embed_query 算问题坐标 ③ cosine_similarity + argsort 取 top-k ④ 返回最近的 k 个 Document(带原文)
图注:绿色点是入库的文档块,橙色是问题。虚线圈就是"top-k 最近邻"——关键词一个不沾也能搜到。
⚠️ 坑:入库和查询必须用同一个 embedding 模型不同模型产出的向量维度、空间都不一样。今天用 A 模型入库、明天换 B 模型查询,等于"用高德的坐标去查百度的地图"——结果全是乱的(维度不同直接报错,维度碰巧相同则悄悄给你错误结果,更可怕)。换 embedding 模型 = 全量重新入库。
L06

串起来 + 今日小结

📝 真实值:一次完整的"入库 → 检索" 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 本尊
明日预告 · Day 12:今天最后那行 as_retriever() 埋了个钩子——把向量库包装成 Retriever 之后,它就成了一个标准 Runnable,能直接用 | 接进 LCEL 链。明天读 BaseRetriever 的模板方法设计,并把 检索 + Prompt + 模型 组装成一条完整可跑的 RAG 问答链。
← Day 10 文本切分 Day 12 · Retriever 与 RAG 链 →