向量检索实操：用词袋向量找相近问题，再用召回率比较方法

用户换个说法，知识库就搜不到了

学习站的常见问题里有一条“如何减少大模型的幻觉”，用户却问“怎么让模型别瞎编”，检索返回了 JSON 格式那一条。本课先用纯 Python 做一个看得见内部的“词袋向量”检索，弄清它为什么会错；再说明真实 embedding 模型与向量数据库各管什么；最后用人工标注算召回率，决定该换方法还是该调阈值。

完成后你会得到四个文件：faq_search.py（词袋检索并保存排名）、store.py（会拒绝混用模型的小向量库）、evaluate.py（召回评测）、rank_from_vectors.py（把真实 embedding 响应变成同格式排名）。准备 Python 3.12 和空目录 embed-lab，只用标准库，不联网。问题库是教学虚构的，编号是稳定标识，日志和评测只认编号。

第一步：把问题变成词袋向量

“向量”就是一串数字。最朴素的做法是数词：每个词一个位置，出现几次就写几。中文没有空格，这里不做分词，而是把相邻两个字当一个特征（字二元组），英文和数字按整词计。两段文本的相似度用余弦：方向越接近越接近 1，完全不共享特征为 0。保存为 faq_search.py：

import json
import math
import re
from collections import Counter

# 教学虚构的常见问题库；编号是稳定标识，评测和日志都用编号而不是问题原文
FAQ = {
    "F1": "如何减少大模型的幻觉",
    "F2": "怎样让模型输出合法的JSON",
    "F3": "提示词太长会不会被截断",
    "F4": "RAG检索不到相关资料怎么办",
    "F5": "温度参数应该设多少",
    "F6": "如何估算调用模型的费用",
}
QUERIES = {
    "Q1": "怎么让模型别瞎编",
    "Q2": "模型输出的JSON不合法",
    "Q3": "调用模型要花多少钱",
    "Q4": "知识库搜不到资料",
    "Q5": "temperature设多少合适",
    "Q6": "输入太长被截断了",
    "Q7": "怎么申请退款",
}


def bag_of_words(text):
    """英文和数字按整词计，中文按相邻两个字计（字二元组），返回 {特征: 次数}。"""
    features = Counter()
    for run in re.findall(r"[a-z0-9]+|[一-鿿]+", text.lower()):
        if run.isascii():
            features[run] += 1
        elif len(run) == 1:
            features[run] += 1
        else:
            features.update(run[i:i + 2] for i in range(len(run) - 1))
    return features


def cosine(a, b):
    dot = sum(a[k] * b[k] for k in a.keys() & b.keys())
    norm = math.sqrt(sum(v * v for v in a.values())) * math.sqrt(sum(v * v for v in b.values()))
    return dot / norm if norm else 0.0


doc_vectors = {fid: bag_of_words(text) for fid, text in FAQ.items()}
rankings = {}
for qid, query in QUERIES.items():
    qv = bag_of_words(query)
    scored = sorted(((cosine(qv, dv), fid) for fid, dv in doc_vectors.items()),
                    key=lambda pair: (-pair[0], pair[1]))
    rankings[qid] = [[fid, round(score, 4)] for score, fid in scored if score > 0]
    top = "  ".join(f"{fid}:{score:.2f}" for fid, score in rankings[qid][:3]) or "（无匹配）"
    print(f"{qid} {query} -> {top}")

print("Q1 的特征：", sorted(bag_of_words(QUERIES["Q1"])))
with open("rank_bow.json", "w", encoding="utf-8") as f:
    json.dump({"method": "字二元组词袋", "rankings": rankings}, f, ensure_ascii=False, indent=1)
print("已保存 rank_bow.json")

assert cosine(bag_of_words("温度"), bag_of_words("温度")) == 1.0
assert cosine(bag_of_words("温度"), bag_of_words("费用")) == 0.0

实际输出：

Q1 怎么让模型别瞎编 -> F2:0.24  F1:0.13  F6:0.12
Q2 模型输出的JSON不合法 -> F2:0.60  F1:0.13  F6:0.12
Q3 调用模型要花多少钱 -> F6:0.34  F5:0.12  F1:0.12
Q4 知识库搜不到资料 -> F4:0.23
Q5 temperature设多少合适 -> F5:0.32
Q6 输入太长被截断了 -> F3:0.36
Q7 怎么申请退款 -> F4:0.13
Q1 的特征： ['么让', '别瞎', '型别', '怎么', '模型', '瞎编', '让模']
已保存 rank_bow.json

逐条看原因，而不是只看对错。Q1 错在“瞎编”和“幻觉”没有一个共同的字，反倒和 F2 共享“让模”“模型”，所以 F2 排第一。Q5 虽然找对了，但靠的是“设多少”，“temperature”和“温度”并没有对上，换个问法就可能失手。Q7 问退款，库里根本没有答案，却因为“怎么”被匹配到 F4。词袋只认字面重合，这三种现象正是语义向量要解决、也是评测要抓住的问题。

第二步：真实 embedding 模型和向量数据库各管什么

embedding 模型把一段文本变成固定长度的稠密向量（常见几百到几千维），训练目标是让意思相近的文本方向相近，所以“别瞎编”和“幻觉”有机会靠在一起。它有几条硬约束：文档和查询必须用同一个模型、同一维度生成，换了模型或维度，旧向量要全部重算；单个维度没有可读含义，不能拿某一维去解释“这句话讲费用”；输入有长度上限，长文要先切分。

向量数据库不理解文字，它负责存“编号、向量、元数据”，按相似度找最近的几条，并支持按元数据过滤、更新和删除。数据量大时它用近似索引加速，结果可能和逐条比较略有出入，所以评测时要以逐条比较为参照。答案是否正确、资料是否过时，仍要靠元数据和你的评测来把关。

下面用四维手工向量把这些分工演示出来。保存为 store.py：

import math


def normalize(vec):
    norm = math.sqrt(sum(x * x for x in vec))
    if norm == 0:
        raise ValueError("零向量没有方向，不能参与比较")
    return [x / norm for x in vec]


def dot(a, b):
    if len(a) != len(b):
        raise ValueError(f"维度不一致：{len(a)} 与 {len(b)}")
    return sum(x * y for x, y in zip(a, b, strict=True))


class TinyVectorStore:
    """教学用：逐条暴力比较。真实向量数据库另外负责索引加速、持久化、并发写入和条件过滤。"""

    def __init__(self, model, dim):
        self.model, self.dim, self.rows = model, dim, {}

    def upsert(self, rid, vector, model, meta):
        if model != self.model or len(vector) != self.dim:
            raise ValueError(f"{rid} 来自 {model}/{len(vector)}维，库要求 {self.model}/{self.dim}维")
        self.rows[rid] = (normalize(vector), meta)

    def search(self, vector, model, k=3, where=None):
        if model != self.model:
            raise ValueError(f"查询用 {model}，库内是 {self.model}，分数不可比")
        q = normalize(vector)
        hits = [(dot(q, v), rid) for rid, (v, meta) in self.rows.items()
                if not where or all(meta.get(key) == value for key, value in where.items())]
        return [(rid, round(score, 3)) for score, rid in sorted(hits, key=lambda h: (-h[0], h[1]))[:k]]


# 教学手工向量：4 个维度人为设定为“可靠性、格式、费用、检索”，便于肉眼核对；
# 真实 embedding 有数百到数千维，单个维度没有可读含义
DOCS = {
    "F1": ([0.9, 0.1, 0.0, 0.3], {"current": True}),
    "F1-旧版": ([0.85, 0.1, 0.0, 0.35], {"current": False}),
    "F2": ([0.2, 0.9, 0.0, 0.0], {"current": True}),
    "F4": ([1.5, 0.0, 0.0, 4.5], {"current": True}),  # 故意放大 5 倍，模拟“长度”很大的向量
    "F6": ([0.0, 0.1, 0.95, 0.0], {"current": True}),
}
query = [0.8, 0.0, 0.0, 0.2]  # 假设“怎么让模型别瞎编”的向量

raw = sorted(((round(dot(query, v), 3), rid) for rid, (v, _) in DOCS.items()), reverse=True)
print("未归一化直接点积：", [(rid, s) for s, rid in raw[:3]])

store = TinyVectorStore("teach-embed-v1", 4)
for rid, (vec, meta) in DOCS.items():
    store.upsert(rid, vec, "teach-embed-v1", meta)
print("归一化后（余弦）：", store.search(query, "teach-embed-v1"))
print("只查当前版本：", store.search(query, "teach-embed-v1", where={"current": True}))

for label, action in [
    ("换了模型的查询", lambda: store.search(query, "teach-embed-v2")),
    ("维度不对的文档", lambda: store.upsert("F9", [0.1] * 8, "teach-embed-v1", {"current": True})),
    ("全零向量", lambda: normalize([0, 0, 0, 0])),
]:
    try:
        action()
    except ValueError as exc:
        print(f"{label}：已拒绝，{exc}")

a, b = [3.0, 4.0, 0.0, 0.0], [0.6, 0.8, 0.0, 0.0]
assert abs(dot(normalize(a), normalize(b)) - 1.0) < 1e-12  # 方向相同，长度不同，余弦为 1

实际输出：

未归一化直接点积： [('F4', 2.1), ('F1', 0.78), ('F1-旧版', 0.75)]
归一化后（余弦）： [('F1', 0.992), ('F1-旧版', 0.984), ('F4', 0.537)]
只查当前版本： [('F1', 0.992), ('F4', 0.537), ('F2', 0.21)]
换了模型的查询：已拒绝，查询用 teach-embed-v2，库内是 teach-embed-v1，分数不可比
维度不对的文档：已拒绝，F9 来自 teach-embed-v1/8维，库要求 teach-embed-v1/4维
全零向量：已拒绝，零向量没有方向，不能参与比较

三个结论对应三个操作。第一，不归一化时，长度被放大的 F4 靠“个头”赢了点积；归一化后点积等于余弦，只比方向，F1 回到第一。平台文档没写明向量是否已归一化时，入库前自己归一化一次。第二，旧版 F1 和新版几乎一样近，相似度分不出新旧，要靠元数据过滤。第三，模型名和维度要随向量一起存，比较前先核对，混用时直接报错，而不是给出一个看似正常的分数。

第三步：用人工标注评估召回

换方法之前先有尺子。为每道查询人工标出“应该找到哪条”，库里没有答案的题标空集合。召回@k：正确答案出现在前 k 条里的比例；MRR：正确答案排第几名取倒数再平均，排第一得 1，第二得 0.5；无答案题：分数都低于阈值、什么也不返回才算对。保存为 evaluate.py，运行 python evaluate.py：

import json
import sys

# 人工标注：每道查询真正应该找到的问题编号；空集合表示库里没有答案，应当什么都不返回
LABELS = {"Q1": {"F1"}, "Q2": {"F2"}, "Q3": {"F6"}, "Q4": {"F4"},
          "Q5": {"F5"}, "Q6": {"F3"}, "Q7": set()}


def evaluate(rankings, threshold):
    r1 = r3 = rr = answerable = 0
    no_answer_ok, no_answer, misses = 0, 0, []
    for qid, relevant in LABELS.items():
        hits = [fid for fid, score in rankings.get(qid, []) if score >= threshold]
        if not relevant:
            no_answer += 1
            no_answer_ok += not hits
            if hits:
                misses.append(f"{qid}误返回{hits[0]}")
            continue
        answerable += 1
        r1 += bool(relevant & set(hits[:1]))
        r3 += bool(relevant & set(hits[:3]))
        rank = next((i for i, fid in enumerate(hits, 1) if fid in relevant), None)
        rr += 1 / rank if rank else 0
        if rank != 1:
            misses.append(f"{qid}第一名是{hits[0] if hits else '空'}")
    return {"召回@1": round(r1 / answerable, 2), "召回@3": round(r3 / answerable, 2),
            "MRR": round(rr / answerable, 2), "无答案题正确": f"{no_answer_ok}/{no_answer}",
            "未排第一": misses}


for path in sys.argv[1:] or ["rank_bow.json"]:
    with open(path, encoding="utf-8") as f:
        data = json.load(f)
    missing = LABELS.keys() - data["rankings"].keys()
    if missing:
        raise SystemExit(f"{path} 缺少查询 {sorted(missing)}，补齐后再比较")
    print(f"== {path}（{data['method']}）")
    for threshold in (0.0, 0.15, 0.3):
        print(f"  阈值 {threshold}: {evaluate(data['rankings'], threshold)}")

实际输出：

== rank_bow.json（字二元组词袋）
  阈值 0.0: {'召回@1': 0.83, '召回@3': 1.0, 'MRR': 0.92, '无答案题正确': '0/1', '未排第一': ['Q1第一名是F2', 'Q7误返回F4']}
  阈值 0.15: {'召回@1': 0.83, '召回@3': 0.83, 'MRR': 0.83, '无答案题正确': '1/1', '未排第一': ['Q1第一名是F2']}
  阈值 0.3: {'召回@1': 0.67, '召回@3': 0.67, 'MRR': 0.67, '无答案题正确': '1/1', '未排第一': ['Q1第一名是空', 'Q4第一名是空']}

阈值是一个取舍：0.15 挡住了退款题的误命中，却也把排第二的 F1 挡掉，Q1 的召回@3 从命中变成未命中；0.3 又多丢了 Q4。七道题只够发现明显问题，正式选阈值要用更多真实查询，并且把“换说法”的题专门写进去。评测脚本只读排名文件，不关心排名来自哪种方法，所以下一节可以直接拿真实 embedding 的结果来比。

换成真实 embedding 再比一次（练习方案（本站未实跑））

这一节需要你自己的模型平台账号，会产生少量费用；本站没有替你调用，下面没有任何真实向量结果。以阿里云百炼文档为例（2026-10 核对官方页面）：text-embedding-v4 可选 2048 到 64 共八档维度，默认 1024；字符串数组每次最多 10 条；区分“查询/文档”的 text_type 参数目前只能通过 DashScope 自己的 SDK 和 API 启用；文档没有说明输出向量是否已归一化。其他平台以各自文档为准。

在平台的 API 调试页或你常用的 HTTP 工具里，按 OpenAI 兼容的 embeddings 接口发两次请求，地址与模型名从文档和你的账号里复制。第一次发六条问题，顺序必须与编号 F1 到 F6 一致：

{
  "model": "<你账号里可用的 embedding 模型名>",
  "input": ["如何减少大模型的幻觉", "怎样让模型输出合法的JSON", "提示词太长会不会被截断",
            "RAG检索不到相关资料怎么办", "温度参数应该设多少", "如何估算调用模型的费用"],
  "dimensions": 1024
}

把完整响应保存为 docs_response.json。第二次把 input 换成 Q1 到 Q7 的七句查询（顺序同上表），其余不变，响应保存为 queries_response.json。然后在同一目录运行下面的 rank_from_vectors.py，它读取响应里 data 数组每项的 embedding，核对条数、模型和维度，归一化后排名：

import json
import math
from pathlib import Path

# 顺序必须与发送请求时 input 数组的顺序一致
DOC_IDS = ["F1", "F2", "F3", "F4", "F5", "F6"]
QUERY_IDS = ["Q1", "Q2", "Q3", "Q4", "Q5", "Q6", "Q7"]


def load_vectors(path, ids):
    data = json.loads(Path(path).read_text(encoding="utf-8"))
    items = sorted(data["data"], key=lambda item: item.get("index", 0))
    if len(items) != len(ids):
        raise SystemExit(f"{path} 有 {len(items)} 条向量，应为 {len(ids)} 条")
    vectors = {}
    for rid, item in zip(ids, items, strict=True):
        vec = item["embedding"]
        norm = math.sqrt(sum(x * x for x in vec))
        if not norm:
            raise SystemExit(f"{path} 中 {rid} 是零向量")
        vectors[rid] = [x / norm for x in vec]  # 文档没说明是否已归一化时，自己再归一化一次
    return data.get("model", "未知模型"), vectors


missing = [p for p in ("docs_response.json", "queries_response.json") if not Path(p).exists()]
if missing:
    print(f"未找到 {missing}：先按练习方案把两次请求的完整响应分别保存到当前目录。")
    raise SystemExit(0)

doc_model, docs = load_vectors("docs_response.json", DOC_IDS)
query_model, queries = load_vectors("queries_response.json", QUERY_IDS)
dims = {len(v) for v in [*docs.values(), *queries.values()]}
if doc_model != query_model or len(dims) != 1:
    raise SystemExit(f"模型或维度不一致：{doc_model}/{query_model}，维度 {sorted(dims)}")

rankings = {}
for qid, qv in queries.items():
    scored = sorted(((sum(a * b for a, b in zip(qv, dv, strict=True)), fid) for fid, dv in docs.items()),
                    key=lambda pair: (-pair[0], pair[1]))
    rankings[qid] = [[fid, round(score, 4)] for score, fid in scored]
    print(qid, "  ".join(f"{fid}:{score:.2f}" for score, fid in scored[:3]))

Path("rank_embed.json").write_text(json.dumps(
    {"method": f"{doc_model}，{dims.pop()}维", "rankings": rankings}, ensure_ascii=False, indent=1),
    encoding="utf-8")
print("已保存 rank_embed.json")

本站在没有响应文件的目录里运行，实际输出是提示信息：

未找到 ['docs_response.json', 'queries_response.json']：先按练习方案把两次请求的完整响应分别保存到当前目录。

拿到两个响应后再运行它，得到 rank_embed.json，接着运行 python evaluate.py rank_bow.json rank_embed.json 并排比较。核对三件事：Q1 是否把 F1 排到第一；Q7 退款题的最高分是多少；七道可答题的分数整体落在什么区间。真实向量的分数分布通常和词袋不同，把 evaluate.py 里的阈值组改成覆盖你实际分数区间的几档再比较，不要照搬 0.15 或别人文章里的数字。结果写进记录，不能先假定 embedding 一定更好。

常见失败与修改

换个说法就搜不到。 在标注集里专门加“同义改写”题，例如每条常见问题配两种口语问法。词袋方案可以补同义词表，但维护成本高；改用 embedding 后，用同一份标注集重测，确认改写题真的变好了。

换了模型或维度，结果突然全乱。 多半是新查询向量和旧文档向量混在一起。向量必须和模型名、维度、生成日期一起存；换模型就全量重算，像 store.py 那样在比较前拒绝不一致的组合。

库里没有答案也返回了结果。 相似度排序永远能排出第一名。用标注集里的无答案题来定阈值，低于阈值就走“未找到”分支，向用户说明并提供人工渠道，而不是硬塞一条最接近的。

旧版资料排在前面。 新旧版本内容相近，向量分不开。入库时写明 current 或生效日期，检索时先按元数据过滤；资料下架时同步删除或标记向量。

整篇长文只有一个向量，命中了却找不到依据。 按小节切分，每块保留标题和来源编号，检索后把具体片段交给生成步骤；分层排查方法见检索实操。

保存和使用结果

保留 faq_search.py、store.py、evaluate.py、rank_from_vectors.py 和每次的排名文件；在记录里写上方法、模型名、维度、日期和七道题的评测结果。以后知识库新增或修改条目、或者换 embedding 模型时，先补标注再重跑评测，指标变差就回退，不凭几次手动搜索下结论。

自测与答案

1. 余弦相似度和点积什么时候相同？两个向量都归一化到长度 1 时相同。
2. 新旧两个 embedding 模型都输出 1024 维，向量能混用吗？不能，维度相同不代表坐标含义相同，必须全量重算。
3. 召回@3 是 1.0，为什么还需要看无答案题？召回只衡量“该找的找没找到”，不衡量“不该返回时是否闭嘴”；Q7 在阈值 0 时就误返回了 F4。
4. 向量数据库能判断答案是否过时吗？不能，它只按相似度和你给的元数据检索，过时要靠元数据和评测把关。

交付物是四个脚本、词袋方法的实际评测输出和一份标注集。词袋结果只代表这七道教学题；真实 embedding 的效果需要你按练习方案自测后才有结论。