别再用嵌入向量瞎猜了
为复杂的金融文档构建高精度的 RAG 流水线
AI模型价格对比 | AI工具导航 | ONNX模型库 | Vibe Coding教程 | PLC在线仿真器 | Tripo 3D | Meshy AI | ElevenLabs | KlingAI | ArtSpace | Phot.AI | InVideo
每家企业最终都会撞上同一堵墙。你有一份又大又密的文档——一份 300 页的年度报告、一个多品牌的产品目录、一本合规手册、一批法律合同——而你需要回答关于它的精确问题。你需要的不是"给我一些相关段落"，而是像这样的问题的答案：
"对比 Mamaearth 与 The Derma Co. 在 Q4 的网红营销支出占其总收入的百分比。"
这是一个看似简单实则极难的问题。文档里有答案。但要准确地检索到它，需要同时尊重两个维度：
- 语义意图，也就是理解"网红营销支出"是什么意思，即使文档把它表述为"创作者主导的 campaign 投资"。
- 结构边界，也就是"Q4 的 Mamaearth"不同于"Q4 的 The Derma Co."。
标准工具在其中一个或两个维度上都会失败。一个普通的 LLM 文档解析器会流畅地产生幻觉，从错误的品牌或错误的季度里抓数字，因为各章节周围的措辞几乎一模一样。而一个普通的向量搜索会自信地返回一堆混杂的近邻结果，恰恰违反了你查询所隐含的约束。
为了具体说明，考虑一个真实世界的数据集：Honasa Consumer Limited 的 FY25 年度报告。Honasa 运营"品牌之家"模式，在一个公司伞下管理着 Mamaearth、The Derma Co.、Aqualogica、BBlunt 等品牌。每个品牌的章节都使用相似的财务词汇。每个季度都使用相似的结构性语言。提取特定品牌、特定季度、特定章节的洞察，正是那种朴素检索会崩溃的问题。
这篇文章将讲解解决它的技术模式：带元数据感知向量搜索的两步检索。本文的完整代码库在这里。
1、纯语义搜索的盲区
让我们快速回顾一下现代语义搜索是如何工作的。当你用 text-embedding-3-small 或 bge-large 这样的模型嵌入文本时，每个块都会变成一个稠密向量——一个由数百个浮点数组成的数组，捕捉它的含义。检索随后找到那些向量与你的查询向量"最接近"的块，通常通过余弦相似度。
在捕捉意图方面这真的很神奇。问"他们是如何获取新客户的？"，它会浮出关于"CAC"、"效果营销"或"漏斗顶部举措"的段落，即使你的查询里没有这些词。
但语义搜索本质上是一种模糊匹配。
考虑两个结构相似的句子，它们可以出现在 Honasa 年度报告的任何地方：
- "The Derma Co. 收入在 Q3 增长 15%，由活性成分驱动的护肤细分市场的扩张推动。"
- "Aqualogica 收入在 Q4 下降 2%，反映出以补水为重点的 SKU 的季节性回调。"
对嵌入模型来说，这两个句子几乎住在同一个街区。相同的句子结构。相同的财务词汇。相同的领域。唯一实质性的差异——品牌名称和季度——恰恰是相似度模型往往会低估的细节。
所以，如果用户搜索"Aqualogica Q4 收入指标"，纯向量搜索会自信地返回一个大杂烩：一些 Aqualogica 块、一些 Derma Co. 块、一些 Mamaearth 块，季度散落在整个财年。它没有严格边界的概念。它不知道"品牌"和"季度"是不可协商的过滤器；它只知道什么看起来相似。
2、两步检索架构
解决方案是一种正在迅速成为严肃文档智能流水线默认选择的范式：两步搜索，有时也叫混合过滤。
思路很简单：
第 1 步：宽泛的语义网 用稠密嵌入找到主题相关的上下文。这里我们让向量做它们擅长的事：找到概念上相关的段落。
第 2 步：精确过滤器（载荷 / 元数据） 应用严格的、数据库式的规则，即时丢弃不相关的块。想象在向量搜索之上叠加 SQL WHERE 子句：
brand_name == "The Derma Co."financial_quarter == "Q4"document_section == "Risk Factors"
神奇之处在于，这些规则不是软偏好，它们是硬约束。一个来自 Mamaearth 的高相似度块，即使得分 0.99，只要品牌过滤器说"The Derma Co."，它就会被扔掉。
历史上，团队用两个系统拼凑出这个方案：Postgres（用于结构化元数据）和一个基本的向量索引（用于嵌入），然后写胶水代码在内存中合并结果。它能用，但慢、脆弱、难以保持同步。
正确的向量数据库可以把这种双系统架构合并成一个。但选哪个呢？
3、为此选择合适的 VectorDB
过去两年向量数据库的版图爆炸式发展，每一个都声称能做混合搜索。那么，对于一个"过滤式语义搜索就是全部价值主张"的用例，你实际上该怎么选？
让我们沿着对两步检索工作负载真正重要的维度来拆解。
3.1 评估标准
对于一个元数据密集的文档智能流水线，我们关心六件事：
- 过滤搜索性能：带元数据约束的搜索有多快、多准？
- 载荷/元数据能力：向量旁边的结构化数据可以有多丰富？
- 混合检索（稠密 + 稀疏）：对组合语义搜索和关键词搜索的一流支持。
- 部署灵活性：自托管、云、嵌入式，我们有选择吗？
- 开发者体验：API 干净吗？SDK 稳定吗？
- 规模化的成本与性能：内存占用、查询延迟、吞吐量。
Pinecone 是家喻户晓的名字，最容易上手。但它只支持云、闭源，而且它的过滤模型历来较弱。元数据过滤器往往在 ANN 搜索检索出候选之后才应用，这意味着高选择性的过滤器即使在相关数据存在的情况下也可能返回稀疏或空的结果集。你最终要过度获取和重新排序。而且在大规模下成本攀升很快。
Weaviate 是一个强有力的竞争者，有设计良好的 schema 系统和内置的 BM25。代价是它僵化的 schema 优先方法在你的元数据演进时会增加摩擦。另外，它 GraphQL 优先的 API 有更陡的学习曲线。
Milvus 是为大规模规模而设计的，想想数十亿向量。这对大多数文档智能流水线来说杀鸡用牛刀，而且运维复杂度相当高。如果你在超大规模下运营，它很棒。对精干团队来说则很痛苦。
pgvector 美在简单：它就是一个 Postgres 扩展。如果你已经在跑 Postgres，加 pgvector 轻而易举，而且你能获得完整的 SQL 来做过滤。然而，它原生不支持稀疏向量（BM25），在超大向量集上的性能明显慢于专用引擎，而且你需要小心地手工调优 HNSW 参数。一个很好的入门选择；在生产中会触到天花板。
Chroma 对开发者友好，在原型阶段很流行（尤其是和 LangChain 一起），但它不是为生产级过滤检索而建的。元数据过滤很基础，它的持久化和扩展故事仍在成熟中。
Elasticsearch 有出色的 BM25 和成熟的过滤能力，并增加了稠密向量支持。但它不是向量优先设计的，你能在查询的易用性、内存使用上感觉到这一点，而且稠密向量操作是硬接在 Lucene 地基之上的。
3.2 为什么 Qdrant 在这个工作负载上胜出
以下是在过滤检索工作负载上选择 Qdrant 的诚实、不浪漫的理由：
- Qdrant 在向量索引旁边构建载荷索引。当你发出过滤查询时，过滤器不是在检索之后应用，而是集成到 HNSW 图遍历本身之中。这意味着高选择性的过滤器（比如品牌="The Derma Co." 且章节="Risk Factors"）不会降低召回率，也不会返回空结果集。这正是企业文档搜索需要的行为。
- Qdrant 支持丰富的、无 schema 的 JSON 载荷。没有僵化的 schema。如果我们下个季度想添加一个新的元数据字段，直接开始往载荷里写就行。嵌套对象、数组、地理数据和日期时间范围——Qdrant 都能处理。对比一下 Pinecone 的扁平键值元数据或 Weaviate 的事先 schema。
- 它用 Rust 编写，内存占用低、尾延迟可预测，没有 JVM 垃圾回收的意外。在独立的基准测试中，Qdrant 在过滤搜索吞吐量方面始终名列前茅。
3.3 决策矩阵
如果要为各种画像浓缩成一句话推荐：
对于任何"沿着结构维度严格过滤是首要需求"的用例，Qdrant 是自然之选。本文后面的一切都假设我们已经有意地做出了这个选择。
4、代码实现
两步检索系统的架构有两个不同的阶段：一个把数据结构化并索引的摄取流水线，和一个把语义搜索与严格元数据过滤结合起来的查询流水线。
首先是摄取端：一个 PDF 被解析成块，每个块被打上结构化元数据标签，然后被嵌入并存储到 Qdrant，它的载荷就放在向量旁边。
4.1 准备索引：分块与元数据提取
让我们动手。首先，我们解析一个大 PDF 文档并把它拆成块。
import fitz # PyMuPDF
from typing import List, Dict
def parse_document(pdf_path: str) -> List[Dict]:
doc = fitz.open(pdf_path)
chunks = []
for page_num, page in enumerate(doc):
text = page.get_text()
# Simple paragraph-level chunking
for para in text.split("\n\n"):
if len(para.strip()) > 100:
chunks.append({
"text": para.strip(),
"page": page_num + 1
})
return chunks
4.2 添加丰富的元数据标签
文本块本身只是成功的一半。在嵌入任何东西之前，我们给每个块打上结构化元数据。你可以用正则规则处理可靠的模式（实体名称、日期标记），用一个 LLM 做更软的分类（章节主题）。
import re
from collections import Counter
# Honasa's actual "house of brands"
BRANDS = ["Mamaearth", "The Derma Co", "Aqualogica", "BBlunt", "Dr. Sheth's", "Ayuga", "Staze"]
PERIOD_PATTERN = r"\bQ[1-4]\b|\bH[1-2]\b"
# Rule-based section classifier. Swap the body for an LLM call if you need
# fuzzier classification
SECTION_KEYWORDS = {
"Risk Management": ["risk management", "principal risks", "mitigation"],
"Financial Statements": ["balance sheet", "statement of profit and loss", "notes to accounts", "cash flow statement"],
"Corporate Governance Report": ["corporate governance report", "board committee", "shareholding pattern"],
"Management Discussion & Analysis": ["management discussion", "industry overview", "outlook"],
"Board's Report": ["board's report", "directors' report"],
"Sustainability": ["sustainability", "esg", "communities", "environment"],
"Corporate Overview": ["house of brands", "founders' message", "we are honasa"],
}
def classify_section(text: str) -> str:
low = text.lower()
for section, keywords in SECTION_KEYWORDS.items():
if any(kw in low for kw in keywords):
return section
return "General"
def enrich_chunk(chunk: Dict) -> Dict:
text = chunk["text"]
# Rule-based entity detection
detected_entities = [b for b in BRANDS if b.lower() in text.lower()]
# Rule-based reporting period detection
detected_periods = re.findall(PERIOD_PATTERN, text)
# Section classification (rule-based here; LLM-driven is a drop-in swap)
section = classify_section(text)
chunk["payload"] = {
"entity_name": detected_entities or ["Unspecified"],
"reporting_period": list(set(detected_periods)) or ["Full Year"],
"document_section": section,
"page": chunk["page"],
"fiscal_year": "FY25",
}
return chunk
enriched_chunks = [enrich_chunk(c) for c in raw_chunks]
# For this demo, focus on chunks that are actually attributable to a brand
# this is where naive retrieval gets confused, and where filtering earns its keep.
brand_chunks = [c for c in enriched_chunks if c["payload"]["entity_name"] != ["Unspecified"]]\
print(f"Total chunks: {len(enriched_chunks)}")
print(f"Brand-tagged chunks: {len(brand_chunks)}")
4.3 写入 Qdrant
现在我们嵌入文本，并让载荷紧挨着向量一起上传这个点。
from qdrant_client import QdrantClient
from qdrant_client.models import (
VectorParams, Distance, SparseVectorParams, PointStruct, SparseVector
)
COLLECTION_NAME = "enterprise_docs"
client = QdrantClient(":memory:")
if client.collection_exists(COLLECTION_NAME):
client.delete_collection(COLLECTION_NAME)
client.create_collection(
collection_name=COLLECTION_NAME,
vectors_config={
"dense": VectorParams(size=VECTOR_SIZE, distance=Distance.COSINE),
},
sparse_vectors_config={
"sparse": SparseVectorParams(),
},
)
points = []
for idx, (chunk, dense_vec, sparse_vec) in enumerate(zip(brand_chunks, dense_vectors, sparse_vectors)):
points.append(
PointStruct(
id=idx,
vector={
"dense": dense_vec.tolist(),
"sparse": SparseVector(
indices=sparse_vec.indices.tolist(),
values=sparse_vec.values.tolist(),
),
},
payload={
"text": chunk["text"],
**chunk["payload"], # entity_name, reporting_period, document_section, page, fiscal_year
},
)
)
client.upsert(collection_name=COLLECTION_NAME, points=points)
print(f"Upserted {len(points)} points into '{COLLECTION_NAME}'")
注意 payload 是如何作为一个字典与向量一起上传的。Qdrant 会给两者都建索引。
5、执行两步查询
现在是查询端。一次 API 调用就把语义相似度、精确关键词匹配和元数据过滤融合在一起。关键在于，载荷过滤器在向量遍历期间运行；任何未通过品牌、章节或财年约束的块都会在打分之前被淘汰，无论它的余弦相似度有多高。
这里就是一切汇聚之处。
from qdrant_client.models import Filter, FieldCondition, MatchValue, MatchAny
query_text = "supply chain risks regarding active ingredients"
query_vector = openai_client.embeddings.create(
model="text-embedding-3-small",
input=query_text
).data[0].embedding
results = client.search(
collection_name="enterprise_docs",
query_vector=query_vector,
query_filter=Filter(
must=[
FieldCondition(
key="entity_name",
match=MatchAny(any=["Brand B"])
),
FieldCondition(
key="document_section",
match=MatchValue(value="Risk Factors")
),
FieldCondition(
key="fiscal_year",
match=MatchValue(value="FY25")
)
]
),
limit=5
)
for r in results:
print(f"[Page {r.payload['page']}] Score: {r.score:.3f}")
print(r.payload["text"][:200], "\n")
5.1 引擎盖下刚刚发生了什么
当用户输入查询：
"Brand B 在 FY25 关于活性成分的供应链风险有哪些？"
我们嵌入这个问题，并与严格的元数据过滤器结合起来。
Qdrant 在向量遍历期间应用了载荷过滤器，而不是在搜索之后做一个朴素的过滤。这意味着：
- 每一个候选块都按照品牌、章节和财年约束进行了检查。
- 任何未通过这些约束的块都被立即淘汰，无论它与"供应链风险"的语义相似度有多高。
- 返回的 top-K 结果保证来自 The Derma Co. 在 FY25 的 Risk Factors 章节，并且在语义上与你查询最相关。
一个关于原料采购的 Mamaearth 段落可能与该查询有 0.94 的余弦相似度。没关系。它从未进入候选名单。
这就是"在演示中惊艳的 AI"和"用户放心把业务关键问题托付的 AI"之间的区别。
6、稠密向量 + 稀疏关键词一次搞定
现在说说边缘情况。假设你的查询是：
"文档在哪里讨论了烟酰胺（Niacinamide）的采购？"
"烟酰胺"是一个特定的化学成分。嵌入模型有时会忽略罕见的技术 token，尤其是当它们在训练数据中出现得不够多时。监管代码（比如一个 SEBI 通告编号）、部件号或特定的 SKU 标识符也是如此。
对于这些情况，你需要精确关键词匹配与语义搜索并肩工作。Qdrant 通过稀疏向量原生支持这一点。
from qdrant_client.models import SparseVector, NamedSparseVector, NamedVector
# Multi-stage query: dense + sparse + metadata filter, all in one API call
results = client.search(
collection_name="enterprise_docs",
query_vector=NamedVector(name="dense", vector=dense_query_vec),
query_filter=Filter(
must=[FieldCondition(key="entity_name", match=MatchAny(any=["Brand B"]))]
),
# ... combined with sparse vector query for "Niacinamide"
limit=5
)
通过一个正确配置的混合设置，一次 API 调用可以执行：
- 语义相似度（稠密向量）——用于主题相关性
- 精确关键词匹配（稀疏向量）——用于罕见 token 和标识符
- 严格元数据过滤（载荷）——用于实体、季度、章节
7、结束语
如果这一切中只有一件事值得带走，那就是：把文本纯粹当作浮点数组来对待，是制造混乱、不准确应用的配方。真实的业务数据——年度报告、合同、合规申报、产品目录——是有结构的。在摄取时把那个结构扔掉，就是把你最好的过滤器扔掉。
两步检索是以下两者的结合：
- 结构约束（数据是什么），以及
- 语义灵活性（查询意味着什么）。
两者结合在一起，就能产出返回精确、可信结果的搜索流水线——无论你在驱动一个财务仪表盘、一个内部问答工具，还是一个下游分析工作流。
而且选对引擎很重要。正如我们在系统性对比中看到的，大多数向量数据库把过滤当作事后才想的事，当作一个从根本上为纯相似度搜索而设计的系统上的附加物。相比之下，Qdrant 是从零开始、把过滤搜索作为首要用例来构建的。
原文链接: Stop Guessing with Embeddings: Exact-M atch Document Retrieval
汇智网翻译整理，转载请标明出处
