嵌入与向量库

本章你将学会

• 嵌入（embeddings）是什么：把「意思」翻译成一串数字，让两段话能像坐标上的两个点一样算远近；余弦相似度怎么读
• OllamaEmbeddings：本地跑 nomic-embed-text，embed_query 和 embed_documents 一个管查询一个管入库
• 批量嵌入与并发控制：JS 版用 p-limit 掐并发，Python 版用 asyncio.Semaphore + gather，收银台只开三个窗口
• InMemoryVectorStore：零依赖的内存向量库，from_documents / add_documents / similarity_search 一条龙
• 向量库的通用接口：今天用内存白板，明天换 FAISS / Chroma 只改构造那几行；顺手把持久化思路聊明白

回忆 RAG 三步曲：切割 → 嵌入 → 入库。06 章我们干完了第一步，把拾光书屋的读书笔记切成了一摞 Document。但有个根本问题一直悬着：计算机不认识字，只认识数。「睡前想读点温柔治愈的」和「有没有适合夜里安静看的温暖书」——人一眼看出是一个意思，机器眼里却是两串完全不同的字符，一个字都对不上。

本章就解决这件事，分两步走：

1. 嵌入：把文本交给嵌入模型，换回一串数字（向量），且意思相近的文本向量也相近——语义从此可计算；
2. 向量库：把这批向量连同原文存起来，等读者来提问时，把问题也变成向量，去库里找「坐标最近」的那几条笔记。

技术基线：Python + LangChain 1.x（langchain + langchain-ollama），本地 Ollama。注意本章要请出第二个模型：qwen3:4b 负责聊天（08 章才上场），nomic-embed-text 负责嵌入——各司其职，单独安装：

ollama pull nomic-embed-text

1. 嵌入是什么：给文字办一张「语义身份证」

作用：嵌入（embedding），也叫向量化，就是把一段文本喂给嵌入模型，换回一个固定长度的浮点数列表。nomic-embed-text 每次吐 768 个数。真正的魔法在于：意思相近的文本，向量也相近。

打个比方：想象书屋有一间 768 维的大书房（别硬想画面，反正比三维多了 763 个方向），每段笔记进来都会被安排到一个精确的坐标上——讲治愈童话的挤在东边一格书架，讲硬核科幻的聚在西边一格。语义成了空间位置，相似度成了距离。找「相关的笔记」，就是找「坐标挨得近的向量」。

最小示例：

from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="nomic-embed-text")

vec = embeddings.embed_query("睡前想读点温柔治愈的书")
print(len(vec))   # 768 —— nomic-embed-text 的固定维度
print(vec[:4])    # [0.0231, -0.1178, ...] 每个数都在 -1 ~ 1 附近晃

一串裸数字看不出名堂，重点从来不是某个数的大小，而是整串数字的方向。

相似度直觉：怎么量化「两个向量近不近」？最常用的是余弦相似度——不管向量多长，只看两个箭头的夹角：同向 = 1（很像），垂直 = 0（没关系），反向 = -1（意思拧着来）。手写只要几行：

import math


def cosine_similarity(a: list[float], b: list[float]) -> float:
    """余弦相似度：越大越像。1 同向，0 无关，-1 反向。"""
    dot = sum(x * y for x, y in zip(a, b))
    norm_a = math.sqrt(sum(x * x for x in a))
    norm_b = math.sqrt(sum(y * y for y in b))
    return dot / (norm_a * norm_b)

拿三句话试试水（完整可跑版在 demo/demo_embed.py）：

q1 = embeddings.embed_query("睡前想读点温柔治愈的书")
q2 = embeddings.embed_query("有没有适合夜里安静看的温暖书")
q3 = embeddings.embed_query("数据库索引为什么用 B+ 树")

print(cosine_similarity(q1, q2))  # 明显更高 —— 同一个意思
print(cosine_similarity(q1, q3))  # 明显更低 —— 八竿子打不着

两个同义问句的分数显著高于无关问句——这就是全部魔法的内核。后面 RAG 里说的「找相关笔记」，翻译成数学就是「找余弦相似度最高的那几个向量」。

常见坑：

• 嵌入模型不是聊天模型。nomic-embed-text 只会把文字变向量，你让它荐书它无能为力；qwen3:4b 会聊天但不产向量。两队人马，别排错岗。
• 不同模型的向量不可混用。A 模型和 B 模型就算都输出 768 维，也是两套完全不同的坐标系——库里存了 A 的向量，查询就必须继续用 A，换了模型等于换了城市，老地址全作废。
• 相似度看相对排序，别迷信绝对数值。0.82 比 0.45 像没问题，但「多少分以上算相关」没有普适及格线——不同模型、不同语料分布下数值整体都会漂移，排序才是硬信息。

2. OllamaEmbeddings：入库一把尺，查询一把尺

作用：OllamaEmbeddings 是 LangChain 对 Ollama 嵌入接口的封装。它实现了 Embeddings 这个基类定下的契约——基类就规定两组核心方法，所有厂商的嵌入类（OpenAIEmbeddings、OllamaEmbeddings……）签名都一样，所以向量库、检索器这些组件可以无感互换嵌入供应商。

签名：

from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(
    model: str = "nomic-embed-text",        # 模型名，需先 ollama pull
    base_url: str = "http://localhost:11434",
)

def embed_query(self, text: str) -> list[float]: ...            # 查询侧：一条进一条出
def embed_documents(self, texts: list[str]) -> list[list[float]]: ...  # 入库侧：一摞进一摞出
# 异步版各有一个：aembed_query / aembed_documents，签名相同

embed_query vs embed_documents 的分工：

一句话记忆：embed_documents 是编目员，把馆藏挨个贴上索书号（一次性、大批量）；embed_query 是查询台，读者每问一次，就把问题也编个号去对暗号（单条、随来随办）。

最小示例：

notes = [
    "小王子在 B-612 星球看了四十四次日落",
    "夜航船是张岱编的明代小百科",
    "山月记里李征变成了老虎",
]

vectors = embeddings.embed_documents(notes)              # 入库用：一次一摞
query_vec = embeddings.embed_query("哪个故事和星球有关")  # 查询用：一次一条

print(len(vectors), len(vectors[0]))  # 3 768 —— 三条文本，每条 768 维
print(len(query_vec))                 # 768

接口背后：Embeddings 基类。假如哪天要接一个 LangChain 没包装的嵌入服务，自己写一个类继承 Embeddings、实现这两个方法，任何向量库都会认它——「面向接口」的甜头：

from langchain_core.embeddings import Embeddings


class MyEmbeddings(Embeddings):
    """自定义嵌入类：实现两个方法，即可被所有 VectorStore 接受。"""

    def embed_documents(self, texts: list[str]) -> list[list[float]]:
        return [self._vec(t) for t in texts]

    def embed_query(self, text: str) -> list[float]:
        return self._vec(text)

    def _vec(self, text: str) -> list[float]:
        ...  # 你的向量化逻辑（调自家接口、或在测试里用哈希打桩）

常见坑：

• 忘了 pull 模型：报 model "nomic-embed-text" not found 之类，先 ollama pull nomic-embed-text，再确认 Ollama 进程活着。
• embed_documents 吃的是纯字符串，不是 Document。06 章切出来的 Document 要先取 .page_content：embed_documents([d.page_content for d in docs])。（下一节用向量库的 add_documents 时可以直接喂 Document，它帮你取。）
• 两个方法换个用「也能跑出向量」，但接口纪律要守：查询侧和文档侧是两个不同的优化方向（缓存、批处理策略都不一样），语义上就该各管各的。
• 事件循环里的代码用异步版 aembed_query / aembed_documents，别拿同步版把整个循环卡住——这个坑和 04 章讲过的 time.sleep 是一家的。

3. 批量嵌入与并发控制：p-limit 的 Python 姿势

作用：语料一多，嵌入的耗时立刻露馅。JS 版课程里做过计时实验：一条条串行请求，时间全花在排队上；用 p-limit 掐住「同时在飞的请求数」后大幅提速。Python 这边的对应物是标准库 asyncio：Semaphore 限流 + gather 收账。

先给写过 JS 的同学一张对照表：

一个好消息：OllamaEmbeddings.embed_documents(texts) 在 Ollama 这一侧本来就是一次请求批量（底层走 /api/embed，整个列表一起发），不像 JS 版手写时一条一请求。所以几十上百条笔记，直接一次调用完事，不用自己造轮子。

那什么时候还需要分批？ 语料上千条时：一是一次请求太大容易超时、内存吃紧；二是想把多批同时发出去，摊薄总耗时。这时「分批 + 并发上限」就派上用场——Semaphore 像收银台只开三个窗口，谁先到谁先办，但同时服务的顾客绝不会超过三个：

最小示例（分批 + Semaphore 控制同时在飞的批数；带计时和并发探针的完整版见 demo/demo_embed.py）：

import asyncio

BATCH_SIZE = 16        # 一批 16 条
MAX_CONCURRENCY = 3    # 同时最多 3 批在飞


async def embed_all(
    embeddings: OllamaEmbeddings, texts: list[str]
) -> list[list[float]]:
    """大语料分批并发嵌入：Semaphore 控并发，gather 保序收账。"""
    semaphore = asyncio.Semaphore(MAX_CONCURRENCY)

    async def embed_batch(batch: list[str]) -> list[list[float]]:
        async with semaphore:  # 领号：窗口满了就排队
            return await embeddings.aembed_documents(batch)

    batches = [texts[i : i + BATCH_SIZE] for i in range(0, len(texts), BATCH_SIZE)]
    results = await asyncio.gather(*(embed_batch(b) for b in batches))
    return [vec for batch in results for vec in batch]  # 摊平，顺序不乱

gather 保证所有批次到齐后一起交账，且结果顺序永远和传入顺序一致——第 i 个位置永远是第 i 批，先完成的后完成的都乖乖排队。

顺带一个并发探针的小技巧（JS 版课堂上也玩了）：在任务体里维护一个 active 计数器，进任务 +1、出任务 -1，随手记下 max_active，跑完看一眼是不是真的没超窗。写爬虫、调批量接口时这是最朴素的自我检查。

常见坑：

• gather 忘了 await。asyncio.gather(...) 返回的是可等待对象，不 await 等于没跑，只留下一句 RuntimeWarning——04 章讲过协程只创建不运行的老坑，这是它的批量版。
• Semaphore 在协程内部创建最稳。像上面那样写在函数体里，天然活在事件循环里，不会遇到循环绑定问题。
• 并发不是越大越好。本地 Ollama 内存和算力有限，开 16 路并发可能把它挤爆反而更慢，3~4 是个谦逊的起点，往上调要看机器脸色。
• 保序的是 gather，不是完成顺序。先回来的批次也排在原位；摊平结果时别「聪明地」按完成时间重排。

4. InMemoryVectorStore：把书屋笔记装上白板书架

作用：向量算出来了，得有地方存、有地方查。LangChain 内置 InMemoryVectorStore——一个纯内存的向量库：零额外依赖、即插即用，本地调试和快速演示的首选。它是一块白板书架：写上去飞快，但打烊一擦就没了（持久化见第 5 节）。

向量库大致分三类，本章先用第一种：

1. 内存型（InMemoryVectorStore）：快、零依赖、不落盘；
2. 本地自托管（FAISS、Chroma 落本地文件/目录）：数据在你手里；
3. 云托管（各种在线向量数据库）：省运维、要花钱。

整体工作流（正好是 RAG 三步曲的后两步）：

1. 把笔记包成 Document（06 章的产出，page_content + metadata）；
2. add_documents 入库——它内部自动调 embed_documents 把文本变向量存好；
3. similarity_search 检索——它内部自动调 embed_query 把问题变向量，再逐条算相似度排序。

签名：

from langchain_core.vectorstores import InMemoryVectorStore

@classmethod
def from_documents(
    documents: list[Document], embedding: Embeddings, **kwargs: Any
) -> Self: ...
# 建库 + 入库一步到位，等价于 InMemoryVectorStore(embedding) 再 add_documents(...)

def add_documents(
    self, documents: list[Document], ids: list[str] | None = None, **kwargs: Any
) -> list[str]: ...

def similarity_search(self, query: str, k: int = 4, **kwargs: Any) -> list[Document]: ...

def similarity_search_with_score(
    self, query: str, k: int = 4, **kwargs: Any
) -> list[tuple[Document, float]]: ...

最小示例（拾光书屋读书笔记节选，完整流程在 demo/demo_vectorstore.py）：

from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(model="nomic-embed-text")

notes = [
    Document(
        page_content="小王子在 B-612 星球看了四十四次日落，狐狸说驯养就是建立联系。",
        metadata={"book": "小王子", "tag": "治愈"},
    ),
    Document(
        page_content="夜航船是张岱编的明代小百科，从天文地理聊到草木鸟兽。",
        metadata={"book": "夜航船", "tag": "杂学"},
    ),
    Document(
        page_content="三体里的黑暗森林法则：宇宙是一座黑暗森林，每个文明都是带枪的猎人。",
        metadata={"book": "三体", "tag": "科幻"},
    ),
]

store = InMemoryVectorStore.from_documents(notes, embeddings)  # ① 入库

for doc in store.similarity_search("睡前想读点温柔的", k=2):  # ② 检索
    print(doc.metadata["book"], doc.page_content)

想知道「有多像」，用带分数的版本：

for doc, score in store.similarity_search_with_score("睡前想读点温柔的", k=3):
    print(f"{score:.4f}  {doc.metadata['book']}")
# 分数高（更靠 1）的排前面 —— InMemoryVectorStore 给的是余弦相似度，越大越像

店里又收到新笔记？不用推倒重建，add_documents 随时往书架上添：

store.add_documents([
    Document(
        page_content="山月记里李征变成了老虎，『我深怕自己本非美玉』读得人脊背发凉。",
        metadata={"book": "山月记", "tag": "经典"},
    ),
])

常见坑：

• 内存型 = 进程退出就清零。脚本跑完、服务重启，书架就擦干净了。演示无所谓，正经用请看第 5 节的搬家与持久化。
• k 默认是 4。不传就永远最多拿 4 条；语料多、想多看几条候选时记得显式传。
• 分数方向要认清：InMemoryVectorStore 给的是余弦相似度，越大越像；下一节换成 FAISS，默认给的是 L2 距离，越小越像——同一套代码换个库，排序方向可能整个掉头。
• from_documents 每调一次就是一座新库。把它写进循环等于反复建空库各塞一点；要累积请持有同一个 store 调 add_documents。
• 检索质量的上游是切块质量：块太大语义一锅粥，太小信息量不够——翻回 06 章看 chunk_size 的讨论，那里埋的因这里结果。

5. 通用接口与搬家路径：今天白板，明天 FAISS

作用：LangChain 把「向量库」抽象成 VectorStore 这套接口：from_documents / add_documents / similarity_search ……方法名全家族统一。InMemoryVectorStore、FAISS、Chroma 只是不同的实现——好比连锁书店不管开在哪，借书流程都一样，店员（你的业务代码）不用重新培训。这也是为什么第 2 节的 Embeddings 基类值得记住：接口统一，零件可换。

换库只改构造那几行：

# 今天：内存白板（零依赖，调试用）
from langchain_core.vectorstores import InMemoryVectorStore
store = InMemoryVectorStore.from_documents(notes, embeddings)

# 明天：FAISS（pip install faiss-cpu，向量落本地文件，检索更快）
from langchain_community.vectorstores import FAISS
store = FAISS.from_documents(notes, embeddings)   # 同一个方法名
store.save_local("book_index")                    # 落盘
store = FAISS.load_local(
    "book_index", embeddings, allow_dangerous_deserialization=True
)

# 后天：Chroma（pip install langchain-chroma，目录即数据库，自动持久化）
from langchain_chroma import Chroma
store = Chroma.from_documents(notes, embeddings, persist_directory="./chroma_db")

注意 from_documents(notes, embeddings) 这一行在三段代码里一个字不用改——业务代码里所有的 similarity_search 也原样能跑，选型差异被接口吃掉了。（小提示：FAISS 的集成在 langchain_community 包里，官方正在把集成逐步迁往独立包，动手装之前查一眼它最新的家在哪。）

内存库的持久化思路：InMemoryVectorStore 自己也留了后门——dump / load，把算好的向量导出成 JSON 文件，下次直接读回来：

store.dump("book_store.json")  # 存：向量 + 原文都在文件里

store = InMemoryVectorStore.load(  # 取：读回来就不用重新算嵌入
    "book_store.json", embeddings
)

省的是「入库那一趟的嵌入计算」——几百条笔记的嵌入时间不用再花一遍；但查询时嵌入模型仍必须在场（新问题还得变向量）。语料再大、要求再高，就换 FAISS / Chroma 这类专业户：FAISS 走索引文件（save_local / load_local），Chroma 走目录还能起服务。

常见坑：

• 换嵌入模型 = 全量重建。向量是模型坐标系里的点，换模型就是换坐标系，旧向量全部作废。dump 出来的文件也必须和「当初那个模型」配套使用，文件名里带上模型名是个好习惯。
• score 方向在换库时会变（重要的事说第二遍）：InMemory 是余弦相似度越大越好，FAISS 默认是 L2 距离越小越好。写「取 top k」时别把方向写死，换库先跑一次对拍。
• FAISS 的 load_local 要传 allow_dangerous_deserialization=True：底层用 pickle，反序列化不可信文件有安全风险——只加载自己亲手存的文件。
• 换库报 ImportError 先看依赖装没装：FAISS 要 pip install faiss-cpu，Chroma 要 pip install langchain-chroma，各是各的包。

6. 本章小结

一张表带走本章所有开关：

配套 demo（都在本章 demo/ 目录）：

• demo_embed.py：嵌入四连——单句看维度、批量嵌入、同义句 vs 无关句的余弦对比、串行 vs Semaphore 限流并发的计时与并发探针——需要本机 Ollama 已启动且拉取了 nomic-embed-text，文件头注释写了全部前提；连不上时自动打印提示并退出。
• demo_vectorstore.py：InMemoryVectorStore 全流程——十条书屋读书笔记入库 → 语义检索 → 带分数输出 → add_documents 追加新笔记再验证 → dump / load 持久化读档——同样依赖嵌入模型，离线自动降级。

到这里，RAG 三步曲走完两步：06 章切块、07 章入库。书架上的向量已经就位，下一章把「检索器」（as_retriever，提前剧透一句：LangChain 1.x 里想改条数要写 as_retriever(search_kwargs={"k": 2})，直接传 k=2 可不生效）接上大模型，让拾光书屋的 AI 真正开口回答读者的问题。

下一章：08 · 检索与 RAG