tjO( ddlmZddlmZddlmZmZmZddlm Z m Z m Z ddl m Z ddlmZddlmZmZddlmZmZdd lmZdd lmZmZdd lmZd Zed d GddZed d GddZGddZ d'dZ!d(dZ"Gddee edZ#ed)ieGdd e eZ$e$d!d!e!e"d!d!d!"Z%d*d+d&Z&d!S),) annotations)Callable) dataclassfieldreplace)AnyGenericcast) BaseStore)BaseUser) TypedDictUnpack)CONFCONFIG_KEY_RUNTIME) get_config) _DC_KWARGS StreamWriter)ContextT)r ExecutionInfo RunControlRuntime ServerInfo get_runtimeT)frozenslotsceZdZUdZded< ded< ded< dZded< dZded < d Zd ed < dZd ed< ddZ dS)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_time overridesrreturnc t|fi|S)zAReturn a new execution info object with selected fields replaced.rselfr(s O/var/www/html/CA-Chatbot/venv/lib/python3.11/site-packages/langgraph/runtime.pypatchzExecutionInfo.patch7st))y))))r(rr)r) __name__ __module__ __qualname____doc____annotations__r"r#r&r'r/r0r.rrsYY6=LLL0 I    HFAL<,00000F******r0rc:eZdZUdZded< ded< dZded<dS)rzmMetadata injected by LangGraph Server. None when running open-source LangGraph without LangSmith deployments.r assistant_idgraph_idNzBaseUser | Noneuser)r1r2r3r4r5r:r6r0r.rr<sHww5MMM1 D    r0rcXeZdZdZdZddZddd Zedd Zedd Z dS)raRun-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_reasonr)Nonecd|_dSNr<r-s r.__init__zRunControl.__init__\s)-r0shutdownreasonrc||_dSr@r<)r-rDs r. request_drainzRunControl.request_drain_s#r0boolc|jduSr@r<rAs r.drain_requestedzRunControl.drain_requestedbs!--r0r!c|jSr@r<rAs r. drain_reasonzRunControl.drain_reasonfs !!r0Nr)r>)rC)rDrr)r>r)rGr)r!) r1r2r3r4 __slots__rBrFpropertyrIrKr6r0r.rrOs#I....$$$$$...X."""X"""r0r_rr)r>cdSr@r6)rQs r._no_op_stream_writerrSkr0cdSr@r6r6r0r._no_op_heartbeatrVnrTr0c`eZdZUded<ded<ded<ded<d ed <d ed <d ed<ded<dS)_RuntimeOverridesrcontextBaseStore | Nonestorer stream_writerCallable[[], None] heartbeatrpreviousrexecution_infoServerInfo | None server_infoRunControl | NonecontrolN)r1r2r3r5r6r0r.rXrXqso!!!!MMM!!!!""""r0rXF)totalc|eZdZUdZedZded< edZded< eeZ ded < ee Z d ed < edZ d ed < edZ ded< edZded< edZded< d dZd!dZd"dZed#dZed$dZdS)%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)defaultrrYrZr[rr\r]r^rr_zExecutionInfo | Noner`rarbrcrdotherRuntime[ContextT]r)c 6t|jp|j|jp|j|jtur|jn|j|jt ur|jn|j|j|jn|j|jp|j|j p|j |j p|j S)zMerge two runtimes together. If a value is not provided in the other runtime, the value from the current runtime is used. N)rYr[r\r^r_r`rbrd) rrYr[r\rSr^rVr_r`rbrd)r-rhs r.mergez Runtime.merges M1T\++"*>>> --#&666oo&+n&+>+>N>>>>2&+U4%8%8%8K8888w!&t!4!4!4G4444     (****     SSSXSOOOXOOOr0rN)rYr[r\r^r_r`rdcontext_schematype[ContextT] | Nonerictttttt }|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)ruruntimes r.rr(s578$jll4&8&<&<=O&P&PQQG Nr0)rQrr)r>rLr6r@)rurvr)ri)' __future__rcollections.abcr dataclassesrrrtypingrr r langgraph.store.baser langgraph_sdk.auth.typesr typing_extensionsr rlanggraph._internal._constantsrrlanggraph.configrlanggraph.typesrrlanggraph.typingr__all__rrrrSrVrXrDEFAULT_RUNTIMErr6r0r.rs""""""$$$$$$1111111111%%%%%%%%%%******------////////CCCCCCCC''''''44444444%%%%%%  $d###*******$#*B $d###$#$""""""""8.---$### 78#4E Z]O]O]O]O]Ogh]O]O]O@'  &  r0