本章必须会用#

• 导入 SentenceTransformer。
• 用模型 ID 创建模型对象。
• 调用 encode() 得到向量。
• 检查结果的类型、形状和一小段数值。
• 解释下载缓存与进程内模型对象的区别。
• 解释为什么建库和查询必须使用兼容的向量方案。

本章必须看懂#

• 项目如何检测设备、选择模型和复用模型对象。
• 单条编码与批量编码的区别。
• Hugging Face Hub、sentence-transformers、transformers 和 huggingface_hub 的边界。

本章只需识别#

• CUDA、MPS、DirectML 的完整设备检测逻辑。
• threading.Lock 的并发加载保护。
• LoRA、量化、训练、分布式推理。
• AutoTokenizer、AutoModel 和手写池化。

四个对象分别是什么#

这里的“类”可以先理解为创建对象的模板：

1
model = SentenceTransformer(MODEL_ID, device="cpu")

• SentenceTransformer 是类，规定这类对象有哪些初始化规则和方法。
• SentenceTransformer(...) 执行创建过程，并加载 MODEL_ID 指定的模型。
• model 是创建出来的实例，也就是当前程序里真正可以调用的模型对象。
• model.encode(TEXT) 调用实例的方法，把文本转换成向量。

这里的“句向量模型”来自英文 sentence embedding model。其中的“句”表示模型把一段文本作为一个整体来表达语义，不严格要求输入必须是带句号的完整句子。它可以处理：

• 一个问题："退款需要几天内申请？"
• 一个标题："退款规则"
• 一个短语："Python 异步生成器"
• 一个段落或 RAG chunk

“句向量模型”也不是所有“向量模型”的全称，而是文本向量模型的一类。图片、音频也可以分别通过视觉或音频 Embedding 模型转换成向量。

(768,) 表示一条包含 768 个数字的一维向量。不要逐个解释浮点数；检索依靠完整向量之间的相对方向或距离。

调用链#

1
MODEL_ID: str
2
  -> SentenceTransformer(MODEL_ID, device="cpu")
3
  -> model: SentenceTransformer
4
  -> model.encode(TEXT)
5
  -> embedding: numpy.ndarray

为什么第一次慢，第二次仍不是“零加载”#

第一次：

1
检查本地缓存
2
  -> 缓存没有模型文件
3
  -> 从 Hub 下载配置和权重
4
  -> 把权重加载进内存
5
  -> 创建 model 对象

下一次启动新进程：

1
检查本地缓存
2
  -> 直接读取已有文件，不必重新下载
3
  -> 仍要把权重加载进内存
4
  -> 仍要创建新的 model 对象

所以：

• 下载缓存保存在磁盘，可跨进程复用。
• 模型对象存在于当前 Python 进程的内存中，进程结束就消失。

为什么它会默认从 Hugging Face 下载#

SentenceTransformer 的第一个参数正式名称是 model_name_or_path，意思是“模型名称或者本地路径”。它会先判断你传入的字符串是什么：

1
SentenceTransformer("./models/my-bge")
2
  -> 这是本地目录
3
  -> 从本地读取模型
4

5
SentenceTransformer("BAAI/bge-base-zh-v1.5")
6
  -> 不是本地目录
7
  -> 按 Hugging Face Hub 的模型 ID 处理
8
  -> 内部通过 huggingface_hub 下载文件
9
  -> 缓存到本地后加载

因此，这是 sentence-transformers 库的默认行为，不是项目在其他地方配置了 huggingface.co。当前安装版本的内部调用关系可以压缩为：

1
SentenceTransformer(model_name_or_path)
2
  -> 检查本地路径
3
  -> huggingface_hub.snapshot_download(repo_id=模型 ID)
4
  -> 得到本地缓存目录
5
  -> 加载模型配置和权重

可以配置的部分包括：

1
model = SentenceTransformer(
2
    "BAAI/bge-base-zh-v1.5",
3
    cache_folder="./model-cache",  # 修改缓存目录
4
    revision="某个分支、标签或提交",  # 固定模型版本
5
    token="访问私有模型所需的令牌",
6
    local_files_only=False,       # True 时禁止联网，只查本地文件
7
)

如果直接传入本地目录，就不会把它当成 Hub 模型 ID：

1
model = SentenceTransformer("/Users/enkidu/models/bge-base-zh-v1.5")

准确规则：非本地路径默认按 Hugging Face Hub 模型 ID 解析；本地路径直接从磁盘加载。

sentence-transformers#

• 来源：Poetry 管理的第三方包，也是项目直接依赖。
• 本章对象：SentenceTransformer 类。
• 本章方法：model.encode(...)。
• 业务作用：把句子或文档转换成适合相似度计算的向量。
• 不负责：保存向量、搜索向量库、生成聊天答案。
• 掌握要求：会用。

1
from sentence_transformers import SentenceTransformer

transformers#

• 来源：Hugging Face 的通用模型调用库，目前由 sentence-transformers 的依赖链带入。
• 当前作用：SentenceTransformer 在内部使用它加载底层 Transformer 模型和文本处理组件。
• 为什么第一关没看到：你调用的是外层封装，内部调用不会自动出现在你的代码里。
• 不负责：连接向量数据库，也不负责保存 RAG chunk。
• 本章掌握要求：只需看懂间接调用位置；后面的 pipeline() 只是代码形状预览，不要求现在独立使用。

概念上的内部代码形状类似：

1
# 这是底层原理预览，不是本章要求你编写的代码。
2
from transformers import AutoModel, AutoTokenizer
3

4

5
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
6
base_model = AutoModel.from_pretrained(MODEL_ID)

sentence-transformers 会继续处理池化等步骤，让你最终只需调用 encode()。

huggingface_hub#

• 来源：连接 Hugging Face Hub 的官方 Python 客户端，也是间接依赖。
• 当前作用：当 MODEL_ID 不是本地路径时，负责查找、下载和缓存模型文件。
• 为什么第一关没看到：SentenceTransformer(...) 在内部调用它。
• 不负责：拿到文件后执行模型数学计算；计算由模型框架完成。
• 本章掌握要求：只需看懂下载调用位置；暂时不要求直接创建 InferenceClient。

当前安装版本内部使用的代码形状类似：

1
# 这是底层原理预览，不是本章要求你编写的代码。
2
from huggingface_hub import snapshot_download
3

4

5
snapshot_download(repo_id=MODEL_ID)

它返回缓存到本机的模型目录，后续再由上层库加载。

所以第一关没有少写代码。你当前真正需要写的是：

1
model = SentenceTransformer(MODEL_ID)
2
embedding = model.encode(TEXT)

另外两层由库自动执行。以后如果业务代码开始直接 import transformers 或 huggingface_hub，再把它们明确写入 pyproject.toml 并正式学习相应 API。

把第二关代码按“谁调用谁”读一遍#

不要把上面三段代码看成三份并列程序。它们是一层包调用下一层包：

1
你写的 model = SentenceTransformer(MODEL_ID)
2
  -> sentence-transformers 内部请求 huggingface_hub 准备模型文件
3
  -> sentence-transformers 内部请求 transformers 创建底层模型组件
4
  -> sentence-transformers 组合组件并返回 model

1
from sentence_transformers import SentenceTransformer
2

3
model = SentenceTransformer(MODEL_ID)
4
embedding = model.encode(TEXT)

1
from transformers import AutoModel, AutoTokenizer
2

3
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
4
base_model = AutoModel.from_pretrained(MODEL_ID)

token ID 到底是什么#

token ID 是 token 在当前模型词表中的整数编号。它不是原文、不是词义向量，也不是相似度分数，只是模型查找词表和处理输入时使用的数字索引。

Tokenizer 是什么#

Tokenizer 可以直译为“分词器”，但它不只做切分。它是模型前面的文本预处理对象，主要负责：

1. 把原始文本拆成 token。
2. 把 token 映射成 token ID。
3. 按模型要求补充特殊 token、截断过长输入或进行填充。
4. 把这些结果整理成模型可以接收的输入数据。

最小调用形状如下：

1
from transformers import AutoTokenizer
2

3

4
tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
5
encoded = tokenizer("退款需要几天内申请？")
6

7
print(encoded["input_ids"])

这里的 encoded 通常是一个字典：input_ids 保存 token ID；实际批量或补齐输入时还可能包含 attention_mask 等字段。Tokenizer 负责把文字变成数字输入，但不负责执行 Transformer 的神经网络计算；真正的模型计算由 AutoModel 或 SentenceTransformer 完成。

1
Tokenizer：文本 -> token -> token ID -> 模型输入字典
2
AutoModel：模型输入字典 -> 隐藏表示

可以用一个简化例子理解：

1
原始文本："退款需要几天"
2
    ↓ tokenizer 拆分
3
tokens： ["退款", "需要", "几天"]
4
    ↓ 查当前模型的词表
5
token IDs： [1205, 884, 3912]   # 仅为示意，真实编号由模型决定
6
    ↓ 交给 Transformer
7
隐藏表示：一组浮点数
8
    ↓ 池化
9
句向量：一条代表整段文本的向量

池化是什么#

Transformer 会为每个 token 产生一条隐藏向量。假设一段文本被拆成 3 个 token，并且每条隐藏向量只有 2 个数字：

1
token 1 -> [1, 2]
2
token 2 -> [3, 4]
3
token 3 -> [5, 6]

向量数据库通常需要“这一整段文本对应一条向量”，所以需要把 3 条向量汇总成 1 条。最简单的平均池化是逐列求平均：

1
([1, 2] + [3, 4] + [5, 6]) / 3
2
  -> [3, 4]

这就是池化的核心：把 token 级表示聚合成文本级表示。它不是对 token ID 求平均，也不是把向量维度随意压小。

一个 token 不一定等于一个词#

token 是 Tokenizer 规定的最小文本片段，可能是：

1
完整词：      "退款"
2
单个汉字：    "退"
3
英文子词：    "play" + "##ing"
4
标点：        "？"
5
特殊标记：    "[CLS]"、"[SEP]"

不同模型的 Tokenizer 词表不同，所以同一段文字可能被拆成不同数量的 token。这里的“多个词”更准确应说成“多个 token”。

为什么要汇总多个 token#

因为模型面对的是一段长度不固定的文本，而向量检索需要一条长度固定的向量：

1
短文本：2 个 token  -> 2 条隐藏向量 -> 1 条 768 维句向量
2
长文本：20 个 token -> 20 条隐藏向量 -> 1 条 768 维句向量

平均池化能把不同长度的 token 序列汇总成相同维度的结果。更重要的是，隐藏向量不是原始词典编号：经过 Transformer 的上下文计算后，每个 token 的表示已经受到周围文本影响，因此汇总它们是在形成整段文本的语义表示。

实际实现不会把填充位置无条件算进去，通常会通过 attention_mask 只聚合真实 token。也不要把平均池化当成唯一正确答案；模型可能配置为 Mean、CLS 或 Max Pooling，具体策略要看模型配置。

常见池化方式有：

• Mean Pooling：对多个 token 的隐藏向量逐维求平均。
• CLS Pooling：只取特殊的 [CLS] token 对应的隐藏向量。
• Max Pooling：每个维度取多个 token 中的最大值。

具体使用哪种方式由模型结构和配置决定。BGE 官方的底层 Transformers 示例使用 [CLS] pooling；使用 SentenceTransformer 时，池化通常已经被封装在模型对象里，所以你只调用 encode() 即可。

1
  多个 token 隐藏向量
2
  -> Pooling
3
  -> 一条固定长度的句向量
4
  -> 存入向量数据库或参与相似度检索

实际拆分结果可能不同：中文可能按词、字或更小的子词拆分，英文也可能把一个单词拆成多个子词。必须使用与模型配套的 Tokenizer，因为模型训练时使用的词表和编号就是固定配套的。

还要区分：

所以 token ID 不是最终向量。它更像模型输入侧的“编号化文本”。

这时还没有得到最终的一条句向量。底层模型先输出每个 token 的隐藏表示，sentence-transformers 再处理这些表示，得到代表整段文本的句向量。这里展示的是底层形状，不是本章要求你现在手写的方案。

1
from huggingface_hub import snapshot_download
2

3
model_path = snapshot_download(repo_id=MODEL_ID)

1
SentenceTransformer(MODEL_ID)
2
  1. huggingface_hub：找到或下载 MODEL_ID 的文件
3
  2. transformers：根据文件创建 tokenizer 和底层模型
4
  3. sentence-transformers：补上池化等句向量处理
5
  4. 返回 model 实例
6

7
model.encode(TEXT)
8
  5. tokenizer：文本 -> token ID
9
  6. base_model：token ID -> token 级隐藏表示
10
  7. sentence-transformers：token 表示 -> 一条句向量

1. 配置只负责“告诉代码用什么”#

模型名称现在固定为：

1
DEFAULT_EMBEDDING_MODEL = "BAAI/bge-small-zh-v1.5"

也可以在 .env 中覆盖：

1
EMBEDDING_MODEL_NAME=BAAI/bge-small-zh-v1.5

设备对应的配置表只保存批量大小：

1
TIER_BATCH_SIZES = {
2
    "mps": 32,
3
    "cpu": 8,
4
}

这里的字典自己不会检测设备，也不会加载模型。它只是保存配置，等其他函数来读取。

为什么不再让 MPS、CUDA、CPU 自动选择不同模型？

1
模型名称 -> 决定生成哪一种向量，不能随硬件变化
2
运行设备 -> 决定在哪里计算，可以变化
3
batch_size -> 决定一次计算几条，可以变化

2. _detect_device() 只负责选择运行设备#

它会检查 CUDA、MPS、DirectML，最后使用 CPU 兜底，返回类似：

1
{
2
    "tier": "mps",
3
    "device_str": "mps",
4
    "name": "MPS: arm64",
5
}

三个值的用途分别是：

本章不要求你默写完整硬件检测，只要知道它的输出会交给后面的加载函数。

3. _load_model() 才真正创建模型对象#

核心代码仍是第二关见过的：

1
model = SentenceTransformer(model_name, device=device)

输入与输出是：

1
输入：模型名称字符串 + 设备字符串
2
  -> SentenceTransformer(...)
3
输出：可以调用 encode() 的 model 对象

项目随后用一条测试文本做试运行，确认这个模型真的能在当前设备上计算。

如果 MPS 或 CUDA 失败，get_embedding_model() 会改用 CPU，再加载一次同一个模型：

1
bge-small 在 MPS 加载失败
2
  -> bge-small 改到 CPU 加载
3
  -> 模型没变，只是运行位置变了

旧实现可能在回退时换成另一个模型，这会破坏已有向量库；新实现已经避免了这个问题。

4. _model 避免每次请求都重新加载#

1
_model: SentenceTransformer | None = None

程序刚启动时，_model 是 None，表示模型还没有加载：

1
第一次调用 get_embedding_model()
2
  -> 发现 _model 是 None
3
  -> 检测设备
4
  -> 加载模型
5
  -> 把模型保存到 _model
6

7
第二次调用 get_embedding_model()
8
  -> 发现 _model 已经有模型
9
  -> 直接返回，不再加载

服务重启后，内存会被清空，_model 又从 None 开始。

threading.Lock 是为了防止两个并发请求同时加载两份模型。本章知道它在保护“首次加载”即可，不要求默写锁的写法。

5. 最后才区分“文档”和“查询”#

现在有三个主要入口：

1
get_document_embedding("退款应在七天内申请。")
2
# 一条文档 -> list[float]
3

4
get_query_embedding("退款需要几天内申请？")
5
# 一条查询 -> list[float]
6

7
get_document_embeddings(["文档 A", "文档 B"])
8
# 多条文档 -> list[list[float]]

它们最终都调用同一个模型的 encode()。区别只发生在文本送入模型之前：

例如：

1
用户原问题：退款需要几天内申请？
2

3
查询入口实际交给模型：
4
为这个句子生成表示以用于检索相关文章：退款需要几天内申请？

这段指令相当于给短文本补充一个“用途说明”：这句话是用来查找相关文档的。文档本身不需要添加。

代码内部用 _encode_one(text, is_query=...) 复用共同逻辑：

1
get_document_embedding(text)
2
  -> _encode_one(text, is_query=False)
3

4
get_query_embedding(text)
5
  -> _encode_one(text, is_query=True)

_encode_one() 前面的下划线表示它是内部辅助函数。业务代码直接调用上面两个名称清楚的公开函数。

6. 两边都使用同一种归一化策略#

文档和查询最终都会执行：

1
model.encode(
2
    text,
3
    convert_to_numpy=True,
4
    normalize_embeddings=True,
5
)

这里先这样理解：

1
normalize_embeddings=True
2
  -> 把每条向量的长度统一缩放为 1
3
  -> 使用同一把尺子比较向量方向

它不会减少向量维度，也不会修改原始文本。最重要的是文档和查询必须采用相同的归一化策略。

7. LangChain 适配层只是“改接口名称”#

LangChain 希望 Embeddings 对象具有两个固定方法：

1
LangChain 调用 embed_query(text)
2
  -> get_query_embedding(text)
3

4
LangChain 调用 embed_documents(texts)
5
  -> get_document_embeddings(texts)

LocalLangChainEmbeddings 就负责这层转接。它不会创建第二个模型，底下仍然调用同一个 get_embedding_model()。

旧函数 get_embedding() 和 get_embeddings() 暂时保留，方便旧教程代码继续运行；它们现在按“文档向量”处理。新代码优先使用含义明确的新函数。

把完整流程重新合起来#

文档入库：

1
chunk_text
2
  -> get_document_embedding()
3
  -> get_embedding_model()
4
  -> model.encode(原始文档，归一化)
5
  -> list[float]
6
  -> 存入 Chroma

用户查询：

1
query_text
2
  -> get_query_embedding()
3
  -> 添加查询指令
4
  -> get_embedding_model()
5
  -> model.encode(带指令的问题，归一化)
6
  -> list[float]
7
  -> 交给 Chroma 检索

两条路线不是使用两个模型。它们是：

1
同一个模型 + 两种输入角色 + 一致的向量规则

第三关立即检查#

1. _detect_device() 会不会加载模型？
2. CPU 回退时，模型名称和运行设备分别怎么变化？
3. 为什么第二次调用 get_embedding_model() 不会重新加载模型？
4. 查询入口和文档入口最终是不是使用同一个模型？

一致性不只是一句“使用同一个模型”#

建库和查询至少要记录并兼容：

当前 .env 应固定模型：

1
EMBEDDING_MODEL_NAME=BAAI/bge-small-zh-v1.5

然后重新建立向量库。不要只更新查询端，却继续长期使用旧文档向量。

推荐迁移顺序：

1. 先用固定评估问题记录当前召回结果。
2. 备份或删除旧的学习用 Chroma collection。
3. 使用固定模型重新导入原始文档，让所有 chunk 重新生成向量。
4. 使用同一组问题重新检索，对比修改前后的 Retrieval Evaluation。

不要机械地把 BGE 查询指令套到所有 Embedding 模型。以后更换模型时，要重新查看对应模型卡，并重新确认查询预处理和归一化要求。

本地调用与远程调用#

本章运行的是本地调用：

1
Python -> 本机 model 对象 -> 本机 CPU/MPS/CUDA -> embedding

InferenceClient 或 TEI 属于远程/服务调用：

1
Python 或 Dify -> HTTP 请求 -> 模型服务 -> HTTP 响应

两者都能得到向量，但部署位置、网络、鉴权、延迟和错误处理不同。

pipeline() 为什么不是本章主角#

1
from transformers import pipeline
2

3
classifier = pipeline("text-classification", model="某个模型 ID")
4
result = classifier("这门课程很清楚")

pipeline() 根据任务名称组合预处理器、模型和后处理器。它适合快速调用多种任务，但当前项目的向量封装已经使用更直接的 SentenceTransformer，所以本章不额外改成 pipeline()。

第一遍：追踪现成代码#

标记：[追踪]

分别从两个项目入口向下追：

1
文档：get_document_embedding(text)
2
查询：get_query_embedding(text)
3
  -> _encode_one(text, is_query=...)
4
  -> get_embedding_model()
5
  -> _detect_device()
6
  -> _get_model_name()
7
  -> _load_model()
8
  -> SentenceTransformer(...)
9
  -> encode(...)

每走一步都回答：输入是什么、返回什么、下一步为何需要它。

第二遍：跟写最小函数#

标记：[跟写]

不要看项目封装，自己写：

1
from sentence_transformers import SentenceTransformer
2

3

4
model = SentenceTransformer("BAAI/bge-base-zh-v1.5", device="cpu")
5

6

7
def get_embedding_demo(text: str) -> list[float]:
8
    embedding = model.encode(
9
        text,
10
        convert_to_numpy=True,
11
        normalize_embeddings=True,
12
    )
13
    return embedding.tolist()

这里把 model 放在函数外，是为了同一进程调用函数多次时不重复加载模型。

第三遍：独立迁移#

标记：[独立做] [验]

业务需求：写一个函数，一次接收多条文本并返回多条向量。

允许使用：本章的 SentenceTransformer、encode()、类型标注和普通函数。

禁止使用：复制 app/embedding.py 的设备检测、线程锁、LangChain 适配器；这些不是本次练习重点。

输入输出契约：

1
def get_embeddings_demo(texts: list[str]) -> list[list[float]]:
2
    ...

运行方式：

1
poetry run python app/你创建的练习文件.py

验收标准：

1. 输入两条字符串，返回外层长度为 2 的列表。
2. 每个元素都是 list[float]。
3. 模型只在文件加载时创建一次，不在函数内部反复创建。
4. 能打印二维结果的行数和每行维度。

失败提示：如果得到 numpy.ndarray 而不是列表，检查是否遗漏 .tolist()；如果每次调用都很慢，检查模型是否写在函数内部。

1. 把模型 ID 当成模型对象#

1
model_id = "BAAI/bge-base-zh-v1.5"
2
model_id.encode("hello")  # 错误

model_id 只是字符串。必须先创建 model。

2. 以为安装包就等于下载了模型#

poetry install 安装 Python 代码；模型权重通常在第一次加载指定模型时下载。

3. 以为缓存存在就不需要加载#

缓存避免重复下载，但每个新进程仍需把权重读进内存。

4. 自己又让建库和查询使用不同模型#

当前封装已经避免按硬件自动换模型，但修改 .env 后仍可能造成不一致。持久化向量库应固定并记录模型配置；修改模型或预处理策略后必须重建索引。

5. 一看到 Hub 就以为必须联网推理#

Hub 可以只负责首次下载；文件缓存后，本地模型可在满足依赖和缓存条件时本地推理。

6. 随手开启 trust_remote_code=True#

该参数允许执行模型仓库中的自定义代码。只有确认仓库可信并检查代码后才使用。