Jump to content

Connect SuperML | Leeroopedia MCP: Equip your AI agents with best practices, code verification, and debugging knowledge. Powered by Leeroo — building Organizational Superintelligence. Contact us at founders@leeroo.com.

Implementation:Langchain ai Langgraph Runtime Class

From Leeroopedia
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

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment