Skip to content

dspy.Embedder

dspy.Embedder(model: str | Callable, batch_size: int = 200, caching: bool = True, **kwargs: dict[str, Any])

DSPy 嵌入类。

用于计算文本输入嵌入的类。该类为以下两者提供统一接口：

  1. 通过litellm集成托管嵌入模型（例如OpenAI的text-embedding-3-small）
  2. 您提供的自定义嵌入函数

对于托管模型，只需将模型名称作为字符串传递（例如，"openai/text-embedding-3-small"）。该类将使用litellm来处理API调用和缓存。

对于自定义嵌入模型，传递一个可调用函数，该函数： - 接收字符串列表作为输入。 - 返回嵌入，可以是： - 一个二维numpy数组，包含float32值 - 一个二维列表，包含float32值 - 每一行应代表一个嵌入向量

参数:

名称 类型 描述 默认值
model str | Callable

要使用的嵌入模型。这可以是一个字符串（表示托管的嵌入模型名称，必须是litellm支持的嵌入模型）或一个表示自定义嵌入模型的可调用对象。

 必填
batch_size int

处理输入时的默认批处理大小。默认为 200。

 200
caching bool

在使用托管模型时是否缓存嵌入响应。默认为 True。

 True
**kwargs dict[str, Any]

传递给嵌入模型的额外默认关键字参数。

 {}

示例:

示例1：使用托管模型。

import dspy

embedder = dspy.Embedder("openai/text-embedding-3-small", batch_size=100)
embeddings = embedder(["hello", "world"])

assert embeddings.shape == (2, 1536)

示例2：使用任何本地嵌入模型，例如来自 https://huggingface.co/models?library=sentence-transformers。

# pip install sentence_transformers
import dspy
from sentence_transformers import SentenceTransformer

# Load an extremely efficient local model for retrieval
model = SentenceTransformer("sentence-transformers/static-retrieval-mrl-en-v1", device="cpu")

embedder = dspy.Embedder(model.encode)
embeddings = embedder(["hello", "world"], batch_size=1)

assert embeddings.shape == (2, 1024)

示例3：使用自定义函数。

import dspy
import numpy as np

def my_embedder(texts):
    return np.random.rand(len(texts), 10)

embedder = dspy.Embedder(my_embedder)
embeddings = embedder(["hello", "world"], batch_size=1)

assert embeddings.shape == (2, 10)
Source code in dspy/clients/embedding.py 
79
80
81
82
83
def __init__(self, model: str | Callable, batch_size: int = 200, caching: bool = True, **kwargs: dict[str, Any]):
    self.model = model
    self.batch_size = batch_size
    self.caching = caching
    self.default_kwargs = kwargs

函数

__call__(inputs: str | list[str], batch_size: int | None = None, caching: bool | None = None, **kwargs: dict[str, Any]) -> np.ndarray

计算给定输入的嵌入向量。

参数:

名称 类型 描述 默认值
inputs str | list[str]

需要计算嵌入的输入，可以是单个字符串或字符串列表。

 必填
batch_size int

处理输入的批次大小。如果为None，则默认为初始化时设置的batch_size。

 None
caching bool

是否在使用托管模型时缓存嵌入响应。如果为None，则默认为初始化时的缓存设置。

 None
kwargs dict[str, Any]

传递给嵌入模型的额外关键字参数。这些将覆盖初始化时提供的默认参数。

 {}

返回：

类型 描述
ndarray

numpy.ndarray: 如果输入是单个字符串，返回一个表示嵌入的一维numpy数组。
ndarray

如果输入是一个字符串列表，则返回一个二维numpy嵌入数组，每行一个嵌入。
Source code in dspy/clients/embedding.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
def __call__(self, inputs: str | list[str], batch_size: int | None = None, caching: bool | None = None, **kwargs: dict[str, Any]) -> np.ndarray:
    """Compute embeddings for the given inputs.

    Args:
        inputs: The inputs to compute embeddings for, can be a single string or a list of strings.
        batch_size (int, optional): The batch size for processing inputs. If None, defaults to the batch_size set
            during initialization.
        caching (bool, optional): Whether to cache the embedding response when using a hosted model. If None,
            defaults to the caching setting from initialization.
        kwargs: Additional keyword arguments to pass to the embedding model. These will override the default
            kwargs provided during initialization.

    Returns:
        numpy.ndarray: If the input is a single string, returns a 1D numpy array representing the embedding.
        If the input is a list of strings, returns a 2D numpy array of embeddings, one embedding per row.
    """
    input_batches, caching, kwargs, is_single_input = self._preprocess(inputs, batch_size, caching, **kwargs)

    compute_embeddings = _cached_compute_embeddings if caching else _compute_embeddings

    embeddings_list = []

    for batch in input_batches:
        embeddings_list.extend(compute_embeddings(self.model, batch, caching=caching, **kwargs))
    return self._postprocess(embeddings_list, is_single_input)

acall(inputs, batch_size=None, caching=None, **kwargs) async

Source code in dspy/clients/embedding.py 
139
140
141
142
143
144
145
146
147
async def acall(self, inputs, batch_size=None, caching=None, **kwargs):
    input_batches, caching, kwargs, is_single_input = self._preprocess(inputs, batch_size, caching, **kwargs)

    embeddings_list = []
    acompute_embeddings = _cached_acompute_embeddings if caching else _acompute_embeddings

    for batch in input_batches:
        embeddings_list.extend(await acompute_embeddings(self.model, batch, caching=caching, **kwargs))
    return self._postprocess(embeddings_list, is_single_input)

:::

优云智算