Implementation:Langchain ai Langgraph StoreEmbedding

Knowledge Sources	Langchain_ai_Langgraph
Domains	Store, Embeddings
Last Updated	2026-02-11 16:00 GMT

Overview

The store embedding module provides utilities for wrapping arbitrary embedding functions (both sync and async) into LangChain's `Embeddings` interface, along with path-based text extraction for structured data.

Description

This module bridges the gap between custom embedding functions and LangChain's standardized `Embeddings` interface. The primary entry point is `ensure_embeddings()`, which accepts multiple input types -- an existing `Embeddings` instance, a synchronous callable, an asynchronous callable, or a provider string (e.g., `"openai:text-embedding-3-small"`) -- and returns a conformant `Embeddings` object. This makes it straightforward to plug any embedding function into LangGraph's store layer for semantic search.

The `EmbeddingsLambda` class serves as the concrete wrapper. When initialized with a synchronous function, it supports both `embed_documents`/`embed_query` for sync usage and falls back to sync for async calls. When initialized with an async function, only async operations (`aembed_documents`/`aembed_query`) are natively supported; calling sync methods raises a `ValueError` with guidance to use the async API instead.

The module also provides `get_text_at_path()` and `tokenize_path()` for extracting text from nested data structures using path expressions. These support dot-separated field access, array indexing (`[0]`, `[*]`, `[-1]`), wildcards (`*`), and multi-field selection (`{field1,field2}`). This enables the store to embed only specific fields from JSON documents rather than the entire structure, saving embedding tokens and improving search relevance.

Usage

Use `ensure_embeddings()` when configuring a store's embedding layer. It is the recommended way to normalize diverse embedding sources into a single interface. Use `get_text_at_path()` when you need to extract specific fields from stored JSON objects for embedding, as specified by the `fields` parameter in the store's index configuration.

Code Reference

Source Location

• Repository: Langchain_ai_Langgraph
• File: libs/checkpoint/langgraph/store/base/embed.py

Signature

EmbeddingsFunc = Callable[[Sequence[str]], list[list[float]]]

AEmbeddingsFunc = Callable[[Sequence[str]], Awaitable[list[list[float]]]]

def ensure_embeddings(
    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str | None,
) -> Embeddings: ...

class EmbeddingsLambda(Embeddings):
    def __init__(self, func: EmbeddingsFunc | AEmbeddingsFunc) -> None: ...
    def embed_documents(self, texts: list[str]) -> list[list[float]]: ...
    def embed_query(self, text: str) -> list[float]: ...
    async def aembed_documents(self, texts: list[str]) -> list[list[float]]: ...
    async def aembed_query(self, text: str) -> list[float]: ...

def get_text_at_path(obj: Any, path: str | list[str]) -> list[str]: ...

def tokenize_path(path: str) -> list[str]: ...

Import

from langgraph.store.base.embed import ensure_embeddings

I/O Contract

Function/Method	Input	Output	Description
`ensure_embeddings`	EmbeddingsFunc | AEmbeddingsFunc | str | None`	`Embeddings`	Normalize any embedding source into LangChain's `Embeddings` interface
`EmbeddingsLambda.embed_documents`	`texts: list[str]`	`list[list[float]]`	Embed a list of texts into vectors (sync)
`EmbeddingsLambda.embed_query`	`text: str`	`list[float]`	Embed a single text into a vector (sync)
`EmbeddingsLambda.aembed_documents`	`texts: list[str]`	`list[list[float]]`	Embed a list of texts into vectors (async)
`EmbeddingsLambda.aembed_query`	`text: str`	`list[float]`	Embed a single text into a vector (async)
`get_text_at_path`	list[str]`	`list[str]`	Extract text values from a nested object using a path expression
`tokenize_path`	`path: str`	`list[str]`	Parse a path expression into token components

Type Alias	Definition	Description
`EmbeddingsFunc`	`Callable[[Sequence[str]], list[list[float]]]`	Synchronous embedding function signature
`AEmbeddingsFunc`	`Callable[[Sequence[str]], Awaitable[list[list[float]]]]`	Asynchronous embedding function signature

Usage Examples

from langgraph.store.base.embed import ensure_embeddings, get_text_at_path

# Wrap a synchronous embedding function
def my_embed_fn(texts):
    return [[0.1, 0.2, 0.3] for _ in texts]

embeddings = ensure_embeddings(my_embed_fn)
result = embeddings.embed_query("hello")  # Returns [0.1, 0.2, 0.3]

# Wrap an asynchronous embedding function
async def my_async_fn(texts):
    return [[0.4, 0.5, 0.6] for _ in texts]

embeddings = ensure_embeddings(my_async_fn)
result = await embeddings.aembed_query("hello")  # Returns [0.4, 0.5, 0.6]

# Initialize embeddings using a provider string (requires langchain>=0.3.9)
embeddings = ensure_embeddings("openai:text-embedding-3-small")

# Extract text from nested data for embedding
data = {"user": {"name": "Alice", "bio": "Engineer"}}
texts = get_text_at_path(data, "user.name")  # Returns ["Alice"]

# Multi-field extraction
texts = get_text_at_path(data, "user.{name,bio}")  # Returns ["Alice", "Engineer"]

# Array indexing
data = {"items": [{"title": "A"}, {"title": "B"}]}
texts = get_text_at_path(data, "items.[*].title")  # Returns ["A", "B"]

Related Pages

• Principle:Langchain_ai_Langgraph_Store_Embedding_Support