Embedding 向量技术入门与应用
AI 导读
Embedding 向量技术入门与应用 从直觉理解到代码实战,掌握 AI 时代最重要的数据表示技术 Maurice | 灵阙学院 前置准备 Python 3.10+ pip install openai numpy scikit-learn matplotlib 一、什么是 Embedding(向量嵌入) 1.1 直觉解释 Embedding 是把"含义"变成"数字"的技术。...
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