Custom embedder

You’ll plug your own embedding model into agent memory, here a local sentence-transformers model that needs no API key. Use this pattern when the built-in embedders don’t fit: loomflow ships OpenAIEmbedder, VoyageEmbedder and CohereEmbedder (plus HashEmbedder for tests) in loomflow.memory, so you only need a custom one for other providers, self-hosted models, or offline use.

Any class with name, dimensions, an async embed(text) and an async embed_batch(texts) satisfies the Embedder protocol. No inheritance required.


import anyio
from loomflow import Agent
from loomflow.memory.vector import VectorMemory
 
class SentenceTransformerEmbedder:
    """Local embeddings with sentence-transformers; no network calls."""
 
    def __init__(self, model_name: str = "all-MiniLM-L6-v2") -> None:
        from sentence_transformers import SentenceTransformer
        self._model = SentenceTransformer(model_name)
        self.name = model_name
        self.dimensions = self._model.get_sentence_embedding_dimension()
 
    async def embed(self, text: str) -> list[float]:
        return (await self.embed_batch([text]))[0]
 
    async def embed_batch(self, texts: list[str]) -> list[list[float]]:
        # encode() is CPU-bound and blocking: run it off the event loop.
        vectors = await anyio.to_thread.run_sync(self._model.encode, texts)
        return [vector.tolist() for vector in vectors]
 
 
memory = VectorMemory(embedder=SentenceTransformerEmbedder())
agent = Agent("...", model="claude-opus-4-7", memory=memory)

Keep embed and embed_batch async, and move blocking work (local inference, a synchronous SDK) to a worker thread as above so it doesn’t stall other runs. The framework only uses the protocol surface, so the same shape works for any provider.