Embedding 与向量检索

文本嵌入、向量数据库和 RAG（检索增强生成）架构详解。

什么是 Embedding？

Embedding（嵌入）是将文本转换为数值向量的技术。向量可以捕捉文本的语义信息，语义相似的文本在向量空间中距离更近。

为什么需要 Embedding？

大语言模型有两个核心限制：

1. 知识截止：训练数据有截止日期，不知道最新信息
2. 上下文限制：无法一次性处理海量文档

Embedding + 向量检索可以解决这些问题：将文档转换为向量存储，查询时检索相关内容，再交给 LLM 处理。

Embedding 的本质：

Embedding 模型将文本映射到高维向量空间。在这个空间中：

• 语义相似的文本距离近
• 语义不同的文本距离远
• 可以用数学方法计算相似度

"猫" → [0.2, 0.5, 0.1, ...]
"狗" → [0.3, 0.4, 0.2, ...]  # 与"猫"接近
"汽车" → [0.8, 0.1, 0.9, ...] # 与"猫"较远

应用场景

flowchart LR
    A[Embedding] --> B[语义搜索]
    A --> C[文档问答 RAG]
    A --> D[推荐系统]
    A --> E[文本分类]
    A --> F[聚类分析]

Embedding API 对比

选择建议：

OpenAI Embedding

基础调用

from openai import OpenAI

client = OpenAI()

response = client.embeddings.create(
    model="text-embedding-3-small",
    input="你好世界"
)

vector = response.data[0].embedding  # 1536 维向量
print(f"向量维度: {len(vector)}")  # 1536

批量处理

批量处理比逐条调用更高效，建议一次处理多条文本：

texts = ["文本1", "文本2", "文本3"]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts
)

vectors = [d.embedding for d in response.data]
# vectors[i] 对应 texts[i]

批量处理建议：

• 单次最多 2048 条文本
• 总 Token 数不超过模型限制
• 大量数据分批处理，避免超时

降维（节省存储）

text-embedding-3 系列支持指定输出维度，可以在精度和存储之间权衡：

# 指定输出维度（仅 text-embedding-3 支持）
response = client.embeddings.create(
    model="text-embedding-3-small",
    input="你好世界",
    dimensions=512  # 从 1536 降到 512
)

降维效果：

相似度计算

余弦相似度

余弦相似度是最常用的向量相似度计算方法，衡量两个向量的方向是否一致：

import numpy as np

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

# 使用
sim = cosine_similarity(vector1, vector2)
# 结果范围 [-1, 1]，越接近 1 越相似

其他相似度方法：

相似度阈值参考

注意： 阈值因模型和场景而异，建议通过实验确定最佳阈值。

向量数据库

向量数据库专门用于存储和检索高维向量，支持快速的相似度搜索。

为什么需要向量数据库？

普通数据库不支持高效的向量相似度搜索。向量数据库使用专门的索引算法（如 HNSW、IVF），可以在毫秒级完成百万级向量的相似度搜索。

主流选择

选择决策：

flowchart TD
    A[选择向量数据库] --> B{数据规模?}
    B -->|< 10万| C[Chroma/SQLite]
    B -->|10万-1000万| D{运维能力?}
    B -->|> 1000万| E[Milvus/Qdrant 集群]
    
    D -->|强| F[Qdrant/Milvus]
    D -->|弱| G[Pinecone 云服务]

Chroma 示例

Chroma 是最简单的向量数据库，适合本地开发和原型：

import chromadb

# 创建客户端
client = chromadb.Client()

# 创建集合
collection = client.create_collection("docs")

# 添加文档
collection.add(
    documents=["文档1内容", "文档2内容"],
    ids=["doc1", "doc2"]
)

# 查询
results = collection.query(
    query_texts=["查询内容"],
    n_results=5
)

Pinecone 示例

from pinecone import Pinecone

pc = Pinecone(api_key="YOUR_KEY")
index = pc.Index("my-index")

# 插入向量
index.upsert(vectors=[
    {"id": "doc1", "values": vector1, "metadata": {"text": "..."}},
    {"id": "doc2", "values": vector2, "metadata": {"text": "..."}}
])

# 查询
results = index.query(vector=query_vector, top_k=5, include_metadata=True)

RAG 架构

RAG（Retrieval-Augmented Generation）通过检索相关文档来增强 LLM 的回答能力。

工作流程

flowchart TD
    A[用户问题] --> B[Embedding]
    B --> C[向量检索]
    C --> D[获取相关文档]
    D --> E[构建 Prompt]
    E --> F[LLM 生成]
    F --> G[返回答案]
    
    H[(向量数据库)] --> C

基础实现

def rag_query(question, collection, llm_client):
    # 1. 检索相关文档
    results = collection.query(query_texts=[question], n_results=3)
    
    # 2. 构建上下文
    context = "\n".join(results["documents"][0])
    
    # 3. 生成回答
    prompt = f"""基于以下信息回答问题：

{context}

问题：{question}
"""
    
    response = llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
    
    return response.choices[0].message.content

文档处理

文本分块策略

长文档需要分块后再生成 Embedding。

flowchart LR
    A[长文档] --> B{分块策略}
    B --> C[固定长度]
    B --> D[按段落]
    B --> E[语义分块]

分块代码

def chunk_text(text, chunk_size=500, overlap=50):
    """重叠分块"""
    chunks = []
    start = 0
    while start < len(text):
        end = start + chunk_size
        chunks.append(text[start:end])
        start = end - overlap
    return chunks

优化技巧

1. 查询优化

# 使用 HyDE：先让 LLM 生成假设答案，再用假设答案检索
def hyde_search(question, collection, llm_client):
    # 生成假设答案
    hypo = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": f"简短回答：{question}"}]
    ).choices[0].message.content
    
    # 用假设答案检索
    return collection.query(query_texts=[hypo], n_results=5)

2. 重排序

# 检索后用 LLM 重排序
def rerank(question, docs, llm_client):
    prompt = f"""对以下文档按与问题的相关性排序，返回排序后的编号：

问题：{question}

文档：
{chr(10).join([f'{i+1}. {d}' for i, d in enumerate(docs)])}
"""
    # ... 解析 LLM 返回的排序

3. 混合检索

结合关键词搜索和向量搜索，取长补短：

flowchart TD
    A[查询] --> B[关键词搜索 BM25]
    A --> C[向量搜索]
    B --> D[结果融合 RRF]
    C --> D
    D --> E[最终结果]

为什么需要混合检索？

def hybrid_search(query, collection, bm25_index, alpha=0.5):
    """混合检索：向量 + BM25"""
    # 向量搜索
    vector_results = collection.query(query_texts=[query], n_results=20)
    
    # BM25 关键词搜索
    bm25_results = bm25_index.search(query, top_k=20)
    
    # RRF 融合（Reciprocal Rank Fusion）
    scores = {}
    k = 60  # RRF 参数
    
    for rank, doc_id in enumerate(vector_results["ids"][0]):
        scores[doc_id] = scores.get(doc_id, 0) + alpha / (k + rank + 1)
    
    for rank, doc_id in enumerate(bm25_results):
        scores[doc_id] = scores.get(doc_id, 0) + (1 - alpha) / (k + rank + 1)
    
    # 按分数排序
    sorted_docs = sorted(scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_docs[:10]

RAG 进阶技巧

1. 查询改写

用户查询可能不够清晰，先改写再检索：

def rewrite_query(query, llm_client):
    """改写查询以提高检索效果"""
    response = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""将以下查询改写为更适合搜索的形式，生成 3 个变体：

原始查询：{query}

改写要求：
1. 保持原意
2. 使用不同的表达方式
3. 包含可能的同义词

输出格式：每行一个查询"""
        }]
    )
    
    queries = response.choices[0].message.content.strip().split("\n")
    return [query] + queries  # 原始查询 + 改写查询

2. 上下文压缩

检索到的文档可能包含无关内容，压缩后再使用：

def compress_context(question, documents, llm_client):
    """压缩上下文，只保留相关部分"""
    response = llm_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{
            "role": "user",
            "content": f"""从以下文档中提取与问题相关的信息：

问题：{question}

文档：
{chr(10).join(documents)}

只输出与问题直接相关的内容，删除无关信息。"""
        }]
    )
    
    return response.choices[0].message.content

3. 多步检索

复杂问题可能需要多步检索：

flowchart TD
    A[复杂问题] --> B[分解子问题]
    B --> C[检索子问题1]
    B --> D[检索子问题2]
    C --> E[合并结果]
    D --> E
    E --> F[生成最终答案]

4. 引用来源

让 LLM 在回答中标注引用来源：

def rag_with_citations(question, documents, llm_client):
    """带引用的 RAG"""
    # 给文档编号
    numbered_docs = "\n".join([f"[{i+1}] {doc}" for i, doc in enumerate(documents)])
    
    response = llm_client.chat.completions.create(
        model="gpt-4o",
        messages=[{
            "role": "user",
            "content": f"""基于以下文档回答问题，在回答中用 [编号] 标注引用来源。

文档：
{numbered_docs}

问题：{question}

要求：
1. 只使用文档中的信息
2. 每个观点都要标注来源
3. 如果文档中没有相关信息，说明"根据提供的文档无法回答"
"""
        }]
    )
    
    return response.choices[0].message.content

常见问题

Q: 检索结果不相关怎么办？

Q: 如何评估 RAG 效果？

Q: 向量数据库如何选择？

flowchart TD
    A[选择向量数据库] --> B{预算}
    B -->|有预算| C{运维能力}
    B -->|无预算| D[开源方案]
    
    C -->|强| E[Milvus/Qdrant]
    C -->|弱| F[Pinecone]
    
    D --> G{数据规模}
    G -->|小| H[Chroma/SQLite]
    G -->|大| I[Qdrant/Milvus]

成本估算

使用 text-embedding-3-small，$0.02/百万 Token 存储成本基于 Pinecone 标准版估算

相关文档

• 20-大模型-知识库问答系统
• 02-OpenAI-Chat-Completions-API
• 11-大模型-Token计算与成本优化
• 24-大模型-LangChain入门