嵌入模型（Embeddings） #

┌─────────────────────────────────────────────────────────────┐
│                    什么是嵌入？                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  文本 ───> 嵌入模型 ───> 向量                               │
│                                                             │
│  "机器学习" ───> [0.1, 0.3, -0.2, 0.5, ...]                │
│                                                             │
│  特点：                                                     │
│  - 语义相似的文本，向量也相似                               │
│  - 可以计算向量之间的相似度                                 │
│  - 支持多语言                                               │
│                                                             │
│  应用：                                                     │
│  ✅ 语义搜索                                                │
│  ✅ 文档检索                                                │
│  ✅ 聚类分析                                                │
│  ✅ 推荐系统                                                │
│  ✅ 异常检测                                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘

from langchain_openai import OpenAIEmbeddings

# 创建嵌入模型
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small"  # 或 text-embedding-3-large
)

# 嵌入单个文本
text = "机器学习是人工智能的一个分支"
vector = embeddings.embed_query(text)

print(f"向量维度: {len(vector)}")  # 1536
print(f"前5个值: {vector[:5]}")

# 嵌入多个文本
texts = [
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络",
    "今天天气很好"
]
vectors = embeddings.embed_documents(texts)

print(f"生成了 {len(vectors)} 个向量")

# text-embedding-3-small
small_embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
# 维度: 1536
# 价格: $0.02/1M tokens

# text-embedding-3-large
large_embeddings = OpenAIEmbeddings(model="text-embedding-3-large")
# 维度: 3072
# 价格: $0.13/1M tokens

# text-embedding-ada-002 (旧版)
ada_embeddings = OpenAIEmbeddings(model="text-embedding-ada-002")
# 维度: 1536

from langchain_openai import OpenAIEmbeddings

embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    # 可选参数
    chunk_size=1000,  # 批处理大小
    max_retries=2,    # 重试次数
    request_timeout=60
)

from langchain_huggingface import HuggingFaceEmbeddings

embeddings = HuggingFaceEmbeddings(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    model_kwargs={'device': 'cpu'},
    encode_kwargs={'normalize_embeddings': True}
)

# 常用模型：
# - all-MiniLM-L6-v2: 快速，384 维
# - all-mpnet-base-v2: 高质量，768 维
# - paraphrase-multilingual-mpnet-base-v2: 多语言

from langchain_ollama import OllamaEmbeddings

embeddings = OllamaEmbeddings(
    model="nomic-embed-text"
)

# 常用模型：
# - nomic-embed-text: 通用嵌入
# - mxbai-embed-large: 高质量

from langchain_cohere import CohereEmbeddings

embeddings = CohereEmbeddings(
    model="embed-english-v3.0",
    cohere_api_key="your-key"
)

from langchain_google_genai import GoogleGenerativeAIEmbeddings

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001"
)

from langchain_openai import AzureOpenAIEmbeddings

embeddings = AzureOpenAIEmbeddings(
    azure_deployment="your-deployment",
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_version="2024-02-15-preview"
)

import numpy as np

def cosine_similarity(v1, v2):
    """计算两个向量的余弦相似度"""
    return np.dot(v1, v2) / (np.linalg.norm(v1) * np.linalg.norm(v2))

# 示例
text1 = "机器学习是人工智能的一个分支"
text2 = "深度学习是机器学习的一个分支"
text3 = "今天天气很好"

v1 = embeddings.embed_query(text1)
v2 = embeddings.embed_query(text2)
v3 = embeddings.embed_query(text3)

print(f"文本1 vs 文本2: {cosine_similarity(v1, v2):.4f}")  # 高相似度
print(f"文本1 vs 文本3: {cosine_similarity(v1, v3):.4f}")  # 低相似度

from langchain_community.utils.math import cosine_similarity

# 计算相似度矩阵
vectors = embeddings.embed_documents([
    "机器学习",
    "深度学习",
    "天气很好"
])

similarity_matrix = cosine_similarity(vectors, vectors)
print(similarity_matrix)

┌─────────────────────────────────────────────────────────────┐
│                    相似度矩阵示例                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│           机器学习    深度学习    天气很好                   │
│  机器学习 [  1.00      0.85       0.32  ]                   │
│  深度学习 [  0.85      1.00       0.28  ]                   │
│  天气很好 [  0.32      0.28       1.00  ]                   │
│                                                             │
│  相似度范围: [-1, 1]                                        │
│  - 1: 完全相同                                              │
│  - 0: 无关联                                                │
│  - -1: 完全相反                                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# 大量文本的批量处理
texts = [f"这是第 {i} 段文本" for i in range(1000)]

# 自动分批处理
vectors = embeddings.embed_documents(texts)

# 控制批处理大小
embeddings = OpenAIEmbeddings(
    model="text-embedding-3-small",
    chunk_size=100  # 每批处理 100 个
)

import asyncio

async def embed_async():
    # 异步嵌入
    vector = await embeddings.aembed_query("文本")
    
    # 异步批量
    vectors = await embeddings.aembed_documents(["文本1", "文本2"])
    
    return vectors

vectors = asyncio.run(embed_async())

from langchain.embeddings import CacheBackedEmbeddings
from langchain_community.storage import LocalFileStore

# 创建底层嵌入模型
underlying_embeddings = OpenAIEmbeddings()

# 创建本地缓存
store = LocalFileStore("./embedding_cache")

# 创建带缓存的嵌入模型
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    store,
    namespace=underlying_embeddings.model
)

# 首次调用会计算并缓存
vector1 = cached_embeddings.embed_query("测试文本")

# 再次调用相同文本会从缓存读取
vector2 = cached_embeddings.embed_query("测试文本")

from langchain_community.storage import RedisStore

store = RedisStore(redis_url="redis://localhost:6379")

cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    underlying_embeddings,
    store,
    namespace="openai-embeddings"
)

from langchain_core.embeddings import Embeddings
from typing import List

class CustomEmbeddings(Embeddings):
    """自定义嵌入模型"""
    
    def __init__(self, model_path: str):
        # 加载模型
        pass
    
    def embed_documents(self, texts: List[str]) -> List[List[float]]:
        """嵌入多个文档"""
        return [self._embed(text) for text in texts]
    
    def embed_query(self, text: str) -> List[float]:
        """嵌入查询"""
        return self._embed(text)
    
    def _embed(self, text: str) -> List[float]:
        """实际嵌入逻辑"""
        # 实现你的嵌入逻辑
        return [0.0] * 768  # 示例
    
    async def aembed_documents(self, texts: List[str]) -> List[List[float]]:
        """异步嵌入多个文档"""
        return self.embed_documents(texts)
    
    async def aembed_query(self, text: str) -> List[float]:
        """异步嵌入查询"""
        return self.embed_query(text)

from langchain_community.vectorstores import FAISS

# 文档库
documents = [
    "机器学习是人工智能的一个分支",
    "深度学习使用多层神经网络",
    "自然语言处理处理人类语言",
    "计算机视觉让机器理解图像",
    "今天天气很好，适合出行"
]

# 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = FAISS.from_texts(documents, embeddings)

# 语义搜索
query = "AI 相关技术"
results = vectorstore.similarity_search(query, k=3)

for doc in results:
    print(doc.page_content)

from sklearn.cluster import KMeans
import numpy as np

# 文档
documents = [
    "机器学习算法",
    "深度学习网络",
    "自然语言处理",
    "计算机视觉",
    "天气预报",
    "美食推荐"
]

# 获取嵌入向量
vectors = embeddings.embed_documents(documents)
vectors = np.array(vectors)

# K-means 聚类
kmeans = KMeans(n_clusters=2, random_state=42)
labels = kmeans.fit_predict(vectors)

# 输出聚类结果
for i, (doc, label) in enumerate(zip(documents, labels)):
    print(f"Cluster {label}: {doc}")

def deduplicate_texts(texts: list, threshold: float = 0.95) -> list:
    """基于语义相似度去重"""
    vectors = embeddings.embed_documents(texts)
    
    unique_texts = []
    unique_vectors = []
    
    for text, vector in zip(texts, vectors):
        is_duplicate = False
        
        for existing_vector in unique_vectors:
            similarity = cosine_similarity(vector, existing_vector)
            if similarity > threshold:
                is_duplicate = True
                break
        
        if not is_duplicate:
            unique_texts.append(text)
            unique_vectors.append(vector)
    
    return unique_texts

# 使用
texts = [
    "机器学习是AI的分支",
    "机器学习是人工智能的一个分支",
    "深度学习使用神经网络"
]

unique = deduplicate_texts(texts)
print(unique)

┌─────────────────────────────────────────────────────────────┐
│                    嵌入模型选择                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  云端模型：                                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ OpenAI text-embedding-3-small                       │   │
│  │ - 平衡性能和成本                                    │   │
│  │ - 1536 维，$0.02/1M tokens                         │   │
│  │ - 推荐日常使用                                      │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ OpenAI text-embedding-3-large                       │   │
│  │ - 最高质量                                          │   │
│  │ - 3072 维，$0.13/1M tokens                         │   │
│  │ - 推荐高质量需求                                    │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  本地模型：                                                  │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ all-MiniLM-L6-v2                                   │   │
│  │ - 快速，轻量                                        │   │
│  │ - 384 维                                           │   │
│  │ - 推荐资源受限场景                                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐   │
│  │ all-mpnet-base-v2                                  │   │
│  │ - 高质量本地模型                                    │   │
│  │ - 768 维                                           │   │
│  │ - 推荐隐私敏感场景                                  │   │
│  └─────────────────────────────────────────────────────┘   │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import re

def preprocess_text(text: str) -> str:
    """文本预处理"""
    # 去除多余空白
    text = re.sub(r'\s+', ' ', text)
    # 去除特殊字符
    text = re.sub(r'[^\w\s\u4e00-\u9fff]', '', text)
    return text.strip()

texts = [preprocess_text(t) for t in raw_texts]
vectors = embeddings.embed_documents(texts)

# 大量文本分批处理
def batch_embed(texts: list, batch_size: int = 100):
    """分批嵌入"""
    all_vectors = []
    
    for i in range(0, len(texts), batch_size):
        batch = texts[i:i + batch_size]
        vectors = embeddings.embed_documents(batch)
        all_vectors.extend(vectors)
    
    return all_vectors

# 对于不变的文本，使用缓存
cached_embeddings = CacheBackedEmbeddings.from_bytes_store(
    OpenAIEmbeddings(),
    LocalFileStore("./cache"),
    namespace="my-app"
)