实用指南：超越语义搜索——高级RAG完整指南
《实用笔记：超越语义搜索——高级RAG完整指南》的操作流程详解：专为采用该模式的团队设计的合同、校验机制以及可直接插入的代码模块。
本指南将逐步构建从原始材料到可运行系统的完整流程，内容来自《超越语义搜索：基于Milvus的先进RAG完全指南》——作者著。重点在于可操作的步骤、明确的检查点，以及可直接放入代码库的代码，无需猜测其用途。 在概览阶段，应在修改代码之前明确输入内容、各步骤的负责人以及完成标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏的状态。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
什么是RAG？它为何存在？
在研究 RAG 及其工作流程时，首先需明确相关规范：所需的输入参数、成功标志，以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改始终符合要求。 应将配置信息与应用程序代码分开。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
User Question
│
▼
[Embed the question] → query vector
│
▼
[Search Vector DB] → top-K relevant document chunks
│
▼
[LLM prompt: "Given these passages, answer: {question}"]
│
▼
Accurate, Grounded Answer
向量数据库的作用
在处理“阶段的作用”这一主题时，首先需列出相关约定：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在调整提示词之前，先使用固定的问题集来衡量检索效果。仅仅更换提示词往往无法解决检索能力薄弱的问题。
理解嵌入模型：密集型与稀疏型
在处理“Understanding Embeddings Dense”阶段时，首先需明确规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非庞大的脚本。当某一步骤失败时，故障应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难改善较差的检索效果。 在处理“Understanding Embeddings Dense”阶段时，首先需明确规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果之外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免从演示环境过渡到共享环境时出现意外费用。
密集嵌入
将“密集嵌入”阶段视为可测量的表面来处理效果最佳。在扩大范围之前，先记录一份理想的转录文本、一个失败案例以及回滚说明。 将配置置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
"sick leave policy" → [0.12, -0.87, 0.34, 0.56, ...] (1024 numbers)
"medical absence entitlement" → [0.13, -0.85, 0.31, 0.54, ...] ← very close
"quarterly revenue target" → [0.91, 0.23, -0.67, 0.02, ...] ← far away
稀疏嵌入（BM25）
将稀疏嵌入BM25阶段视为可度量的模型表面时，其效果最佳。在扩大范围之前，先记录一个成功的处理案例、一个失败案例以及回滚说明。同时记录正常流程与恢复流程的细节。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的功能。应将分块策略与检索策略分开设计，当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
"sick leave policy" → {word_index_for_"sick": 0.82, word_index_for_"leave": 0.91, ...}
为何需要两者兼备
“为何需要两者”这一阶段若被视为可度量的对象，效果会更好。在扩大范围之前，需记录一份最佳示例、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障点应指向单一责任模块，而非复杂的流程链。 应将分块策略与检索策略分开。当质量指标发生变化时，调整其中一项不应迫使重新编写另一项。 “为何需要两者”这一阶段若被视为可度量的对象，效果会更好。在扩大范围之前，需记录一份最佳示例、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
项目设置与依赖项
在项目设置与依赖关系阶段，应在修改代码之前明确输入内容、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 需引用实际作为答案依据的段落。若没有引用，操作人员就无法区分是虚假信息还是索引缺失所致。
pip install --upgrade pymilvus
pip install "pymilvus[model]"
pip install sentence-transformers
pip install langchain-text-splitters
pip install langchain-openai
pip install langchain-community
pip install scipy
pip install nltk
import uuid
from tqdm import tqdm
from pymilvus import (
MilvusClient, DataType,
AnnSearchRequest, RRFRanker
)
from pymilvus.model.sparse import BM25EmbeddingFunction
from pymilvus.model.sparse.bm25.tokenizers import build_default_analyzer
from langchain_community.document_loaders import PyPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
import scipy.sparse as sp
import re, json
import nltk
nltk.download('stopwords')
配置
在配置阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的内容。 必须引用那些真正作为答案依据的段落。如果没有引用，操作人员就无法区分是虚假信息还是索引缺失导致的错误。
PDF_PATH = "./data/sample_employee_handbook.pdf" # path of you document
COLLECTION_NAME = "rag_documents_hybrid"
MILVUS_DB_PATH = "./db/milvus_demo.db"
API_KEY = "sk-..."
EMBEDDING_MODEL = "text-embedding-3-large"
EMBEDDING_DIM = 1024
CHUNK_SIZE = 500
CHUNK_OVERLAP = 100
TOP_K = 5
构建索引处理流程
在构建索引管道阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 优先选择小型、可测试的单元，而非冗长的脚本。当某一步骤失败时，故障应指向单一责任点，而非复杂的管道结构。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是幻觉内容还是索引缺失问题。 在构建索引管道阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况可以避免在流程从演示模式转为正式运行时出现意外费用。
红色环境。PDF → Pages → Chunks → Dense Embeddings
→ Sparse Embeddings
→ Milvus Collection
步骤1和步骤2：初始化模型并建立连接
在执行“步骤1和步骤2：初始化”阶段时，首先写下相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改不会出错。 将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，这样操作人员无需查看整个系统结构即可进行审计。 缓存稳定的系统指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。
# Dense embedding model here we'll be using OpenAI's embedding model
embedding_obj = OpenAIEmbeddings(
model=EMBEDDING_MODEL,
api_key=API_KEY,
dimensions=EMBEDDING_DIM
)
# Milvus Lite - single file, no server needed
client = MilvusClient(MILVUS_DB_PATH)
print("Models and DB connection ready.")
步骤3和步骤4：加载文档并分块处理
在处理第3步和第4步的加载阶段时，首先写下相关契约：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在调整提示词之前，先使用固定的问题集来衡量检索效果。仅仅更换提示词很难解决检索能力不足的问题。
# Load PDF — one Document object per page
loader = PyPDFLoader(PDF_PATH)
documents = loader.load()
print(f"Loaded {len(documents)} pages.")
# Split into overlapping chunks
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=CHUNK_SIZE,
chunk_overlap=CHUNK_OVERLAP,
separators=["\n\n", "\n", ".", " ", ""]
)
chunks = text_splitter.split_documents(documents)
print(f"Created {len(chunks)} chunks.")
第5步：生成两种类型的嵌入向量
在执行第5步“生成两个版本”时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改有据可依。 相比冗长的脚本，应优先选择小型且易于测试的单元。当某一步骤失败时，故障应指向单一责任模块，而非复杂的流程链。 在调整提示词之前，需先使用固定的问题集来衡量召回率。仅仅更换提示词往往无法解决检索效果不佳的问题。
texts = [doc.page_content for doc in chunks]
# Dense embeddings - one API call for the entire corpus
print("Generating dense embeddings...")
dense_embeddings = embedding_obj.embed_documents(texts)
print(f"Dense dimension: {len(dense_embeddings[0])}")
# Sparse embeddings - BM25 must be fit on YOUR corpus first
print("Fitting BM25 on corpus...")
analyzer = build_default_analyzer(language="en") # for this you will require nltk-stopwords
bm25_ef = BM25EmbeddingFunction(analyzer)
bm25_ef.fit(texts) # Builds vocabulary from your documents
sparse_embeddings = bm25_ef.encode_documents(texts)
print("Sparse embeddings generated.")
在执行第5步“生成两个版本”时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这份清单能确保后续的代码修改有据可依。 除了功能结果外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
第6步：使用模式与索引创建集合
将第6步视为可度量的工作面最为有效。在扩大范围之前，先记录一份最佳案例、一个失败案例以及回滚说明。 配置应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个架构即可进行审计。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
# Drop and recreate for a clean state
if COLLECTION_NAME in client.list_collections():
client.drop_collection(COLLECTION_NAME)
# Define schema
schema = client.create_schema()
schema.add_field("id", DataType.VARCHAR, is_primary=True, max_length=100)
schema.add_field("vector", DataType.FLOAT_VECTOR, dim=EMBEDDING_DIM)
schema.add_field("sparse_vector", DataType.SPARSE_FLOAT_VECTOR)
schema.add_field("text", DataType.VARCHAR, max_length=65535)
schema.add_field("page_number", DataType.INT64)
schema.add_field("source", DataType.VARCHAR, max_length=500)
schema.add_field("chunk_id", DataType.INT64)
# Create the collection
client.create_collection(collection_name=COLLECTION_NAME, schema=schema)
# Build indexes separately
index_params = client.prepare_index_params()
index_params.add_index(
field_name="vector",
index_type="FLAT", # Exact search - swap to HNSW for production
metric_type="COSINE"
)
index_params.add_index(
field_name="sparse_vector",
index_type="SPARSE_INVERTED_INDEX",
metric_type="IP" # Inner Product is the only valid metric for sparse
)
client.create_index(collection_name=COLLECTION_NAME, index_params=index_params)
print("Collection and indexes created.")
第7步和第8步：准备记录并插入数据
将第7、8步的准备阶段视为可度量的对象来处理效果最佳。在扩大范围之前，先记录一个成功的用例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程的文档。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。应将分块策略与检索策略分开，当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
def sparse_to_dict(s_emb) -> dict:
"""Convert a scipy sparse row into Milvus-compatible {index: value} dict."""
if sp.issparse(s_emb):
coo = s_emb.tocoo()
return {int(col): float(val) for col, val in zip(coo.col, coo.data)}
elif isinstance(s_emb, dict):
return s_emb
else:
return {int(i): float(v) for i, v in enumerate(s_emb) if v != 0.0}
# Build the records list
data = []
for idx, (chunk, d_emb) in enumerate(tqdm(zip(chunks, dense_embeddings), total=len(chunks))):
sparse_dict = sparse_to_dict(sparse_embeddings[idx])
if not sparse_dict:
print(f"Warning: empty sparse vector at chunk {idx}, skipping.")
continue
data.append({
"id": str(uuid.uuid4()),
"vector": d_emb,
"sparse_vector": sparse_dict,
"text": chunk.page_content,
"page_number": int(chunk.metadata.get("page", -1)),
"source": PDF_PATH,
"chunk_id": idx
})
# Insert into Milvus
res = client.insert(collection_name=COLLECTION_NAME, data=data)
print(f"Inserted {res['insert_count']} records.")
# Load into memory - required before any search operation
client.load_collection(COLLECTION_NAME)
print(f"Load state: {client.get_load_state(COLLECTION_NAME)}")
基础RAG：密集向量搜索
将 Basic RAG Dense Vector 阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的输出样本、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且易于测试的单元。当某个步骤出现故障时，故障点应指向单一责任主体，而非复杂的流程链。 应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。 将 Basic RAG Dense Vector 阶段视为可度量的对象来处理时，其效果最佳。在扩大范围之前，先记录一份理想的输出样本、一个故障案例以及回滚说明。 除了功能结果外，还需记录处理时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
# ════════════════════════════════════════════════════════════
# Dense Vector Search
# ════════════════════════════════════════════════════════════
query = "What is the leave policy?"
# Step 1: Embed the query using the same model used at index time
query_dense_embedding = embedding_obj.embed_query(query)
# Step 2: Search
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_dense_embedding],
anns_field="vector",
search_param={"metric_type": "COSINE"},
limit=TOP_K,
output_fields=["text", "page_number", "source"]
)
# Step 3: Display results
for idx, hit in enumerate(results[0], start=1):
entity = hit["entity"]
print(f"Rank {idx} | Cosine Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
print(f" {entity['text'][:300]}\n")
更优的 RAG：混合搜索（密集型 + 稀疏型）
在改进版的RAG混合搜索阶段，应在修改代码之前明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审核。 需注明实际作为答案依据的段落。如果没有引用，操作人员就无法区分是幻觉内容还是索引缺失导致的错误。
# ════════════════════════════════════════════════════════════
# Hybrid Search (Dense + Sparse)
# ════════════════════════════════════════════════════════════
query = "leave policy?"
# Dense query vector
query_dense = embedding_obj.embed_query(query)
# Sparse query vector - uses the same BM25 model fitted on the corpus
sparse_raw = bm25_ef.encode_queries([query])
sparse_dict = sparse_to_dict(sparse_raw[0])
print(f"Sparse query terms: {len(sparse_dict)}") # Should be > 0
# Build two separate ANN search requests
dense_req = AnnSearchRequest(
data=[query_dense],
anns_field="vector",
param={"metric_type": "COSINE"},
limit=TOP_K
)
sparse_req = AnnSearchRequest(
data=[sparse_dict],
anns_field="sparse_vector",
param={"metric_type": "IP"},
limit=TOP_K
)
# Execute hybrid search with RRF fusion
results = client.hybrid_search(
collection_name=COLLECTION_NAME,
reqs=[dense_req, sparse_req],
ranker=RRFRanker(k=60),
limit=TOP_K,
output_fields=["text", "page_number", "source"]
)
for idx, hit in enumerate(results[0], start=1):
entity = hit["entity"]
print(f"Rank {idx} | RRF Score: {hit['distance']:.4f} | Page: {entity['page_number']}")
print(f" {entity['text'][:300]}\n")
高级RAG——四种检索技术
在高级RAG四阶段检索中，修改代码之前需明确输入内容、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品本身的组成部分，而非后续需要补充的功能。 必须标注出真正作为答案依据的段落。如果没有引用说明，操作人员就无法区分幻觉内容与索引缺失问题。
元数据过滤
在元数据过滤阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某个步骤失败时，故障原因应能明确指向单一责任模块，而非复杂的流程链。 需引用实际作为答案依据的段落。没有引用的话，操作人员就无法区分是虚假信息还是索引缺失导致的错误。 在元数据过滤阶段，修改代码之前需明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
# ════════════════════════════════════════════════════════════
# METADATA FILTERING
# ════════════════════════════════════════════════════════════
def search_with_metadata_filter(
client, collection_name, embedding_obj, bm25_ef,
query: str,
page_range: tuple = None,
source_file: str = None,
top_k: int = 5
):
filter_parts = []
if page_range:
lo, hi = page_range
filter_parts.append(f"page_number >= {lo} && page_number <= {hi}")
if source_file:
filter_parts.append(f'source == "{source_file}"')
filter_expr = " && ".join(filter_parts) if filter_parts else None
print(f"\n[Metadata Filter] Query : '{query}'")
print(f"[Metadata Filter] Filter: {filter_expr or 'None (unfiltered)'}")
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=query,
top_k=top_k,
filters=filter_expr
)
return results
# ── Run ──────────────────────────────────────────────────────
meta_results = search_with_metadata_filter(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
page_range = (1, 30),
source_file= None,
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nMETADATA-FILTERED RESULTS")
print("=" * 55)
if not meta_results or not meta_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(meta_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
'page_number >= 1 && page_number <= 30' # page range
'source == "hr_policy.pdf"' # exact source
'category in ["leave", "performance"]' # in a list
'source like "hr%"' # prefix match
查询重写
在处理查询重写阶段时，首先需明确相关规范：所需的输入参数、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 应将配置信息置于应用程序代码之外。环境文件、密钥存储以及功能开关应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 在调整提示词之前，需先使用固定的问题集来评估召回率。仅仅更换提示词往往无法改善较差的检索效果。
# ════════════════════════════════════════════════════════════
# QUERY REWRITING
# ════════════════════════════════════════════════════════════
import re, json
REWRITE_PROMPT = """You are an expert at reformulating search queries to improve document retrieval.
Given a user query, produce {n} alternative search queries that:
- Use formal, document-style language
- Include relevant keywords and synonyms
- Cover different angles of the same question
User query: {query}
Respond ONLY with a JSON array of strings. Example:
["rewritten query 1", "rewritten query 2", "rewritten query 3"]"""
def rewrite_query(query: str, n: int = 3) -> list[str]:
prompt = REWRITE_PROMPT.format(query=query, n=n)
response = llm.invoke(prompt)
raw = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
try:
variants = json.loads(raw)
return [query] + variants # always keep the original
except json.JSONDecodeError:
print("Warning: Could not parse rewrites, using original query only.")
return [query]
def search_with_query_rewriting(
client, collection_name, embedding_obj, bm25_ef,
query: str,
n_rewrites: int = 3,
top_k: int = 5
):
variants = rewrite_query(query, n=n_rewrites)
print(f"\n[Query Rewriting] Original : '{query}'")
for i, v in enumerate(variants[1:], 1):
print(f"[Query Rewriting] Variant {i} : '{v}'")
seen_ids = {}
rank_scores = {}
for variant in variants:
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=variant,
top_k=top_k
)
if not results or not results[0]:
continue
for rank, hit in enumerate(results[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
return [merged]
# ── Run ──────────────────────────────────────────────────────
rewrite_results = search_with_query_rewriting(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
n_rewrites = 3,
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nQUERY-REWRITTEN RESULTS")
print("=" * 55)
if not rewrite_results or not rewrite_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(rewrite_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
Input: "how many days off do I get?"
Output variants:
1. "annual leave entitlement number of days employee handbook"
2. "vacation days accrual policy full-time employee"
3. "paid time off PTO allowance per calendar year"
HyDE — 假设性文档嵌入
在处理 HyDE 假设文档嵌入阶段时，首先写下相关约定：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的组成部分，而非后续需要补充的功能。 在调整提示词之前，先使用固定的问题集来衡量召回率。仅仅更换提示词很难改善较差的检索效果。
# ════════════════════════════════════════════════════════════
# HyDE (Hypothetical Document Embeddings)
# ════════════════════════════════════════════════════════════
HYDE_PROMPT = """You are a corporate policy document writer.
Write a 2-3 paragraph excerpt from an official HR policy or company document
that would DIRECTLY ANSWER the following question.
Write in formal document style. Do not mention the question itself.
Question: {query}
Document excerpt:"""
def generate_hypothetical_document(query: str) -> str:
response = llm.invoke(HYDE_PROMPT.format(query=query))
return response.content.strip()
def search_with_hyde(
client, collection_name, embedding_obj, bm25_ef,
query: str,
top_k: int = 5
):
hypothetical_doc = generate_hypothetical_document(query)
print(f"\n[HyDE] Query : '{query}'")
print(f"[HyDE] Hypothetical doc :\n {hypothetical_doc[:300]}...\n")
# Search using the hypothetical document's embedding
hyde_results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=hypothetical_doc, # embed the answer, not the question
top_k=top_k
)
# Also search with the original query and merge both via RRF
original_results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=query,
top_k=top_k
)
seen_ids = {}
rank_scores = {}
for result_set in [hyde_results, original_results]:
if not result_set or not result_set[0]:
continue
for rank, hit in enumerate(result_set[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
return [merged]
# ── Run ──────────────────────────────────────────────────────
hyde_results = search_with_hyde(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy?",
top_k = 5
)
# ── Print Results ─────────────────────────────────────────────
print("\nHyDE RESULTS")
print("=" * 55)
if not hyde_results or not hyde_results[0]:
print("No results returned.")
else:
for idx, hit in enumerate(hyde_results[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
查询分解
在处理查询分解阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 优先选择小型、可测试的单元，而非冗长的脚本。当某个步骤出错时，错误应指向单一责任点，而非复杂的流程链。 在调整提示词之前，先使用固定的问题集来测试召回率。仅仅更换提示词很难改善较差的检索效果。 在处理查询分解阶段时，首先需明确相关规范：所需输入、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果之外，还需记录执行时间以及token或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
# ════════════════════════════════════════════════════════════
# QUERY DECOMPOSITION
# ════════════════════════════════════════════════════════════
DECOMPOSE_PROMPT = """You are an expert at breaking down complex questions for document retrieval.
Decompose the following question into 2-4 simple, self-contained sub-questions.
Each sub-question should target a single distinct piece of information.
Complex question: {query}
Respond ONLY with a JSON array of strings. Example:
["sub-question 1", "sub-question 2", "sub-question 3"]"""
def decompose_query(query: str) -> list[str]:
response = llm.invoke(DECOMPOSE_PROMPT.format(query=query))
raw = re.sub(r"^```json|^```|```quot;, "", response.content.strip(), flags=re.MULTILINE).strip()
try:
return json.loads(raw)
except json.JSONDecodeError:
print("Warning: Could not parse decomposition, using original query.")
return [query]
def search_with_decomposition(
client, collection_name, embedding_obj, bm25_ef,
query: str,
top_k: int = 5
):
sub_questions = decompose_query(query)
print(f"\n[Decomposition] Original query : '{query}'")
for i, sq in enumerate(sub_questions, 1):
print(f"[Decomposition] Sub-question {i} : '{sq}'")
per_subquery_results = {}
seen_ids = {}
rank_scores = {}
for sq in sub_questions:
results = hybrid_search(
client, collection_name, embedding_obj, bm25_ef,
query_text=sq,
top_k=top_k
)
per_subquery_results[sq] = results
if not results or not results[0]:
continue
for rank, hit in enumerate(results[0], start=1):
hit_id = hit["id"]
rank_scores[hit_id] = rank_scores.get(hit_id, 0) + 1.0 / (60 + rank)
if hit_id not in seen_ids:
seen_ids[hit_id] = hit
merged = sorted(seen_ids.values(), key=lambda h: rank_scores[h["id"]], reverse=True)[:top_k]
# Per sub-question breakdown
print("\n── Per Sub-question Results ──")
for sq, res in per_subquery_results.items():
print(f"\n SUB-QUERY: '{sq[:60]}'")
if res and res[0]:
for i, hit in enumerate(res[0], start=1):
print(f" {i}. Page {hit['entity']['page_number']} | Score {hit['distance']:.4f} | {hit['entity']['text'][:150]}")
return {"per_subquery": per_subquery_results, "merged": [merged]}
# ── Run ──────────────────────────────────────────────────────
decomp_results = search_with_decomposition(
client, COLLECTION_NAME, embedding_obj, bm25_ef,
query = "What is the leave policy and how does it affect salary deductions?",
top_k = 5
)
# ── Print Merged Results ──────────────────────────────────────
print("\nDECOMPOSED — MERGED FINAL RESULTS")
print("=" * 55)
merged_hits = decomp_results["merged"]
if not merged_hits or not merged_hits[0]:
print("No results returned.")
else:
for idx, hit in enumerate(merged_hits[0], start=1):
entity = hit["entity"]
print(f"\nRank : {idx}")
print(f"Score : {hit['distance']:.4f}")
print(f"Page : {entity['page_number']}")
print(f"Text :\n{entity['text'][:400]}")
Input: "What is the leave policy and how does performance review affect salary?"
Sub-questions:
1. "What is the annual leave policy?"
2. "How many sick days are employees entitled to?"
3. "How does performance review affect salary?"
4. "What is the performance review schedule?"
交叉编码器重排序
将交叉编码器重排序阶段视为可度量的模型表面时，其效果最佳。在扩大范围之前，需记录一个理想案例、一个失败案例以及回滚说明。 配置应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放，以便操作人员无需查看整个系统结构即可进行审计。 将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制要求重新编写另一项。
# ============================================================
# RERANKING WITH CROSS-ENCODER
# ============================================================
from sentence_transformers import CrossEncoder
# Huggingface: cross-encoder/ms-marco-MiniLM-L12-v2
cross_encoder = CrossEncoder("cross-encoder/ms-marco-MiniLM-L12-v2")
query = "What is the leave policy?"
RETRIEVAL_K = 20 # fetch more than you need
FINAL_K = 5 # rerank down to this
# Step 1: Broad retrieval - fetch 20 candidates
query_dense = embedding_obj.embed_query(query)
results = client.search(
collection_name=COLLECTION_NAME,
data=[query_dense],
anns_field="vector",
search_param={"metric_type": "COSINE"},
limit=RETRIEVAL_K,
output_fields=["text", "page_number", "source"]
)
hits = results[0]
print(f"Retrieved {len(hits)} candidates for reranking.")
# Step 2: Score each (query, chunk) pair with the cross-encoder
pairs = [[query, hit["entity"]["text"]] for hit in hits]
rerank_scores = cross_encoder.predict(pairs)
# Step 3: Sort by cross-encoder score
for hit, score in zip(hits, rerank_scores):
hit["rerank_score"] = float(score)
reranked = sorted(hits, key=lambda x: x["rerank_score"], reverse=True)[:FINAL_K]
# Step 4: Display
for idx, hit in enumerate(reranked, start=1):
entity = hit["entity"]
print(f"Rank {idx} | Rerank: {hit['rerank_score']:.4f} | Vector: {hit['distance']:.4f}")
print(f" Page {entity['page_number']}: {entity['text'][:300]}\n")
各技术的对比
将“技术对比方式”这一环节视为可度量的指标会更为有效。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。同时记录正常流程和恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。应将分块策略与检索策略分开处理；当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
结论
将“结论阶段”视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想的输出样本、一个故障案例以及回滚说明。 相较于庞大的脚本，应优先选择小型且可测试的单元。当某一步骤出现故障时，故障原因应能明确指向某个特定责任方，而非复杂的流程链。 应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。 将“结论阶段”视为可度量的对象来处理效果最佳。在扩大范围之前，需记录一份理想的输出样本、一个故障案例以及回滚说明。 除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。
操作检查清单
将操作检查清单阶段视为可度量的标准，效果最佳。在扩大范围之前，先记录一份完美的测试用例、一个故障案例以及回滚说明。
把这一阶段视为输入与已验证输出之间的契约。为相关文档命名，明确成功标准，绝不允许出现悄无声息的半完成状态。
将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应迫使重新编写另一项。
在预算允许的情况下，使用测试环境而非真实的付费 API，在持续集成过程中添加能够检测关键路径的冒烟测试。
在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外账单。
应将分块策略与检索策略分开。当质量指标发生变化时，修改其中一项不应强制重新编写另一项。
在推广该技术栈之前，需冻结版本，为关键路径生成标准转录文本，并明确回滚步骤。共享环境需要设置速率限制、进行租户检查，同时指定专人负责密钥轮换工作。与其追求花哨的一次性演示，不如注重扎实的可靠性。
c9664ffe2213的批处理说明：不要将提供商密钥放入代码仓库，为每个会话设置令牌上限，并将转录文本存储在评估用文件旁，以便后续更换模型时保持数据可比性。
部署说明1（c9664ffe2213）：锁定图像版本，设定请求预算，在大规模部署前在测试环境中验证租户隔离效果。
部署说明2（c9664ffe2213）：锁定图像版本，设定请求预算，在大规模部署前在测试环境中验证租户隔离效果。
部署注意事项3（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署注意事项4（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署注意事项5（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署注意事项6（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署注意事项7（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署注意事项8（c9664ffe2213）：在更大范围推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署说明9（c9664ffe2213）：在更广泛的推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。
部署说明10（c9664ffe2213）：在更广泛的推广之前，先锁定镜像、设置请求预算，并在测试环境中验证租户隔离功能。