Embedders

An embedder turns text into a vector, so memory and vector stores can find text by meaning rather than exact words. Memory picks one for you automatically; choose one yourself when you want a specific provider, or write your own for a provider the framework doesn’t ship.

The framework ships four:

from loomflow import HashEmbedder
from loomflow.memory.embedder import CohereEmbedder, OpenAIEmbedder, VoyageEmbedder
 
embedder = OpenAIEmbedder()                              # text-embedding-3-small (default)
embedder = OpenAIEmbedder("text-embedding-3-large")
embedder = VoyageEmbedder("voyage-3")
embedder = CohereEmbedder(api_key="...")
embedder = HashEmbedder()                                # zero-key, deterministic

Default selection

When you don’t pass embedder= to the memory resolver, it picks one in this order:

1. The LOOMFLOW_EMBEDDER environment variable, if set (hash, openai, openai-large, voyage or cohere). Set LOOMFLOW_EMBEDDER=hash when your chat model isn’t OpenAI’s and you don’t want memory recall calling the OpenAI embeddings API.
2. OpenAIEmbedder("text-embedding-3-small") if OPENAI_API_KEY is set.
3. HashEmbedder() otherwise.

A fresh pip install loomflow works without any keys; adding OPENAI_API_KEY switches the default to a real embedder. Vector stores don’t auto-pick: embedder is a required argument there.

The Embedder protocol

Any class with two properties and two methods satisfies the protocol, no inheritance:

from typing import Protocol, runtime_checkable
 
@runtime_checkable
class Embedder(Protocol):
    name: str                      # model name for telemetry / logs
    dimensions: int                # vector size
 
    async def embed(self, text: str) -> list[float]: ...
    async def embed_batch(self, texts: list[str]) -> list[list[float]]: ...

embed_batch is the hot path: memory backends and vector stores call it when adding many chunks. Real implementations should batch the network call; the slow fallback is [await self.embed(t) for t in texts].

Custom embedder

class CustomEmbedder:
    name: str = "custom-v1"
    dimensions: int = 768
 
    def __init__(self, api_key: str) -> None:
        self._client = ...        # your client
 
    async def embed(self, text: str) -> list[float]:
        return await self._client.embed(text)
 
    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
        return await self._client.embed_batch(texts)

That’s it. Pass an instance as embedder= to any memory backend or vector store.

See Custom embedder recipe for a full Cohere-via-third-party-SDK example.

Why batching matters

memory.remember(episode) embeds one episode at a time. Indexing a corpus is different: store.add(chunks) hands every chunk to embed_batch at once, and a batched API sends many texts per network round trip instead of one request per chunk. For thousands of chunks that is the difference between seconds and many minutes.

Implement embed_batch properly even if it’s just sugar over embed; the framework expects it.

Pinning the embedder

The embedder used to write vectors must match the embedder used to search them. Mismatched embedders silently produce near-zero similarity (vectors of different sizes score 0.0). The usual way to hit this is relying on the auto-pick across environments:

from loomflow import Agent
 
# Run 1: OPENAI_API_KEY is set, so the resolver picks OpenAIEmbedder.
agent = Agent("...", model="claude-opus-4-7", memory="sqlite:./bot.db")
 
# Run 2, same file, but the key is missing: the resolver picks HashEmbedder,
# and recall over the episodes from run 1 returns near-zero scores.
agent = Agent("...", model="claude-opus-4-7", memory="sqlite:./bot.db")

Pin the embedder explicitly to avoid this, either in the memory spec or with LOOMFLOW_EMBEDDER:

from loomflow import Agent
 
agent = Agent(
    "...",
    model="claude-opus-4-7",
    memory={"backend": "sqlite", "path": "./bot.db", "embedder": "openai"},
)

For a persisted vector store, pass the same embedder every time you reopen it.