RAG 系列 3：使用 Qwen3 Embedding 实现评论数据语义搜索

上一篇已经完成 Qwen3-Embedding-0.6B 的本地部署，也分别使用了 LangChain 的 embed_documents() 和 embed_query()。前文只比较了几句话，这一篇把这些知识放进一个更接近真实使用场景的案例中。

我们会下载 1000 条美食评论，从中取 100 条生成向量，然后输入中文或英文问题，找出语义最接近的 3 条评论。

这个案例只使用 CSV、Embedding 和余弦相似度。它不是 RAG，也没有使用 FAISS、Chroma、Retriever 或其他向量数据库。

1. 关键词搜索和语义搜索有什么区别

假设评论中写的是：

My kids love these snacks. They are healthier than traditional snack bars.

用户搜索的是：

适合孩子吃的健康零食

两段文本的语言不同，也没有完全相同的单词。普通关键词匹配很难把它们联系起来，但 Embedding 模型可以分别把评论和查询转换成向量，再比较两个向量在语义空间中的方向。

因此，语义搜索比较的不是“文字是否一样”，而是“表达的意思是否接近”。

2. 整体流程

整个程序分为两个阶段。

第一个阶段是离线准备：

1. 下载原始评论 CSV。
2. 清洗 Summary 和 Text。
3. 使用 embed_documents() 批量生成评论向量。
4. 把评论与向量一起写入新的 CSV。

第二个阶段是在线查询：

1. 使用 embed_query() 生成查询向量。
2. 从 CSV 读取评论，将向量字符串还原为 NumPy 数组。
3. 将查询向量与每条评论向量计算余弦相似度。
4. 使用 nlargest() 取得分数最高的 3 条评论。

这里的“在线查询”指程序收到问题时执行的查询阶段，不是说程序需要调用外网服务。本文的模型和评论数据都保存在本机。

评论向量不需要在每次查询时重新生成。只要原始评论、文本合并方式、Embedding 模型版本、文档端编码配置和归一化配置都没有改变，就可以重复使用已经生成的向量文件。

3. 准备运行环境

本篇继续使用 RAG 系列 2 创建的 .venv_embedding，不再重复下载 Qwen/Qwen3-Embedding-0.6B。

目录中需要存在模型入口：

llm_learning/embedding_model

进入项目并安装本文依赖：

cd source/_posts/llm_learning

source .venv_embedding/bin/activate
python -m pip install \
  -r rag/p03_review_semantic_search/requirements.txt
python -m pip check

requirements.txt 的内容为：

-r ../p02_local_qwen3_embeddings/requirements.txt
pandas==2.3.3
numpy==2.2.6

这里复用 RAG 系列 2 的 LangChain、SentenceTransformer 和 Transformers 版本，只增加数据处理所需的 Pandas 与 NumPy。

本机检查结果为：

No broken requirements found.

4. 下载评论数据

本文使用 OpenAI Cookbook 中的 Fine Food Reviews 数据。为了避免上游文件发生变化，下载地址固定到了具体 Git 提交，并使用 SHA-256 校验文件内容。

代码文件为 01_download_review_data.py：

# 这个文件下载评论语义搜索使用的美食评论数据。
# 下载完成后会检查 SHA-256、记录数量和字段，确保数据可以稳定复现。

import csv
import hashlib
from pathlib import Path
from urllib.error import URLError
from urllib.request import ProxyHandler, build_opener


PROJECT_ROOT = Path(__file__).resolve().parents[2]
DATA_DIR = PROJECT_ROOT / "data" / "p03_review_semantic_search"
DATA_FILE = DATA_DIR / "fine_food_reviews_1k.csv"

# 固定到 OpenAI Cookbook 的具体提交，避免上游文件变化影响运行结果。
DATA_URL = (
    "https://raw.githubusercontent.com/openai/openai-cookbook/"
    "2515ddc7b8905e56160693e1a1fe08a6683436e4/"
    "examples/data/fine_food_reviews_1k.csv"
)
EXPECTED_SHA256 = "716853f9ba5d857aef56fe1f3d46189a20e94403a950c8640a7b6b8851cc92c0"
EXPECTED_ROWS = 1000
EXPECTED_FIELDS = ["", "Time", "ProductId", "UserId", "Score", "Summary", "Text"]


def calculate_sha256(file_path: Path) -> str:
    """分块计算文件的 SHA-256，避免一次把整个文件读入内存。"""
    digest = hashlib.sha256()
    with file_path.open("rb") as file:
        for block in iter(lambda: file.read(1024 * 1024), b""):
            digest.update(block)
    return digest.hexdigest()


DATA_DIR.mkdir(parents=True, exist_ok=True)

# 文件不存在或校验失败时重新下载。先尝试直连，失败后再使用系统网络配置。
if not DATA_FILE.exists() or calculate_sha256(DATA_FILE) != EXPECTED_SHA256:
    print("开始下载评论数据……")
    direct_opener = build_opener(ProxyHandler({}))
    try:
        with direct_opener.open(DATA_URL, timeout=10) as response:
            DATA_FILE.write_bytes(response.read())
        print("下载方式：直连")
    except (TimeoutError, URLError):
        print("直连下载失败，改用系统网络配置重试。")
        with build_opener().open(DATA_URL, timeout=60) as response:
            DATA_FILE.write_bytes(response.read())
        print("下载方式：系统网络配置")
else:
    print("数据文件已存在且校验通过，不重复下载。")

actual_sha256 = calculate_sha256(DATA_FILE)
if actual_sha256 != EXPECTED_SHA256:
    raise ValueError(f"数据文件 SHA-256 不正确：{actual_sha256}")

# 使用标准库读取 CSV，只检查结构，不在下载步骤引入 Pandas 处理逻辑。
with DATA_FILE.open("r", encoding="utf-8", newline="") as file:
    reader = csv.DictReader(file)
    rows = list(reader)
    fields = reader.fieldnames

if len(rows) != EXPECTED_ROWS:
    raise ValueError(f"数据记录数不正确：期望 {EXPECTED_ROWS}，实际 {len(rows)}")

if fields != EXPECTED_FIELDS:
    raise ValueError(f"数据字段不正确：{fields}")

print("数据文件：", DATA_FILE)
print("文件大小：", DATA_FILE.stat().st_size, "bytes")
print("SHA-256：", actual_sha256)
print("记录数量：", len(rows))
print("字段列表：", fields)

运行：

python rag/p03_review_semantic_search/01_download_review_data.py

本机直连 GitHub Raw 地址失败，程序自动改用系统网络配置后下载成功：

开始下载评论数据……
直连下载失败，改用系统网络配置重试。
下载方式：系统网络配置
文件大小： 436148 bytes
SHA-256： 716853f9ba5d857aef56fe1f3d46189a20e94403a950c8640a7b6b8851cc92c0
记录数量： 1000
字段列表： ['', 'Time', 'ProductId', 'UserId', 'Score', 'Summary', 'Text']

如果文件已经存在且哈希正确，脚本不会重复下载。

5. 认识评论数据

CSV 中真正需要使用的字段如下：

字段	含义
Time	评论时间戳
ProductId	商品编号
UserId	用户编号
Score	评分，取值为 1 到 5
Summary	评论标题或摘要
Text	评论正文

原始 CSV 的第一列没有字段名，它保存的是原数据行号。因此 Pandas 读取时使用：

source_df = pd.read_csv(SOURCE_FILE, index_col=0)

这样第一列会成为 DataFrame 索引，不会被误当成业务字段。

6. 清洗并合并评论文本

一条评论的语义同时存在于标题和正文中。只使用标题可能丢失细节，只使用正文又可能忽略总结，所以把两者合并成统一文本：

Summary: 评论标题; Text: 评论正文

在合并之前先做三项处理：

1. 删除标题或正文为空的记录。
2. 去掉文本两端空格。
3. 默认截取前 100 条，降低学习和重复测试的等待时间。

将 REVIEW_LIMIT 从 100 改为 1000，就可以处理完整数据集。

7. 批量生成评论向量

代码文件为 02_build_review_embeddings.py：

# 这个文件读取美食评论，清洗文本并批量生成评论向量。
# 生成结果会写入新的 CSV，供后续语义搜索脚本直接读取。

from pathlib import Path
from time import perf_counter

import pandas as pd
import torch
from langchain_huggingface import HuggingFaceEmbeddings


PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_PATH = PROJECT_ROOT / "embedding_model"
DATA_DIR = PROJECT_ROOT / "data" / "p03_review_semantic_search"
SOURCE_FILE = DATA_DIR / "fine_food_reviews_1k.csv"
OUTPUT_FILE = DATA_DIR / "reviews_with_qwen3_embeddings.csv"

# 默认处理前 100 条，便于学习和测试；改成 1000 可以处理完整数据集。
REVIEW_LIMIT = 100

if not MODEL_PATH.exists():
    raise FileNotFoundError("没有找到 embedding_model，请先完成 RAG 系列 2 的模型部署。")

if not SOURCE_FILE.exists():
    raise FileNotFoundError("没有找到评论数据，请先运行 01_download_review_data.py。")

device = "mps" if torch.backends.mps.is_available() else "cpu"

# 文档向量做归一化；查询端配置留给下一步的 embed_query() 使用。
embeddings_model = HuggingFaceEmbeddings(
    model_name=str(MODEL_PATH),
    model_kwargs={"device": device},
    encode_kwargs={"normalize_embeddings": True, "batch_size": 8},
    query_encode_kwargs={
        "prompt_name": "query",
        "normalize_embeddings": True,
    },
)

# 第一列是原数据保存的行号，因此使用 index_col=0 读取。
source_df = pd.read_csv(SOURCE_FILE, index_col=0)
reviews_df = source_df[["Time", "ProductId", "UserId", "Score", "Summary", "Text"]].copy()

# 摘要或正文为空时无法形成完整评论文本，先删除再截取指定数量。
reviews_df = reviews_df.dropna(subset=["Summary", "Text"])
reviews_df["Summary"] = reviews_df["Summary"].astype(str).str.strip()
reviews_df["Text"] = reviews_df["Text"].astype(str).str.strip()
reviews_df = reviews_df[(reviews_df["Summary"] != "") & (reviews_df["Text"] != "")]
reviews_df = reviews_df.head(REVIEW_LIMIT).copy()

# 将摘要和正文合并成一段文本，让一个向量同时表达两部分语义。
reviews_df["text_content"] = (
    "Summary: " + reviews_df["Summary"] + "; Text: " + reviews_df["Text"]
)

start_time = perf_counter()
vectors = embeddings_model.embed_documents(reviews_df["text_content"].tolist())
elapsed = perf_counter() - start_time

# Pandas 会把 Python 列表序列化为 CSV 单元格中的字符串。
reviews_df["embedding"] = vectors
reviews_df.to_csv(OUTPUT_FILE, index=False)

first_vector_norm = sum(value * value for value in vectors[0]) ** 0.5
print("运行设备：", device)
print("原始记录数：", len(source_df))
print("生成向量数：", len(vectors))
print("向量维度：", len(vectors[0]))
print(f"首条向量范数：{first_vector_norm:.6f}")
print(f"向量生成耗时：{elapsed:.2f} 秒")
print(f"输出文件大小：{OUTPUT_FILE.stat().st_size / 1024 / 1024:.2f} MiB")
print("输出文件：", OUTPUT_FILE)

运行：

python rag/p03_review_semantic_search/02_build_review_embeddings.py

本机使用 Apple MPS 的真实输出为：

运行设备： mps
原始记录数： 1000
生成向量数： 100
向量维度： 1024
首条向量范数：1.000000
向量生成耗时：6.69 秒
输出文件大小：2.25 MiB

按照 LangChain Embeddings 接口的划分，embed_documents() 接收多条文本，因此输入类型是 list[str]，返回类型是 list[list[float]]。100 条评论会得到 100 个向量，每个向量包含 1024 个浮点数。

normalize_embeddings=True 把向量归一化到长度约为 1。实测首条向量范数为 1.000000。

8. 为什么 CSV 中的向量变成了字符串

在 Python 中，一条向量是浮点数列表：

[0.0123, -0.0456, 0.0789]

CSV 只有行、列和文本单元格，并没有“浮点数列表”这种字段类型。Pandas 写入 CSV 后，整个列表会变成一段字符串：

"[0.0123, -0.0456, 0.0789]"

重新读取时不能直接参与 NumPy 运算，需要先还原：

from ast import literal_eval

vector = literal_eval("[0.0123, -0.0456, 0.0789]")

不要使用 eval() 解析外部数据。literal_eval() 只接受 Python 的基础字面量，风险更小。即便如此，生产系统仍应保证数据文件来源可信。

9. 计算余弦相似度

查询向量为 (A)，评论向量为 (B)，余弦相似度公式为：

$$
\cos(\theta) = \frac{A \cdot B}{\lVert A \rVert \lVert B \rVert}
$$

对应代码：

def cosine_similarity(vector_a: np.ndarray, vector_b: np.ndarray) -> float:
    denominator = np.linalg.norm(vector_a) * np.linalg.norm(vector_b)
    if denominator == 0:
        raise ValueError("余弦相似度不能用于零向量。")
    return float(np.dot(vector_a, vector_b) / denominator)

当前查询向量和评论向量都已经归一化。理论上两个向量的模长都是 1，此时点积就等于余弦相似度。不过这里仍保留完整公式，便于理解，也提醒关闭归一化后需要同步修改计算方法。

10. 使用 embed_query() 生成查询向量

评论使用 embed_documents()，查询使用 embed_query()。两者调用的是同一个模型，但 Qwen3 Embedding 为查询端提供了专门的 query prompt：

query_encode_kwargs={
    "prompt_name": "query",
    "normalize_embeddings": True,
}

文档端不应使用查询提示。否则生成向量时，查询和文档的角色会混在一起，可能降低检索效果。

还需要满足两个条件：

• 查询和评论必须使用同一个 Embedding 模型。
• 两边必须使用兼容的维度、归一化和提示配置。

不同模型生成的向量通常不在同一个向量空间中，不能直接计算相似度。

11. 完整搜索代码

代码文件为 03_search_review_embeddings.py：

# 这个文件把用户问题转换成向量，再从评论数据中找出语义最接近的内容。
# 示例使用完整的余弦相似度公式，方便理解语义搜索的计算过程。

from ast import literal_eval
from pathlib import Path

import numpy as np
import pandas as pd
import torch
from langchain_huggingface import HuggingFaceEmbeddings


PROJECT_ROOT = Path(__file__).resolve().parents[2]
MODEL_PATH = PROJECT_ROOT / "embedding_model"
DATA_FILE = (
    PROJECT_ROOT
    / "data"
    / "p03_review_semantic_search"
    / "reviews_with_qwen3_embeddings.csv"
)

if not DATA_FILE.exists():
    raise FileNotFoundError("没有找到评论向量文件，请先运行 02_build_review_embeddings.py。")

device = "mps" if torch.backends.mps.is_available() else "cpu"

# 查询向量使用 Qwen3 Embedding 内置的 query prompt，文档向量不使用该提示。
embeddings_model = HuggingFaceEmbeddings(
    model_name=str(MODEL_PATH),
    model_kwargs={"device": device},
    encode_kwargs={"normalize_embeddings": True, "batch_size": 8},
    query_encode_kwargs={
        "prompt_name": "query",
        "normalize_embeddings": True,
    },
)


def cosine_similarity(vector_a: np.ndarray, vector_b: np.ndarray) -> float:
    """按照点积除以两个向量模长的公式计算余弦相似度。"""
    denominator = np.linalg.norm(vector_a) * np.linalg.norm(vector_b)
    if denominator == 0:
        raise ValueError("余弦相似度不能用于零向量。")
    return float(np.dot(vector_a, vector_b) / denominator)


def search_reviews(query: str, top_n: int = 3) -> None:
    """搜索评论并打印相似度最高的前 top_n 条结果。"""
    reviews_df = pd.read_csv(DATA_FILE)

    # CSV 中的向量是字符串，literal_eval 可以安全还原成 Python 列表。
    reviews_df["embedding_vector"] = reviews_df["embedding"].apply(
        lambda value: np.asarray(literal_eval(value), dtype=np.float32)
    )

    query_vector = np.asarray(embeddings_model.embed_query(query), dtype=np.float32)
    reviews_df["similarity"] = reviews_df["embedding_vector"].apply(
        lambda vector: cosine_similarity(vector, query_vector)
    )

    results = reviews_df.nlargest(top_n, "similarity")

    print(f"\n查询：{query}")
    for rank, (_, row) in enumerate(results.iterrows(), start=1):
        # 合并连续空白并截断正文，避免终端被一整篇评论占满。
        text = " ".join(str(row["Text"]).split())
        if len(text) > 220:
            text = text[:220] + "..."

        print(f"{rank}. 相似度：{row['similarity']:.4f}，评分：{int(row['Score'])}")
        print("   标题：", row["Summary"])
        print("   评论：", text)


queries = [
    "delicious beans",
    "适合孩子吃的健康零食",
    "I like juicy barbecued meat.",
]

for query_text in queries:
    search_reviews(query_text)

运行：

python rag/p03_review_semantic_search/03_search_review_embeddings.py

12. 三组真实查询结果

12.1 查询 delicious beans

查询：delicious beans
1. 相似度：0.5121，评分：5
   标题： Delicious!
2. 相似度：0.4874，评分：3
   标题： This price is ridiculous!
3. 相似度：0.3984，评分：5
   标题： yummy

第一条评论正文明确提到了 white beans 和 seasoning，与查询语义一致。第二条虽然标题看不出来，但正文讨论了不同口味的 Beanitos，因此也得到了较高分数。

12.2 查询适合孩子吃的健康零食

查询：适合孩子吃的健康零食
1. 相似度：0.6257，评分：5
   标题： Yum
2. 相似度：0.5089，评分：5
   标题： Woody and Arlo's favorites!
3. 相似度：0.5026，评分：5
   标题： Tasty, and the yogurt coating is a nice addition

第一条英文评论写到孩子喜欢这些零食，而且比传统零食棒更健康。查询是中文，评论是英文，仍然排在第一位，说明模型能够进行跨语言语义匹配。

第二条实际是健康、低脂的宠物零食。它命中了“健康零食”的含义，却不满足“给孩子吃”这个业务条件。这是单纯向量相似度的典型局限。

12.3 查询 I like juicy barbecued meat.

查询：I like juicy barbecued meat.
1. 相似度：0.5491，评分：5
   标题： Makes me drool just thinking of them
2. 相似度：0.5491，评分：5
   标题： Makes me drool just thinking of them
3. 相似度：0.4694，评分：5
   标题： Tasty and Quick Pasta

前两条内容相同，正文提到了 steak and onions，因此与肉类查询接近。重复结果不是计算错误，而是原始 100 条数据中存在重复评论。

13. 为什么有些结果不够理想

语义相似度只是“接近程度”，不是事实判断或业务规则判断。本次结果暴露了几个实际问题。

13.1 数据量太小

默认只搜索 100 条评论。如果数据中没有真正匹配的内容，程序仍会返回分数最高的 3 条，但“最高”不代表“一定正确”。

13.2 原始数据存在重复

重复评论会占用多个 Top-N 位置。正式使用前应根据评论 ID、正文哈希或业务主键去重。

13.3 缺少业务过滤条件

“健康的宠物零食”和“健康的儿童零食”在向量空间中可能接近。真实系统通常还会结合商品类别、评分、时间等结构化字段过滤。

13.4 没有最低分数阈值

当前代码无论分数高低都会返回 3 条。实际业务可以设置最低相似度，低于阈值时直接提示没有合适结果。阈值不能凭感觉固定，应使用真实测试数据评估后确定。

14. CSV 暴力检索适合什么场景

本例每次查询都会执行这些操作：

1. 从 CSV 读取全部评论。
2. 把每个向量字符串还原成数组。
3. 将查询向量与全部评论逐条计算相似度。
4. 使用 nlargest() 从全部分数中选出 Top-N。

100 条或 1000 条数据用于学习完全足够，代码也非常直观。但当数据达到几十万或几百万条时，逐条比较的速度和内存成本都会明显增加。

因此，CSV 暴力检索适合：

• 学习 Embedding 和余弦相似度。
• 验证模型是否适合某类数据。
• 小规模离线实验。
• 排查向量和查询配置问题。

它不适合大规模生产检索。后续学习向量数据库时，再解决索引、近似最近邻、增量更新和持久化查询等问题。

15. 常见问题

15.1 下载一直超时

当前脚本先尝试直连，10 秒内失败后使用系统网络配置重试。本机就是通过第二种方式下载成功的。

如果两种方式都失败，先检查浏览器能否打开数据链接，再检查 VPN、代理和 DNS。不要跳过 SHA-256 校验，否则无法确认文件是否完整。

15.2 找不到 embedding_model

说明还没有完成 RAG 系列 2，或者模型符号链接失效。进入 llm_learning 后检查：

ls -l embedding_model

15.3 找不到评论数据或向量文件

三个脚本必须按编号执行：

python rag/p03_review_semantic_search/01_download_review_data.py
python rag/p03_review_semantic_search/02_build_review_embeddings.py
python rag/p03_review_semantic_search/03_search_review_embeddings.py

15.4 literal_eval 解析失败

先确认读取的是 reviews_with_qwen3_embeddings.csv，并且 embedding 列没有被手工修改。如果写入过程被中断，重新运行第二个脚本生成文件。

15.5 查询和评论向量维度不同

最常见原因是两边使用了不同模型，或者更换模型后没有重新生成评论向量。修改模型后必须删除旧向量文件并重新执行第二步。

16. 总结

这次完成了一个最小但完整的语义搜索流程：

1. 使用 Pandas 清洗真实评论。
2. 使用 LangChain 的 embed_documents() 批量生成文档向量。
3. 将评论和向量保存到 CSV。
4. 使用 literal_eval() 还原向量。
5. 使用 embed_query() 生成查询向量。
6. 使用 NumPy 计算余弦相似度。
7. 按分数降序返回 Top-N 评论。

这个案例把 RAG 系列 1 的向量原理和 RAG 系列 2 的本地 Qwen3 Embedding 接口真正组合了起来。同时也能看到，Embedding 解决的是语义表示问题，不会自动解决数据去重、业务过滤和大规模索引问题。

检查项	本次结果
原始数据	1000 条美食评论
默认参与检索的数据	100 条
Embedding 模型	Qwen/Qwen3-Embedding-0.6B
LangChain 文档接口	embed_documents()
LangChain 查询接口	embed_query()
运行设备	Apple MPS
向量维度	1024
首条向量范数	1.000000
100 条向量生成耗时	6.69 秒
向量 CSV 大小	2.25 MiB
检索方式	NumPy 余弦相似度全量比较
返回数量	每次查询 Top 3