From 5312edc830df9ad6a268aa626dcd2a1468d39d35 Mon Sep 17 00:00:00 2001 From: Sydney Runkle Date: Thu, 24 Jul 2025 17:28:58 -0400 Subject: [PATCH] more runtime details --- docs/docs/reference/runtime.md | 18 +++++++++ libs/langgraph/langgraph/runtime.py | 58 +++++++++++++++++++++++++++-- 2 files changed, 72 insertions(+), 4 deletions(-) create mode 100644 docs/docs/reference/runtime.md diff --git a/docs/docs/reference/runtime.md b/docs/docs/reference/runtime.md new file mode 100644 index 000000000..f326e78b8 --- /dev/null +++ b/docs/docs/reference/runtime.md @@ -0,0 +1,18 @@ +# Runtime + +::: langgraph.runtime.Runtime + options: + show_root_heading: true + show_root_full_path: false + members: + - context + - store + - stream_writer + - previous + +::: langgraph.runtime + options: + members: + - get_runtime + + diff --git a/libs/langgraph/langgraph/runtime.py b/libs/langgraph/langgraph/runtime.py index 2d430ebcd..5c1053a78 100644 --- a/libs/langgraph/langgraph/runtime.py +++ b/libs/langgraph/langgraph/runtime.py @@ -11,9 +11,9 @@ from langgraph.store.base import BaseStore from langgraph.types import _DC_KWARGS, StreamWriter from langgraph.typing import ContextT - __all__ = ("Runtime", "get_runtime") + def _no_op_stream_writer(_: Any) -> None: ... @@ -26,11 +26,61 @@ class _RuntimeOverrides(TypedDict, Generic[ContextT], total=False): @dataclass(**_DC_KWARGS) class Runtime(Generic[ContextT]): - """Convenience class that bundles run-scoped context and graph configuration. + """Convenience class that bundles run-scoped context and other runtime utilities. !!! version-added "Added in version v0.6.0" - TODO: write a compelling example with the new API. + Example: + + ```python + 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. Define a schema for the runtime context. + 2. Create a store to persist memories and other information. + 3. Use the runtime context to access the user_id. """ context: ContextT = field(default=None) # type: ignore[assignment] @@ -81,7 +131,7 @@ DEFAULT_RUNTIME = Runtime( def get_runtime(context_schema: type[ContextT] | None = None) -> Runtime[ContextT]: """Get the runtime for the current graph run. - + Args: context_schema: Optional schema used for type hinting the return type of the runtime.