ޠjO( SSKJr SSKJr SSKJrJrJr SSKJ r J r J r SSK J r SSKJr SSKJrJr SSKJrJr SS KJr SS KJrJr SS KJr S r\"S S S9"SS55r\"S S S9"SS55r"SS5r S Sjr!S!Sjr""SS\\ \SS9r#\"S"0\D6"SS\ \55r$\$"SS\!\"SSSS9r%S#S$Sjjr&g)%) annotations)Callable) dataclassfieldreplace)AnyGenericcast) BaseStore)BaseUser) TypedDictUnpack)CONFCONFIG_KEY_RUNTIME) get_config) _DC_KWARGS StreamWriter)ContextT)r ExecutionInfo RunControlRuntime ServerInfo get_runtimeT)frozenslotsc\rSrSr%SrS\S'S\S'S\S'SrS\S 'SrS\S 'S rS \S 'Sr S\S'SSjr Sr g)rzORead-only execution info/metadata for the execution of current thread/run/node.str checkpoint_id checkpoint_nstask_idN str | None thread_idrun_idint node_attemptz float | Nonenode_first_attempt_timec [U40UD6$)zAReturn a new execution info object with selected fields replaced.rself overridess K/var/www/html/gaurav/venv/lib/python3.13/site-packages/langgraph/runtime.pypatchExecutionInfo.patch7st)y)))r-rreturnr) __name__ __module__ __qualname____firstlineno____doc____annotations__r#r$r'r(r/__static_attributes__r2r1r.rrscY6= L0 Iz HFJAL#<,0\0F*r1rc@\rSrSr%SrS\S'S\S'SrS\S'S rg) r<zmMetadata injected by LangGraph Server. None when running open-source LangGraph without LangSmith deployments.r assistant_idgraph_idNzBaseUser | Noneuserr2)r4r5r6r7r8r9r?r:r2r1r.rr<s%w5M1 D/ r1rc\\rSrSrSrSrS SjrS S Sjjr\S Sj5r \S Sj5r Sr g)rOavRun-scoped control surface for cooperative draining. Intended for a single graph run. Create a fresh `RunControl` per run; reusing a control after `request_drain()` leaves it drained. Safe to call from any thread: the drain request is represented by a single attribute write, so no lock is needed for this signal. If more mutable state is added here, add synchronization.  _drain_reasoncSUlgNrBr,s r.__init__RunControl.__init__\s )-r1cXlgrErB)r,reasons r. request_drainRunControl.request_drain_s#r1cURSL$rErBrFs r.drain_requestedRunControl.drain_requestedbs!!--r1cUR$rErBrFs r. drain_reasonRunControl.drain_reasonfs!!!r1Nr3None)shutdown)rJrr3rTr3boolr3r") r4r5r6r7r8 __slots__rGrKpropertyrNrQr:r2r1r.rrOs@#I.$..""r1rcgrEr2)_s r._no_op_stream_writerr]ks#r1cgrEr2r2r1r._no_op_heartbeatr_nsr1cf\rSrSr%S\S'S\S'S\S'S\S 'S \S 'S \S 'S\S'S\S'Srg)_RuntimeOverridesqrcontextBaseStore | Nonestorer stream_writerCallable[[], None] heartbeatrpreviousrexecution_infoServerInfo | None server_infoRunControl | Nonecontrolr2N)r4r5r6r7r9r:r2r1r.raraqs1  !!M!!"" r1raF)totalc(\rSrSr%Sr\"SS9rS\S'\"SS9rS\S'\"\ S9r S \S '\"\ S9r S \S '\"SS9r S \S'\"SS9rS\S'\"SS9rS\S'\"SS9rS\S'SSjrSSjrSSjr\SSj5r\SSj5rSrg) r|aConvenience class that bundles run-scoped context and other runtime utilities. This class is injected into graph nodes and middleware. It provides access to `context`, `store`, `stream_writer`, `previous`, and `execution_info`. !!! note "Accessing `config`" `Runtime` does not include `config`. To access `RunnableConfig`, you can inject it directly by adding a `config: RunnableConfig` parameter to your node function (recommended), or use `get_config()` from `langgraph.config`. !!! note `ToolRuntime` (from `langgraph.prebuilt`) is a subclass that provides similar functionality but is designed specifically for tools. It shares `context`, `store`, and `stream_writer` with `Runtime`, and adds tool-specific attributes like `config`, `state`, and `tool_call_id`. !!! version-added "Added in version v0.6.0" 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`. N)defaultrrcrdrerrfrgrhrrizExecutionInfo | Nonerjrkrlrmrnc .[UR=(d URUR=(d URUR[La URO URUR [ La UR O UR URc URO URUR=(d URUR=(d URUR=(d URS9$)z{Merge two runtimes together. If a value is not provided in the other runtime, the value from the current runtime is used. )rcrerfrhrirjrlrn) rrcrerfr]rhr_rirjrlrn)r,others r.merge Runtime.merges MM1T\\+++""*>> --##&66oo&+nn&N(>2&+4%8K"8w!&t!4G 4  (*>* *  SSOOr1rN)rcrerfrhrirjrncx[[[[5[R [ 55nU$)zGet the runtime for the current graph run. Args: context_schema: Optional schema used for type hinting the return type of the runtime. Returns: The runtime for the current graph run. )r rrrrgetr)context_schemaruntimes r.rr(s-78$jl4&8&<&<=O&PQG Nr1)r\rr3rTrSr2rE)rztype[ContextT] | Noner3r)' __future__rcollections.abcr dataclassesrrrtypingrr r langgraph.store.baser langgraph_sdk.auth.typesr typing_extensionsr rlanggraph._internal._constantsrrlanggraph.configrlanggraph.typesrrlanggraph.typingr__all__rrrr]r_rarDEFAULT_RUNTIMErr2r1r.rs"$11%%*-/C'4%  $d#**$*B $d#$$""8.$ 78#4E Z]Ogh]O]O@  &  r1