为生产环境中的RAG系统选择并调整嵌入模型。
了解嵌入模型如何将文本转换为可检索的向量，领域词汇为何会破坏语义搜索，以及如何为生产环境中的RAG选择、压缩和微调模型。
这是关于构建生产级检索增强生成系统系列文章的第五篇，讲述了如何从原始文档发展到能够回答实际问题的系统。该系列的早期文章介绍了对提取内容进行清洗和标准化处理，再通过分块将其拆分为检索单元。拿到这些分块后，接下来的任务就是将它们转化为搜索索引可以实际比较的形式。
想象一下之前提到的那位关系经理，他仍在为某家高风险企业客户争取1200万欧元的信贷额度。要满足这一需求，就需要在政策条款中找到相关的强化尽职调查要求，在表格行中找出审批阈值，以及在流程图中找到的控制步骤。分块阶段已经将这些内容拆分成独立的、可追溯的片段。
即便如此，这些内容目前还都无法通过向量索引进行检索。
嵌入模型负责将每个文本片段转换成固定长度的数值向量。同一个模型还会把输入的查询转换为向量，然后索引会找出在该向量空间中与之距离最近的那些文本片段。政策文件中的“集团信贷委员会审批”与用户问题中的“GCC批准阈值”是否真的会出现在彼此相近的位置并不确定——这完全取决于所选的模型以及它在训练过程中学到的内容。
本系列内容的重点就在探讨这种依赖关系。
选择嵌入模型实际上就是在判断文档的词汇和表达方式与用户查询有多少相似之处。如果为所在领域选错了模型，你将花费数周时间去排查看似检索错误实则为表示问题的问题——向量本身的位置有误，因此再怎么调整索引也无法解决。
嵌入模型究竟计算什么
从根本上说，嵌入模型接收一个词元序列并输出一个密集向量，其维度通常在384到3072之间，具体取决于所用模型。这个向量旨在以压缩形式表达输入内容的含义。
检索背后的基本假设是，含义相似的输入在对应空间中的向量会彼此靠近。衡量这种接近程度最常用的方法是余弦相似度，它关注的是两个向量之间的夹角而非原始距离——这一特性使得该方法对文本长度的差异不敏感。
假设有一项合规规则规定，被归类为高风险的企业客户在提交信贷申请以供审核之前必须经过强化尽职调查。一个功能强大的通用模型很可能会将其向量置于与其他银行类似合规条款、有关尽职调查义务的法律文件以及关于管理高风险客户的监管指南相近的位置。
不过，关系经理可能会用完全不同的方式表达同样的核心需求，比如会询问在提交信贷申请之前需要完成哪些步骤。这种日常用语是否与正式政策表述相近，实际上取决于有多少训练数据将非正式的操作性问题与正式的合规用语相结合。那些主要基于通用网络文本进行训练的模型，在面对专业领域时往往无法掌握这种特定的对应关系。
日常查询用语与专业文档用语之间的这种差异被称为词汇不匹配，它是企业RAG系统中检索质量问题的主要根源。
分词与上下文窗口
在嵌入模型开始计算之前，它会先使用自身的内部词汇表将输入文本拆分为标记。标记数量与单词数或字符数并无直接对应关系。英文中500个标记可能相当于350到400个单词，而在德语中——由于单词常常是复合词——同样的标记数量所代表的独立概念可能更少。
每个嵌入模型都会设定最大上下文长度，超过该长度的内容要么会被截断，要么需要特殊处理。Sentence-transformers模型通常将最大长度限制在256到512个标记之间。OpenAI的text-embedding-3-large模型可处理多达8,191个标记，BGE-M3则支持高达8,192个标记，Jina embeddings v3同样支持最多8,192个标记。
在实际应用中，生成式 RAG 的关键要点在于：在处理流程早期确定的文本块边界必须保持在所选嵌入模型规定的上下文限制之内。任何超出该限制的文本块都会被无声地截断，由此产生的向量仅能反映原始文本的一部分内容——这种故障模式不会出现在处理流程的日志中。
语义空间及其失效情形
目前大多数嵌入模型都是通过对比学习目标训练的变换器编码器：含义相似的文本对会在向量空间中靠得更近，而含义不同的文本对则会被分开。经过足够多的训练后，模型会形成一种几何结构，其中距离大小可用来表示语义上的相似性。
只要查询内容和文档在术语、写作风格以及概念框架上与模型训练数据一致，这种布局就能良好运行。但对于企业级RAG系统而言，它往往会在几种可预见的情形下出现故障：
- 领域特定术语会导致容易被忽视的故障。如果模型从未学习到“结构化可疑活动报告提交标准”与“SAR文件结构化提交阈值”这两个表述的含义相同，那么询问相关问题的分析师就可能找不到匹配结果。
了解通用嵌入模型在哪些方面容易出错，与知道哪款模型能在公开基准测试中名列前茅同样重要。
2025年如何选择嵌入模型
嵌入模型领域已大幅缩小。以下的对比内容涵盖了截至2025年中对银行业和金融服务领域的企业RAG系统而言最重要的模型。
在各种企业应用场景中并不存在通用的最佳解决方案。您的选择取决于需要支持的语言种类、可用的延迟预算与基础设施、是否可以选择本地推理，以及您的领域术语与通用模型训练数据之间的差异大小——这一差异决定了微调是否值得投入精力。
非对称检索与告知模型输入类型
许多团队容易忽视的一个区别是非对称检索。在检索段落时，查询语句与用于匹配的文本片段在结构上存在很大差异。查询语句通常较短，以问题形式呈现，且往往缺少正确答案中出现的许多词汇；而文本片段则较长，以事实形式呈现，且包含大量领域专用术语。
某些模型被设计为能够直接识别这种不对称性。E5模型会在输入文本前加上“query:”或“passage:”前缀，以便模型知晓其所嵌入的内容属于哪种类型。Cohere的Embed v3通过input_type参数来实现这一点，允许的值包括“search_query”、“search_document”、“classification”和“clustering”。
在索引或查询时使用错误的输入类型，会悄无声息地影响相似度得分，且很难追溯到根本原因。如果将文档当作查询来嵌入，生成的向量就会符合查询的几何结构而非文本段落的几何结构。虽然检索不会完全失败，但精度会下降，而在随意测试时很容易忽略这一点。
在生产环境中，不要依赖开发人员记得正确设置此项——应通过配置来强制要求。只要所使用的模型支持相应选项，无论是索引时的嵌入调用还是查询时的嵌入调用，都应明确指定输入类型。
import cohere
from typing import List
co = cohere.Client(api_key="your_api_key")
def embed_documents(chunks: List[str]) -> List[List[float]]:
"""Embed document chunks for indexing with explicit document input type."""
response = co.embed(
texts=chunks,
model="embed-english-v3.0",
input_type="search_document",
embedding_types=["float"]
)
return response.embeddings.float
def embed_query(query: str) -> List[float]:
"""Embed a search query with explicit query input type."""
response = co.embed(
texts=[query],
model="embed-english-v3.0",
input_type="search_query",
embedding_types=["float"]
)
return response.embeddings.float[0]
稀疏向量：关键词匹配优于语义搜索
密集嵌入能够捕捉含义，而稀疏表示则用于标识存在哪些术语以及应给予它们多大的权重。在企业级RAG系统中，对于大部分有意义的查询类型而言，稀疏检索的表现明显优于密集检索；在大多数生产系统中，结合使用这两种方法的效果也优于单独使用其中任何一种。
BM25作为可靠的基准
BM25依然是基于关键词的检索标准方法。它通过计算术语在文档中出现的频率、在整个语料库中的稀有程度，以及一个考虑文档长度的归一化因子来评估相关性。该方法无需训练模型，不需要GPU，也不涉及嵌入式API调用。
以“CRD-EU-047 approval authority threshold”这样的查询为例，BM25会为包含这些确切术语的任何片段赋予较高排名。而密集模型则可能不会展示这类片段，除非其训练语料库恰好建立了该特定政策代码与审批权限概念之间的强关联。
from rank_bm25 import BM25Okapi
import re
from typing import List, Tuple
def tokenise(text: str) -> List[str]:
"""Simple whitespace and punctuation tokeniser for BM25."""
return re.findall(r'\b\w+\b', text.lower())
class BM25Index:
def __init__(self, documents: List[str]):
self.documents = documents
tokenised = [tokenise(doc) for doc in documents]
self.bm25 = BM25Okapi(tokenised)
def search(self, query: str, top_k: int = 10) -> List[Tuple[int, float]]:
"""Return (doc_index, score) pairs for the top_k results."""
tokens = tokenise(query)
scores = self.bm25.get_scores(tokens)
ranked = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
return ranked[:top_k]
密集检索擅长捕捉概念上的相似性，而稀疏检索则善于找到术语和标识符的精确匹配。将两者结合——即混合检索——在银行领域往往能取得非常好的效果，因为该领域的监管语言极为精确且包含大量标识符。
在基于稳定、明确定义的监管语言构建的语料库中，仅使用BM25在进行有限的事实查询时，其召回率往往能与密集检索相媲美，且基础设施开销要低得多。它的主要缺陷在于同义词处理：如果查询中使用的是“EDD requirements”，除非恰好有完全相同的短语出现，否则无法检索到仅写有“Enhanced Due Diligence requirements”的内容。
SPLADE：能够实现词汇扩展的稀疏向量
SPLADE（稀疏词汇与扩展模型）介于简单的关键词匹配和完全密集检索之间。在索引阶段，会使用掩码语言模型为文档和查询添加语义相关的词汇，这些词汇不一定存在于原始文本中。最终得到的是一个稀疏向量，其各个维度对应一个特定的词汇标记，并根据该标记对输入内容的重要性进行加权。
因此，经过SPLADE编码且涉及EDD要求的文本，在遇到“客户尽职调查”、“风险评估”和“身份验证”等术语时，即便原文并未出现这些确切表述，也会赋予其更高的权重。这种扩展机制能够提升对同义词查询的召回率，同时仍保持稀疏结构所带来的高效性与可解释性。
其代价是在构建索引时需要额外的推理开销，且存储占用也会大于普通的BM25算法。但对于金融服务领域的语料库而言，由于不同司法管辖区和文档版本中对同一政策概念的表述各异，这种扩展机制能够显著扩大检索范围。
马特罗什卡嵌入：通过可调向量大小实现成本控制
马特里奥什卡表示学习（MRL）生成的嵌入中，前N个维度就已经构成了输入的完整、自包含的表示——后续维度用于添加更细粒度的信息，而非覆盖之前的内容。
该技术得名于俄罗斯的套娃：一个1536维的马特里奥什卡向量在其前256个位置中包含一个功能完备的256维表示，在前512个位置中包含一个功能完备的512维表示，依此类推。
OpenAI的text-embedding-3模型系列通过维度参数直接支持这种方式。
from openai import OpenAI
from typing import List
client = OpenAI()
def embed_with_matryoshka(
texts: List[str],
dimensions: int = 256,
model: str = "text-embedding-3-large"
) -> List[List[float]]:
"""
Embed texts at a specified sub-dimension.
Lower dimensions reduce storage and index cost.
Measure retrieval quality drop before committing to a dimension.
"""
response = client.embeddings.create(
input=texts,
model=model,
dimensions=dimensions
)
return [item.embedding for item in response.data]
马特里奥什卡嵌入将越来越精细的表示嵌套在同一个向量中：前256个维度就已经能生成可用的检索表示，而每增加一层都会以相应的存储成本提升语义精度。
真正的生产优势在于能够根据需求调整存储与质量之间的平衡，而无需重新训练模型或从头构建索引。对于银行政策语料库，你可以分别在256、512、1024和3072维度下测试检索效果，会发现512维度即可实现97%的完整召回率，同时仅需要全向量所需存储空间的17%。
在实践中，这种平衡往往不像基准测试图表中显示的那样理想。细微的领域差异——尤其是那些密切相关的监管概念之间的差异——通常存在于向量的高维部分。在确定用于生产的较低维度之前，应先用自己的语料库和实际查询模式进行测试。
在几乎不牺牲准确率的前提下压缩嵌入向量
标准嵌入模型将每个维度存储为32位浮点数。若要将数据扩展到一百万个文档块，且每个块有1536个维度，那么在计入任何索引开销之前，原始向量的存储容量大约为6GB。在企业级应用中，这样的存储需求及其相应的内存成本已不再是可以忽略的微小误差。
量化技术通过减少表示每个维度所需的比特数来解决这一问题。实际应用中主要有三种技术：标量量化（将float32转换为int8）、二进制量化（将float32转换为单个比特）以及乘积量化（将每个向量压缩为更短的编码）。
标量量化：int8
标量量化将连续的float32数值范围映射为256个离散的整数值。每个维度的存储大小从4字节缩减到1字节，存储需求因此降低75%。由于高维嵌入模型会将信息分散在多个维度中，单个维度本身并不承载太多权重，因此这种四舍五入带来的精度损失通常很小。
import numpy as np
from typing import Tuple
def quantise_to_int8(
embeddings: np.ndarray
) -> Tuple[np.ndarray, float, float]:
"""
Scalar quantisation to int8.
Returns quantised array plus the scale and zero_point needed for dequantisation.
"""
min_val = embeddings.min()
max_val = embeddings.max()
scale = (max_val - min_val) / 255.0
zero_point = -round(min_val / scale)
quantised = np.clip(
np.round(embeddings / scale) + zero_point,
0, 255
).astype(np.uint8)
return quantised, scale, zero_point
def dequantise_from_int8(
quantised: np.ndarray,
scale: float,
zero_point: float
) -> np.ndarray:
"""Reconstruct approximate float32 embeddings from int8."""
return ((quantised.astype(np.float32) - zero_point) * scale)
二进制量化
二进制量化则更为极致，它将每个维度简化为单个比特，仅用于记录原始浮点数值是正数还是负数。与float32相比，这种方式可使存储需求减少约97%。由于数值表示不再连续，相似度计算便改用汉明距离而非余弦相似度。
该技术在那些输出分布天然处于中心位置的模型上效果最佳，这样对于任意给定的输入，大约有一半的数值会位于零点两侧。如果模型的数值分布偏斜而非平衡，二进制量化会导致更明显的质量损失。Cohere在开发Embed v3时便考虑了这一限制，Anthropic对该模型的测试结果显示，在其测试集上检索精度下降幅度低于1%，同时存储空间减少了97%。请将这一数据视为参考值而非绝对保证，在实际应用前需用自己的语料库进行验证。
import numpy as np
def quantise_to_binary(embeddings: np.ndarray) -> np.ndarray:
"""
Binary quantisation: positive dimensions become 1, negative become 0.
Packs 8 dimensions per byte using numpy packbits.
"""
binary_matrix = (embeddings > 0).astype(np.uint8)
return np.packbits(binary_matrix, axis=1)
def hamming_similarity(
query_binary: np.ndarray,
corpus_binary: np.ndarray
) -> np.ndarray:
"""Compute normalised Hamming similarity for binary embeddings."""
n_bits = corpus_binary.shape[1] * 8
xor = np.bitwise_xor(
query_binary,
corpus_binary
)
hamming_distances = np.unpackbits(xor, axis=1).sum(axis=1)
return 1.0 - (hamming_distances / n_bits)
在生产环境中，二进制量化通常作为两步检索流程的第一阶段来使用：先通过二进制索引实现快速、大规模的候选项召回，然后再通过全精度处理对最顶层的结果进行重新评分。这种架构在节省大部分存储空间的同时，能够在最终排序阶段恢复所需的精度。
针对特定领域的 RAG 进行微调
一旦确认通用嵌入模型在您的数据上确实表现不佳，微调便是正确的解决方案。其目标是让模型明白，特定于您所在领域的词汇、缩写以及概念关联在语义空间中应当彼此靠近。
微调并非总是必要的，也未必是正确的解决方案。如果如本系列前文所述，检索问题源于糟糕的分块方式，调整嵌入模型并无帮助。若根本原因在于重排机制的配置或提示语的构建方式，微调则完全针对了系统的错误层面。在投入资源之前，应按查询类型分析检索失败的原因，以确定实际问题所在。
通用嵌入模型失效时
尤其是在银行领域的RAG应用中，存在几种反复出现的故障模式，使得微调成为合理的投资选择：
领域特定的缩写容易被误解。通用模型可能会将“NPA”与国家公园协会联系起来，而非不良资产；同时它也可能只是弱地将“KYC”与实际上在银行查询中更为常见的合规及客户准入概念关联起来。
不同文档中相关概念之间的联系会丢失。搜索“设施重组条款”本应能找到关于“贷款调整框架”的政策文本，但若模型主要基于通用网络内容训练，则可能从未见过这些表述足够频繁地出现在一起，从而无法建立相应的关联。
监管代码和标识符并未得到足够的重视。政策版本号、监管代码以及管辖区域标记本应切实影响排序结果，但现有的通用嵌入模型往往将它们视为价值较低的符号，几乎不携带语义信息。
数值阈值也失去了其监管背景。像“1000万欧元”这样的数字若单独出现，不应自动与关于“大额风险敞口的审批权限”的查询相匹配——只有当模型在将数字与其监管含义相关联的领域数据上经过训练后，这种关联才会形成。
利用领域特定数据构建训练对
在对比学习目标下微调句子转换模型依赖于正样本对：即将查询与模型应视为相关的段落配对起来的示例。负样本可以手动挑选，也可以从相关语料库中自动提取。
在银行领域的RAG应用中，这些正样本对可以从以下几个实际来源获取：
合规团队和信贷团队已生成的现有问答集，其中每个问题都与其对应的来源段落相连。
政策文件的自然结构，标题与下方的段落组合即可形成现成的正样本对。
分析师的查询记录，以及当答案正确时实际被检索到的相关段落。
大语言模型会针对每段文本生成机器生成的查询，以该文本本身作为匹配的正样本。
在标注数据极为匮乏的情况下，生成合成查询往往是最为可行的方法。
from openai import OpenAI
import json
from typing import List, Dict
client = OpenAI()
def generate_training_queries(
chunk: str,
chunk_metadata: Dict,
n_queries: int = 3
) -> List[Dict]:
"""
Generate synthetic query-passage pairs for fine-tuning.
The chunk itself is the positive passage for each generated query.
"""
prompt = f"""You are generating training data for a banking RAG system.
Given the following policy passage, generate {n_queries} realistic questions
that a credit analyst, compliance officer, or relationship manager might ask
that this passage directly answers. Each question should use natural language
and may use different terminology than the passage itself.
Passage:
{chunk}
Return a JSON array of objects with keys "query" and "difficulty".
Difficulty should be "narrow" (single fact) or "synthesis" (multiple facts).
Return only the JSON array, no other text."""
response = client.chat.completions.create(
model="gpt-4o-mini",
messages=[{"role": "user", "content": prompt}],
response_format={"type": "json_object"}
)
try:
result = json.loads(response.choices[0].message.content)
queries = result.get("queries", result) if isinstance(result, dict) else result
return [
{
"query": q["query"],
"passage": chunk,
"document_id": chunk_metadata.get("document_id"),
"chunk_id": chunk_metadata.get("chunk_id"),
"difficulty": q.get("difficulty", "narrow")
}
for q in queries
]
except (json.JSONDecodeError, KeyError):
return []
基于三元组损失函数的对比训练
对于以检索为导向的嵌入模型而言，最强的训练目标便是对比学习，可通过批量内负样本或刻意挑选的难负样本来实现。Sentence-transformers通过MultipleNegativesRankingLoss功能支持这种训练方式，该函数会将训练批次中的其他样本作为给定锚点-正样本对的隐含负样本。
from sentence_transformers import SentenceTransformer, InputExample
from sentence_transformers.losses import MultipleNegativesRankingLoss
from torch.utils.data import DataLoader
from typing import List, Dict
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
def build_training_examples(
pairs: List[Dict]
) -> List[InputExample]:
"""
Convert query-passage pairs into InputExample objects.
MultipleNegativesRankingLoss expects (anchor, positive) pairs.
Negatives are sampled automatically from other items in the batch.
"""
return [
InputExample(texts=[pair["query"], pair["passage"]])
for pair in pairs
if pair.get("query") and pair.get("passage")
]
def fine_tune_embedding_model(
base_model_name: str,
training_pairs: List[Dict],
output_path: str,
epochs: int = 3,
batch_size: int = 16,
warmup_steps: int = 100
) -> SentenceTransformer:
"""
Fine-tune a sentence-transformers model on domain-specific query-passage pairs.
base_model_name: HuggingFace model identifier or local path.
training_pairs: List of dicts with "query" and "passage" keys.
output_path: Directory to save the fine-tuned model.
"""
model = SentenceTransformer(base_model_name)
logger.info(f"Loaded base model: {base_model_name}")
logger.info(f"Training on {len(training_pairs)} query-passage pairs")
examples = build_training_examples(training_pairs)
loader = DataLoader(examples, shuffle=True, batch_size=batch_size)
loss = MultipleNegativesRankingLoss(model)
total_steps = len(loader) * epochs
logger.info(f"Training for {epochs} epochs, {total_steps} total steps")
model.fit(
train_objectives=[(loader, loss)],
epochs=epochs,
warmup_steps=warmup_steps,
output_path=output_path,
show_progress_bar=True,
checkpoint_path=output_path,
checkpoint_save_steps=len(loader)
)
logger.info(f"Fine-tuned model saved to: {output_path}")
return model
基准测试结果：微调前后的银行领域语料库
以针对企业信用政策语料库进行的检索基准测试为例，该语料库由从政策文件、审批矩阵、强化尽职调查程序以及反洗钱指南中提取的847个文本片段组成。评估集包含120个查询，涵盖四类类型：精确事实检索、阈值问题、多证据验证问题以及综合分析问题。
领域微调的收益在涉及阈值类型的查询中最为显著，例如询问适用于欧盟高风险企业客户的审批阈值。这类查询中的银行缩写及监管术语与通用模型在预训练阶段接触过的内容差异最大，因此缩小这一词汇差距能带来最高的召回率提升。多证据与综合查询单独经过微调后提升效果有限，但一旦叠加混合检索技术，其性能就能得到显著改善。
请将这些数字视为在标注规范的银行数据集上成功完成微调项目可能取得的成果示例，而非承诺。您自己的语料构成、查询类型组合以及标注精度都会影响最终结果。最重要的是按查询类型分别评估检索质量，而非仅给出一个综合分数，因为不同类型的查询往往因不同原因出现故障。
大规模批量嵌入的吞吐量考量
一次仅处理10万个数据块并通过嵌入API发送，既缓慢又成本高昂。生产级嵌入流程则会将输入批量处理，从而提升吞吐量、控制在速率限制范围内、从故障中顺利恢复，并确保输出结果的确定性。
通过服务提供商API进行批量处理
OpenAI的嵌入接口允许在单次调用中处理最多2,048个输入。Cohere的Embed接口每次请求最多支持96篇文本，若需处理更多数据，则必须使用其专用的批量API。而使用sentence-transformers在本地进行推理时，可以自定义批量大小，仅受可用GPU内存的限制。
import time
import logging
from typing import List, Optional
from openai import OpenAI, RateLimitError, APIError
logger = logging.getLogger(__name__)
client = OpenAI()
def embed_in_batches(
texts: List[str],
model: str = "text-embedding-3-large",
batch_size: int = 512,
max_retries: int = 3,
retry_delay: float = 2.0,
dimensions: Optional[int] = None
) -> List[List[float]]:
"""
Embed a large list of texts using batched API calls with retry logic.
texts: Pre-chunked text strings. Caller is responsible for ensuring
no text exceeds the model's token limit.
batch_size: Number of texts per API call. Stay well below the API limit
to avoid hitting per-request token limits.
dimensions: Optional Matryoshka dimension reduction for supported models.
"""
all_embeddings: List[List[float]] = []
total_batches = (len(texts) + batch_size - 1) // batch_size
for batch_idx in range(0, len(texts), batch_size):
batch = texts[batch_idx: batch_idx + batch_size]
current_batch = batch_idx // batch_size + 1
logger.info(f"Embedding batch {current_batch}/{total_batches} "
f"({len(batch)} texts)")
kwargs = {
"input": batch,
"model": model
}
if dimensions is not None:
kwargs["dimensions"] = dimensions
attempt = 0
while attempt < max_retries:
try:
response = client.embeddings.create(**kwargs)
# Preserve input order: API returns items sorted by index
sorted_items = sorted(response.data, key=lambda x: x.index)
all_embeddings.extend([item.embedding for item in sorted_items])
break
except RateLimitError:
attempt += 1
wait = retry_delay * (2 ** attempt)
logger.warning(f"Rate limit hit on batch {current_batch}. "
f"Waiting {wait:.1f}s before retry {attempt}/{max_retries}")
time.sleep(wait)
except APIError as e:
attempt += 1
logger.error(f"API error on batch {current_batch}: {e}. "
f"Retry {attempt}/{max_retries}")
if attempt >= max_retries:
raise
time.sleep(retry_delay)
logger.info(f"Embedding complete. Total vectors: {len(all_embeddings)}")
return all_embeddings
使用sentence-transformers在本地进行推理
某些机构因数据驻留要求而无法将政策文件发送到外部API，这种情况下，使用sentence-transformers在本地进行推理是最佳解决方案。
from sentence_transformers import SentenceTransformer
import numpy as np
from typing import List, Optional
import logging
logger = logging.getLogger(__name__)
class LocalEmbeddingPipeline:
"""
Production-ready local embedding pipeline using sentence-transformers.
Suitable for data-residency-constrained banking environments.
"""
def __init__(
self,
model_name_or_path: str,
device: str = "cpu",
batch_size: int = 64,
normalise: bool = True
):
self.model = SentenceTransformer(model_name_or_path, device=device)
self.batch_size = batch_size
self.normalise = normalise
self.device = device
logger.info(f"Loaded model: {model_name_or_path} on {device}")
def embed(
self,
texts: List[str],
show_progress: bool = True
) -> np.ndarray:
"""
Embed a list of texts. Returns an (N, D) numpy array.
Normalises to unit length if normalise=True (required for cosine similarity).
"""
embeddings = self.model.encode(
texts,
batch_size=self.batch_size,
show_progress_bar=show_progress,
normalize_embeddings=self.normalise,
convert_to_numpy=True
)
logger.info(f"Embedded {len(texts)} texts. "
f"Output shape: {embeddings.shape}")
return embeddings
def embed_query(self, query: str) -> np.ndarray:
"""Embed a single query. Returns a 1D array."""
return self.embed([query], show_progress=False)[0]
在嵌入之前检查令牌长度
如果某个片段的长度超过模型的令牌限制，它会在没有任何警告的情况下被截断。在策略语料库中，这种无声的截断可能会恰好删掉让该片段值得被检索的那些条款或数值阈值。在嵌入之前验证令牌数量，可以在问题影响到索引之前就将其解决。
from transformers import AutoTokenizer
from typing import List, Tuple
import logging
logger = logging.getLogger(__name__)
def validate_chunk_lengths(
chunks: List[str],
model_name: str,
max_tokens: int,
truncation_strategy: str = "warn"
) -> Tuple[List[str], List[int]]:
"""
Validate that all chunks are within the model's token limit.
truncation_strategy:
"warn" - Log a warning for oversized chunks and include them (will be truncated by model).
"skip" - Remove oversized chunks and return only valid ones.
"raise" - Raise ValueError on the first oversized chunk.
Returns (validated_chunks, oversized_indices).
"""
tokeniser = AutoTokenizer.from_pretrained(model_name)
oversized = []
for idx, chunk in enumerate(chunks):
token_count = len(tokeniser.encode(chunk, add_special_tokens=True))
if token_count > max_tokens:
oversized.append(idx)
msg = (f"Chunk {idx} has {token_count} tokens, "
f"exceeds model limit of {max_tokens}. "
f"First 80 chars: {chunk[:80]!r}")
if truncation_strategy == "raise":
raise ValueError(msg)
else:
logger.warning(msg)
if truncation_strategy == "skip" and oversized:
valid = [c for i, c in enumerate(chunks) if i not in set(oversized)]
logger.info(f"Removed {len(oversized)} oversized chunks. "
f"{len(valid)} chunks remain.")
return valid, oversized
return chunks, oversized
为嵌入向量添加元数据与来源信息
对于需要符合监管要求的RAG系统而言，仅靠原始的嵌入向量是不足以正常运行的。每个向量都需要附带结构化的元数据，这样后续的检索、重排序和生成阶段才能确定内容的来源，实施访问权限控制，按管辖区域限制结果，并指向权威的源文档。
回到那项1200万欧元的信贷方案场景，每个嵌入块至少应包含此处所示的字段：
from dataclasses import dataclass, field
from typing import Optional, List
import uuid
@dataclass
class EmbeddedChunk:
"""
Production embedding record for a banking policy RAG system.
The vector enables retrieval. The metadata enables everything else.
"""
# Vector
vector: List[float]
vector_dimensions: int
embedding_model: str
embedding_model_version: str
# Content
text: str
content_type: str # "narrative", "table_row", "proposition", "image_description"
# Provenance
document_id: str
document_version: str # e.g. "7.2"
policy_id: Optional[str] # e.g. "CRD-EU-047"
jurisdiction: Optional[str] # e.g. "EU"
effective_date: Optional[str]
# Chunk structure
chunk_id: str = field(default_factory=lambda: str(uuid.uuid4()))
parent_id: Optional[str] = None
section: Optional[str] = None
page_number: Optional[int] = None
source_artifact_path: Optional[str] = None # path to original image/table
# Access control
classification: str = "INTERNAL" # "PUBLIC", "INTERNAL", "CONFIDENTIAL"
permitted_roles: List[str] = field(default_factory=list)
# Indexing
indexed_at: Optional[str] = None
indexing_pipeline_version: Optional[str] = None
在从数据分块、嵌入到向量索引的整个流程中始终保留这些元数据，不仅是一种良好的实践。在受监管的银行业环境中，若依据过时的政策版本得出技术上正确的答案，即视为合规失误。向量的作用是找到对应的数据块；而元数据的作用则是确认该数据块来自正确且最新的源版本。
评估嵌入质量
像MTEB这样的公开排行榜会展示在各种学术数据集上的通用检索得分。这些数值有助于筛选出表现明显不佳的模型。然而，当任务是为银行内部政策库这类专业语料库挑选最佳模型时，这些指标就不够用了。
唯一真正重要的衡量标准是：模型使用您自己的查询，针对您自行标注的相关性判断，对您自己的文档进行检索时的表现如何。
构建检索评估集
为银行RAG嵌入流程构建的检索测试集需要涵盖多种查询类型：
那些能对应到唯一权威信息块的简单事实查询，例如询问高风险企业客户至少多久需要进行一次年度审查。
基于阈值的问题，将特定的数值条件与其对应的治理规则相结合，例如询问当欧盟企业的资产规模超过1000万欧元时需要哪一级别的审批权限。
多证据问题，其完整答案需要整合多个信息片段，例如询问在提交高风险企业信贷申请之前必须进行哪些审核。
综合问题，需要同时从多个部分提取内容，例如要求对用于管理高风险企业贷款的反洗钱控制框架进行完整描述。
跨文档问题，适用于政策在不同文档之间相互引用的情况。
import numpy as np
from typing import List, Dict, Set
def recall_at_k(
retrieved_ids: List[str],
relevant_ids: Set[str],
k: int
) -> float:
"""
Compute Recall@k for a single query.
relevant_ids is the ground truth set of chunk identifiers.
retrieved_ids is the ordered list of retrieved chunk identifiers.
"""
if not relevant_ids:
return 0.0
top_k_retrieved = set(retrieved_ids[:k])
return len(top_k_retrieved & relevant_ids) / len(relevant_ids)
def mean_reciprocal_rank(
retrieved_ids: List[str],
relevant_ids: Set[str]
) -> float:
"""Compute MRR for a single query."""
for rank, chunk_id in enumerate(retrieved_ids, start=1):
if chunk_id in relevant_ids:
return 1.0 / rank
return 0.0
def evaluate_embedding_model(
model_name: str,
evaluation_queries: List[Dict],
corpus_chunks: List[Dict],
k_values: List[int] = [1, 5, 10, 20]
) -> Dict:
"""
Evaluate an embedding model on a labelled retrieval dataset.
evaluation_queries: List of dicts with "query" and "relevant_chunk_ids" keys.
corpus_chunks: List of dicts with "chunk_id" and "text" keys.
Returns per-query-type and aggregate retrieval metrics.
"""
from sentence_transformers import SentenceTransformer
model = SentenceTransformer(model_name)
corpus_texts = [c["text"] for c in corpus_chunks]
corpus_ids = [c["chunk_id"] for c in corpus_chunks]
corpus_embeddings = model.encode(corpus_texts, normalize_embeddings=True)
results_by_type: Dict[str, List] = {}
all_recall: Dict[int, List[float]] = {k: [] for k in k_values}
all_mrr: List[float] = []
for query_item in evaluation_queries:
query = query_item["query"]
relevant = set(query_item["relevant_chunk_ids"])
query_type = query_item.get("query_type", "unspecified")
query_embedding = model.encode(query, normalize_embeddings=True)
scores = corpus_embeddings @ query_embedding
ranked_indices = np.argsort(scores)[::-1]
retrieved = [corpus_ids[i] for i in ranked_indices]
mrr = mean_reciprocal_rank(retrieved, relevant)
all_mrr.append(mrr)
for k in k_values:
r = recall_at_k(retrieved, relevant, k)
all_recall[k].append(r)
if query_type not in results_by_type:
results_by_type[query_type] = {"mrr": [], "recall": {k: [] for k in k_values}}
results_by_type[query_type]["mrr"].append(mrr)
for k in k_values:
results_by_type[query_type]["recall"][k].append(
recall_at_k(retrieved, relevant, k)
)
aggregate = {
"model": model_name,
"n_queries": len(evaluation_queries),
"mrr": float(np.mean(all_mrr)),
"recall": {k: float(np.mean(all_recall[k])) for k in k_values}
}
per_type = {
qt: {
"mrr": float(np.mean(data["mrr"])),
"recall": {k: float(np.mean(data["recall"][k])) for k in k_values},
"n_queries": len(data["mrr"])
}
for qt, data in results_by_type.items()
}
return {"aggregate": aggregate, "by_query_type": per_type}
不应孤立地评判检索指标，而应考虑它们对最终答案质量的影响。假设Recall@5提升了3分，但被检索到的近似重复内容的比例上升了30％——这种权衡实际上可能无助于大型语言模型，因为提供三段几乎相同的文本而非一段有用的文本，并不能增加真正的有效证据。
正确的做法是评估整个流程，从最初的查询到最终呈现给用户的答案。嵌入模型的作用仅限于将合适的证据放入上下文窗口中，而这些证据能否生成既准确又符合要求的答案，则取决于其之后的每一个环节。
作为基础设施的嵌入处理流程
一旦某个数据块完成嵌入处理流程，它就必须带有一个向量、完整的元数据以及一个稳定的标识符从另一端输出，这样才能在后续检索、更新或删除该数据块时，不会破坏索引中相邻的记录。
这属于基础设施的一部分，而非一次性脚本。面向受监管银行环境的生产级嵌入处理流程需要满足以下要求：
- 幂等性。如果因模型版本更替而重新对某个数据块进行嵌入处理，应覆盖现有的记录，而非生成重复记录。
一旦进入生产环境，这些都不是可选的。正是这些特性区分了仅在演示环境中运行的流程与能够在受监管环境下长期运行、审计和维护的流程。
返回1200万欧元信贷提案
客户经理最初的问题依然没有变化：这笔交易需要什么审批权限，且在提交之前必须通过哪些审核流程？
- 第四部分介绍了如何设计能够保持这些答案原始形式的检索单元：EDD政策条款、明确GCC权限的审批矩阵行，以及阐述提交前控制措施的流程图。
- 在本部分中，我们利用针对银行领域术语进行过基准测试的模型，将这些检索单元转换为可搜索的向量，并对该模型进行了微调，使其能够理解监管缩写与其合规背景之间的关联；同时还会嵌入完整的来源元数据，以便日后通过检索确认政策版本和适用管辖区域。
当查询到来时，向量索引会找到涵盖风险敞口超过1000万欧元、EDD条款以及控制流程图表的审批矩阵行，并将它们连同元数据一起返回，这些元数据可证明这三者均源自政策CRD-EU-047的7.2版本，适用欧盟司法管辖区，自2026年1月15日起生效。
传递到生成层的都是准确、完整且可追溯的证据。
这正是那种仅将数据块加载到向量数据库中的嵌入流程，与那些能够保留所有必要信息以生成可靠且可审计的答案的嵌入流程之间的区别。
在采用向量索引之前：检查清单
在将嵌入的数据块放入向量索引之前，请确认以下事项：
- 在用于索引查询和检索查询的嵌入模型中，输入类型参数是否设置正确？这种不匹配会悄悄降低检索精度。
- 在进行嵌入处理之前，是否检查过每个分块的标记长度？无声的截断会改变嵌入文本的实际含义，而整个处理流程中不会产生任何错误提示。
- 每条向量记录是否包含完整的来源信息——文档版本、策略编号、管辖区域以及生效日期？
- 该嵌入模型是否真的在您自己的领域语料库和查询模式上经过测试，而非仅仅依据公开基准测试的排名来选择？
- 如果您对模型进行了微调，是否在保留的查询集上测量了调整前后的检索效果数据？
如果缺少其中任何一点，向量索引就无法为您发现问题——它只会毫无异议地存储您输入的任何内容。数据库中不会对由截断数据块生成的向量、包含错误输入类型的查询，或是使用与索引其他部分不同模型版本嵌入的数据块发出警告。这些缺陷不会自行暴露，日后会以检索质量问题的形式重新出现，从表面上看就像是大型语言模型本身的问题。
在本系列的前文第4部分中，我们介绍了如何在不破坏文档结构的前提下将其转换为可检索单元。本部分则展示了如何在保留这些单元来源信息的同时，将它们转化为可搜索的向量。
接下来的内容将探讨这些向量如何与索引相结合：密集向量搜索、稀疏检索、混合方法、近似最近邻算法，以及用于筛选哪些向量有资格参与检索的元数据过滤机制。
相关阅读
- 生产级企业代理式AI系统的参考架构 — 了解将代理式AI从原型发展为可靠的企业级生产系统所需的核心架构层、内存策略、检索机制及约束条件。
- 向量数据库详解：RAG与AI搜索背后的技术原理 — 了解向量数据库如何将文本转换为嵌入向量，为语义搜索和RAG流程提供支持，并推动推荐系统等实际AI应用的发展。