\ǻi%4ddlmZddlmZmZmZddlmZmZm Z m Z ddl m Z ddl mZmZddlmZmZddlmZddlmZmZdd lmZd ZGd d e Zdd ZGddeeedZedieGddeeZeddedeZdddZ y)) annotations) dataclassfieldreplace)AnyGeneric NamedTuplecast) BaseStore) TypedDictUnpack)CONFCONFIG_KEY_RUNTIME) get_config) _DC_KWARGS StreamWriter)ContextT) ExecutionInfoRuntime get_runtimec:eZdZUdZdZded< dZded< d dZy) rzORead-only execution info/metadata for the execution of current thread/run/node.int node_attemptNz float | Nonenode_first_attempt_timec &|jdi|S)zAReturn a new execution info object with selected fields replaced.)_replaceself overridess O/opt/lhia/marcimex/agent/venv/lib/python3.12/site-packages/langgraph/runtime.pypatchzExecutionInfo.patchst}})y)))r!rreturnr)__name__ __module__ __qualname____doc__r__annotations__rr#rr$r"rrs%YL#<,0\0F*r$rcyNr)_s r"_no_op_stream_writerr.sr$c@eZdZUded<ded<ded<ded<d ed <y ) _RuntimeOverridesrcontextBaseStore | Nonestorer stream_writerrpreviousrexecution_infoN)r&r'r(r*rr$r"r0r0"s  M!!r$r0F)totalceZdZUdZedZded< edZded< eeZ ded < edZ d ed < ee Z d ed< ddZ ddZddZy)ra Convenience 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)defaultrr1r2r3rr4rr5)default_factoryrr6c*t|jxs |j|jxs |j|jtur |jn |j|j |j n |j |j S)zMerge two runtimes together. If a value is not provided in the other runtime, the value from the current runtime is used. r1r3r4r5r6)rr1r3r4r.r5r6)r others r"mergez Runtime.mergess MM1T\\+++""*>> --##&+nn&r@rBrr$r"rr*sFPd+GX+5$D1E 1C"'0D"EM<E4$'Hc' %*-$HNMHL  *>* *  r$rNr<cxtttttj t }|S)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"rrs-78$jl4&8&<&<=O&PQG Nr$)r-rr%Nonerr,)rFztype[ContextT] | Noner%rC)! __future__r dataclassesrrrtypingrrr r langgraph.store.baser typing_extensionsr r langgraph._internal._constantsrrlanggraph.configrlanggraph.typesrrlanggraph.typingr__all__rr.r0rDEFAULT_RUNTIMErrr$r"rTs"1111*/C'4% 5 *J *." 78#4E" Zw ghw w t  &  ? r$