A'hddlZddlZddlmZddlmZddlmZddlm Z m Z m Z ddl m Z ddlmZded dfd Zd efd Zd e fd Zd efd Zy)N)Any)RunnableConfig)var_child_runnable_config)CONFCONFIG_KEY_STORECONFIG_KEY_STREAM_WRITER) BaseStore) StreamWritercreturncy)N)r s W/home/kushmeetdev/Regenta/Chatbot/venv/lib/python3.12/site-packages/langgraph/config.py_no_op_stream_writerr sctjdkr! tjr t d t j x}r|St d#t$rY.wxYw)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_configrsr '! ##%"S& /2244z4LMM    sA A$#A$c6t}|ttS)a Access 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 ``` )rrrconfigs r get_storer sj\F $<( ))rcXt}|tjttS)aAccess 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!'} ``` )rrrrrrs rget_stream_writerr"ys%~\F $<  46J KKr)rrtypingrlangchain_core.runnablesrlangchain_core.runnables.configrlanggraph.constantsrrrlanggraph.store.baser langgraph.typesr rrr r"rrrr)s` 3EPP*( C D  NN NV*9V*r@L<@Lr