嵌入模型 API

from langchain_core.embeddings import Embeddings

class Embeddings(ABC):
    """嵌入模型基类"""

    @abstractmethod
    def embed_documents(
        self,
        texts: List[str],
    ) -> List[List[float]]:
        """
        嵌入文档列表

        Args:
            texts: 文本列表

        Returns:
            嵌入向量列表（每个向量是浮点数列表）
        """

    @abstractmethod
    def embed_query(self, text: str) -> List[float]:
        """
        嵌入查询文本

        Args:
            text: 查询文本

        Returns:
            嵌入向量
        """

    async def aembed_documents(
        self,
        texts: List[str],
    ) -> List[List[float]]:
        """异步嵌入文档列表"""
        return await asyncio.to_thread(self.embed_documents, texts)

    async def aembed_query(self, text: str) -> List[float]:
        """异步嵌入查询文本"""
        return await asyncio.to_thread(self.embed_query, text)

    def embed_documents_asynchronously(
        self,
        texts: List[str],
        batch_size: int = 100,
    ) -> List[List[float]]:
        """
        批量异步嵌入文档

        Args:
            texts: 文本列表
            batch_size: 批处理大小

        Returns:
            嵌入向量列表
        """

from langchain_openai import OpenAIEmbeddings

class OpenAIEmbeddings(BaseModel, Embeddings):
    """OpenAI 嵌入模型"""

    def __init__(
        self,
        model: str = "text-embedding-3-small",
        embedding_ctx_length: Optional[int] = None,
        chunk_size: int = 1000,
        max_retries: int = 6,
        request_timeout: Optional[Union[float, Tuple[float, float]]] = None,
        openai_api_key: Optional[str] = None,
        openai_organization: Optional[str] = None,
        openai_proxy: Optional[str] = None,
        openai_api_base: Optional[str] = None,
        dimensions: Optional[int] = None,
    ):
        """
        初始化 OpenAI 嵌入

        Args:
            model: 模型名称
                - text-embedding-3-small: 1536 维
                - text-embedding-3-large: 3072 维
                - text-embedding-ada-002: 1536 维
            embedding_ctx_length: 最大上下文长度
            chunk_size: 批处理大小
            max_retries: 最大重试次数
            request_timeout: 请求超时
            openai_api_key: API 密钥
            openai_organization: 组织 ID
            openai_proxy: 代理地址
            openai_api_base: API 基础 URL
            dimensions: 输出维度（仅 text-embedding-3 支持）
        """

from langchain_huggingface import HuggingFaceEmbeddings

class HuggingFaceEmbeddings(BaseModel, Embeddings):
    """Hugging Face 嵌入模型"""

    def __init__(
        self,
        model_name: str = "sentence-transformers/all-mpnet-base-v2",
        cache_folder: Optional[str] = None,
        model_kwargs: Optional[Dict[str, Any]] = None,
        encode_kwargs: Optional[Dict[str, Any]] = None,
        multi_process: bool = False,
        show_progress: bool = True,
    ):
        """
        初始化 HuggingFace 嵌入

        Args:
            model_name: 模型名称
                - sentence-transformers/all-mpnet-base-v2
                - sentence-transformers/all-MiniLM-L6-v2
                - sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2
            cache_folder: 模型缓存目录
            model_kwargs: 传递给模型的参数
                - device: "cuda" 或 "cpu"
            encode_kwargs: 编码参数
                - batch_size: 批处理大小
                - normalize_embeddings: 是否归一化
            multi_process: 是否使用多进程
            show_progress: 是否显示进度
        """

from langchain_cohere import CohereEmbeddings

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

# 可用模型:
# - embed-english-v3.0: 英文
# - embed-multilingual-v3.0: 多语言
# - embed-english-light-v3.0: 轻量英文
# - embed-multilingual-light-v3.0: 轻量多语言

from langchain_aws import BedrockEmbeddings

embeddings = BedrockEmbeddings(
    model_id="amazon.titan-embed-text-v1",
    region_name="us-east-1"
)

# 可用模型:
# - amazon.titan-embed-text-v1
# - amazon.titan-embed-image-v1
# - cohere.embed-english-v3 (通过 Bedrock)

from langchain_google_genai import GoogleGenerativeAIEmbeddings

embeddings = GoogleGenerativeAIEmbeddings(
    model="models/embedding-001",
    google_api_key="your-api-key"
)

from langchain_community.embeddings import JinaEmbeddings

# 使用云端 API
embeddings = JinaEmbeddings(
    jina_api_key="your-api-key",
    model_name="jina-embeddings-v2"  # 或 jina-embeddings-v2-base
)

# 使用本地模型
embeddings = JinaEmbeddings(
    model_name="jina-embeddings-v2-base",
    # 自动下载并使用本地模型
)

from langchain_community.embeddings import SpacyEmbeddings

embeddings = SpacyEmbeddings(
    model_name="en_core_web_md"  # 需要先下载: python -m spacy download en_core_web_md
)

# 可用模型:
# - en_core_web_md: 中等英文模型
# - en_core_web_lg: 大型英文模型
# - zh_core_web_md: 中等中文模型

from langchain_community.embeddings import HuggingFaceInstructEmbeddings

embeddings = HuggingFaceInstructEmbeddings(
    model_name="hkunlp/instructor-large",
    model_kwargs={"device": "cuda"},
    encode_kwargs={
        "normalize_embeddings": True
    }
)

# 带指令嵌入
query_vector = embeddings.embed_query(
    "Represent the sentence for retrieving relevant documents:"
    "What is Python?"
)
doc_vector = embeddings.embed_documents(
    ["Python is a programming language"]
)[0]

from langchain_community.embeddings import LlamaCppEmbeddings

embeddings = LlamaCppEmbeddings(
    model_path="path/to/model.gguf",
    n_ctx=2048,
    n_threads=4
)

query_vector = embeddings.embed_query("查询文本")

from langchain_community.embeddings import FakeEmbeddings

embeddings = FakeEmbeddings(size=1536)

# 返回固定大小的随机向量
query_vector = embeddings.embed_query("任意文本")