
    <i                         S SK r S SKrS SKJr  S SKJr  S SKJr  S SKJ	r	  S SK
JrJr  S SKJr  S\S	S4S
 jrS	\4S jrS	\	4S jrS	\4S jrg)    N)Any)RunnableConfig)var_child_runnable_config)	BaseStore)CONFCONFIG_KEY_RUNTIME)StreamWritercreturnc                     g )N )r
   s    `/home/dmtnaga/Documents/work/airagagent/rag_env/lib/python3.13/site-packages/langgraph/config.py_no_op_stream_writerr      s        c                      [         R                  S:  a'   [        R                  " 5       (       a  [	        S5      e [
        R                  " 5       =n (       a  U $ [	        S5      e! [         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_infoasynciocurrent_taskRuntimeErrorr   get)
var_configs    r   
get_configr      su    
'!	##%%"S  & /2244z4LMM  		s   %A% %
A21A2c                  F    [        5       [           [           R                  $ )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
    ```
)r   r   r   storer   r   r   	get_storer       s    v <01777r   c                  J    [        5       [           [           n U R                  $ )a>  Access 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!"}
    ```
)r   r   r   stream_writer)runtimes    r   get_stream_writerr"   ~   s#    J l4 !34G   r   )r   r   typingr   langchain_core.runnablesr   langchain_core.runnables.configr   langgraph.store.baser   langgraph._internal._constantsr   r   langgraph.typesr	   r   r   r   r"   r   r   r   <module>r)      s[     
  3 E * C (	C 	D 	NN N[89 [8|F!< F!r   