存储

langgraph.store.base ¶

用于持久化键值存储的基类和类型。

存储提供了跨线程和对话持久化的长期记忆。支持分层命名空间、键值存储和可选的向量搜索。

核心类型

BaseStore：具有同步/异步操作的存储接口
Item：带有元数据的已存储键值对
Op：获取/放置/搜索/列出操作

函数	描述
`ensure_embeddings`	确保嵌入函数符合 LangChain 的 Embeddings 接口。
`get_text_at_path`	使用路径表达式或预分词的路径从对象中提取文本。
`tokenize_path`	将路径分词为组件。

NamespacePath `module-attribute` ¶

NamespacePath = tuple[str | Literal['*'], ...]

表示命名空间路径的元组，可以包含通配符。

示例

("users",)  # Exact users namespace
("documents", "*")  # Any sub-namespace under documents
("cache", "*", "v1")  # Any cache category with v1 version

NamespaceMatchType `module-attribute` ¶

NamespaceMatchType = Literal['prefix', 'suffix']

指定如何匹配命名空间路径。

值

"prefix"：从命名空间开始匹配 "suffix"：从命名空间末尾匹配

Embeddings ¶

基类: ABC

嵌入模型的接口。

这是一个用于实现文本嵌入模型的接口。

文本嵌入模型用于将文本映射到一个向量（n维空间中的一个点）。

相似的文本通常会被映射到该空间中彼此接近的点。至于何为“相似”以及如何在该空间中衡量“距离”的具体细节，则取决于特定的嵌入模型。

这个抽象包含一个用于嵌入文档列表的方法和一个用于嵌入查询文本的方法。查询文本的嵌入预期是一个单一向量，而文档列表的嵌入预期是一个向量列表。

通常情况下，查询嵌入与文档嵌入是相同的，但该抽象允许独立处理它们。

除了同步方法，此接口还提供了这些方法的异步版本。

默认情况下，异步方法是使用同步方法实现的；但是，实现者可以选择使用异步原生实现来覆盖异步方法，以提高性能。

方法	描述
`embed_documents`	嵌入搜索文档。
`embed_query`	嵌入查询文本。
`aembed_documents`	异步嵌入搜索文档。
`aembed_query`	异步嵌入查询文本。

embed_documents `abstractmethod` ¶

embed_documents(texts: list[str]) -> list[list[float]]

嵌入搜索文档。

参数	描述
`texts`	要嵌入的文本列表。类型: `list[str]`

返回	描述
`list[list[float]]`	嵌入列表。

embed_query `abstractmethod` ¶

embed_query(text: str) -> list[float]

嵌入查询文本。

参数	描述
`text`	要嵌入的文本。类型： `str`

返回	描述
`list[float]`	嵌入。

aembed_documents `async` ¶

aembed_documents(texts: list[str]) -> list[list[float]]

异步嵌入搜索文档。

参数	描述
`texts`	要嵌入的文本列表。类型: `list[str]`

返回	描述
`list[list[float]]`	嵌入列表。

aembed_query `async` ¶

aembed_query(text: str) -> list[float]

异步嵌入查询文本。

参数	描述
`text`	要嵌入的文本。类型： `str`

返回	描述
`list[float]`	嵌入。

NotProvided ¶

哨兵单例。

Item ¶

表示带有元数据的已存储项目。

参数	描述
`value`	以字典形式存储的数据。键是可过滤的。类型： `dict[str, Any]`
`key`	命名空间内的唯一标识符。类型： `str`
`namespace`	定义此文档所在集合的分层路径。表示为字符串元组，允许嵌套分类。例如：`("documents", 'user123')` 类型： `tuple[str, ...]`
`created_at`	项目创建的时间戳。类型： `datetime`
`updated_at`	上次更新的时间戳。类型： `datetime`

SearchItem ¶

基类： Item

表示从搜索操作返回的带有附加元数据的项目。

方法	描述
`__init__`	初始化一个结果项。

init ¶

__init__(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    created_at: datetime,
    updated_at: datetime,
    score: float | None = None,
) -> None

初始化一个结果项。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`
`value`	存储的值。类型： `dict[str, Any]`
`created_at`	项目首次创建的时间。类型： `datetime`
`updated_at`	项目上次更新的时间。类型： `datetime`
`score`	如果来自排名操作，则为相关性/相似度分数。类型： `float \| None` 默认值： `None`

GetOp ¶

基类：NamedTuple

通过其命名空间和键检索特定项目的操作。

此操作允许使用存储项的完整路径（命名空间）和唯一标识符（键）组合进行精确检索。

示例

基本项目检索

GetOp(namespace=("users", "profiles"), key="user123")
GetOp(namespace=("cache", "embeddings"), key="doc456")

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

唯一标识项目位置的分层路径。

示例

("users",)  # Root level users namespace
("users", "profiles")  # Profiles within users namespace

key `instance-attribute` ¶

key: str

项目在其特定命名空间内的唯一标识符。

示例

"user123"  # For a user profile
"doc456"  # For a document

refresh_ttl `class-attribute` `instance-attribute` ¶

refresh_ttl: bool = True

是否刷新返回项目的 TTL。

如果原始项目未指定 TTL，或者您的适配器未启用 TTL 支持，则此参数将被忽略。

SearchOp ¶

基类：NamedTuple

在指定命名空间层次结构内搜索项目的操作。

此操作支持在给定命名空间前缀内进行结构化过滤和自然语言搜索。它通过 limit 和 offset 参数提供分页功能。

注意

自然语言搜索支持取决于您的存储实现。

示例

带筛选和分页的搜索

SearchOp(
    namespace_prefix=("documents",),
    filter={"type": "report", "status": "active"},
    limit=5,
    offset=10
)

自然语言搜索

SearchOp(
    namespace_prefix=("users", "content"),
    query="technical documentation about APIs",
    limit=20
)

namespace_prefix `instance-attribute` ¶

namespace_prefix: tuple[str, ...]

定义搜索范围的分层路径前缀。

示例

()  # Search entire store
("documents",)  # Search all documents
("users", "content")  # Search within user content

filter `class-attribute` `instance-attribute` ¶

filter: dict[str, Any] | None = None

用于根据精确匹配或比较运算符筛选结果的键值对。

筛选器支持精确匹配和基于运算符的比较。

支持的运算符

$eq: 等于（与直接值比较相同）
$ne: 不等于
$gt：大于
$gte：大于或等于
$lt：小于
$lte：小于或等于

示例

简单精确匹配

{"status": "active"}

比较运算符

{"score": {"$gt": 4.99}}  # Score greater than 4.99

多个条件

{
    "score": {"$gte": 3.0},
    "color": "red"
}

limit `class-attribute` `instance-attribute` ¶

limit: int = 10

在搜索结果中返回的最大项目数。

offset `class-attribute` `instance-attribute` ¶

offset: int = 0

用于分页的要跳过的匹配项目数。

query `class-attribute` `instance-attribute` ¶

query: str | None = None

用于语义搜索功能的自然语言搜索查询。

示例

“关于 REST API 的技术文档”
“2023年的机器学习论文”

refresh_ttl `class-attribute` `instance-attribute` ¶

refresh_ttl: bool = True

是否刷新返回项目的 TTL。

如果原始项目未指定 TTL，或者您的适配器未启用 TTL 支持，则此参数将被忽略。

MatchCondition ¶

基类：NamedTuple

表示在存储中匹配命名空间的模式。

该类将匹配类型（前缀或后缀）与可以包含通配符的命名空间路径模式相结合，以灵活地匹配不同的命名空间层次结构。

示例

前缀匹配

MatchCondition(match_type="prefix", path=("users", "profiles"))

带通配符的后缀匹配

MatchCondition(match_type="suffix", path=("cache", "*"))

简单后缀匹配

MatchCondition(match_type="suffix", path=("v1",))

match_type `instance-attribute` ¶

match_type: NamespaceMatchType

要执行的命名空间匹配类型。

path `instance-attribute` ¶

path: NamespacePath

可以包含通配符的命名空间路径模式。

ListNamespacesOp ¶

基类：NamedTuple

列出和过滤存储中命名空间的操作。

此操作允许探索数据组织、查找特定集合以及导航命名空间层次结构。

示例

列出 "documents" 路径下的所有命名空间

ListNamespacesOp(
    match_conditions=(MatchCondition(match_type="prefix", path=("documents",)),),
    max_depth=2
)

列出所有以 "v1" 结尾的命名空间

ListNamespacesOp(
    match_conditions=(MatchCondition(match_type="suffix", path=("v1",)),),
    limit=50
)

match_conditions `class-attribute` `instance-attribute` ¶

match_conditions: tuple[MatchCondition, ...] | None = None

用于过滤命名空间的可选条件。

示例

所有用户命名空间

(MatchCondition(match_type="prefix", path=("users",)),)

所有以 "docs" 开头并以 "draft" 结尾的命名空间

(
    MatchCondition(match_type="prefix", path=("docs",)),
    MatchCondition(match_type="suffix", path=("draft",))
)

max_depth `class-attribute` `instance-attribute` ¶

max_depth: int | None = None

要返回的命名空间层次结构的最大深度。

注意

比此级别更深的命名空间将被截断。

limit `class-attribute` `instance-attribute` ¶

limit: int = 100

要返回的最大命名空间数。

offset `class-attribute` `instance-attribute` ¶

offset: int = 0

用于分页的要跳过的命名空间数。

PutOp ¶

基类：NamedTuple

在存储中存储、更新或删除项目的操作。

此类表示一个修改存储内容的单一操作，无论是添加新项、更新现有项还是删除它们。

namespace `instance-attribute` ¶

namespace: tuple[str, ...]

标识项目位置的分层路径。

命名空间的作用类似于文件夹结构，用于组织项目。元组中的每个元素代表层次结构中的一个级别。

示例

根级文档

("documents",)

用户特定文档

("documents", "user123")

嵌套缓存结构

("cache", "embeddings", "v1")

key `instance-attribute` ¶

key: str

项目在其命名空间内的唯一标识符。

键在特定命名空间内必须是唯一的，以避免冲突。它与命名空间一起构成项目的完整路径。

示例

如果命名空间是 ("documents", "user123") 且键是 "report1"，则完整路径实际上是 "documents/user123/report1"

value `instance-attribute` ¶

value: dict[str, Any] | None

要存储的数据，或 None 表示标记该项目以供删除。

值必须是具有字符串键和可 JSON 序列化值的字典。将其设置为 None 表示应删除该项目。

示例

{ "field1": "字符串值", "field2": 123, "nested": {"can": "contain", "any": "可序列化数据"} }

index `class-attribute` `instance-attribute` ¶

index: Literal[False] | list[str] | None = None

控制如何为搜索操作索引项目的字段。

索引配置决定了如何通过搜索找到项目

None（默认值）：使用存储的默认索引配置（如果提供）
False：为此项目禁用索引
list[str]: 指定要为搜索索引哪些 json 路径字段

无论索引如何，该项目仍然可以通过直接的 get() 操作访问。索引后，可以使用自然语言查询通过向量相似性搜索来搜索字段（如果存储实现支持）。

路径语法

简单字段访问："field"
嵌套字段："parent.child.grandchild"
数组索引
- 特定索引："array[0]"
- 最后一个元素："array[-1]"
- 所有元素（每个元素独立）："array[*]"

示例

None - 使用存储默认值（整个项目）
list[str] - 要索引的字段列表

[
    "metadata.title",                    # Nested field access
    "context[*].content",                # Index content from all context as separate vectors
    "authors[0].name",                   # First author's name
    "revisions[-1].changes",             # Most recent revision's changes
    "sections[*].paragraphs[*].text",    # All text from all paragraphs in all sections
    "metadata.tags[*]",                  # All tags in metadata
]

ttl `class-attribute` `instance-attribute` ¶

ttl: float | None = None

控制项目的 TTL（生存时间），单位为分钟。

如果提供，并且您使用的存储支持此功能，则该项目将在上次访问后这么多分钟后过期。过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新。当 TTL 过期时，该项目将被安排以尽力而为的方式删除。默认为 None（无过期）。

InvalidNamespaceError ¶

基类： ValueError

提供的命名空间无效。

TTLConfig ¶

基类：TypedDict

存储中 TTL（生存时间）行为的配置。

refresh_on_read `instance-attribute` ¶

refresh_on_read: bool

在读取操作（GET 和 SEARCH）上刷新 TTL 的默认行为。

如果为 True，则默认情况下会在读取操作（get/search）时刷新 TTL。这可以通过显式设置 refresh_ttl 在每个操作中覆盖。如果未配置，则默认为 True。

default_ttl `instance-attribute` ¶

default_ttl: float | None

新项目的默认 TTL（生存时间），以分钟为单位。

如果提供，新项目将在最后一次访问后这么多分钟后过期。过期计时器在读取和写入操作时都会刷新。默认为 None（无过期）。

sweep_interval_minutes `instance-attribute` ¶

sweep_interval_minutes: int | None

TTL 清扫操作之间的间隔（以分钟为单位）。

如果提供，存储将根据 TTL 定期删除过期的项目。默认为 None（不进行清扫）。

IndexConfig ¶

基类：TypedDict

用于在存储中为语义搜索索引文档的配置。

如果未提供给存储，则存储将不支持向量搜索。在这种情况下，所有传递给 put() 和 aput() 操作的 index 参数都将被忽略。

dims `instance-attribute` ¶

dims: int

嵌入向量中的维度数。

常见的嵌入模型具有以下维度

openai:text-embedding-3-large: 3072
openai:text-embedding-3-small: 1536
openai:text-embedding-ada-002: 1536
cohere:embed-english-v3.0: 1024
cohere:embed-english-light-v3.0: 384
cohere:embed-multilingual-v3.0: 1024
cohere:embed-multilingual-light-v3.0: 384

embed `instance-attribute` ¶

embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str

用于从文本生成嵌入的可选函数。

可以通过三种方式指定

一个 LangChain Embeddings 实例
一个同步嵌入函数 (EmbeddingsFunc)
一个异步嵌入函数 (AEmbeddingsFunc)
提供商字符串（例如，"openai:text-embedding-3-small"）

示例

使用 LangChain 的初始化与 InMemoryStore

from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": init_embeddings("openai:text-embedding-3-small")
    }
)

使用自定义嵌入函数与 InMemoryStore

from openai import OpenAI
from langgraph.store.memory import InMemoryStore

client = OpenAI()

def embed_texts(texts: list[str]) -> list[list[float]]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [e.embedding for e in response.data]

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": embed_texts
    }
)

使用异步嵌入函数与 InMemoryStore

from openai import AsyncOpenAI
from langgraph.store.memory import InMemoryStore

client = AsyncOpenAI()

async def aembed_texts(texts: list[str]) -> list[list[float]]:
    response = await client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )
    return [e.embedding for e in response.data]

store = InMemoryStore(
    index={
        "dims": 1536,
        "embed": aembed_texts
    }
)

fields `instance-attribute` ¶

fields: list[str] | None

用于提取文本以生成嵌入的字段。

控制存储项目的哪些部分被嵌入以进行语义搜索。遵循 JSON 路径语法

["$"]: 将整个 JSON 对象嵌入为一个向量（默认）
["field1", "field2"]：嵌入特定的顶层字段
["parent.child"]：使用点表示法嵌入嵌套字段
["array[*].field"]：分别嵌入每个数组元素的字段

注意

您始终可以在存储项目时使用 put 或 aput 操作中的 index 参数来覆盖此行为。

示例

# Embed entire document (default)
fields=["$"]

# Embed specific fields
fields=["text", "summary"]

# Embed nested fields
fields=["metadata.title", "content.body"]

# Embed from arrays
fields=["messages[*].content"]  # Each message content separately
fields=["context[0].text"]      # First context item's text

注意

文档中缺少的字段将被跳过
数组表示法为每个元素创建单独的嵌入
支持复杂的嵌套路径（例如，"a.b[*].c.d"）

BaseStore ¶

基类: ABC

持久化键值存储的抽象基类。

存储实现了持久性和内存，可以在线程之间共享，并可限定于用户ID、助手ID或其他任意命名空间。某些实现可能通过可选的 index 配置支持语义搜索功能。

注意

语义搜索功能因实现而异，并且通常默认禁用。支持此功能的存储可以通过在创建时提供 index 配置来进行配置。如果没有此配置，语义搜索将被禁用，任何传递给存储操作的 index 参数都将无效。

同样，TTL（生存时间）支持默认是禁用的。子类必须显式设置 supports_ttl = True 来启用此功能。

方法	描述
`batch`	在单个批次中同步执行多个操作。
`abatch`	在单个批次中异步执行多个操作。
`get`	检索单个项目。
`search`	在命名空间前缀内搜索项目。
`put`	在存储中存储或更新一个项目。
`delete`	删除一个项目。
`list_namespaces`	列出并筛选存储中的命名空间。
`aget`	异步检索单个项目。
`asearch`	在命名空间前缀内异步搜索项目。
`aput`	异步存储或更新存储中的项目。
`adelete`	异步删除一个项目。
`alist_namespaces`	异步列出和过滤存储中的命名空间。

batch `abstractmethod` ¶

batch(ops: Iterable[Op]) -> list[Result]

在单个批次中同步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

abatch `abstractmethod` `async` ¶

abatch(ops: Iterable[Op]) -> list[Result]

在单个批次中异步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

get ¶

get(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的默认 `refresh_ttl` 设置。如果未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

search ¶

search(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果未指定 TTL，则忽略此参数。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = store.search(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed. Defaults to ["$"]
    }
)

# Search for semantically similar documents

results = store.search(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

put ¶

put(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

在存储中存储或更新一个项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

store.put(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=["memory"])

delete ¶

delete(namespace: tuple[str, ...], key: str) -> None

删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

list_namespaces ¶

list_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

列出并筛选存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

???+ example "示例"

Setting `max_depth=3`. Given the namespaces:

```python
# Example if you have the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
store.list_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
```

aget `async` ¶

aget(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

异步检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

asearch `async` ¶

asearch(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内异步搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的 `TTLConfig.refresh_default` 设置。如果未提供 `TTLConfig` 或未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = await store.asearch(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed
    }
)

# Search for semantically similar documents

results = await store.asearch(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

aput `async` ¶

aput(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

异步存储或更新存储中的项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

await store.aput(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

await store.aput(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段（如果存储配置为索引项目）

await store.aput(
    ("docs",),
    "report",
    {
        "memory": "Will likes ai",
        "context": [{"content": "..."}, {"content": "..."}]
    },
    index=["memory", "context[*].content"]
)

adelete `async` ¶

adelete(namespace: tuple[str, ...], key: str) -> None

异步删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

alist_namespaces `async` ¶

alist_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

异步列出和过滤存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断至此深度。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

示例

使用现有命名空间设置 max_depth=3

# Given the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")

await store.alist_namespaces(prefix=("a", "b"), max_depth=3)
# Returns: [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]

ensure_embeddings ¶

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

确保嵌入函数符合 LangChain 的 Embeddings 接口。

此函数包装任意嵌入函数，使其与 LangChain 的 Embeddings 接口兼容。它处理同步和异步函数。

参数	描述
`embed`	可以是现有的 Embeddings 实例，也可以是将文本转换为嵌入的函数。如果函数是异步的，它将用于同步和异步操作。类型： `Embeddings \| EmbeddingsFunc \| AEmbeddingsFunc \| str \| None`

返回	描述
`嵌入`	一个包装了所提供函数的 Embeddings 实例。

示例

包装一个同步嵌入函数

def my_embed_fn(texts):
    return [[0.1, 0.2] for _ in texts]

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

包装一个异步嵌入函数

async def my_async_fn(texts):
    return [[0.1, 0.2] for _ in texts]

embeddings = ensure_embeddings(my_async_fn)
result = await embeddings.aembed_query("hello")  # Returns [0.1, 0.2]

使用提供商字符串初始化嵌入

# Requires langchain>=0.3.9 and langgraph-checkpoint>=2.0.11
embeddings = ensure_embeddings("openai:text-embedding-3-small")
result = embeddings.embed_query("hello")

get_text_at_path ¶

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

使用路径表达式或预分词的路径从对象中提取文本。

参数	描述
`obj`	要从中提取文本的对象类型： `Any`
`path`	可以是路径字符串或预分词的路径列表。类型: `str \| list[str]`

处理的路径类型

简单路径: "field1.field2"
数组索引: "[0]", "[*]", "[-1]"
通配符: "*"
多字段选择: "{field1,field2}"
多字段中的嵌套路径: "{field1,nested.field2}"

tokenize_path ¶

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

将路径分词为组件。

处理的类型

简单路径: "field1.field2"
数组索引: "[0]", "[*]", "[-1]"
通配符: "*"
多字段选择: "{field1,field2}"

langgraph.store.postgres ¶

AsyncPostgresStore ¶

基类： AsyncBatchedBaseStore, BasePostgresStore[Conn]

异步的 Postgres 支持的存储，可选使用 pgvector 进行向量搜索。

示例

基本设置和用法

from langgraph.store.postgres import AsyncPostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

async with AsyncPostgresStore.from_conn_string(conn_string) as store:
    await store.setup()  # Run migrations. Done once

    # Store and retrieve data
    await store.aput(("users", "123"), "prefs", {"theme": "dark"})
    item = await store.aget(("users", "123"), "prefs")

使用 LangChain 嵌入进行向量搜索

from langchain.embeddings import init_embeddings
from langgraph.store.postgres import AsyncPostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

async with AsyncPostgresStore.from_conn_string(
    conn_string,
    index={
        "dims": 1536,
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "fields": ["text"]  # specify which fields to embed. Default is the whole serialized value
    }
) as store:
    await store.setup()  # Run migrations. Done once

    # Store documents
    await store.aput(("docs",), "doc1", {"text": "Python tutorial"})
    await store.aput(("docs",), "doc2", {"text": "TypeScript guide"})
    await store.aput(("docs",), "doc3", {"text": "Other guide"}, index=False)  # don't index

    # Search by similarity
    results = await store.asearch(("docs",), query="programming guides", limit=2)

使用连接池以获得更好性能

from langgraph.store.postgres import AsyncPostgresStore, PoolConfig

conn_string = "postgresql://user:pass@localhost:5432/dbname"

async with AsyncPostgresStore.from_conn_string(
    conn_string,
    pool_config=PoolConfig(
        min_size=5,
        max_size=20
    )
) as store:
    await store.setup()  # Run migrations. Done once
    # Use store with connection pooling...

警告

请确保： 1. 在首次使用前调用 setup() 来创建必要的表和索引 2. 有 pgvector 扩展可用以使用向量搜索 3. 使用 Python 3.10+ 以支持异步功能

注意

语义搜索默认是禁用的。您可以通过在创建存储时提供 index 配置来启用它。如果没有此配置，所有传递给 put 或 aput 的 index 参数都将无效。

注意

如果您提供了 TTL 配置，则必须显式调用 start_ttl_sweeper() 来启动移除过期项目的后台任务。当您完成存储的使用时，调用 stop_ttl_sweeper() 以正确清理资源。

方法	描述
`batch`	在单个批次中同步执行多个操作。
`get`	检索单个项目。
`search`	在命名空间前缀内搜索项目。
`put`	在存储中存储或更新一个项目。
`delete`	删除一个项目。
`list_namespaces`	列出并筛选存储中的命名空间。
`aget`	异步检索单个项目。
`asearch`	在命名空间前缀内异步搜索项目。
`aput`	异步存储或更新存储中的项目。
`adelete`	异步删除一个项目。
`alist_namespaces`	异步列出和过滤存储中的命名空间。
`abatch`	在单个批次中异步执行多个操作。
`from_conn_string`	从连接字符串创建一个新的 AsyncPostgresStore 实例。
`setup`	异步设置存储数据库。
`sweep_ttl`	根据 TTL 删除过期的存储项。
`start_ttl_sweeper`	根据 TTL 定期删除过期的存储项。
`stop_ttl_sweeper`	如果 TTL 清扫任务正在运行，则停止它。

batch ¶

batch(ops: Iterable[Op]) -> list[Result]

在单个批次中同步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

get ¶

get(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的默认 `refresh_ttl` 设置。如果未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

search ¶

search(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果未指定 TTL，则忽略此参数。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = store.search(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed. Defaults to ["$"]
    }
)

# Search for semantically similar documents

results = store.search(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

put ¶

put(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

在存储中存储或更新一个项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

store.put(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=["memory"])

delete ¶

delete(namespace: tuple[str, ...], key: str) -> None

删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

list_namespaces ¶

list_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

列出并筛选存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

???+ example "示例"

Setting `max_depth=3`. Given the namespaces:

```python
# Example if you have the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
store.list_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
```

aget `async` ¶

aget(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

异步检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

asearch `async` ¶

asearch(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内异步搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的 `TTLConfig.refresh_default` 设置。如果未提供 `TTLConfig` 或未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = await store.asearch(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed
    }
)

# Search for semantically similar documents

results = await store.asearch(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

aput `async` ¶

aput(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

异步存储或更新存储中的项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

await store.aput(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

await store.aput(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段（如果存储配置为索引项目）

await store.aput(
    ("docs",),
    "report",
    {
        "memory": "Will likes ai",
        "context": [{"content": "..."}, {"content": "..."}]
    },
    index=["memory", "context[*].content"]
)

adelete `async` ¶

adelete(namespace: tuple[str, ...], key: str) -> None

异步删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

alist_namespaces `async` ¶

alist_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

异步列出和过滤存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断至此深度。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

示例

使用现有命名空间设置 max_depth=3

# Given the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")

await store.alist_namespaces(prefix=("a", "b"), max_depth=3)
# Returns: [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]

abatch `async` ¶

abatch(ops: Iterable[Op]) -> list[Result]

在单个批次中异步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

from_conn_string `async` `classmethod` ¶

from_conn_string(
    conn_string: str,
    *,
    pipeline: bool = False,
    pool_config: PoolConfig | None = None,
    index: PostgresIndexConfig | None = None,
    ttl: TTLConfig | None = None,
) -> AsyncIterator[AsyncPostgresStore]

从连接字符串创建一个新的 AsyncPostgresStore 实例。

参数	描述
`conn_string`	Postgres 连接信息字符串。类型： `str`
`pipeline`	是否使用 AsyncPipeline（仅适用于单个连接）类型： `bool` 默认值： `False`
`pool_config`	连接池的配置。如果提供，将创建一个连接池并使用它，而不是单个连接。这将覆盖 `pipeline` 参数。类型： `PoolConfig \| None` 默认值： `None`
`index`	嵌入配置。类型： `PostgresIndexConfig \| None` 默认值： `None`

返回	描述
`AsyncPostgresStore`	一个新的 AsyncPostgresStore 实例。类型： `AsyncIterator[AsyncPostgresStore]`

setup `async` ¶

setup() -> None

异步设置存储数据库。

此方法在 Postgres 数据库中创建必要的表（如果它们尚不存在）并运行数据库迁移。用户在首次使用存储时必须直接调用它。

sweep_ttl `async` ¶

sweep_ttl() -> int

根据 TTL 删除过期的存储项。

返回	描述
`int`	已删除项目的数量。类型： `int`

start_ttl_sweeper `async` ¶

start_ttl_sweeper(sweep_interval_minutes: int | None = None) -> Task[None]

根据 TTL 定期删除过期的存储项。

返回	描述
`Task[None]`	可以被等待或取消的任务。

stop_ttl_sweeper `async` ¶

stop_ttl_sweeper(timeout: float | None = None) -> bool

如果 TTL 清扫任务正在运行，则停止它。

参数	描述
`timeout`	等待任务停止的最大时间（秒）。如果为 `None`，则无限期等待。类型： `float \| None` 默认值： `None`

返回	描述
`bool`	如果任务成功停止或未运行，则为 True；如果在任务停止前达到超时，则为 False。类型： `bool`

PoolConfig ¶

基类：TypedDict

PostgreSQL 连接的连接池设置。

控制连接生命周期和资源利用率： - 小型池（1-5）适用于低并发工作负载 - 较大的池处理并发请求，但消耗更多资源 - 设置 max_size 可防止负载下资源耗尽

min_size `instance-attribute` ¶

min_size: int

连接池中维护的最小连接数。默认为 1。

max_size `instance-attribute` ¶

max_size: int | None

连接池中允许的最大连接数。None 表示无限制。

kwargs `instance-attribute` ¶

kwargs: dict

传递给池中每个连接的附加连接参数。

自动设置的默认 kwargs：- autocommit: True - prepare_threshold: 0 - row_factory: dict_row

PostgresStore ¶

基类： BaseStore, BasePostgresStore[Conn]

由 Postgres 支持的存储，可选使用 pgvector 进行向量搜索。

示例

基本设置和用法

from langgraph.store.postgres import PostgresStore
from psycopg import Connection

conn_string = "postgresql://user:pass@localhost:5432/dbname"

# Using direct connection
with Connection.connect(conn_string) as conn:
    store = PostgresStore(conn)
    store.setup() # Run migrations. Done once

    # Store and retrieve data
    store.put(("users", "123"), "prefs", {"theme": "dark"})
    item = store.get(("users", "123"), "prefs")

或使用便捷的 from_conn_string 辅助函数

from langgraph.store.postgres import PostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

with PostgresStore.from_conn_string(conn_string) as store:
    store.setup()

    # Store and retrieve data
    store.put(("users", "123"), "prefs", {"theme": "dark"})
    item = store.get(("users", "123"), "prefs")

使用 LangChain 嵌入进行向量搜索

from langchain.embeddings import init_embeddings
from langgraph.store.postgres import PostgresStore

conn_string = "postgresql://user:pass@localhost:5432/dbname"

with PostgresStore.from_conn_string(
    conn_string,
    index={
        "dims": 1536,
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "fields": ["text"]  # specify which fields to embed. Default is the whole serialized value
    }
) as store:
    store.setup() # Do this once to run migrations

    # Store documents
    store.put(("docs",), "doc1", {"text": "Python tutorial"})
    store.put(("docs",), "doc2", {"text": "TypeScript guide"})
    store.put(("docs",), "doc2", {"text": "Other guide"}, index=False) # don't index

    # Search by similarity
    results = store.search(("docs",), query="programming guides", limit=2)

注意

语义搜索默认是禁用的。您可以通过在创建存储时提供 index 配置来启用它。如果没有此配置，所有传递给 put 或 aput 的 index 参数都将无效。

警告

请确保在首次使用前调用 setup() 来创建必要的表和索引。必须有 pgvector 扩展才能使用向量搜索。

注意

如果您提供了 TTL 配置，则必须显式调用 start_ttl_sweeper() 来启动移除过期项目的后台线程。当您完成存储的使用时，调用 stop_ttl_sweeper() 以正确清理资源。

方法	描述
`get`	检索单个项目。
`search`	在命名空间前缀内搜索项目。
`put`	在存储中存储或更新一个项目。
`delete`	删除一个项目。
`list_namespaces`	列出并筛选存储中的命名空间。
`aget`	异步检索单个项目。
`asearch`	在命名空间前缀内异步搜索项目。
`aput`	异步存储或更新存储中的项目。
`adelete`	异步删除一个项目。
`alist_namespaces`	异步列出和过滤存储中的命名空间。
`from_conn_string`	从连接字符串创建一个新的 PostgresStore 实例。
`sweep_ttl`	根据 TTL 删除过期的存储项。
`start_ttl_sweeper`	根据 TTL 定期删除过期的存储项。
`stop_ttl_sweeper`	如果 TTL 清扫线程正在运行，则停止它。
`__del__`	确保在对象被垃圾回收时停止 TTL 清扫线程。
`batch`	在单个批次中同步执行多个操作。
`abatch`	在单个批次中异步执行多个操作。
`setup`	设置存储数据库。

get ¶

get(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的默认 `refresh_ttl` 设置。如果未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

search ¶

search(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果未指定 TTL，则忽略此参数。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = store.search(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed. Defaults to ["$"]
    }
)

# Search for semantically similar documents

results = store.search(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

put ¶

put(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

在存储中存储或更新一个项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

store.put(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段

store.put(("docs",), "report", {"memory": "Will likes ai"}, index=["memory"])

delete ¶

delete(namespace: tuple[str, ...], key: str) -> None

删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

list_namespaces ¶

list_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

列出并筛选存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

???+ example "示例"

Setting `max_depth=3`. Given the namespaces:

```python
# Example if you have the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")
store.list_namespaces(prefix=("a", "b"), max_depth=3)
# [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]
```

aget `async` ¶

aget(
    namespace: tuple[str, ...], key: str, *, refresh_ttl: bool | None = None
) -> Item | None

异步检索单个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

返回	描述
`Item \| None`	检索到的项目，如果未找到则为 `None`。

asearch `async` ¶

asearch(
    namespace_prefix: tuple[str, ...],
    /,
    *,
    query: str | None = None,
    filter: dict[str, Any] | None = None,
    limit: int = 10,
    offset: int = 0,
    refresh_ttl: bool | None = None,
) -> list[SearchItem]

在命名空间前缀内异步搜索项目。

参数	描述
`namespace_prefix`	要搜索的分层路径前缀。类型： `tuple[str, ...]`
`query`	用于自然语言搜索的可选查询。类型: `str \| None` 默认值: `None`
`filter`	用于筛选结果的键值对。类型: `dict[str, Any] \| None` 默认值: `None`
`limit`	返回的最大项目数。类型： `int` 默认值： `10`
`offset`	在返回结果之前跳过的项目数。类型： `int` 默认值： `0`
`refresh_ttl`	是否刷新返回项目的 TTL。如果为 `None`，则使用存储的 `TTLConfig.refresh_default` 设置。如果未提供 `TTLConfig` 或未指定 TTL，则此参数将被忽略。类型： `bool \| None` 默认值： `None`

返回	描述
`list[SearchItem]`	符合搜索条件的项的列表。

示例

基本筛选

# Search for documents with specific metadata
results = await store.asearch(
    ("docs",),
    filter={"type": "article", "status": "published"}
)

自然语言搜索（需要向量存储实现）

# Initialize store with embedding configuration
store = YourStore( # e.g., InMemoryStore, AsyncPostgresStore
    index={
        "dims": 1536,  # embedding dimensions
        "embed": your_embedding_function,  # function to create embeddings
        "fields": ["text"]  # fields to embed
    }
)

# Search for semantically similar documents

results = await store.asearch(
    ("docs",),
    query="machine learning applications in healthcare",
    filter={"type": "research_paper"},
    limit=5
)

注意

自然语言搜索支持取决于您的存储实现，并需要正确的嵌入配置。

aput `async` ¶

aput(
    namespace: tuple[str, ...],
    key: str,
    value: dict[str, Any],
    index: Literal[False] | list[str] | None = None,
    *,
    ttl: float | None | NotProvided = NOT_PROVIDED,
) -> None

异步存储或更新存储中的项目。

参数	描述
`namespace`	项目的分层路径，表示为字符串元组。例如：`("documents", "user123")` 类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。与命名空间一起构成项目的完整路径。类型： `str`
`value`	包含项目数据的字典。必须包含字符串键和可 JSON 序列化的值。类型： `dict[str, Any]`
`index`	控制如何为搜索索引项目的字段 None（默认）：使用您在创建存储时配置的 `fields`（如果有）。如果您在初始化存储时未启用索引功能，`index` 参数将被忽略。 False：为此项目禁用索引 `list[str]`：要索引的字段路径列表，支持嵌套字段: `"metadata.title"` 数组访问：`"chapters[].content"`（每个单独索引）特定索引：`"authors[0].name"` 类型：* `Literal[False] \| list[str] \| None` 默认值： `None`
`ttl`	生存时间（以分钟为单位）。对此参数的支持取决于您的存储适配器。如果指定，该项目将在最后一次访问后这么多分钟后过期。None 表示永不过期。过期的运行将择机删除。默认情况下，过期计时器会在读取操作（get/search）和写入操作（put/update）时刷新，只要该项目包含在操作中。类型： `float \| None \| NotProvided` 默认值： `NOT_PROVIDED`

注意

索引支持取决于您的存储实现。如果您未使用索引功能初始化存储，index 参数将被忽略。

同样，TTL 支持取决于具体的存储实现。一些实现可能不支持项目的过期。

示例

存储项目。索引取决于您如何配置存储

await store.aput(("docs",), "report", {"memory": "Will likes ai"})

不要为语义搜索索引项目。仍然可以通过 get() 和 search() 操作访问，但不会有向量表示。

await store.aput(("docs",), "report", {"memory": "Will likes ai"}, index=False)

为搜索索引特定字段（如果存储配置为索引项目）

await store.aput(
    ("docs",),
    "report",
    {
        "memory": "Will likes ai",
        "context": [{"content": "..."}, {"content": "..."}]
    },
    index=["memory", "context[*].content"]
)

adelete `async` ¶

adelete(namespace: tuple[str, ...], key: str) -> None

异步删除一个项目。

参数	描述
`namespace`	项目的分层路径。类型： `tuple[str, ...]`
`key`	命名空间内的唯一标识符。类型： `str`

alist_namespaces `async` ¶

alist_namespaces(
    *,
    prefix: NamespacePath | None = None,
    suffix: NamespacePath | None = None,
    max_depth: int | None = None,
    limit: int = 100,
    offset: int = 0,
) -> list[tuple[str, ...]]

异步列出和过滤存储中的命名空间。

用于探索数据组织、查找特定集合或导航命名空间层次结构。

参数	描述
`prefix`	过滤以此路径开头的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`后缀`	筛选以此路径结尾的命名空间。类型： `NamespacePath \| None` 默认值： `None`
`max_depth`	返回层次结构中达到此深度的命名空间。比此级别更深的命名空间将被截断至此深度。 TYPE: `int \| None` DEFAULT: `None`
`limit`	要返回的最大命名空间数。类型： `int` 默认值： `100`
`offset`	用于分页的要跳过的命名空间数。类型： `int` 默认值： `0`

返回	描述
`list[tuple[str, ...]]`	符合条件的命名空间元组列表。每个元组代表一个完整的命名空间路径，最深达到 `max_depth`。

示例

使用现有命名空间设置 max_depth=3

# Given the following namespaces:
# ("a", "b", "c")
# ("a", "b", "d", "e")
# ("a", "b", "d", "i")
# ("a", "b", "f")
# ("a", "c", "f")

await store.alist_namespaces(prefix=("a", "b"), max_depth=3)
# Returns: [("a", "b", "c"), ("a", "b", "d"), ("a", "b", "f")]

from_conn_string `classmethod` ¶

from_conn_string(
    conn_string: str,
    *,
    pipeline: bool = False,
    pool_config: PoolConfig | None = None,
    index: PostgresIndexConfig | None = None,
    ttl: TTLConfig | None = None,
) -> Iterator[PostgresStore]

从连接字符串创建一个新的 PostgresStore 实例。

参数	描述
`conn_string`	Postgres 连接信息字符串。类型： `str`
`pipeline`	是否使用 Pipeline 类型： `bool` 默认值： `False`
`pool_config`	连接池的配置。如果提供，将创建一个连接池并使用它，而不是单个连接。这将覆盖 `pipeline` 参数。类型： `PoolConfig \| None` 默认值： `None`
`index`	存储的索引配置。类型： `PostgresIndexConfig \| None` 默认值： `None`
`ttl`	存储的 TTL 配置。类型： `TTLConfig \| None` 默认值： `None`

返回	描述
`PostgresStore`	一个新的 PostgresStore 实例。类型： `Iterator[PostgresStore]`

sweep_ttl ¶

sweep_ttl() -> int

根据 TTL 删除过期的存储项。

返回	描述
`int`	已删除项目的数量。类型： `int`

start_ttl_sweeper ¶

start_ttl_sweeper(sweep_interval_minutes: int | None = None) -> Future[None]

根据 TTL 定期删除过期的存储项。

返回	描述
`Future[None]`	可以等待或取消的 Future 对象。

stop_ttl_sweeper ¶

stop_ttl_sweeper(timeout: float | None = None) -> bool

如果 TTL 清扫线程正在运行，则停止它。

参数	描述
`timeout`	等待线程停止的最长时间（秒）。如果为 `None`，则无限期等待。类型： `float \| None` 默认值： `None`

返回	描述
`bool`	如果线程已成功停止或未在运行，则为 True；如果在线程停止前达到超时，则为 False。类型： `bool`

del ¶

__del__() -> None

确保在对象被垃圾回收时停止 TTL 清扫线程。

batch ¶

batch(ops: Iterable[Op]) -> list[Result]

在单个批次中同步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

abatch `async` ¶

abatch(ops: Iterable[Op]) -> list[Result]

在单个批次中异步执行多个操作。

参数	描述
`ops`	要执行的操作的可迭代对象。类型： `Iterable[Op]`

返回	描述
`list[Result]`	一个结果列表，其中每个结果对应输入中的一个操作。
`list[Result]`	结果的顺序与输入操作的顺序匹配。

setup ¶

setup() -> None

设置存储数据库。

此方法在 Postgres 数据库中创建必要的表（如果它们尚不存在）并运行数据库迁移。用户在首次使用存储时必须直接调用它。

存储

langgraph.store.base ¶

NamespacePath module-attribute ¶

NamespaceMatchType module-attribute ¶

Embeddings ¶

embed_documents abstractmethod ¶

embed_query abstractmethod ¶

aembed_documents async ¶

aembed_query async ¶

NotProvided ¶

Item ¶

SearchItem ¶

__init__ ¶

GetOp ¶

namespace instance-attribute ¶

key instance-attribute ¶

refresh_ttl class-attribute instance-attribute ¶

SearchOp ¶

namespace_prefix instance-attribute ¶

filter class-attribute instance-attribute ¶

limit class-attribute instance-attribute ¶

offset class-attribute instance-attribute ¶

query class-attribute instance-attribute ¶

refresh_ttl class-attribute instance-attribute ¶

MatchCondition ¶

match_type instance-attribute ¶

path instance-attribute ¶

ListNamespacesOp ¶

match_conditions class-attribute instance-attribute ¶

max_depth class-attribute instance-attribute ¶

limit class-attribute instance-attribute ¶

offset class-attribute instance-attribute ¶

PutOp ¶

namespace instance-attribute ¶

key instance-attribute ¶

value instance-attribute ¶

index class-attribute instance-attribute ¶

ttl class-attribute instance-attribute ¶

InvalidNamespaceError ¶

TTLConfig ¶

refresh_on_read instance-attribute ¶

default_ttl instance-attribute ¶

sweep_interval_minutes instance-attribute ¶

IndexConfig ¶

dims instance-attribute ¶

embed instance-attribute ¶

fields instance-attribute ¶

BaseStore ¶

batch abstractmethod ¶

abatch abstractmethod async ¶

get ¶

search ¶

put ¶

delete ¶

list_namespaces ¶

aget async ¶

asearch async ¶

aput async ¶

adelete async ¶

alist_namespaces async ¶

ensure_embeddings ¶

get_text_at_path ¶

tokenize_path ¶

langgraph.store.postgres ¶

AsyncPostgresStore ¶

batch ¶

get ¶

search ¶

put ¶

delete ¶

list_namespaces ¶

aget async ¶

asearch async ¶

aput async ¶

adelete async ¶

alist_namespaces async ¶

abatch async ¶

from_conn_string async classmethod ¶

setup async ¶

sweep_ttl async ¶

NamespacePath `module-attribute` ¶

NamespaceMatchType `module-attribute` ¶

embed_documents `abstractmethod` ¶

embed_query `abstractmethod` ¶

aembed_documents `async` ¶

aembed_query `async` ¶

init ¶

namespace `instance-attribute` ¶

key `instance-attribute` ¶

refresh_ttl `class-attribute` `instance-attribute` ¶

namespace_prefix `instance-attribute` ¶

filter `class-attribute` `instance-attribute` ¶

limit `class-attribute` `instance-attribute` ¶

offset `class-attribute` `instance-attribute` ¶

query `class-attribute` `instance-attribute` ¶

refresh_ttl `class-attribute` `instance-attribute` ¶

match_type `instance-attribute` ¶

path `instance-attribute` ¶

match_conditions `class-attribute` `instance-attribute` ¶

max_depth `class-attribute` `instance-attribute` ¶

limit `class-attribute` `instance-attribute` ¶

offset `class-attribute` `instance-attribute` ¶

namespace `instance-attribute` ¶

key `instance-attribute` ¶

value `instance-attribute` ¶

index `class-attribute` `instance-attribute` ¶

ttl `class-attribute` `instance-attribute` ¶

refresh_on_read `instance-attribute` ¶

default_ttl `instance-attribute` ¶

sweep_interval_minutes `instance-attribute` ¶

dims `instance-attribute` ¶

embed `instance-attribute` ¶

fields `instance-attribute` ¶

batch `abstractmethod` ¶

abatch `abstractmethod` `async` ¶

aget `async` ¶

asearch `async` ¶

aput `async` ¶

adelete `async` ¶

alist_namespaces `async` ¶

aget `async` ¶

asearch `async` ¶

aput `async` ¶

adelete `async` ¶

alist_namespaces `async` ¶

abatch `async` ¶

from_conn_string `async` `classmethod` ¶

setup `async` ¶

sweep_ttl `async` ¶

start_ttl_sweeper `async` ¶

stop_ttl_sweeper `async` ¶

min_size `instance-attribute` ¶

max_size `instance-attribute` ¶

kwargs `instance-attribute` ¶

aget `async` ¶

asearch `async` ¶

aput `async` ¶

adelete `async` ¶

alist_namespaces `async` ¶

from_conn_string `classmethod` ¶

del ¶

abatch `async` ¶