存储

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

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

核心类型

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

模块

名称	描述
batch	用于在后台任务中批量处理操作的工具。
embed	用于处理嵌入函数和 LangChain 的 Embeddings 接口的工具。

类

名称	描述
嵌入	嵌入模型的接口。
项	表示带有元数据的存储项。
获取操作	根据命名空间和键检索特定项的操作。
搜索操作	在指定命名空间层级内搜索项的操作。
匹配条件	表示用于匹配存储中命名空间的模式。
列出命名空间操作	在存储中列出和过滤命名空间的操作。
放置操作	在存储中存储、更新或删除项的操作。
基础存储	持久化键值存储的抽象基类。

函数

名称	描述
确保嵌入	确保嵌入函数符合 LangChain 的 Embeddings 接口。
按路径获取文本	使用路径表达式或预标记化路径从对象中提取文本。
标记化路径	将路径标记化为组件。

属性

名称	类型	描述
命名空间路径		表示可包含通配符的命名空间路径的元组。
命名空间匹配类型		指定如何匹配命名空间路径。

NamespacePath module-attribute

NamespacePath = tuple[Union[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 abstractmethod

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

嵌入搜索文档。

参数

名称	类型	描述	默认
texts	列表[字符串]	要嵌入的文本列表。	必需

返回

类型	描述
列表[列表[浮点数]]	嵌入列表。

embed_query abstractmethod

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

嵌入查询文本。

参数

名称	类型	描述	默认
text	字符串	要嵌入的文本。	必需

返回

类型	描述
列表[浮点数]	嵌入。

aembed_documents async

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

异步嵌入搜索文档。

参数

名称	类型	描述	默认
texts	列表[字符串]	要嵌入的文本列表。	必需

返回

类型	描述
列表[列表[浮点数]]	嵌入列表。

aembed_query async

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

异步嵌入查询文本。

参数

名称	类型	描述	默认
text	字符串	要嵌入的文本。	必需

返回

类型	描述
列表[浮点数]	嵌入。

NotProvided

哨兵单例。

Item

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

参数

名称	类型	描述	默认
value	字典[字符串, 任意类型]	存储为字典的数据。键可用于过滤。	必需
key	字符串	命名空间内的唯一标识符。	必需
namespace	元组[字符串, ...]	定义此文档所在集合的分层路径。表示为字符串元组，允许嵌套分类。例如：("documents", 'user123')	必需
created_at	datetime	项创建的时间戳。	必需
updated_at	datetime	最后更新的时间戳。	必需

SearchItem

基类：项

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

方法

名称	描述
__init__	初始化结果项。

__init__

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

初始化结果项。

参数

名称	类型	描述	默认
namespace	元组[字符串, ...]	项的分层路径。	必需
key	字符串	命名空间内的唯一标识符。	必需
value	字典[字符串, 任意类型]	存储的值。	必需
created_at	datetime	项首次创建的时间。	必需
updated_at	datetime	项最后更新的时间。	必需
score	可选[浮点数]	如果来自排序操作，则为相关性/相似性得分。	None

GetOp

基类：NamedTuple

根据命名空间和键检索特定项的操作。

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

示例

基本项检索

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

属性

名称	类型	描述
namespace	元组[字符串, ...]	唯一标识项位置的分层路径。
key	字符串	项在其特定命名空间内的唯一标识符。
refresh_ttl	布尔值	是否为返回的项刷新 TTL。

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	元组[字符串, ...]	定义搜索范围的分层路径前缀。
filter	可选[字典[字符串, 任意类型]]	用于根据精确匹配或比较运算符过滤结果的键值对。
limit	整数	搜索结果中返回的最大项数。
offset	整数	用于分页跳过的匹配项数。
query	可选[字符串]	用于语义搜索功能的自然语言搜索查询。
refresh_ttl	布尔值	是否为返回的项刷新 TTL。

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: Optional[dict[str, Any]] = 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: Optional[str] = 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	命名空间匹配类型	要执行的命名空间匹配类型。
path	命名空间路径	可包含通配符的命名空间路径模式。

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	可选[元组[匹配条件, ...]]	用于过滤命名空间的可选条件。
max_depth	可选[整数]	返回的命名空间层级的最大深度。
limit	整数	返回的最大命名空间数量。
offset	整数	用于分页跳过的命名空间数量。

match_conditions class-attribute instance-attribute

match_conditions: Optional[tuple[MatchCondition, ...]] = (
    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: Optional[int] = None

返回的命名空间层级的最大深度。

注意

深于此层级的命名空间将被截断。

limit class-attribute instance-attribute

limit: int = 100

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

offset class-attribute instance-attribute

offset: int = 0

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

PutOp

基类：NamedTuple

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

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

属性

名称	类型	描述
namespace	元组[字符串, ...]	标识项位置的分层路径。
key	字符串	项在其命名空间内的唯一标识符。
value	可选[字典[字符串, 任意类型]]	要存储的数据，或 None（表示将项标记为删除）。
index	可选[联合类型[文字类型[布尔值], 列表[字符串]]]	控制项的字段如何为搜索操作建立索引。
ttl	可选[浮点数]	控制项的 TTL（存活时间），以分钟为单位。

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: Optional[dict[str, Any]]

要存储的数据，或 None（表示将项标记为删除）。

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

示例

{ "field1": "string value", "field2": 123, "nested": {"can": "contain", "any": "serializable data"} }

index class-attribute instance-attribute

index: Optional[Union[Literal[False], list[str]]] = 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: Optional[float] = None

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

如果提供此值，并且您使用的存储支持此功能，则该项在最后一次访问后将在这么长时间（以分钟为单位）后过期。过期计时器在读操作 (get/search) 和写操作 (put/update) 时都会刷新。当 TTL 过期时，该项将按尽力而为的方式安排删除。默认为 None（不设过期时间）。

InvalidNamespaceError

基类：ValueError

提供的命名空间无效。

TTLConfig

基类：TypedDict

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

属性

名称	类型	描述
refresh_on_read	布尔值	读取操作（GET 和 SEARCH）时刷新 TTL 的默认行为。
default_ttl	可选[浮点数]	新项的默认 TTL（存活时间），以分钟为单位。
sweep_interval_minutes	可选[整数]	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: Optional[float]

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

如果提供此值，新项在最后一次访问后将在这么长时间（以分钟为单位）后过期。过期计时器在读写操作时都会刷新。默认为 None（不设过期时间）。

sweep_interval_minutes instance-attribute

sweep_interval_minutes: Optional[int]

TTL 清理操作之间的间隔，以分钟为单位。

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

IndexConfig

基类：TypedDict

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

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

属性

名称	类型	描述
dims	整数	嵌入向量中的维度数量。
embed	Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc, str]	可选的从文本生成嵌入的函数。
字段	Optional[list[str]]	用于从文本中提取字段以生成嵌入。

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: Union[
    Embeddings, EmbeddingsFunc, AEmbeddingsFunc, str
]

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

可以通过三种方式指定

1. LangChain Embeddings 实例
2. 同步嵌入函数 (EmbeddingsFunc)
3. 异步嵌入函数 (AEmbeddingsFunc)
4. 提供者字符串（例如，“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: Optional[list[str]]

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

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

- ["$"]: Embeds the entire JSON object as one vector  (default)
- ["field1", "field2"]: Embeds specific top-level fields
- ["parent.child"]: Embeds nested fields using dot notation
- ["array[*].field"]: Embeds field from each array element separately

注意

在使用 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: Optional[bool] = None
) -> Optional[Item]

检索单个项。

参数

名称	类型	描述	默认
namespace	元组[字符串, ...]	项的分层路径。	必需
key	字符串	命名空间内的唯一标识符。	必需
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果为 None（默认值），则使用存储库的默认 refresh_ttl 设置。如果未指定 TTL，则忽略此参数。	None

返回

类型	描述
Optional[Item]	检索到的项；如果未找到，则为 None。

search

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

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

参数

名称	类型	描述	默认
namespace_prefix	元组[字符串, ...]	要搜索的分层路径前缀。	必需
query	可选[字符串]	可选的自然语言搜索查询。	None
filter	可选[字典[字符串, 任意类型]]	用于过滤结果的键值对。	None
limit	整数	要返回的最大项数。	10
offset	整数	在返回结果之前要跳过的项数。	0
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果未指定 TTL，则忽略此参数。	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: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    *,
    ttl: Union[Optional[float], NotProvided] = NOT_PROVIDED
) -> None

在存储库中存储或更新项。

参数

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

list_namespaces

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

列出并过滤存储库中的命名空间。

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

参数

名称	类型	描述	默认
prefix	Optional[NamespacePath]	过滤以此路径开头的命名空间。	None
suffix	Optional[NamespacePath]	过滤以此路径结尾的命名空间。	None
max_depth	可选[整数]	在层次结构中返回最大深度的命名空间。深度超过此级别的命名空间将被截断。	None
limit	整数	要返回的最大命名空间数量（默认 100）。	100
offset	整数	分页时要跳过的命名空间数量（默认 0）。	0

返回

类型	描述
list[tuple[str, ...]]	List[Tuple[str, ...]]：匹配条件的命名空间元组列表。
list[tuple[str, ...]]	每个元组表示直到 max_depth 的完整命名空间路径。

???+ 示例 "示例"：设置 max_depth=3。给定命名空间

# 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: Optional[bool] = None
) -> Optional[Item]

异步检索单个项。

参数

名称	类型	描述	默认
namespace	元组[字符串, ...]	项的分层路径。	必需
key	字符串	命名空间内的唯一标识符。	必需

返回

类型	描述
Optional[Item]	检索到的项；如果未找到，则为 None。

asearch async

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

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

参数

名称	类型	描述	默认
namespace_prefix	元组[字符串, ...]	要搜索的分层路径前缀。	必需
query	可选[字符串]	可选的自然语言搜索查询。	None
filter	可选[字典[字符串, 任意类型]]	用于过滤结果的键值对。	None
limit	整数	要返回的最大项数。	10
offset	整数	在返回结果之前要跳过的项数。	0
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果为 None（默认值），则使用存储库的 TTLConfig.refresh_default 设置。如果未提供 TTLConfig 或未指定 TTL，则忽略此参数。	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: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    *,
    ttl: Union[Optional[float], NotProvided] = NOT_PROVIDED
) -> None

异步在存储库中存储或更新项。

参数

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

alist_namespaces async

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

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

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

参数

名称	类型	描述	默认
prefix	Optional[NamespacePath]	过滤以此路径开头的命名空间。	None
suffix	Optional[NamespacePath]	过滤以此路径结尾的命名空间。	None
max_depth	可选[整数]	在层次结构中返回最大深度的命名空间。深度超过此级别的命名空间将被截断到此深度。	None
limit	整数	要返回的最大命名空间数量（默认 100）。	100
offset	整数	分页时要跳过的命名空间数量（默认 0）。	0

返回

类型	描述
list[tuple[str, ...]]	List[Tuple[str, ...]]：匹配条件的命名空间元组列表。
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: Union[
        Embeddings,
        EmbeddingsFunc,
        AEmbeddingsFunc,
        str,
        None,
    ],
) -> Embeddings

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

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

参数

名称	类型	描述	默认
embed	Union[Embeddings, EmbeddingsFunc, AEmbeddingsFunc, str, None]	现有的 Embeddings 实例，或将文本转换为嵌入的函数。如果函数是异步的，它将用于同步和异步操作。	必需

返回

类型	描述
嵌入	一个包装提供的函数(s) 的 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: Union[str, list[str]]
) -> list[str]

使用路径表达式或预标记化路径从对象中提取文本。

参数

名称	类型	描述	默认
obj	Any	要从中提取文本的对象	必需
path	Union[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}"

模块

名称	描述
aio
base

类

名称	描述
AsyncPostgresStore	使用 pgvector 实现异步 Postgres 后端存储，可选支持向量搜索。
PostgresStore	使用 pgvector 实现 Postgres 后端存储，可选支持向量搜索。

AsyncPostgresStore

Bases: AsyncBatchedBaseStore, BasePostgresStore[Conn]

使用 pgvector 实现异步 Postgres 后端存储，可选支持向量搜索。

示例

基本设置和用法

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() 来正确清理资源。

方法

名称	描述
from_conn_string	从连接字符串创建新的 AsyncPostgresStore 实例。
setup	异步设置存储数据库。
sweep_ttl	根据 TTL 删除过期存储项。
start_ttl_sweeper	根据 TTL 定期删除过期存储项。
stop_ttl_sweeper	如果 TTL sweeper 任务正在运行，则停止它。

from_conn_string async classmethod

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

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

参数

名称	类型	描述	默认
conn_string	字符串	Postgres 连接信息字符串。	必需
pipeline	布尔值	是否使用 AsyncPipeline（仅适用于单连接）	False
pool_config	Optional[PoolConfig]	连接池的配置。如果提供，将创建连接池并使用它，而不是单个连接。这将覆盖 pipeline 参数。	None
index	Optional[PostgresIndexConfig]	嵌入配置。	None

返回

名称	类型	描述
AsyncPostgresStore	AsyncIterator[AsyncPostgresStore]	新的 AsyncPostgresStore 实例。

setup async

setup() -> None

异步设置存储数据库。

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

sweep_ttl async

sweep_ttl() -> int

根据 TTL 删除过期存储项。

返回

名称	类型	描述
整数	整数	删除的项数。

start_ttl_sweeper async

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

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

返回

类型	描述
Task[None]	可以等待或取消的任务。

stop_ttl_sweeper async

stop_ttl_sweeper(timeout: Optional[float] = None) -> bool

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

参数

名称	类型	描述	默认
timeout	可选[浮点数]	等待任务停止的最长时间，以秒为单位。如果为 None，则无限期等待。	None

返回

名称	类型	描述
布尔值	布尔值	如果任务成功停止或未运行，则为 True；如果在任务停止前达到超时，则为 False。

PostgresStore

Bases: BaseStore, BasePostgresStore[Conn]

使用 pgvector 实现 Postgres 后端存储，可选支持向量搜索。

示例

基本设置和用法

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 sweeper 线程正在运行，则停止它。
__del__	确保在对象被垃圾回收时停止 TTL sweeper 线程。
setup	设置存储数据库。

get

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

检索单个项。

参数

名称	类型	描述	默认
namespace	元组[字符串, ...]	项的分层路径。	必需
key	字符串	命名空间内的唯一标识符。	必需
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果为 None（默认值），则使用存储库的默认 refresh_ttl 设置。如果未指定 TTL，则忽略此参数。	None

返回

类型	描述
Optional[Item]	检索到的项；如果未找到，则为 None。

search

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

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

参数

名称	类型	描述	默认
namespace_prefix	元组[字符串, ...]	要搜索的分层路径前缀。	必需
query	可选[字符串]	可选的自然语言搜索查询。	None
filter	可选[字典[字符串, 任意类型]]	用于过滤结果的键值对。	None
limit	整数	要返回的最大项数。	10
offset	整数	在返回结果之前要跳过的项数。	0
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果未指定 TTL，则忽略此参数。	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: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    *,
    ttl: Union[Optional[float], NotProvided] = NOT_PROVIDED
) -> None

在存储库中存储或更新项。

参数

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

list_namespaces

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

列出并过滤存储库中的命名空间。

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

参数

名称	类型	描述	默认
prefix	Optional[NamespacePath]	过滤以此路径开头的命名空间。	None
suffix	Optional[NamespacePath]	过滤以此路径结尾的命名空间。	None
max_depth	可选[整数]	在层次结构中返回最大深度的命名空间。深度超过此级别的命名空间将被截断。	None
limit	整数	要返回的最大命名空间数量（默认 100）。	100
offset	整数	分页时要跳过的命名空间数量（默认 0）。	0

返回

类型	描述
list[tuple[str, ...]]	List[Tuple[str, ...]]：匹配条件的命名空间元组列表。
list[tuple[str, ...]]	每个元组表示直到 max_depth 的完整命名空间路径。

???+ 示例 "示例"：设置 max_depth=3。给定命名空间

# 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: Optional[bool] = None
) -> Optional[Item]

异步检索单个项。

参数

名称	类型	描述	默认
namespace	元组[字符串, ...]	项的分层路径。	必需
key	字符串	命名空间内的唯一标识符。	必需

返回

类型	描述
Optional[Item]	检索到的项；如果未找到，则为 None。

asearch async

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

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

参数

名称	类型	描述	默认
namespace_prefix	元组[字符串, ...]	要搜索的分层路径前缀。	必需
query	可选[字符串]	可选的自然语言搜索查询。	None
filter	可选[字典[字符串, 任意类型]]	用于过滤结果的键值对。	None
limit	整数	要返回的最大项数。	10
offset	整数	在返回结果之前要跳过的项数。	0
refresh_ttl	Optional[bool]	是否为返回的项刷新 TTL。如果为 None（默认值），则使用存储库的 TTLConfig.refresh_default 设置。如果未提供 TTLConfig 或未指定 TTL，则忽略此参数。	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: Optional[
        Union[Literal[False], list[str]]
    ] = None,
    *,
    ttl: Union[Optional[float], NotProvided] = NOT_PROVIDED
) -> None

异步在存储库中存储或更新项。

参数

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

alist_namespaces async

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

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

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

参数

名称	类型	描述	默认
prefix	Optional[NamespacePath]	过滤以此路径开头的命名空间。	None
suffix	Optional[NamespacePath]	过滤以此路径结尾的命名空间。	None
max_depth	可选[整数]	在层次结构中返回最大深度的命名空间。深度超过此级别的命名空间将被截断到此深度。	None
limit	整数	要返回的最大命名空间数量（默认 100）。	100
offset	整数	分页时要跳过的命名空间数量（默认 0）。	0

返回

类型	描述
list[tuple[str, ...]]	List[Tuple[str, ...]]：匹配条件的命名空间元组列表。
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: Optional[PoolConfig] = None,
    index: Optional[PostgresIndexConfig] = None,
    ttl: Optional[TTLConfig] = None
) -> Iterator[PostgresStore]

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

参数

名称	类型	描述	默认
conn_string	字符串	Postgres 连接信息字符串。	必需
pipeline	布尔值	是否使用 Pipeline	False
pool_config	Optional[PoolConfig]	连接池的配置。如果提供，将创建连接池并使用它，而不是单个连接。这将覆盖 pipeline 参数。	None
index	Optional[PostgresIndexConfig]	存储库的索引配置。	None

返回

名称	类型	描述
PostgresStore	Iterator[PostgresStore]	新的 PostgresStore 实例。

sweep_ttl

sweep_ttl() -> int

根据 TTL 删除过期存储项。

返回

名称	类型	描述
整数	整数	删除的项数。

start_ttl_sweeper

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

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

返回

类型	描述
Future[None]	可以等待或取消的 Future。

stop_ttl_sweeper

stop_ttl_sweeper(timeout: Optional[float] = None) -> bool

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

参数

名称	类型	描述	默认
timeout	可选[浮点数]	等待线程停止的最长时间，以秒为单位。如果为 None，则无限期等待。	None

返回

名称	类型	描述
布尔值	布尔值	如果线程成功停止或未运行，则为 True；如果在线程停止前达到超时，则为 False。

__del__

__del__() -> None

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

setup

setup() -> None

设置存储数据库。

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

评论