ޠjSSKrSSKrSSKJr SSKJr SSKJr SSKJ r SSK J r J r SSK Jr S\S S4S jrS \4S jrS \ 4S jrS \4S jrg)N)Any)RunnableConfig)var_child_runnable_config) BaseStore)CONFCONFIG_KEY_RUNTIME) StreamWritercreturncg)N)r s J/var/www/html/gaurav/venv/lib/python3.13/site-packages/langgraph/config.py_no_op_stream_writerr sc[RS:a'[R"5(a [ S5e[ R "5=n(aU$[ S5e![a N5f=f)N) z=Python 3.11 or later required to use this in an async contextz/Called get_config outside of a runnable context)sys version_infoasyncio current_task RuntimeErrorrget) var_configs r get_configrsu '! ##%%"S& /2244z4LMM    s%A%% A21A2cF[5[[R$)aAccess LangGraph store from inside a graph node or entrypoint task at runtime. Can be called from inside any [`StateGraph`][langgraph.graph.StateGraph] node or functional API [`task`][langgraph.func.task], as long as the `StateGraph` or the [`entrypoint`][langgraph.func.entrypoint] was initialized with a store, e.g.: ```python # with StateGraph graph = ( StateGraph(...) ... .compile(store=store) ) # or with entrypoint @entrypoint(store=store) def workflow(inputs): ... ``` !!! warning "Async with Python < 3.11" If you are using Python < 3.11 and are running LangGraph asynchronously, `get_store()` won't work since it uses [`contextvar`](https://docs.python.org/3/library/contextvars.html) propagation (only available in [Python >= 3.11](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task)). Example: Using with `StateGraph` ```python from typing_extensions import TypedDict from langgraph.graph import StateGraph, START from langgraph.store.memory import InMemoryStore from langgraph.config import get_store store = InMemoryStore() store.put(("values",), "foo", {"bar": 2}) class State(TypedDict): foo: int def my_node(state: State): my_store = get_store() stored_value = my_store.get(("values",), "foo").value["bar"] return {"foo": stored_value + 1} graph = ( StateGraph(State) .add_node(my_node) .add_edge(START, "my_node") .compile(store=store) ) graph.invoke({"foo": 1}) ``` ```pycon {"foo": 3} ``` Example: Using with functional API ```python from langgraph.func import entrypoint, task from langgraph.store.memory import InMemoryStore from langgraph.config import get_store store = InMemoryStore() store.put(("values",), "foo", {"bar": 2}) @task def my_task(value: int): my_store = get_store() stored_value = my_store.get(("values",), "foo").value["bar"] return stored_value + 1 @entrypoint(store=store) def workflow(value: int): return my_task(value).result() workflow.invoke(1) ``` ```pycon 3 ``` )rrrstorer rr get_storer sv < 0 1 7 77rcJ[5[[nUR$)aHAccess LangGraph [`StreamWriter`][langgraph.types.StreamWriter] from inside a graph node or entrypoint task at runtime. Can be called from inside any [`StateGraph`][langgraph.graph.StateGraph] node or functional API [`task`][langgraph.func.task]. !!! warning "Async with Python < 3.11" If you are using Python < 3.11 and are running LangGraph asynchronously, `get_stream_writer()` won't work since it uses [`contextvar`](https://docs.python.org/3/library/contextvars.html) propagation (only available in [Python >= 3.11](https://docs.python.org/3/library/asyncio-task.html#asyncio.create_task)). Example: Using with `StateGraph` ```python from typing_extensions import TypedDict from langgraph.graph import StateGraph, START from langgraph.config import get_stream_writer class State(TypedDict): foo: int def my_node(state: State): my_stream_writer = get_stream_writer() my_stream_writer({"custom_data": "Hello!"}) return {"foo": state["foo"] + 1} graph = ( StateGraph(State) .add_node(my_node) .add_edge(START, "my_node") .compile(store=store) ) for chunk in graph.stream({"foo": 1}, stream_mode="custom"): print(chunk) ``` ```pycon {"custom_data": "Hello!"} ``` Example: Using with functional API ```python from langgraph.func import entrypoint, task from langgraph.config import get_stream_writer @task def my_task(value: int): my_stream_writer = get_stream_writer() my_stream_writer({"custom_data": "Hello!"}) return value + 1 @entrypoint(store=store) def workflow(value: int): return my_task(value).result() for chunk in workflow.stream(1, stream_mode="custom"): print(chunk) ``` ```pycon {"custom_data": "Hello!"} ``` )rrr stream_writer)runtimes rget_stream_writerr"~s#Jl4 !34G   r)rrtypingrlangchain_core.runnablesrlangchain_core.runnables.configrlanggraph.store.baserlanggraph._internal._constantsrrlanggraph.typesr rrrr"r rrr)s[ 3E*C( C D  NN N[89[8|F!<F!r