Implementation:Langchain ai Langgraph Runtime Class
| Attribute | Value |
|---|---|
| Source | `libs/langgraph/langgraph/runtime.py` (161 lines) |
| Domain | Runtime, Configuration |
| Principle | Runtime_Context |
| Library | langgraph |
| Import | `from langgraph.runtime import Runtime, get_runtime` |
Overview
The `runtime.py` module defines the `Runtime` dataclass and the `get_runtime` helper function. `Runtime` bundles run-scoped context and utilities (context, store, stream writer, previous return value) that are automatically injected into graph nodes and middleware during execution. It was introduced in LangGraph v0.6.0.
Description
`Runtime(Generic[ContextT])` is a frozen-style dataclass that provides graph nodes with access to runtime utilities without requiring them to be part of the graph state:
- `context: ContextT` -- Static context for the graph run, such as `user_id`, database connections, or other dependencies. Defaults to `None`. Set via the `context` parameter when invoking the graph.
- `store: BaseStore | None` -- An optional `BaseStore` instance that provides persistence and memory capabilities. Set via the `store` parameter when compiling the graph.
- `stream_writer: StreamWriter` -- A function that writes data to the custom stream. Defaults to a no-op function.
- `previous: Any` -- The previous return value for the given thread. Only available with the functional API when a checkpointer is provided. Defaults to `None`.
The class provides two mutation methods:
- `merge(other: Runtime[ContextT]) -> Runtime[ContextT]` -- Creates a new `Runtime` by merging two instances, preferring non-default values from `other`.
- `override(**overrides) -> Runtime[ContextT]` -- Creates a new `Runtime` with specific fields replaced using `dataclasses.replace`.
`DEFAULT_RUNTIME` -- A module-level singleton `Runtime` instance with all fields at their default values.
`get_runtime(context_schema=None) -> Runtime[ContextT]` -- Retrieves the current `Runtime` from the active graph configuration. The optional `context_schema` parameter is used only for type hinting the return type and does not affect runtime behavior.
The module also defines `_RuntimeOverrides`, a `TypedDict` used to type the keyword arguments accepted by `override()`.
Usage
from langgraph.runtime import Runtime, get_runtime
from dataclasses import dataclass
@dataclass
class Context:
user_id: str
def my_node(state: dict, runtime: Runtime[Context]) -> dict:
user_id = runtime.context.user_id
if runtime.store:
data = runtime.store.get(("users",), user_id)
return state
Code Reference
Runtime Dataclass
| Field | Type | Default | Description |
|---|---|---|---|
| `context` | `ContextT` | `None` | Static context for the graph run (user_id, db_conn, etc.). |
| `store` | None` | `None` | Store for persistence and memory. |
| `stream_writer` | `StreamWriter` | `_no_op_stream_writer` | Function that writes to the custom stream. |
| `previous` | `Any` | `None` | Previous return value for the thread (functional API only). |
Runtime Methods
| Method | Signature | Description |
|---|---|---|
| `merge` | `(other: Runtime[ContextT]) -> Runtime[ContextT]` | Merge two runtimes, preferring non-default values from `other`. |
| `override` | `(**overrides) -> Runtime[ContextT]` | Replace specific fields via `dataclasses.replace`. |
get_runtime Function
| Parameter | Type | Default | Description |
|---|---|---|---|
| `context_schema` | None` | `None` | Optional schema for type hinting the return type. |
| Returns | `Runtime[ContextT]` | -- | The runtime for the current graph run. |
I/O Contract
| Aspect | Detail |
|---|---|
| Input | `Runtime` is constructed by the LangGraph runtime and injected into node functions. `get_runtime` reads from the current config via `get_config()`. |
| Output | A `Runtime` instance providing access to context, store, stream writer, and previous values. |
| Side Effects | `stream_writer` may write to external streams when called. `store` may perform I/O. |
| Thread Safety | `Runtime` is designed for single-run scope. Each graph invocation receives its own `Runtime` instance. |
Usage Examples
Injecting Runtime into a Node
from typing import TypedDict
from dataclasses import dataclass
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from langgraph.store.memory import InMemoryStore
@dataclass
class Context:
user_id: str
class State(TypedDict, total=False):
response: str
store = InMemoryStore()
store.put(("users",), "user_123", {"name": "Alice"})
def personalized_greeting(state: State, runtime: Runtime[Context]) -> State:
user_id = runtime.context.user_id
name = "unknown_user"
if runtime.store:
if memory := runtime.store.get(("users",), user_id):
name = memory.value["name"]
return {"response": f"Hello {name}!"}
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"))
# {'response': 'Hello Alice!'}
Using get_runtime Programmatically
from langgraph.runtime import get_runtime
def my_node(state: dict) -> dict:
runtime = get_runtime()
if runtime.store:
runtime.store.put(("logs",), "entry_1", {"action": "processed"})
return state
Writing to the Custom Stream
from langgraph.runtime import Runtime
def streaming_node(state: dict, runtime: Runtime) -> dict:
runtime.stream_writer({"status": "processing", "progress": 50})
# ... do work ...
runtime.stream_writer({"status": "done", "progress": 100})
return state
Related Pages
- Langchain_ai_Langgraph_Typing_Module -- `ContextT` type variable used to parameterize `Runtime`.
- Langchain_ai_Langgraph_PregelProtocol -- The protocol that creates and manages `Runtime` instances.
- Langchain_ai_Langgraph_ServerRuntime -- Server-side runtime variant for the LangGraph Agent Server.
- Langchain_ai_Langgraph_Public_Constants -- Internal constants (`CONF`, `CONFIG_KEY_RUNTIME`) used by `get_runtime`.