RAG Embedding 模型替换与索引回归

RAG Embedding 模型替换与索引回归

文章目录

• RAG Embedding 模型替换与索引回归
• • 1. 换模型名不等于完成了向量层升级
  • 2. 先说结论
  • 3. 演示：TF-IDF 换成哈希 Embedding
  • 4. 演示：弱模型替换被阻断
  • 5. 策略与后端说明
  • 6. Embedding 后端实现
  • 7. 按后端构建索引包
  • 8. 双层评测
  • 9. 换模闸门
  • 10. 一键换模与回归
  • 11. 人工核对清单
  • 12. 常见错误
  • • 12.1 改模型配置但不重建索引
    • 12.2 只验收原话层
    • 12.3 model_id 不写进元数据
    • 12.4 把演示哈希向量当成生产模型
    • 12.5 换模时改动题集口径
    • 12.6 切分参数随手改
  • 13. 术语速查
  • 14. 小结
  • 15. 相关阅读

摘要：《RAG 向量索引重建与回归》把索引包做成带指纹的版本制品后，下一步常是把演示用的 TF-IDF 换成真实 Embedding 模型。换模若只改配置名、不重建索引、不做双层回归，变体问法会悄悄退步。本文用同一套题集与闸门口径，演示从 tfidf_cosine_v1 替换到定长哈希 Embedding（接口对齐真实模型），并给出弱模型替换被阻断的对照；标准库即可跑通，上线时只替换向量层实现。

说明：承接 RAG 向量索引重建与回归。上一篇解决“索引何时重建”；本篇解决“Embedding 模型怎么换、怎样回归才允许切换”。

承接前文：

• RAG 向量索引重建与回归
• RAG 文档更新后的全链路闸门验收
• RAG 关键词检索与向量检索的切换边界
• RAG 可回归评测问法的业务采集与标注

建议目录：

mkdir -p ~/kb-embedding-swap/{notes,scripts,configs,fixtures/{docs,eval},images,logs/embedding_swap}
cd ~/kb-embedding-swap

文中配置与脚本均全文给出。哈希 Embedding 用于演示定长向量接口；上线可换成 sentence-transformers 或云端 Embedding API，题集、扩展表、指纹字段与闸门输出保持不变。

1. 换模型名不等于完成了向量层升级

《RAG 关键词检索与向量检索的切换边界》给出切换建议，《RAG 向量索引重建与回归》要求文档与扩展表变更后重建索引。工程上还有第三种变更：Embedding 后端本身换了。

常见缺口有三类：

图1. 换 Embedding 必须重建索引包，并做与基线的双层对比。

本篇硬规则：

Embedding 后端变更后，必须重建索引包并过 EMBEDDING_SWAP_ALLOWED；未达标不得切换线上模型或索引指针。

2. 先说结论

本篇用到的说法，先统一说明：

四条落地判断：

图2. 建基线 → 换后端 → 重建 → 双层对比 → 闸门。

3. 演示：TF-IDF 换成哈希 Embedding

python3 scripts/run_embedding_swap.py \\
–policy configs/embedding_swap_policy.example.json \\
–docs-dir fixtures/docs \\
–expansion configs/query_expansion.example.json \\
–canonical fixtures/eval/questions_canonical.jsonl \\
–paraphrase fixtures/eval/questions_paraphrase.jsonl \\
–workdir .

本机输出节选：

EMBED_INDEX_OK backend=tfidf_cosine_v1 model_id=demo-tfidf-v1
EMBED_INDEX_OK backend=hashing_embedding_v1 model_id=demo-hashing-emb-v1
EMBED_EVAL … canonical=100.0% paraphrase=100.0%
COMPARE paraphrase 100.0% -> 100.0% drop=0.0% regressed=0
EMBEDDING_SWAP_ALLOWED
swap tfidf_cosine_v1 -> hashing_embedding_v1
EMBEDDING_SWAP_OK

图3. 后端与 model_id 都变化，双层命中率不掉，闸门放行。

4. 演示：弱模型替换被阻断

把目标后端改成 bag_of_chars_v0（无二元字、忽略扩展表）：

python3 scripts/run_embedding_swap.py \\
–policy configs/embedding_swap_policy.example.json \\
–docs-dir fixtures/docs \\
–expansion configs/query_expansion.example.json \\
–canonical fixtures/eval/questions_canonical.jsonl \\
–paraphrase fixtures/eval/questions_paraphrase.jsonl \\
–workdir . \\
–target-backend bag_of_chars_v0

本机输出节选：

EMBED_EVAL … canonical=100.0% paraphrase=60.0%
COMPARE paraphrase 100.0% -> 60.0% drop=40.0% regressed=2
EMBEDDING_SWAP_BLOCKED
FAIL: paraphrase_hit_rate 0.600 < min_paraphrase_hit_rate 0.85
FAIL: regressed_count 2 > max_regressed_questions 0

原话层仍 100%，变体层掉到 60%，V2、V3 退步。这说明：换模验收必须以变体层为主约束之一。

图4. 原话层满分不能掩盖变体层掉点。

5. 策略与后端说明

保存为 configs/embedding_swap_policy.example.json：

{
"baseline_backend": "tfidf_cosine_v1",
"target_backend": "hashing_embedding_v1",
"min_canonical_hit_rate": 1.0,
"min_paraphrase_hit_rate": 0.85,
"max_regressed_questions": 0,
"max_hit_drop": 0.15,
"require_model_id_change": true,
"require_rebuild_after_swap": true,
"canonical_top_k": 3,
"paraphrase_top_k": 1,
"max_chars": 240,
"overlap_chars": 40
}

保存为 configs/embedding_backends.example.json：

{
"backends": {
"tfidf_cosine_v1": {
"model_id": "demo-tfidf-v1",
"kind": "sparse",
"note": "演示基线，非神经网络"
},
"hashing_embedding_v1": {
"model_id": "demo-hashing-emb-v1",
"kind": "dense",
"dim": 128,
"note": "定长哈希向量，接口对齐真实 embedding；上线可换成 sentence-transformers / API"
},
"bag_of_chars_v0": {
"model_id": "demo-bag-chars-v0",
"kind": "sparse_weak",
"note": "弱模型，仅用于阻断演示"
}
}
}

图5. 题集、扩展表、闸门字段固定；可替换的是 embed 实现。

6. Embedding 后端实现

保存为 scripts/embed_backends.py：

#!/usr/bin/env python3
"""Embedding backends for demo: TF-IDF sparse and hashing dense (stdlib only)."""

from __future__ import annotations

import hashlib
import math
import re
from collections import Counter
from typing import Any

WORD_RE = re.compile(r"[A-Za-z0-9_./-]+|[\\u4e00-\\u9fff]+", re.UNICODE)

def tokenize(text: str) –> list[str]:
tokens: list[str] = []
for piece in WORD_RE.findall(text):
if re.fullmatch(r"[A-Za-z0-9_./-]+", piece):
if len(piece) > 1:
tokens.append(piece.lower())
continue
chars = list(piece)
tokens.extend(chars)
tokens.extend(chars[i] + chars[i + 1] for i in range(len(chars) – 1))
return tokens

def expand_query(query: str, expansion: dict[str, list[str]]) –> str:
if not expansion:
return query
extra: list[str] = []
q_lower = query.lower()
for _key, aliases in expansion.items():
for alias in aliases:
a = alias.lower()
if a in q_lower or alias in query:
extra.extend(aliases)
break
if not extra:
return query
return query + " " + " ".join(dict.fromkeys(extra))

def _signed_bucket(token: str, dim: int) –> tuple[int, float]:
digest = hashlib.md5(token.encode("utf-8")).hexdigest()
idx = int(digest[:8], 16) % dim
sign = 1.0 if int(digest[8:10], 16) % 2 == 0 else –1.0
return idx, sign

def l2_normalize(vec: list[float]) –> list[float]:
norm = math.sqrt(sum(x * x for x in vec)) or 1.0
return [x / norm for x in vec]

def dense_cosine(a: list[float], b: list[float]) –> float:
return sum(x * y for x, y in zip(a, b))

def sparse_cosine(a: dict[str, float], b: dict[str, float]) –> float:
if not a or not b:
return 0.0
keys = set(a) & set(b)
dot = sum(a[k] * b[k] for k in keys)
na = math.sqrt(sum(v * v for v in a.values())) or 1.0
nb = math.sqrt(sum(v * v for v in b.values())) or 1.0
return dot / (na * nb)

def embed_tfidf(chunks: list[dict]) –> tuple[list[dict[str, float]], dict[str, float]]:
df: Counter[str] = Counter()
tfs: list[Counter[str]] = []
for row in chunks:
tf = Counter(tokenize(row["text"]))
tfs.append(tf)
for term in tf:
df[term] += 1
n = max(len(chunks), 1)
idf = {t: math.log((n + 1) / (df[t] + 1)) + 1.0 for t in df}
vectors = [{t: w * idf.get(t, 0.0) for t, w in tf.items()} for tf in tfs]
return vectors, idf

def embed_query_tfidf(query: str, idf: dict[str, float], expansion: dict[str, list[str]]) –> dict[str, float]:
q = expand_query(query, expansion)
tf = Counter(tokenize(q))
return {t: w * idf.get(t, 0.0) for t, w in tf.items()}

def embed_hashing(texts: list[str], dim: int = 128) –> list[list[float]]:
out: list[list[float]] = []
for text in texts:
vec = [0.0] * dim
for tok in tokenize(text):
idx, sign = _signed_bucket(tok, dim)
vec[idx] += sign
out.append(l2_normalize(vec))
return out

def embed_query_hashing(query: str, dim: int, expansion: dict[str, list[str]]) –> list[float]:
q = expand_query(query, expansion)
return embed_hashing([q], dim=dim)[0]

def embed_bag_of_chars(texts: list[str]) –> list[dict[str, float]]:
"""Weak backend: only single Chinese chars / ascii tokens, no bigrams — for BLOCKED demo."""
vectors: list[dict[str, float]] = []
for text in texts:
toks = []
for piece in WORD_RE.findall(text):
if re.fullmatch(r"[A-Za-z0-9_./-]+", piece):
if len(piece) > 1:
toks.append(piece.lower())
else:
toks.extend(list(piece))
tf = Counter(toks)
norm = math.sqrt(sum(v * v for v in tf.values())) or 1.0
vectors.append({t: v / norm for t, v in tf.items()})
return vectors

def embed_query_bag_of_chars(query: str, expansion: dict[str, list[str]]) –> dict[str, float]:
# deliberately ignore expansion to simulate a bad model swap
_ = expansion
return embed_bag_of_chars([query])[0]

BACKENDS = {
"tfidf_cosine_v1": {
"kind": "sparse",
"model_id": "demo-tfidf-v1",
"description": "演示用 TF-IDF 余弦（非神经网络）",
},
"hashing_embedding_v1": {
"kind": "dense",
"model_id": "demo-hashing-emb-v1",
"dim": 128,
"description": "定长哈希向量，演示替换真实 embedding 时的接口与闸门",
},
"bag_of_chars_v0": {
"kind": "sparse_weak",
"model_id": "demo-bag-chars-v0",
"description": "弱模型：无二元字、忽略扩展表，用于阻断演示",
},
}

def build_backend(name: str, chunks: list[dict], expansion: dict[str, list[str]]) –> dict[str, Any]:
meta = BACKENDS.get(name)
if not meta:
raise SystemExit(f"unknown backend: {name}; choose from {list(BACKENDS)}")
texts = [c["text"] for c in chunks]
if name == "tfidf_cosine_v1":
vectors, idf = embed_tfidf(chunks)
return {
"backend": name,
"kind": "sparse",
"model_id": meta["model_id"],
"idf": idf,
"vectors": vectors,
"chunk_ids": [c.get("chunk_id") for c in chunks],
"uses_expansion_at_query": True,
}
if name == "hashing_embedding_v1":
dim = int(meta["dim"])
# bake expansion aliases into chunk side lightly by appending known aliases present in text
enriched = []
for t in texts:
extra = []
for aliases in expansion.values():
if any(a in t for a in aliases):
extra.extend(aliases)
enriched.append(t + (" " + " ".join(dict.fromkeys(extra)) if extra else ""))
vectors = embed_hashing(enriched, dim=dim)
return {
"backend": name,
"kind": "dense",
"model_id": meta["model_id"],
"dim": dim,
"vectors": vectors,
"chunk_ids": [c.get("chunk_id") for c in chunks],
"uses_expansion_at_query": True,
}
if name == "bag_of_chars_v0":
vectors = embed_bag_of_chars(texts)
return {
"backend": name,
"kind": "sparse_weak",
"model_id": meta["model_id"],
"vectors": vectors,
"chunk_ids": [c.get("chunk_id") for c in chunks],
"uses_expansion_at_query": False,
}
raise SystemExit(f"unhandled backend: {name}")

def score_query(
query: str,
chunks: list[dict],
payload: dict[str, Any],
expansion: dict[str, list[str]],
) –> list[tuple[float, dict]]:
kind = payload.get("kind")
vectors = payload["vectors"]
if kind == "sparse":
qv = embed_query_tfidf(query, payload.get("idf") or {}, expansion)
ranked = [(sparse_cosine(qv, vectors[i]), chunks[i]) for i in range(len(chunks))]
elif kind == "dense":
qv = embed_query_hashing(query, int(payload.get("dim") or 128), expansion)
ranked = [(dense_cosine(qv, vectors[i]), chunks[i]) for i in range(len(chunks))]
elif kind == "sparse_weak":
qv = embed_query_bag_of_chars(query, expansion)
ranked = [(sparse_cosine(qv, vectors[i]), chunks[i]) for i in range(len(chunks))]
else:
raise SystemExit(f"unknown kind: {kind}")
ranked.sort(key=lambda x: x[0], reverse=True)
return ranked

上线替换真实模型时，优先新增一个 backend 名（例如 sbert_minilm_v1），在 build_backend / score_query 里接模型推理；不要改闸门字段名。

7. 按后端构建索引包

保存为 scripts/build_embedding_index.py：

#!/usr/bin/env python3
"""Build embedding index package for a chosen backend."""

from __future__ import annotations

import argparse
import hashlib
import json
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))
from embed_backends import BACKENDS, build_backend # noqa: E402
from retrieve_vector import load_chunks, load_expansion # noqa: E402

def sha256_file(path: Path) –> str:
return hashlib.sha256(path.read_bytes()).hexdigest()

def sha256_text(text: str) –> str:
return hashlib.sha256(text.encode("utf-8")).hexdigest()

def main() –> None:
parser = argparse.ArgumentParser(description="Build embedding index for one backend")
parser.add_argument("–docs-dir", type=Path, required=True)
parser.add_argument("–expansion", type=Path, required=True)
parser.add_argument("–backend", required=True, choices=sorted(BACKENDS.keys()))
parser.add_argument("–out-dir", type=Path, required=True)
parser.add_argument("–max-chars", type=int, default=240)
parser.add_argument("–overlap-chars", type=int, default=40)
parser.add_argument("–label", default="current")
args = parser.parse_args()

out_dir = args.out_dir
out_dir.mkdir(parents=True, exist_ok=True)
chunks_path = out_dir / "chunks.jsonl"
meta_path = out_dir / "index_meta.json"
vectors_path = out_dir / "vectors.json"
manifest_path = out_dir / "docs_manifest.json"
py = sys.executable

subprocess.run(
[py, str(ROOT / "fingerprint_docs.py"), "–docs-dir", str(args.docs_dir), "–out", str(manifest_path)],
check=True,
)
subprocess.run(
[
py,
str(ROOT / "chunk_docs.py"),
"–docs-dir",
str(args.docs_dir),
"–out",
str(chunks_path),
"–max-chars",
str(args.max_chars),
"–overlap-chars",
str(args.overlap_chars),
],
check=True,
)

chunks = load_chunks(chunks_path)
expansion = load_expansion(args.expansion)
payload = build_backend(args.backend, chunks, expansion)
vectors_path.write_text(json.dumps(payload, ensure_ascii=False) + "\\n", encoding="utf-8")

manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
docs_fp = sha256_text(json.dumps(manifest, sort_keys=True, ensure_ascii=False))
expansion_fp = sha256_file(args.expansion)
model_fp = sha256_text(json.dumps({"backend": args.backend, "model_id": payload["model_id"]}, sort_keys=True))

meta = {
"label": args.label,
"backend": args.backend,
"model_id": payload["model_id"],
"model_fingerprint": model_fp,
"kind": payload.get("kind"),
"built_at": datetime.now(timezone.utc).isoformat(),
"docs_dir": str(args.docs_dir.resolve()),
"expansion_path": str(args.expansion.resolve()),
"max_chars": args.max_chars,
"overlap_chars": args.overlap_chars,
"chunk_count": len(chunks),
"docs_fingerprint": docs_fp,
"expansion_fingerprint": expansion_fp,
"chunks_fingerprint": sha256_file(chunks_path),
"docs_manifest": manifest,
}
meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2) + "\\n", encoding="utf-8")
print(
f"EMBED_INDEX_OK label={args.label} backend={args.backend} "
f"model_id={payload['model_id']} chunks={len(chunks)} -> {out_dir}"
)

if __name__ == "__main__":
main()

index_meta.json 至少包含：backend、model_id、model_fingerprint、文档指纹、扩展表指纹、切分参数。

8. 双层评测

保存为 scripts/eval_embedding_index.py：

#!/usr/bin/env python3
"""Evaluate an embedding index package on canonical + paraphrase sets."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent
sys.path.insert(0, str(ROOT))
from embed_backends import score_query # noqa: E402
from retrieve_vector import load_chunks, load_expansion # noqa: E402

def load_jsonl(path: Path) –> list[dict]:
rows = []
for line in path.read_text(encoding="utf-8").splitlines():
line = line.strip()
if line:
rows.append(json.loads(line))
return rows

def eval_one(chunks: list[dict], payload: dict, questions: list[dict], expansion: dict, top_k: int) –> dict:
details = []
hits = 0
for q in questions:
ranked = score_query(q["question"], chunks, payload, expansion)[:top_k]
got_docs = [row["doc_id"] for _, row in ranked]
gold = set(q.get("gold_docs") or [])
ok = bool(gold & set(got_docs))
if ok:
hits += 1
details.append(
{
"id": q.get("id"),
"ok": ok,
"gold_docs": sorted(gold),
"got_docs": got_docs,
"question": q.get("question"),
}
)
total = len(questions) or 1
return {
"total": len(questions),
"hits": hits,
"hit_rate": round(hits / total, 4),
"top_k": top_k,
"failed_ids": [d["id"] for d in details if not d["ok"]],
"details": details,
}

def main() –> None:
parser = argparse.ArgumentParser(description="Evaluate embedding index")
parser.add_argument("–index-dir", type=Path, required=True)
parser.add_argument("–canonical", type=Path, required=True)
parser.add_argument("–paraphrase", type=Path, required=True)
parser.add_argument("–expansion", type=Path, required=True)
parser.add_argument("–canonical-top-k", type=int, default=3)
parser.add_argument("–paraphrase-top-k", type=int, default=1)
parser.add_argument("–out", type=Path, required=True)
args = parser.parse_args()

chunks = load_chunks(args.index_dir / "chunks.jsonl")
payload = json.loads((args.index_dir / "vectors.json").read_text(encoding="utf-8"))
meta = json.loads((args.index_dir / "index_meta.json").read_text(encoding="utf-8"))
expansion = load_expansion(args.expansion)

c_rep = eval_one(chunks, payload, load_jsonl(args.canonical), expansion, args.canonical_top_k)
p_rep = eval_one(chunks, payload, load_jsonl(args.paraphrase), expansion, args.paraphrase_top_k)

summary = {
"backend": meta.get("backend") or payload.get("backend"),
"model_id": meta.get("model_id") or payload.get("model_id"),
"canonical_top_k": args.canonical_top_k,
"paraphrase_top_k": args.paraphrase_top_k,
"canonical_hit_rate": c_rep["hit_rate"],
"paraphrase_hit_rate": p_rep["hit_rate"],
"canonical_failed_ids": c_rep["failed_ids"],
"paraphrase_failed_ids": p_rep["failed_ids"],
"canonical_total": c_rep["total"],
"paraphrase_total": p_rep["total"],
}
args.out.parent.mkdir(parents=True, exist_ok=True)
args.out.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\\n", encoding="utf-8")
print(f"wrote {args.out}")
print(
f"EMBED_EVAL backend={summary['backend']} model_id={summary['model_id']} "
f"canonical={summary['canonical_hit_rate']:.1%} paraphrase={summary['paraphrase_hit_rate']:.1%}"
)
for d in c_rep["details"]:
print(f"[C {'PASS' if d['ok'] else 'FAIL'}] {d['id']}: got={d['got_docs']} gold={d['gold_docs']}")
for d in p_rep["details"]:
print(f"[P {'PASS' if d['ok'] else 'FAIL'}] {d['id']}: got={d['got_docs']} gold={d['gold_docs']}")

if __name__ == "__main__":
main()

对比脚本沿用前篇的 compare_index_evals.py（字段同为 canonical_hit_rate / paraphrase_hit_rate）。

9. 换模闸门

保存为 scripts/embedding_swap_gate.py：

#!/usr/bin/env python3
"""Gate embedding model swap against policy."""

from __future__ import annotations

import argparse
import json
from pathlib import Path

def main() –> None:
parser = argparse.ArgumentParser(description="Gate embedding model swap")
parser.add_argument("–policy", type=Path, required=True)
parser.add_argument("–baseline-eval", type=Path, required=True)
parser.add_argument("–target-eval", type=Path, required=True)
parser.add_argument("–compare-report", type=Path, required=True)
parser.add_argument("–baseline-meta", type=Path, default=None)
parser.add_argument("–target-meta", type=Path, default=None)
parser.add_argument("–json", action="store_true")
args = parser.parse_args()

policy = json.loads(args.policy.read_text(encoding="utf-8"))
base = json.loads(args.baseline_eval.read_text(encoding="utf-8"))
cur = json.loads(args.target_eval.read_text(encoding="utf-8"))
cmp = json.loads(args.compare_report.read_text(encoding="utf-8"))
errors: list[str] = []

min_c = float(policy.get("min_canonical_hit_rate", 1.0))
min_p = float(policy.get("min_paraphrase_hit_rate", 0.85))
c_hit = float(cur.get("canonical_hit_rate") or 0)
p_hit = float(cur.get("paraphrase_hit_rate") or 0)
if c_hit < min_c:
errors.append(f"canonical_hit_rate {c_hit:.3f} < min_canonical_hit_rate {min_c}")
if p_hit < min_p:
errors.append(f"paraphrase_hit_rate {p_hit:.3f} < min_paraphrase_hit_rate {min_p}")

max_reg = int(policy.get("max_regressed_questions", 0))
max_drop = float(policy.get("max_hit_drop", 1.0))
reg = int(cmp.get("regressed_count") or 0)
drop = max(float(cmp.get("canonical_hit_drop") or 0), float(cmp.get("paraphrase_hit_drop") or 0))
if reg > max_reg:
errors.append(f"regressed_count {reg} > max_regressed_questions {max_reg}")
if drop > max_drop:
errors.append(f"hit_drop {drop:.3f} > max_hit_drop {max_drop}")

if policy.get("require_model_id_change") and args.baseline_meta and args.target_meta:
bmeta = json.loads(args.baseline_meta.read_text(encoding="utf-8"))
tmeta = json.loads(args.target_meta.read_text(encoding="utf-8"))
if bmeta.get("model_id") == tmeta.get("model_id"):
errors.append("model_id unchanged; swap did not change embedding model")
if bmeta.get("backend") == tmeta.get("backend"):
errors.append("backend unchanged; expected a different embedding backend")
if policy.get("require_rebuild_after_swap"):
if bmeta.get("model_fingerprint") == tmeta.get("model_fingerprint"):
errors.append("model_fingerprint unchanged; rebuild required after swap")

allow = not errors
out = {
"allow": allow,
"baseline_backend": base.get("backend"),
"target_backend": cur.get("backend"),
"baseline_model_id": base.get("model_id"),
"target_model_id": cur.get("model_id"),
"canonical_hit_rate": c_hit,
"paraphrase_hit_rate": p_hit,
"regressed_count": reg,
"errors": errors,
}
if args.json:
print(json.dumps(out, ensure_ascii=False, indent=2))
else:
print("EMBEDDING_SWAP_ALLOWED" if allow else "EMBEDDING_SWAP_BLOCKED")
for e in errors:
print(f"FAIL: {e}")
if allow:
print(
f"swap {base.get('backend')} -> {cur.get('backend')} "
f"canonical={c_hit:.1%} paraphrase={p_hit:.1%}"
)
raise SystemExit(0 if allow else 2)

if __name__ == "__main__":
main()

python3 scripts/embedding_swap_gate.py \\
–policy configs/embedding_swap_policy.example.json \\
–baseline-eval logs/embedding_swap/baseline_eval.json \\
–target-eval logs/embedding_swap/target_eval.json \\
–compare-report logs/embedding_swap/compare_report.json \\
–baseline-meta logs/embedding_swap/index_baseline/index_meta.json \\
–target-meta logs/embedding_swap/index_target/index_meta.json

10. 一键换模与回归

保存为 scripts/run_embedding_swap.py：

#!/usr/bin/env python3
"""One-shot: build baseline + target embedding indexes, eval, compare, gate."""

from __future__ import annotations

import argparse
import json
import shutil
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent

def run(cmd: list[str], cwd: Path, check: bool = True) –> int:
proc = subprocess.run(cmd, cwd=str(cwd), capture_output=True, text=True)
text = (proc.stdout or "") + (proc.stderr or "")
if text.strip():
print(text[–2200:])
if check and proc.returncode != 0:
raise SystemExit(proc.returncode)
return proc.returncode

def main() –> None:
parser = argparse.ArgumentParser(description="Swap embedding backend and regress")
parser.add_argument("–policy", type=Path, required=True)
parser.add_argument("–docs-dir", type=Path, required=True)
parser.add_argument("–expansion", type=Path, required=True)
parser.add_argument("–canonical", type=Path, required=True)
parser.add_argument("–paraphrase", type=Path, required=True)
parser.add_argument("–workdir", type=Path, default=Path("."))
parser.add_argument("–baseline-backend", default=None)
parser.add_argument("–target-backend", default=None)
args = parser.parse_args()

workdir = args.workdir.resolve()
policy = json.loads(args.policy.read_text(encoding="utf-8"))
log = workdir / "logs/embedding_swap"
log.mkdir(parents=True, exist_ok=True)
py = sys.executable

baseline_backend = args.baseline_backend or policy.get("baseline_backend", "tfidf_cosine_v1")
target_backend = args.target_backend or policy.get("target_backend", "hashing_embedding_v1")
max_chars = int(policy.get("max_chars", 240))
overlap = int(policy.get("overlap_chars", 40))
c_k = int(policy.get("canonical_top_k", 3))
p_k = int(policy.get("paraphrase_top_k", 1))

baseline_dir = log / "index_baseline"
target_dir = log / "index_target"
for d in (baseline_dir, target_dir):
if d.exists():
shutil.rmtree(d)

for label, backend, out_dir in [
("baseline", baseline_backend, baseline_dir),
("target", target_backend, target_dir),
]:
run(
[
py,
str(ROOT / "build_embedding_index.py"),
"–docs-dir",
str(args.docs_dir),
"–expansion",
str(args.expansion),
"–backend",
backend,
"–out-dir",
str(out_dir),
"–max-chars",
str(max_chars),
"–overlap-chars",
str(overlap),
"–label",
label,
],
workdir,
)

for name, index_dir, out_name in [
("baseline", baseline_dir, "baseline_eval.json"),
("target", target_dir, "target_eval.json"),
]:
run(
[
py,
str(ROOT / "eval_embedding_index.py"),
"–index-dir",
str(index_dir),
"–canonical",
str(args.canonical),
"–paraphrase",
str(args.paraphrase),
"–expansion",
str(args.expansion),
"–canonical-top-k",
str(c_k),
"–paraphrase-top-k",
str(p_k),
"–out",
str(log / out_name),
],
workdir,
)

# reuse compare_index_evals from day51 (same field names)
run(
[
py,
str(ROOT / "compare_index_evals.py"),
"–baseline",
str(log / "baseline_eval.json"),
"–current",
str(log / "target_eval.json"),
"–out",
str(log / "compare_report.json"),
],
workdir,
)

gate = run(
[
py,
str(ROOT / "embedding_swap_gate.py"),
"–policy",
str(args.policy),
"–baseline-eval",
str(log / "baseline_eval.json"),
"–target-eval",
str(log / "target_eval.json"),
"–compare-report",
str(log / "compare_report.json"),
"–baseline-meta",
str(baseline_dir / "index_meta.json"),
"–target-meta",
str(target_dir / "index_meta.json"),
],
workdir,
check=False,
)
print("EMBEDDING_SWAP_OK" if gate == 0 else "EMBEDDING_SWAP_FAIL")
raise SystemExit(gate)

if __name__ == "__main__":
main()

成功时末尾输出 EMBEDDING_SWAP_OK。

11. 人工核对清单

保存为 notes/embedding_swap_checklist.md：

Embedding 模型替换与索引回归核对清单

替换前

• 基线索引包已归档（index_baseline/ + baseline_eval.json）
• 原话层与变体层题集与线上一致
• 查询扩展表版本已锁定
• 策略文件写明 baseline_backend 与 target_backend

构建与元数据

• 目标后端重建后 index_meta.json 含 backend、model_id、model_fingerprint
• model_id 相对基线已变化
• 切分参数与基线一致（max_chars / overlap）

双层回归

• 原话层前三条检索命中率 ≥ 策略阈值
• 变体层排名第一命中率 ≥ 策略阈值
• 相对基线退步问法条数 ≤ max_regressed_questions
• 命中率掉点 ≤ max_hit_drop

闸门与切换

• embedding_swap_gate 输出 EMBEDDING_SWAP_ALLOWED
• 未达标时不得切换线上索引指针 / 模型配置
• 发布记录归档 compare_report.json 与两侧 index_meta.json

收尾

• 将目标索引包设为下一轮基线
• 真实 embedding（本地或 API）只替换 embed_backends 中对应实现，闸门字段保持不变

12. 常见错误

12.1 改模型配置但不重建索引

旧 vectors.json 仍按旧 backend 生成。换模必须走构建脚本，写出新的 model_fingerprint。

12.2 只验收原话层

弱模型对照里原话层仍 100%。变体层才是 Embedding 质量的主信号。

12.3 model_id 不写进元数据

没有 model_id，线上事故无法回答“当前向量是哪一版模型算的”。

12.4 把演示哈希向量当成生产模型

hashing_embedding_v1 只证明接口与闸门。生产应换成真实 Embedding，并沿用同一套评测与闸门。

12.5 换模时改动题集口径

题集与阈值应先冻结；否则无法判断掉点来自模型还是来自题目变更。

12.6 切分参数随手改

基线 240 字、目标 360 字，对比失去意义。切分参数与换模正交，应分开变更单。

13. 术语速查

14. 小结

Embedding 升级的正确动作是换后端 → 重建索引包 → 双层回归 → 闸门放行：

上文已给出策略、后端实现、构建、评测、闸门与一键脚本全文。可把本阶段接到《RAG 文档更新后的全链路闸门验收》：在向量索引重建之后增加换模闸门，再进入总闸门。

15. 相关阅读

• RAG 向量索引重建与回归
• RAG 文档更新后的全链路闸门验收
• RAG 关键词检索与向量检索的切换边界
• RAG 可回归评测问法的业务采集与标注
• 本地知识库落地顺序：先做可验收的 RAG

《RAG 向量索引重建与回归》回答文档变更后如何重建；《RAG Embedding 模型替换与索引回归》回答向量实现本身如何替换。闸门口径稳定后，再把哈希后端换成真实 Embedding，不会把验收链路一起打乱。

如果本篇对你有帮助，欢迎点赞、收藏，也欢迎关注后续更新。