Embedding 向量化：阿里百炼
一、 embedding 在 RAG 里到底干了什么
把 RAG 拆成四步：文档 → chunk → 向量 → 检索。
embedding 就是第三步——把一段文本变成一个固定长度的浮点数数组。比如"王建国教授教数据结构"变成 1024 个 float32：
# 输入：一段中文文本
text = "王建国教授教数据结构"
# 输出：一个 1024 维的浮点数向量
embedding = [0.023, -0.041, 0.008, ..., 0.015] # 1024 个数字
这个数组有什么用？两条语义相近的文本，它们的向量在空间中距离近。 用户问"谁教数据结构"，这个 query 的向量和"王建国教授教数据结构"的向量余弦相似度很高，查出来就是它。
这就是为什么 embedding 模型的选择直接影响 RAG 的检索质量——如果模型不理解中文、不理解专业术语，向量再快也没用。
二、阿里百炼的文本向量
前面几篇文章用到的阿里百炼生态——百炼做 chunk 拆分（32-bailian-chunking.md）。embedding 这一环继续用阿里百炼的 DashScope，整个技术栈统一。
百炼目前最新的文本向量模型是 text-embedding-v4，关键指标：
- 1024 维输出（默认，支持从 64 到 2048 可调）
- 最大输入 8192 token，一条 chunk 最长能塞约 5000 个中文字
- 支持 100+ 语种，中英文混合文档不需要换模型
- 支持
text_type参数区分 query 和 document，检索精度有明显提升 - 北京地域延迟约 20-40ms，跟调 OpenAI 差不多
百炼文本向量模型历史：
| 模型 | 维度 | 最大 Token | 价格(每千Token) | 状态 |
|---|---|---|---|---|
| text-embedding-v1 | 1536 | 2,048 | 已下架 | 别用了 |
| text-embedding-v2 | 1536 | 2,048 | ¥0.0007 | 老模型，不推荐 |
| text-embedding-v3 | 1024(默认)/768/512/256/128/64 | 8,192 | ¥0.0005 | 上一代主力 |
| text-embedding-v4 | 1024(默认)/2048/1536/768/512/256/128/64 | 8,192 | ¥0.0005 | 当前推荐 |
v4 比 v3 的主要提升是语言覆盖（50+ → 100+）和检索精度（C-MTEB 中文榜单涨了约 2 个点）。价格一样，直接上 v4。
新开通百炼的用户有 90 天免费额度，v4 模型赠送 100 万 token。够把你整个知识库 embed 好几遍了。
三、接入——OpenAI 兼容模式，不用装额外 SDK
百炼的文本向量支持 OpenAI 兼容接口，直接用 openai 包调：
# pip install openai
from openai import OpenAI
client = OpenAI(
api_key="sk-你的百炼APIKey", # 在 bailian.console.aliyun.com → API-KEY 管理里创建
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
base_url 这个地址是关键——北京地域用 dashscope.aliyuncs.com，新加坡地域用 dashscope-intl.aliyuncs.com。国内服务器用北京地域，延迟低。
如果用原生 DashScope SDK（pip install dashscope），调用方式稍微不一样，但效果一样。我个人习惯用 OpenAI 兼容模式——跟调 DeepSeek 的 LLM 用同一个 openai 包，依赖少。
四、单条和批量调用
单条文本嵌入：
def embed_text(text: str, text_type: str = "query") -> list[float]:
"""
把单条文本转成向量。
text_type="query" —— 表示这是用户查询，模型会调整向量使其更匹配 document
text_type="document" —— 表示这是知识库文档，模型会调整向量使其更匹配 query
这个参数是 v3/v4 的重要特性。传对了检索精度有明显提升。
"""
resp = client.embeddings.create(
model="text-embedding-v4",
input=text,
dimensions=1024, # 输出维度，默认 1024，范围 64-2048
extra_body={"text_type": text_type}, # query 或 document
)
return resp.data[0].embedding # list[float]，长度 1024
实际入库时几百几千条 chunk，一条一条调会慢死。API 支持一次传多条：
def embed_batch(texts: list[str], text_type: str = "document", batch_size: int = 50) -> list[list[float]]:
"""
批量 embedding。
百炼 API 单次允许最多传多少条文档没写死，但 batch_size 设 50 是经验值。
为什么是 50 而不是 100？
1. 中文 chunk 通常 400-500 字，约 600-800 token。50 条就是 30000-40000 token。
加上 v4 的输入限制是每条 8192 token，设 50 条时即使某条特别肥也不超限。
2. API 单次请求的延迟和总 token 量正相关。50 条约 2-3 秒，
对离线入库来说能接受。设 100 也不会报错，但单次要等 4-5 秒，
中途网络抖动全 batch 重来，反而更慢。
"""
all_embeddings = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
resp = client.embeddings.create(
model="text-embedding-v4",
input=batch,
dimensions=1024,
extra_body={"text_type": text_type},
)
# API 返回的数据顺序和 input 顺序严格一致
batch_embeddings = [item.embedding for item in resp.data]
all_embeddings.extend(batch_embeddings)
print(f" [{i+len(batch)}/{len(texts)}] token消耗: {resp.usage.total_tokens}")
return all_embeddings
API 返回的顺序和传入的顺序一致——resp.data[0] 对应 texts[0]。不需要自己加 index。
五、text_type 是什么——一个容易被忽略但很有用的参数
v3/v4 支持传 text_type，告诉模型"这段文本是查询还是文档"。模型内部会根据这个标记调整向量空间——query 向量会往 document 向量集中的区域拉，提高匹配精度。
# 入库时：所有 chunk 都标记为 document
embeddings = embed_batch(chunks, text_type="document")
# 查询时：用户的 query 标记为 query
query_vec = embed_text("谁教数据结构", text_type="query")
如果不传 text_type，模型用默认行为，query 和 document 的向量不会有针对性调整。在我的测试中，传对了 text_type 后 recall@10 提升了约 2-3 个点。改动成本为零——只是多传了一个参数。
注意：v1/v2 不支持 text_type，传了会被忽略。只有 v3/v4 有这个能力。
六、维度选择——1024 是甜点
v4 的输出维度可调，从 64 到 2048。维度越高，向量能编码的语义信息越多，但存储和检索引擎的计算开销也越大。
# 低维度（512）：省存储，召回率略低，适合百万级以上的大数据量
resp = client.embeddings.create(
model="text-embedding-v4", input=text, dimensions=512
)
# 默认维度（1024）：各项均衡，推荐
resp = client.embeddings.create(
model="text-embedding-v4", input=text, dimensions=1024
)
# 高维度（2048）：召回率最高，适合十万级以下、质量要求高的场景
resp = client.embeddings.create(
model="text-embedding-v4", input=text, dimensions=2048
)
各个维度的存储和计算开销：
维度 一条向量占用 10万条占用 相似度计算耗时
512 2KB 200MB ~0.5ms
1024 4KB 400MB ~1ms
1536 6KB 600MB ~2ms
2048 8KB 800MB ~3ms
我的选择是 1024。 湖北理工官网的知识库约 1500 条 chunk——这个量级下 1024 维和 2048 维的存储差异无关紧要（差 6MB），但 1024 维的检索延迟更低，而且刚好是一般 GPU 最友好的维度。如果你在做一个百万级文档的知识库，而且对延迟敏感，降到 512 维——存储量砍半，召回率只掉 1-2 个点。
选维度的原则：看数据量，别只看模型排名。 10 万条以内 1024 随便用，50 万条以上优先 512。
七、token 怎么算——控制成本的关键
embedding API 按 token 计费（¥0.0005/千 token），不是按条数。一条 400 字的 chunk 和一条 4000 字的 chunk 成本差 10 倍。
v4 最大输入 8192 token，超过直接报错。入库前估算一下 token 数：
# dashscope 提供了 token 计数工具
# pip install dashscope
from dashscope import get_tokenizer
tokenizer = get_tokenizer("qwen-turbo") # 百炼的 tokenizer 通用
def count_tokens(text: str) -> int:
return len(tokenizer.encode(text))
text = "王建国教授教数据结构与算法课程"
print(count_tokens(text)) # 大概 15-18 个 token
中文文本：1 个中文字 ≈ 1.2-1.5 个 token。一条 500 字的 chunk 大概 600-750 token。v4 的 8192 token 上限轻松装下。
如果你不想装 dashscope SDK 只为了数 token，用粗略估算法：中文字数 × 1.5 ≈ token 数。精确度差 5-10%，但做成本估算够用了。
八、完整 pipeline——从 chunk 到检索一条线穿起来
把前面几篇文章的内容和这里的 embedding 串起来：
import hashlib
from openai import OpenAI
from pymilvus import connections, Collection, CollectionSchema, FieldSchema, DataType
client = OpenAI(
api_key="sk-你的百炼APIKey",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
)
# ===== 第 0 步：连接 Milvus =====
connections.connect(host="localhost", port="19530")
# 定义 Collection（只执行一次）
fields = [
FieldSchema(name="id", dtype=DataType.VARCHAR, max_length=64, is_primary=True),
FieldSchema(name="text", dtype=DataType.VARCHAR, max_length=8192),
FieldSchema(name="embedding", dtype=DataType.FLOAT_VECTOR, dim=1024), # v4 默认 1024 维
FieldSchema(name="doc_id", dtype=DataType.VARCHAR, max_length=64),
]
schema = CollectionSchema(fields, description="学院官网 RAG 知识库")
collection = Collection(name="hubei_cs_kb", schema=schema)
# ===== 第 1 步：文档 → chunk =====
# 这一块在 22-multimodal-chunking.md 和 32-bailian-chunking.md 里写得很详细
chunks = [
{"text": "王建国教授负责数据结构与算法(CS201)、算法设计与分析(CS301)两门课程。", "doc_id": "teacher_001"},
{"text": "数据结构与算法考试包含期末笔试70%+平时作业20%+实验10%。", "doc_id": "course_cs201"},
{"text": "实验课旷课2次取消考试资格，需提前向教务处报备（2025版规定）。", "doc_id": "policy_exam"},
# ... 几百条
]
# ===== 第 2 步：chunk → 向量（阿里百炼 embedding）=====
texts = [c["text"] for c in chunks]
embeddings = embed_batch(texts, text_type="document", batch_size=50)
# ===== 第 3 步：向量 + 元数据 → Milvus =====
ids = [hashlib.md5(c["text"].encode()).hexdigest() for c in chunks]
doc_ids = [c["doc_id"] for c in chunks]
BATCH = 1000
for i in range(0, len(chunks), BATCH):
collection.insert([
ids[i:i+BATCH],
texts[i:i+BATCH],
embeddings[i:i+BATCH],
doc_ids[i:i+BATCH],
])
collection.flush()
collection.create_index(
field_name="embedding",
index_params={
"index_type": "HNSW",
"metric_type": "COSINE",
"params": {"M": 16, "efConstruction": 200},
}
)
collection.load()
print(f"入库完成：{len(chunks)} 条 chunk")
# ===== 第 4 步：查询 =====
def search(query: str, top_k: int = 5) -> list[dict]:
"""用户查询 → 向量 → Milvus 检索 → 返回 top-k chunk"""
query_vec = embed_text(query, text_type="query") # 注意：text_type="query"
results = collection.search(
data=[query_vec],
anns_field="embedding",
param={"metric_type": "COSINE", "params": {"ef": 128}},
limit=top_k,
output_fields=["text", "doc_id"],
)
return [
{
"text": hit.entity.get("text"),
"doc_id": hit.entity.get("doc_id"),
"score": hit.distance, # COSINE 的 distance 即相似度
}
for hit in results[0]
]
# 测试
results = search("谁教数据结构")
for r in results:
print(f" [{r['score']:.3f}] {r['text'][:80]}...")
输出：
[0.921] 王建国教授负责数据结构与算法(CS201)、算法设计与分析(CS301)两门课程。
[0.756] 数据结构与算法考试包含期末笔试70%+平时作业20%+实验10%。
[0.634] 实验课旷课2次取消考试资格，需提前向教务处报备（2025版规定）。
第一条完美命中，后面两条语义相关但不精确。score=0.921 在 COSINE 下已经很好了。
九、为什么要用 MD5 做 chunk ID
上面用 MD5 生成 chunk ID，不用 UUID。原因是：
同一段文本永远生成同一个 ID → 自动去重。 UUID 每次入库生成新 ID，同一条 chunk 会存两份。Milvus 的 VARCHAR 主键是唯一约束——ID 重复时新数据覆盖旧数据，天然支持增量更新。
# UUID：每次生成新 ID，同一条 chunk 存两遍
id = str(uuid.uuid4()) # "a1b2c3d4-..."
# MD5：同一段文本永远生成同一个 ID，自动去重
id = hashlib.md5(text.encode()).hexdigest() # "3f8a9b2c..."
增量更新时特别有用：文档改了某一段，重新 embed 新 chunk，MD5 ID 不变，insert 进去自动覆盖旧的。不需要找出来旧的先 delete。
十、embedding 文本和检索返回的文本可以不一样
这是一个经验技巧。
embedding 模型需要干净、语义密集的文本才能产出好向量。但检索返回给 LLM 的文本需要信息完整、可读。两者的最佳形态可能不同：
# 给 embedding 用的文本 —— 拼接了标题和关键词，语义更聚焦
text_for_embedding = "教师 王建国 数据结构 CS201 课程 计算机学院"
# 检索返回给 LLM 的文本 —— 完整的原文
text_for_retrieval = (
"王建国教授负责数据结构与算法(CS201)，"
"上课地点在计算机楼301，每周二周四第3-4节。"
)
入库时用 text_for_embedding 调 API 生成向量，但 Milvus 里 text 字段存 text_for_retrieval。查询时搜到的是原文，可以直接喂给 LLM。
在我的实测中，embedding 文本里拼上标题、课程编号、分类标签后，recall@10 高了约 3 个点。改动成本为零——只是 embed 的时候换了段文本。
十一、缓存 embedding——同样的文本不要嵌两遍
知识库会频繁更新，但大部分 chunk 没变。每次都重新 embed 所有 chunk 是浪费。最简单的缓存：
import json
import sqlite3
cache_conn = sqlite3.connect("embedding_cache.db")
cache_conn.execute("""
CREATE TABLE IF NOT EXISTS emb_cache (
text_hash TEXT PRIMARY KEY, -- 文本的 MD5
embedding TEXT -- JSON 序列化后的向量
)
""")
def embed_with_cache(text: str, text_type: str = "document") -> list[float]:
"""同样的文本不重复调 API"""
text_hash = hashlib.md5(text.encode()).hexdigest()
row = cache_conn.execute(
"SELECT embedding FROM emb_cache WHERE text_hash=?", (text_hash,)
).fetchone()
if row:
return json.loads(row[0])
embedding = embed_text(text, text_type=text_type)
cache_conn.execute(
"INSERT INTO emb_cache VALUES (?, ?)",
(text_hash, json.dumps(embedding))
)
cache_conn.commit()
return embedding
增量更新时——每天新增 10 篇通知，只有这 10 篇需要调 API，已有的几百篇从 SQLite 缓存直接读。
十二、Batch 模式——大批量入库时价格减半
百炼支持 Batch 模式，价格是实时 API 的一半（¥0.00025/千 token）。缺点是异步——提交 job 后要等几分钟到几小时。
适合的场景是"首次入库"——你有一整批文档要处理，不赶时间。每天增量更新那几条不值得用 Batch——实时 API 几秒出结果。
Batch 模式的使用方式跟实时 API 一样，只是调用的 endpoint 不同。具体可以参考百炼官方文档的 Batch 调用章节。
十三、实际成本估算
以湖北理工官网助手为例：
文档量：200 篇官网页面
chunk 数：约 1500 条（每篇 7-8 个 chunk）
每 chunk 约 500 字，约 750 token
总 token：1500 × 750 = 1,125,000
用 text-embedding-v4 实时 API：
首次入库：1,125,000 token × ¥0.0005/千token ≈ ¥0.56
用 Batch 模式：
首次入库：1,125,000 token × ¥0.00025/千token ≈ ¥0.28
每天增量更新（新增 5 篇，约 40 条 chunk）：
实时 API：40 × 750 × ¥0.0005/千token ≈ ¥0.015
90 天免费额度（100 万 token）：
免费用完了才计费。首次入库 112 万 token，用掉免费额度，
超出 12.5 万 token，实际付费 ¥0.06
这个量级下 embedding 的成本基本忽略不计。真正花钱的是 LLM 生成回答。如果知识库大到百万级 chunk，那才需要考虑成本优化——用 Batch 模式或者降到 512 维换存储成本。