AI/ML 详解：RAG 体系

就像把一句话『hash 到一个 1536 维的空间』——只是这个 hash 不再追求"碰撞越少越好"，而是追求"语义越近的句子坐标越接近"。两个向量的余弦距离就是它们的"语义相似度"。

传统全文搜索（grep、ElasticSearch）只能匹配"字面"，搜"汽车"找不到"轿车"，搜"如何省钱"找不到"理财技巧"。Embedding 把语义编码成数值向量后，"找意思相近的文档"就变成了"找坐标相近的点"——一个数据库问题。这是整个 RAG（Retrieval-Augmented Generation）体系的地基。

训练时用大量"句子对+相似/不相似标签"，让模型学到一个映射 f(text) → vector，使得"语义相近 → 向量靠近"。常见维度是 384 / 768 / 1536 / 3072。用的时候只需一次 API 调用就把任意字符串变成向量。

# 直觉演示：相似度 ≈ 向量夹角的余弦
"猫喜欢吃鱼"        → [0.12, -0.34, 0.88, ...]   ┐
"我家猫爱吃鱼罐头"  → [0.15, -0.30, 0.91, ...]   ├ 这两条夹角很小 ⇒ 相似度 ≈ 0.92
"利率上调影响楼市"  → [0.71, 0.05, -0.42, ...]   ┘ 与上面夹角大 ⇒ 相似度 ≈ 0.08

类比：Word2Vec 那个经典例子——king - man + woman ≈ queen。语义关系被编码成了向量的方向。

from openai import OpenAI
import numpy as np

client = OpenAI()

def embed(text):
    # 一行调用：字符串 → 1536 维向量
    r = client.embeddings.create(model="text-embedding-3-small", input=text)
    return np.array(r.data[0].embedding)

def cos_sim(a, b):
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

a = embed("如何控制孩子玩手机的时间")
b = embed("管理儿童屏幕使用时长的方法")
c = embed("美联储加息对股市的影响")

print(cos_sim(a, b))  # ~0.85 高
print(cos_sim(a, c))  # ~0.12 低

• Jay Alammar 的 Illustrated Word2Vec — 经典图解
• OpenAI 官方 Embeddings 文档

就像把 MySQL 的 WHERE 子句从"=、LIKE"换成"按余弦距离 ORDER BY 相似度 LIMIT 10"——专门为"找最近的几百万个向量"做了索引优化的数据库。

有了 embedding 之后，"找最相似的 K 个文档"变成"在千万个 1536 维向量里算 cosine similarity 取 top-K"。暴力算每次都要 O(N) 次浮点运算，千万级别就要几秒——线上不可接受。向量数据库（Pinecone、Weaviate、Milvus、pgvector）用近似最近邻（ANN, Approximate Nearest Neighbor）索引把查询压到几毫秒。

最常见的索引是 HNSW（Hierarchical Navigable Small World）：把向量组织成多层图，上层稀疏只放"枢纽点"，下层稠密放所有点。查询时先在上层快速跳到大致区域，再下沉到下层精细比对。类比 skip list 的多层指针。

# 传统 SQL                          # 向量 DB
SELECT * FROM docs                  collection.query(
WHERE category = 'finance'    ←→      query_vector=embed(q),
ORDER BY title;                       filter={"category":"finance"},
                                      top_k=10)
# 关键差别：
# 传统 DB 索引 = B+ Tree，按精确字段排序
# 向量 DB 索引 = HNSW/IVF，按"高维空间距离"近似排序

注意"近似"两个字——为了速度它不保证 100% 召回真正的 top-K，但通常 95%+ 召回足够日常使用。

import chromadb
from openai import OpenAI

openai = OpenAI()
client = chromadb.Client()
col = client.create_collection("my_notes")

# 1) 写入：每条文本 + 自动 embedding + metadata
docs = ["复利是世界第八大奇迹", "屏幕时间影响儿童注意力", ...]
embeds = [openai.embeddings.create(model="text-embedding-3-small",
                                   input=d).data[0].embedding for d in docs]
col.add(documents=docs, embeddings=embeds,
        metadatas=[{"topic": "finance"}, {"topic": "parenting"}],
        ids=["n1", "n2"])

# 2) 查询：自然语言 → 找出最相关的 K 条
q_emb = openai.embeddings.create(model="text-embedding-3-small",
                                 input="如何让钱滚钱").data[0].embedding
hits = col.query(query_embeddings=[q_emb], n_results=3,
                 where={"topic": "finance"})  # metadata 过滤
print(hits["documents"])  # → ["复利是世界第八大奇迹", ...]

• Pinecone 的 HNSW 图解教程 — 友好讲清楚 ANN 索引
• Chroma 官方文档 — 最易上手的本地向量库

就像写 SQL 查询要先 EXPLAIN：单纯"embedding + top-K"在简单场景能跑，但生产里要混搭 keyword 搜索、metadata 过滤、查询改写、分块策略，像调一个多层缓存命中率一样调出"召回 → 排序 → 生成"的整条流水线。

朴素 RAG 上线后第一个发现：模型答错的原因 80% 不是 LLM 不行，而是"根本没召回到正确文档"。这就是 Retrieval 策略要解决的：怎么切文档（chunking）？怎么写查询（query rewriting）？纯向量 + 关键词 + metadata 怎么混合（hybrid search）？多次检索还是一次到位（multi-hop）？这些都是为了把"我想问的问题"和"知识库里真正相关的那块"对齐。

生产级 RAG 的典型 pipeline：

用户问题 q
   ↓ 查询改写（query rewriting）
      用 LLM 把"上周那篇关于复利的笔记呢？"改成
      "复利 投资 长期收益"，更利于 embedding 召回
   ↓ 多路召回（hybrid retrieval）
      ① 向量召回 top 20  ─┐
      ② BM25 关键词 top 20 ─┼→ 取并集（约 30-40 条）
      ③ metadata 过滤      ─┘
   ↓ 重排序（re-ranking，下一节细讲）
      用更精细的模型把 30-40 条排成 top 5
   ↓ 拼接进 prompt → LLM 生成答案

另一个关键设计是 chunking（切块）：把长文档切成 200-800 token 的小块再 embedding。切得太大召回的内容噪音多，切得太小又会丢上下文。常见策略是"按语义边界切 + 相邻 chunk 留 50-100 token 的 overlap"。

from langchain.retrievers import EnsembleRetriever, BM25Retriever
from langchain_community.vectorstores import Chroma
from langchain_openai import OpenAIEmbeddings

# 1) 准备向量召回 + BM25 召回两条管道
vec_retriever = Chroma(embedding_function=OpenAIEmbeddings()).as_retriever(k=10)
bm25_retriever = BM25Retriever.from_documents(docs, k=10)

# 2) Hybrid：加权融合两路结果（向量 0.6 + 关键词 0.4）
hybrid = EnsembleRetriever(
    retrievers=[vec_retriever, bm25_retriever],
    weights=[0.6, 0.4]
)

# 3) 查询改写：先让 LLM 把用户原话改成更"可搜索"的表述
def retrieve(user_question):
    rewritten = llm.invoke(f"把这个问题改写为关键词搜索：{user_question}")
    return hybrid.invoke(rewritten)

• Anthropic 的 Contextual Retrieval 工程笔记 — 实战派
• LangChain Retrieval 概念文档 各种策略对比

就像两阶段招聘：第一轮 HR 用关键词海选 100 人（粗排），第二轮面试官精读简历挑出 5 人（精排）。Re-ranker 就是 RAG 流水线里的"面试官"——用更慢但更准的模型，把召回的 30-50 条精筛成最相关的 top 5。

Embedding 检索快但"粗糙"——它把整段话压成单一向量，丢失了细节。Re-ranker 反其道而行：把"查询 + 候选文档"成对喂给一个更精细的模型（通常是 cross-encoder），直接输出"这条文档跟查询的相关性分数"。算力代价大，但只对召回的几十条算，总成本可控。结果是 RAG 的"答案正确率"经常因为加一层 re-ranking 提升 10-20 个点。

Embedding（bi-encoder）和 re-ranker（cross-encoder）的核心区别：

# Bi-encoder（embedding）：query 和 doc 各自独立 embed
emb_q   = encode(query)        ─┐
emb_doc = encode(doc)          ─┼→ cosine_similarity → 分数
# 优点：doc 可以预先 embed 入库，查询时只 embed query；O(1) 比较
# 缺点：query 和 doc 互不知道对方存在，细粒度匹配差

# Cross-encoder（re-ranker）：query 和 doc 拼起来一起喂模型
score = model(f"[CLS] {query} [SEP] {doc}")
# 优点：模型每一层都能让 query 和 doc 的 token 互相 attend，匹配更准
# 缺点：每对都要跑一次模型，N×M 复杂度，无法全库扫

所以工程上是"两阶段"：bi-encoder 快速从百万级粗筛到几十条，cross-encoder 在这几十条上精细排序。

from sentence_transformers import CrossEncoder
import cohere  # 或用 Cohere 托管的 reranker API

# 方案 A：本地 cross-encoder 模型（开源）
reranker = CrossEncoder("BAAI/bge-reranker-base")
query = "如何让孩子专注做作业"
candidates = [d1, d2, ..., d30]   # 从向量库召回的 30 条

scores = reranker.predict([(query, d) for d in candidates])
top5 = sorted(zip(candidates, scores), key=lambda x: -x[1])[:5]

# 方案 B：托管 API（更省事）
co = cohere.Client()
result = co.rerank(query=query, documents=candidates,
                   top_n=5, model="rerank-multilingual-v3.0")
top5 = [r.document["text"] for r in result.results]

# top5 拼进 prompt，再交给 LLM 生成最终答案

• Pinecone 的 Rerankers 教程 — 概念到代码全覆盖
• Cohere Rerank 文档 — 最常用的托管 reranker API