pregel runtime

1
2

# 提问
上面是 Langgraph Runtime 的定义，请给我解释一下 Runtime 的语义，并以表格的形式列举 Runtime 的属性，并以另一个表格列举 Runtime 的方法

 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

from typing import TypedDict
from langgraph.graph import StateGraph
from dataclasses import dataclass
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore


@dataclass
class Context:  # (1)!
    user_id: str


class State(TypedDict, total=False):
    response: str


store = InMemoryStore()  # (2)!
store.put(("users",), "user_123", {"name": "Alice"})


def personalized_greeting(state: State, runtime: Runtime[Context]) -> State:
    '''Generate personalized greeting using runtime context and store.'''
    user_id = runtime.context.user_id  # (3)!
    name = "unknown_user"
    if runtime.store:
        if memory := runtime.store.get(("users",), user_id):
            name = memory.value["name"]

    response = f"Hello {name}! Nice to see you again."
    return {"response": response}


graph = (
    StateGraph(state_schema=State, context_schema=Context)
    .add_node("personalized_greeting", personalized_greeting)
    .set_entry_point("personalized_greeting")
    .set_finish_point("personalized_greeting")
    .compile(store=store)
)

result = graph.invoke({}, context=Context(user_id="user_123"))
print(result)
# > {'response': 'Hello Alice! Nice to see you again.'}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

@dataclass(**_DC_KWARGS)
class Runtime(Generic[ContextT]):
    context: ContextT = field(default=None)  # type: ignore[assignment]
    """Static context for the graph run, like user_id, db_conn, etc.
    
    Can also be thought of as 'run dependencies'."""

    store: BaseStore | None = field(default=None)
    """Store for the graph run, enabling persistence and memory."""

    stream_writer: StreamWriter = field(default=_no_op_stream_writer)
    """Function that writes to the custom stream."""

    previous: Any = field(default=None)
    """The previous return value for the given thread.
    
    Only available with the functional API when a checkpointer is provided.
    """

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

runtime = Runtime(
    # 对 context 做类型转换，对应示例传入的 context_schema
    context=_coerce_context(self.context_schema, context),
    # BaseStore
    store=store,
    # def stream_writer(c: Any) -> None:
    # 内部会调用 SyncQueue.put 方法，往消息队列提交自定义消息
    # pregel.stream 可以从 SyncQueue 读取到对应的消息
    stream_writer=stream_writer,
    previous=None,
)
parent_runtime = config[CONF].get(CONFIG_KEY_RUNTIME, DEFAULT_RUNTIME)
# 从 configurable 获取 __pregel_runtime 并合并
runtime = parent_runtime.merge(runtime)
# 更新 runtime
config[CONF][CONFIG_KEY_RUNTIME] = runtime

1
2

# 是 Python 3.11+ 中 带 default 参数的 TypeVar 泛型变量定义，用于类型系统中的泛型约束。
ContextT = TypeVar("ContextT", bound=Union[StateLike, None], default=None)

1
2

# 是 Python 中的 类型别名 定义，用来为复杂类型起一个更简洁、更语义化的名字。
StateLike: TypeAlias = Union[TypedDictLikeV1, TypedDictLikeV2, DataclassLike, BaseModel]

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

class TypedDictLikeV1(Protocol):
    """Protocol to represent types that behave like TypedDicts

    Version 1: using `ClassVar` for keys."""

    __required_keys__: ClassVar[frozenset[str]]
    __optional_keys__: ClassVar[frozenset[str]]


class TypedDictLikeV2(Protocol):
    """Protocol to represent types that behave like TypedDicts

    Version 2: not using `ClassVar` for keys."""

    __required_keys__: frozenset[str]
    __optional_keys__: frozenset[str]


class DataclassLike(Protocol):
    """Protocol to represent types that behave like dataclasses.

    Inspired by the private _DataclassT from dataclasses that uses a similar protocol as a bound."""

    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]

1
2
3
4

class TypedDictLikeV1(Protocol):
    """Version 1: using `ClassVar` for keys."""
    __required_keys__: ClassVar[frozenset[str]]
    __optional_keys__: ClassVar[frozenset[str]]

1
2
3

class MyTD:
    __required_keys__ = frozenset({"a", "b"})
    __optional_keys__ = frozenset({"c"})

1
2
3
4

class TypedDictLikeV2(Protocol):
    """Version 2: not using `ClassVar` for keys."""
    __required_keys__: frozenset[str]
    __optional_keys__: frozenset[str]

1
2
3

class DataclassLike(Protocol):
    """行为类似 dataclass 的类型"""
    __dataclass_fields__: ClassVar[dict[str, Field[Any]]]

1
2
3
4
5
6
7
8

from dataclasses import dataclass

@dataclass
class User:
    name: str
    age: int

# User 类自动有 __dataclass_fields__ 属性

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

@dataclass(**_DC_KWARGS)
class Runtime(Generic[ContextT]):
    def merge(self, other: Runtime[ContextT]) -> Runtime[ContextT]:
        """Merge two runtimes together.

        If a value is not provided in the other runtime, the value from the current runtime is used.
        """
        return Runtime(
            context=other.context or self.context,
            store=other.store or self.store,
            stream_writer=other.stream_writer
            if other.stream_writer is not _no_op_stream_writer
            else self.stream_writer,
            previous=other.previous or self.previous,
        )

    def override(
        self, **overrides: Unpack[_RuntimeOverrides[ContextT]]
    ) -> Runtime[ContextT]:
        """Replace the runtime with a new runtime with the given overrides."""
        return replace(self, **overrides)

1
2
3
4
5
6
7
8

DEFAULT_RUNTIME = Runtime(
    context=None,
    store=None,
    stream_writer=_no_op_stream_writer,
    previous=None,
)

def _no_op_stream_writer(_: Any) -> None: ...