跳到主要内容
版本：V1.1.0

嵌入函数概览

在 seekdb 中，嵌入函数（Embedding Functions，简称 EF） 用于将文本等输入转换为向量，以支持语义搜索（向量相似性检索）等能力。seekdb 提供了对多种主流平台/模型的嵌入函数封装，您可以直接使用内置实现，也可以按接口规范实现自定义嵌入函数。

工作方式​

您可以在创建或获取 Collection 时通过 embedding_function 参数绑定 EF。绑定后，Collection 的写入与查询会按以下规则自动生成向量：

  • 写入：调用 add / update / upsert 等接口时，如果传入的是 documents（而不是 embeddings），则使用 EF 将文本转换为向量并写入。
  • 查询：调用 query 接口时，如果传入的是 query_texts（而不是 query_embeddings），则使用 EF 将查询文本转换为查询向量并执行检索。

支持的平台​

当前 seekdb 已支持以下平台的嵌入函数：

典型用法（与 Collection 绑定）​

下面示例展示了 EF 与 Collection 的典型绑定方式。其中 ef 可以替换为任一平台的嵌入函数实例。

import pyseekdb
# 以某个已实现的 EF 为例
from pyseekdb.utils.embedding_functions import SomePlatformEmbeddingFunction

# 1. 初始化嵌入函数
ef = SomePlatformEmbeddingFunction(api_key="your-api-key")

# 2. 在创建集合时绑定该函数
db = pyseekdb.Client(path="./seekdb.db")
collection = db.create_collection(
    name="my_collection", 
    embedding_function=ef
)

# 3. 后续操作将自动处理向量化
# 您只需提供 documents，seekdb 会调用 ef 自动生成向量并存储
collection.add(
    ids=["1", "2"], 
    documents=["Hello world", "How are you?"]
)

# 查询时同样只需提供查询文本
results = collection.query("How are you?", n_results=1)

默认嵌入函数：DefaultEmbeddingFunction（all-MiniLM-L6-v2）​

如果在创建 Collection 时未指定 embedding_function，seekdb 默认使用 DefaultEmbeddingFunction。

默认模型为 Sentence Transformers 的 all-MiniLM-L6-v2（维度 384）。该模型在本地运行，首次使用可能会自动下载模型文件。

import pyseekdb

db = pyseekdb.Client(path="./seekdb.db")

# 未指定 embedding_function 时，默认使用 DefaultEmbeddingFunction
collection = db.create_collection(name="default_collection")

有关默认嵌入函数的更多信息，请参考 默认嵌入式函数。

自定义嵌入函数​

如果内置嵌入函数无法满足需求，你可以实现 EmbeddingFunction 协议创建自定义 EF。自定义 EF 需要满足以下约束：

  • 实现 __call__ 方法：输入为 str | List[str]，返回为 List[List[float]] -（推荐）实现 dimension 属性：返回向量维度（int），便于创建 collection 时进行一致性校验

下面是一个最小示例：

from typing import List, Union
from pyseekdb import EmbeddingFunction

Documents = Union[str, List[str]]
Embeddings = List[List[float]]

@register_embedding_function
class MyCustomEmbeddingFunction(EmbeddingFunction[Documents]):
    def __init__(self, model_name: str = "my-model"):
        self.model_name = model_name

    def __call__(self, input: list[str] | str) -> list[list[float]]:
        # 在这里实现您的向量化逻辑（例如调用内部模型服务）
        return [[0.1, 0.2, 0.3] for _ in (input if isinstance(input, list) else [input])]

    @property
    def dimension(self) -> int:
        return 3  # 替换为该模型的维度

    @staticmethod
    def name() -> str:
        return "my_custom_embedding"

    def get_config(self) -> Dict[str, Any]:
        return {"model_name": self.model_name}

    @staticmethod
    def build_from_config(config: Dict[str, Any]) -> "MyCustomEmbeddingFunction":
        return MyCustomEmbeddingFunction(model_name=config.get("model_name", "my-model"))

有关自定义嵌入函数的更多信息，请参考：

Contents

编辑此页