pregel store

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

class BaseStore(ABC):
    supports_ttl: bool = False
    ttl_config: TTLConfig | None = None

    __slots__ = ("__weakref__",)

class TTLConfig(TypedDict, total=False):
    """Configuration for TTL (time-to-live) behavior in the store."""

    refresh_on_read: bool
    """Default behavior for refreshing TTLs on read operations (GET and SEARCH).
    
    If True, TTLs will be refreshed on read operations (get/search) by default.
    This can be overridden per-operation by explicitly setting refresh_ttl.
    Defaults to True if not configured.
    """
    default_ttl: float | None
    """Default TTL (time-to-live) in minutes for new items.
    
    If provided, new items will expire after this many minutes after their last access.
    The expiration timer refreshes on both read and write operations.
    Defaults to None (no expiration).
    """
    sweep_interval_minutes: int | None
    """Interval in minutes between TTL sweep operations.
    
    If provided, the store will periodically delete expired items based on TTL.
    Defaults to None (no sweeping).
    """

1
2
3
4
5
6
7
8
9

Op = Union[GetOp, SearchOp, PutOp, ListNamespacesOp]

class BaseStore(ABC):
    @abstractmethod
    def batch(self, ops: Iterable[Op]) -> list[Result]:
        pass

    @abstractmethod
    async def abatch(self, ops: Iterable[Op]) -> list[Result]:

1
2
3
4
5
6
7
8
9

# 提问
我在Langgraph store 的源码中，看到了如下对象定义，请以表格的形式给我讲解它们的语义:
1. Item
2. SearchItem
3. GetOp
4. SearchOp
5. PutOp
6. ListNamespacesOp
7. MatchCondition

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

class Item:
    """Represents a stored item with metadata.

    Args:
        value: The stored data as a dictionary. Keys are filterable.
        key: Unique identifier within the namespace.
        namespace: Hierarchical path defining the collection in which this document resides.
            Represented as a tuple of strings, allowing for nested categorization.
            For example: ("documents", 'user123')
        created_at: Timestamp of item creation.
        updated_at: Timestamp of last update.
    """

    __slots__ = ("value", "key", "namespace", "created_at", "updated_at")

    def __init__(
        self,
        *,
        value: dict[str, Any],
        key: str,
        namespace: tuple[str, ...],
        created_at: datetime,
        updated_at: datetime,
    ):
        self.value = value
        self.key = key
        # The casting from json-like types is for if this object is
        # deserialized.
        self.namespace = tuple(namespace)
        self.created_at = (
            datetime.fromisoformat(cast(str, created_at))
            if isinstance(created_at, str)
            else created_at
        )
        self.updated_at = (
            datetime.fromisoformat(cast(str, updated_at))
            if isinstance(updated_at, str)
            else updated_at
        )

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

class SearchItem(Item):
    """Represents an item returned from a search operation with additional metadata."""

    __slots__ = ("score",)

    def __init__(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict[str, Any],
        created_at: datetime,
        updated_at: datetime,
        score: float | None = None,
    ) -> None:
        """Initialize a result item.

        Args:
            namespace: Hierarchical path to the item.
            key: Unique identifier within the namespace.
            value: The stored value.
            created_at: When the item was first created.
            updated_at: When the item was last updated.
            score: Relevance/similarity score if from a ranked operation.
        """
        super().__init__(
            value=value,
            key=key,
            namespace=namespace,
            created_at=created_at,
            updated_at=updated_at,
        )
        self.score = score

1
2
3
4
5
6
7

class GetOp(NamedTuple):
    namespace: tuple[str, ...] # 分层命名空间路径
    key: str # 查询的键
    refresh_ttl: bool = True # 是否刷新返回项的 TTL

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

class SearchOp(NamedTuple):
    namespace_prefix: tuple[str, ...]
    """
        Supported Operators:
        - $eq: Equal to (same as direct value comparison)
        - $ne: Not equal to
        - $gt: Greater than
        - $gte: Greater than or equal to
        - $lt: Less than
        - $lte: Less than or equal to
        {"score": {"$gt": 4.99}}
        {
            "score": {"$gte": 3.0},
            "color": "red"
        }
    """
    filter: dict[str, Any] | None = None
    limit: int = 10
    offset: int = 0
    # 自然语言的搜索查询
    query: str | None = None
    # 是否刷新返回项的 TTL
    refresh_ttl: bool = True

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

1
2
3
4
5
6

class PutOp(NamedTuple):
    namespace: tuple[str, ...]
    key: str
    value: dict[str, Any] | None
    index: Literal[False] | list[str] | None = None
    ttl: float | None = None

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

class MatchCondition(NamedTuple):
    match_type: NamespaceMatchType # 匹配前缀，还是后缀
    path: NamespacePath # 命名空间路径，支持通配符 *

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

NamespacePath = tuple[Union[str, Literal["*"]], ...]
"""
example:
    ("users",)  # Exact users namespace
    ("documents", "*")  # Any sub-namespace under documents
    ("cache", "*", "v1")  # Any cache category with v1 version
"""

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

class ListNamespacesOp(NamedTuple):
    match_conditions: tuple[MatchCondition, ...] | None = None
    max_depth: int | None = None
    limit: int = 100
    offset: int = 0

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

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35

    def get(
        self,
        namespace: tuple[str, ...],
        key: str,
        *,
        refresh_ttl: bool | None = None,
    ) -> Item | None:
        return self.batch(
            [GetOp(namespace, str(key), _ensure_refresh(self.ttl_config, refresh_ttl))]
        )[0]

    def search(
        self,
        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]:
        
        return self.batch(
            [
                SearchOp(
                    namespace_prefix,
                    filter,
                    limit,
                    offset,
                    query,
                    _ensure_refresh(self.ttl_config, refresh_ttl),
                )
            ]
        )[0]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

    def put(
        self,
        namespace: tuple[str, ...],
        key: str,
        value: dict[str, Any],
        index: Literal[False] | list[str] | None = None,
        *,
        ttl: float | None | NotProvided = NOT_PROVIDED,
    ) -> None:
        
        _validate_namespace(namespace)
        if ttl not in (NOT_PROVIDED, None) and not self.supports_ttl:
            raise NotImplementedError(
                f"TTL is not supported by {self.__class__.__name__}. "
                f"Use a store implementation that supports TTL or set ttl=None."
            )
        self.batch(
            [
                PutOp(
                    namespace,
                    str(key),
                    value,
                    index=index,
                    ttl=_ensure_ttl(self.ttl_config, ttl),
                )
            ]
        )

    def delete(self, namespace: tuple[str, ...], key: str) -> None:
        """Delete an item.

        Args:
            namespace: Hierarchical path for the item.
            key: Unique identifier within the namespace.
        """
        self.batch([PutOp(namespace, str(key), None, ttl=None)])

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

class IndexConfig(TypedDict, total=False):
    dims: int # embedding 的维度
    embed: Embeddings | EmbeddingsFunc | AEmbeddingsFunc | str
    """Optional function to generate embeddings from text.
    
    Can be specified in three ways:
        1. A LangChain Embeddings instance
        2. A synchronous embedding function (EmbeddingsFunc)
        3. An asynchronous embedding function (AEmbeddingsFunc)
        4. A provider string (e.g., "openai:text-embedding-3-small")
    """
    fields: list[str] | None
    """Fields to extract text from for embedding generation.
    
    Controls which parts of stored items are embedded for semantic search. Follows JSON path syntax:

        - ["$"]: 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
    """

class InMemoryStore(BaseStore):
    __slots__ = (
        "_data",
        "_vectors",
        "index_config",
        "embeddings",
    )

    def __init__(self, *, index: IndexConfig | None = None) -> None:
        # Both _data and _vectors are wrapped in the In-memory API
        # Do not change their names
        self._data: dict[tuple[str, ...], dict[str, Item]] = defaultdict(dict)
        # [ns][key][path]
        self._vectors: dict[tuple[str, ...], dict[str, dict[str, list[float]]]] = (
            defaultdict(lambda: defaultdict(dict))
        )
        self.index_config = index
        if self.index_config:
            self.index_config = self.index_config.copy()
            self.embeddings: Embeddings | None = ensure_embeddings(
                self.index_config.get("embed"),
            )
            self.index_config["__tokenized_fields"] = [
                (p, tokenize_path(p)) if p != "$" else (p, p)
                for p in (self.index_config.get("fields") or ["$"])
            ]

        else:
            self.index_config = None
            self.embeddings = None