Embedding 向量技术入门与应用

从直觉理解到代码实战,掌握 AI 时代最重要的数据表示技术 Maurice | 灵阙学院

前置准备

  • Python 3.10+
  • pip install openai numpy scikit-learn matplotlib

一、什么是 Embedding(向量嵌入)

1.1 直觉解释

Embedding 是把"含义"变成"数字"的技术。

传统计算机理解文本的方式:
"猫" = UTF-8 编码 [0xE7, 0x8C, 0xAB]  --> 只知道编码,不知道含义

Embedding 理解文本的方式:
"猫" = [0.23, -0.15, 0.87, 0.42, ...]  --> 768 个数字,编码了"含义"
"狗" = [0.21, -0.12, 0.85, 0.39, ...]  --> 和"猫"很接近(都是宠物)
"桌子" = [-0.45, 0.67, 0.12, -0.33, ...]  --> 和"猫"差别很大

核心思想:语义相近的内容,在向量空间中距离也相近

1.2 中文示例

from openai import OpenAI
import numpy as np

client = OpenAI()

def get_embedding(text: str) -> list[float]:
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return resp.data[0].embedding

def cosine_similarity(a, b):
    a, b = np.array(a), np.array(b)
    return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b))

# 获取几组词的向量
words = {
    "国王": get_embedding("国王"),
    "王后": get_embedding("王后"),
    "男人": get_embedding("男人"),
    "女人": get_embedding("女人"),
    "苹果": get_embedding("苹果"),
}

# 计算相似度
print("国王 vs 王后:", round(cosine_similarity(words["国王"], words["王后"]), 4))
print("国王 vs 男人:", round(cosine_similarity(words["国王"], words["男人"]), 4))
print("国王 vs 苹果:", round(cosine_similarity(words["国王"], words["苹果"]), 4))

预期输出:

国王 vs 王后: 0.8934
国王 vs 男人: 0.7521
国王 vs 苹果: 0.3102

二、Embedding 技术演进

年代 技术 维度 特点
2013 Word2Vec 100-300 静态词向量,一词一向量
2018 BERT 768 上下文相关,同词不同义
2022 E5 / BGE 768-1024 专为检索优化
2024-26 OpenAI v3 256-3072 可调维度,多语言

关键进步:

  • Word2Vec 时代:"苹果"只有一个向量,无法区分水果和公司
  • BERT 之后:"苹果很甜"和"苹果发布新品"中的"苹果"向量不同
  • 现代模型:支持长文本(8K+ tokens),多语言零样本迁移

三、主流 Embedding 模型对比

pip install openai sentence-transformers

3.1 对比实验

from openai import OpenAI
from sentence_transformers import SentenceTransformer
import numpy as np

openai_client = OpenAI()

# BGE 中文模型(本地运行,免费)
bge_model = SentenceTransformer("BAAI/bge-small-zh-v1.5")

query = "如何提高代码质量?"
docs = [
    "代码审查是提升软件质量的有效方法",        # 相关
    "单元测试覆盖率应该达到 80% 以上",          # 相关
    "今天天气不错,适合去公园散步",              # 无关
    "重构可以降低技术债务,提升可维护性",        # 相关
]

# OpenAI Embedding
def openai_search(query, docs):
    all_texts = [query] + docs
    resp = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=all_texts
    )
    embeddings = [d.embedding for d in resp.data]
    q_emb = np.array(embeddings[0])
    scores = []
    for i, d_emb in enumerate(embeddings[1:]):
        sim = np.dot(q_emb, np.array(d_emb)) / (
            np.linalg.norm(q_emb) * np.linalg.norm(np.array(d_emb))
        )
        scores.append((docs[i], round(float(sim), 4)))
    return sorted(scores, key=lambda x: -x[1])

# BGE 本地 Embedding
def bge_search(query, docs):
    q_emb = bge_model.encode(query)
    d_embs = bge_model.encode(docs)
    scores = []
    for i, d_emb in enumerate(d_embs):
        sim = np.dot(q_emb, d_emb) / (
            np.linalg.norm(q_emb) * np.linalg.norm(d_emb)
        )
        scores.append((docs[i], round(float(sim), 4)))
    return sorted(scores, key=lambda x: -x[1])

print("=== OpenAI text-embedding-3-small ===")
for doc, score in openai_search(query, docs):
    print(f"  {score:.4f}  {doc}")

print("\n=== BGE-small-zh ===")
for doc, score in bge_search(query, docs):
    print(f"  {score:.4f}  {doc}")

预期输出:

=== OpenAI text-embedding-3-small ===
  0.8234  代码审查是提升软件质量的有效方法
  0.7891  重构可以降低技术债务,提升可维护性
  0.7456  单元测试覆盖率应该达到 80% 以上
  0.2103  今天天气不错,适合去公园散步

=== BGE-small-zh ===
  0.8567  代码审查是提升软件质量的有效方法
  0.8102  重构可以降低技术债务,提升可维护性
  0.7823  单元测试覆盖率应该达到 80% 以上
  0.1876  今天天气不错,适合去公园散步

3.2 模型选型指南

模型 维度 中文 价格 适用场景
OpenAI text-embedding-3-small 1536 $0.02/1M tokens 通用,性价比高
OpenAI text-embedding-3-large 3072 $0.13/1M tokens 高精度检索
BAAI/bge-small-zh-v1.5 512 优秀 免费(本地) 中文场景首选
BAAI/bge-large-zh-v1.5 1024 优秀 免费(本地) 中文高精度
Jina-embeddings-v3 1024 免费(本地) 多语言
Cohere embed-v4 1024 $0.1/1M tokens 多语言检索

四、实战应用

4.1 语义搜索

from openai import OpenAI
import numpy as np

client = OpenAI()

# 知识库
knowledge_base = [
    "Python 是一种解释型、面向对象的高级编程语言",
    "机器学习是人工智能的一个子领域",
    "Docker 是一个开源的容器化平台",
    "Git 是分布式版本控制系统",
    "RESTful API 使用 HTTP 方法进行资源操作",
    "向量数据库专门用于存储和检索高维向量",
    "Transformer 架构是现代 NLP 的基础",
    "微服务架构将应用拆分为独立的小服务",
]

# 预计算所有文档的向量
resp = client.embeddings.create(
    model="text-embedding-3-small",
    input=knowledge_base
)
doc_embeddings = [d.embedding for d in resp.data]

def search(query: str, top_k: int = 3) -> list[tuple[str, float]]:
    q_resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    )
    q_emb = np.array(q_resp.data[0].embedding)

    scores = []
    for i, d_emb in enumerate(doc_embeddings):
        sim = np.dot(q_emb, np.array(d_emb)) / (
            np.linalg.norm(q_emb) * np.linalg.norm(np.array(d_emb))
        )
        scores.append((knowledge_base[i], float(sim)))

    scores.sort(key=lambda x: -x[1])
    return scores[:top_k]

# 测试
results = search("如何管理代码版本?")
for doc, score in results:
    print(f"  {score:.4f}  {doc}")

预期输出:

  0.7823  Git 是分布式版本控制系统
  0.4512  Docker 是一个开源的容器化平台
  0.3967  微服务架构将应用拆分为独立的小服务

4.2 文本聚类

from sklearn.cluster import KMeans
import numpy as np

# 假设已有 doc_embeddings(接上面的代码)
embeddings_array = np.array(doc_embeddings)

# K-Means 聚类
kmeans = KMeans(n_clusters=3, random_state=42, n_init=10)
labels = kmeans.fit_predict(embeddings_array)

# 输出聚类结果
clusters = {}
for i, label in enumerate(labels):
    if label not in clusters:
        clusters[label] = []
    clusters[label].append(knowledge_base[i])

for cluster_id, docs in sorted(clusters.items()):
    print(f"\n--- 聚类 {cluster_id} ---")
    for doc in docs:
        print(f"  {doc}")

预期输出:

--- 聚类 0 ---
  Python 是一种解释型、面向对象的高级编程语言
  Git 是分布式版本控制系统
  Docker 是一个开源的容器化平台

--- 聚类 1 ---
  机器学习是人工智能的一个子领域
  Transformer 架构是现代 NLP 的基础
  向量数据库专门用于存储和检索高维向量

--- 聚类 2 ---
  RESTful API 使用 HTTP 方法进行资源操作
  微服务架构将应用拆分为独立的小服务

4.3 异常检测

import numpy as np

def detect_anomalies(embeddings: list[list[float]],
                     texts: list[str],
                     threshold: float = 2.0) -> list[str]:
    """基于向量距离的异常检测"""
    emb_array = np.array(embeddings)
    centroid = emb_array.mean(axis=0)

    distances = []
    for i, emb in enumerate(emb_array):
        dist = np.linalg.norm(emb - centroid)
        distances.append((texts[i], dist))

    mean_dist = np.mean([d[1] for d in distances])
    std_dist = np.std([d[1] for d in distances])

    anomalies = []
    for text, dist in distances:
        z_score = (dist - mean_dist) / std_dist
        if z_score > threshold:
            anomalies.append(text)
            print(f"  [ANOMALY] z={z_score:.2f}  {text}")

    return anomalies

# 检测混入的无关文档
test_docs = knowledge_base + ["今日大盘收跌,沪指报 3100 点"]
resp = client.embeddings.create(
    model="text-embedding-3-small",
    input=test_docs
)
test_embeddings = [d.embedding for d in resp.data]

anomalies = detect_anomalies(test_embeddings, test_docs, threshold=1.5)

预期输出:

  [ANOMALY] z=2.34  今日大盘收跌,沪指报 3100 点

五、降维可视化

from sklearn.manifold import TSNE
import matplotlib.pyplot as plt
import matplotlib
import numpy as np

matplotlib.rcParams['font.sans-serif'] = ['Arial Unicode MS']

# 准备数据
categories = {
    "编程": ["Python编程", "Java开发", "前端框架", "数据库设计"],
    "AI":   ["深度学习", "自然语言处理", "计算机视觉", "强化学习"],
    "美食": ["川菜做法", "日料寿司", "法式甜点", "烘焙入门"],
}

all_texts = []
all_labels = []
for label, texts in categories.items():
    all_texts.extend(texts)
    all_labels.extend([label] * len(texts))

# 获取向量
resp = client.embeddings.create(
    model="text-embedding-3-small",
    input=all_texts
)
embeddings = np.array([d.embedding for d in resp.data])

# t-SNE 降维到 2D
tsne = TSNE(n_components=2, random_state=42, perplexity=5)
coords = tsne.fit_transform(embeddings)

# 绘图
colors = {"编程": "#FF6B6B", "AI": "#4ECDC4", "美食": "#FFE66D"}
plt.figure(figsize=(10, 8))
for i, (x, y) in enumerate(coords):
    label = all_labels[i]
    plt.scatter(x, y, c=colors[label], s=100, zorder=5)
    plt.annotate(all_texts[i], (x, y), fontsize=9,
                 textcoords="offset points", xytext=(5, 5))

# 图例
for label, color in colors.items():
    plt.scatter([], [], c=color, label=label, s=100)
plt.legend(fontsize=12)
plt.title("Embedding 2D Visualization (t-SNE)", fontsize=14)
plt.tight_layout()
plt.savefig("embedding_tsne.png", dpi=150)
plt.show()
print("Saved: embedding_tsne.png")

可视化结果中,同一类别的点会自然聚集在一起,不同类别之间有明显的距离。


六、性能优化建议

6.1 批量处理

# 单条调用(慢)
for text in texts:
    embedding = get_embedding(text)  # 每次一个 HTTP 请求

# 批量调用(快,最多 2048 条/次)
resp = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts[:2048]  # 一次请求
)
embeddings = [d.embedding for d in resp.data]

6.2 维度压缩

OpenAI v3 模型支持指定输出维度:

# 1536 维(默认)
resp = client.embeddings.create(model="text-embedding-3-small", input="test")

# 256 维(存储节省 83%,精度略降)
resp = client.embeddings.create(
    model="text-embedding-3-small",
    input="test",
    dimensions=256
)

6.3 缓存策略

import hashlib
import json
import os

CACHE_DIR = ".embedding_cache"
os.makedirs(CACHE_DIR, exist_ok=True)

def get_embedding_cached(text: str) -> list[float]:
    key = hashlib.md5(text.encode()).hexdigest()
    cache_path = os.path.join(CACHE_DIR, f"{key}.json")
    if os.path.exists(cache_path):
        with open(cache_path) as f:
            return json.load(f)
    embedding = get_embedding(text)
    with open(cache_path, "w") as f:
        json.dump(embedding, f)
    return embedding

常见问题

Q1: Embedding 和 LLM 有什么区别? Embedding 模型将文本转为固定维度的向量(用于搜索/分类),LLM 模型生成文本。Embedding 成本是 LLM 的 1/100,延迟是 1/10。

Q2: 中文场景选哪个模型? 优先 BGE 系列(本地运行、免费、中文优化)。如果需要 API 调用的便捷性,OpenAI text-embedding-3-small 性价比最高。

Q3: 向量维度越高越好吗? 不一定。512-1024 维通常够用。更高维度增加存储和计算成本,收益递减。可以先用低维度快速验证,再根据需要提升。

Q4: 如何评估 Embedding 质量? 用 MTEB 排行榜(Massive Text Embedding Benchmark)作为参考。实际项目中,用自己的测试集评估检索准确率(Recall@K)最可靠。

Q5: 向量需要定期更新吗? 如果知识库内容频繁变化,需要增量更新向量。Embedding 模型本身不变,但如果切换模型版本,需要重新计算所有向量。


Maurice | maurice_wen@proton.me