Providers

EndpointProvider

chatbot_plugin_sdk.providers.endpoint.EndpointProvider

EndpointProvider(
    url: str,
    response_key: str = "dense",
    api_key: str | None = None,
    dimension: int | None = None,
    timeout: float = 60.0,
    rate_limit: "RateLimitStrategy | None" = None,
    retries: int = 3,
    retry_delay: float = 2.0,
)

HTTP embedding provider。適用於外部 API（如 Google AI Studio） 與內部 sidecar microservice（如自架 fastembed HTTP service）， 兩者呼叫方式相同，差別只在 URL。

Parameters:

Name	Type	Description	Default
url	str	embedding service 的 base URL（如 "http://localhost:8080"）。	required
response_key	str	API response 中存放向量的 key（dense 用 "dense"，sparse 用 "sparse"）。	'dense'
api_key	str | None	若 service 需要 Bearer token，在此傳入。	None
dimension	int | None	dense 向量的維度。使用 response_key="dense" 時必填； response_key="sparse" 時可省略（sparse 無固定維度）。	None
timeout	float	HTTP 請求 timeout（秒），預設 60。	60.0
rate_limit	'RateLimitStrategy | None'	選填的 rate limiting 策略（如 SlidingWindowStrategy）。 使用外部 API（如 Google AI Studio）時建議設定； 內部 service 或自架 model 可省略（傳 None）。	None

Usage::

from chatbot_plugin_sdk import EndpointProvider
from chatbot_plugin_sdk.rate_limit import SlidingWindowStrategy

# 搭配 Google AI Studio（有 rpm/tpm/rpd 限制）
dense = EndpointProvider(
    url="https://generativelanguage.googleapis.com/...",
    dimension=768,
    api_key="AIza...",
    rate_limit=SlidingWindowStrategy(rpm=10, tpm=40_000, rpd=1_500),
)

# 內部 sidecar（無限流）
sparse = EndpointProvider(url="http://embed:8080", response_key="sparse")

embed async

embed(texts: list[str]) -> list

送出 POST /embed 請求，回傳對應 response_key 的向量列表。

Request body: {"texts": ["text1", ...]} Expected response: {"dense": [[...], ...], "sparse": [{...}, ...]}

Retries on connection errors (e.g. serverless cold start) with exponential backoff. HTTP status errors are not retried.

LocalProvider

chatbot_plugin_sdk.providers.local.LocalProvider

LocalProvider(
    fn: Callable[[list[str]], list],
    dimension: int | None = None,
)

In-process embedding provider，接受 sync 或 async callable。

適用於在同一個 Python process 內直接呼叫 embedding function， 例如自行初始化的 fastembed 模型。 Sync callable 會被包進 asyncio executor 以避免 block event loop。

Parameters:

Name	Type	Description	Default
fn	Callable[[list[str]], list]	接受 list[str] 並回傳向量列表的 callable。 Dense: fn(texts) -> list[list[float]] Sparse: fn(texts) -> list[dict[str, float]]	required
dimension	int | None	dense 向量維度。用於 dense 場景時必填；sparse 可省略。	None

Usage::

from fastembed import TextEmbedding
model = TextEmbedding("BAAI/bge-small-en")

dense = LocalProvider(
    fn=lambda texts: [v.tolist() for v in model.embed(texts)],
    dimension=384,
)

embed async

embed(texts: list[str]) -> list

呼叫 fn(texts)，自動處理 sync/async 差異。

Protocols

chatbot_plugin_sdk.protocols.DenseEmbeddingProvider

Bases: Protocol

HTTP endpoint 或 in-process callable，輸出 dense 向量。

dimension 屬性供 ensure_ready() 在首次建表時決定 VECTOR(N) 的 N。

chatbot_plugin_sdk.protocols.SparseEmbeddingProvider

Bases: Protocol

HTTP endpoint 或 in-process callable，輸出 sparse 向量（token_id → weight）。

dimension 屬性為詞彙表大小（SPLADE / BERT: 30522），供 setup() 建立 SPARSEVEC(N) 欄位使用。