codekingpro/portable-devtools

1�

2��j����SS
KrSS
K
r
SSKJr SSKJr SSKJr SSKJ	r	 SSK3JrJr SSK
Jr S\S	S
4S4jrS	\4SjrS	\	4SjrS	\4S
jrg
)�N)
�Any)
�RunnableConfig)
�var_child_runnable_config)
�	BaseStore)�CONF�CONFIG_KEY_RUNTIME)
�StreamWriter�
c�returnc

��g)
N�)
r5s
 �VD:\code\apps\devtools\python\user_packages\Python313\site-packages\langgraph/config.py�_no_op_stream_writerr
s���c���[RS
:a'[R"5(a[	S5
e
[6R"5=
n(aU$[	S5
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)
�7var_configs
 r�8get_configrsu��9���'�!�	��#�#�%�%�"�S���&�/�2�2�4�4�z�4����L�M�M���
	��
	�s�%A%�%10A2�1
A2c�F�[
5[[R$)
a�Access LangGraph store from inside a graph node or entrypoint task at runtime.11 12Can be called from inside any [`StateGraph`][langgraph.graph.StateGraph] node or13functional API [`task`][langgraph.func.task], as long as the `StateGraph` or the [`entrypoint`][langgraph.func.entrypoint]14was initialized with a store, e.g.:15 16```python17# with StateGraph18graph = (19    StateGraph(...)20    ...21    .compile(store=store)22)23 24# or with entrypoint25@entrypoint(store=store)26def workflow(inputs):27    ...28```29 30!!! warning "Async with Python < 3.11"31 32    If you are using Python < 3.11 and are running LangGraph asynchronously,33    `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)).34 35 36Example: Using with `StateGraph`37    ```python38    from typing_extensions import TypedDict39    from langgraph.graph import StateGraph, START40    from langgraph.store.memory import InMemoryStore41    from langgraph.config import get_store42 43    store = InMemoryStore()44    store.put(("values",), "foo", {"bar": 2})45 46 47    class State(TypedDict):48        foo: int49 50 51    def my_node(state: State):52        my_store = get_store()53        stored_value = my_store.get(("values",), "foo").value["bar"]54        return {"foo": stored_value + 1}55 56 57    graph = (58        StateGraph(State)59        .add_node(my_node)60        .add_edge(START, "my_node")61        .compile(store=store)62    )63 64    graph.invoke({"foo": 1})65    ```66 67    ```pycon68    {"foo": 3}69    ```70 71Example: Using with functional API72    ```python73    from langgraph.func import entrypoint, task74    from langgraph.store.memory import InMemoryStore75    from langgraph.config import get_store76 77    store = InMemoryStore()78    store.put(("values",), "foo", {"bar": 2})79 80 81    @task82    def my_task(value: int):83        my_store = get_store()84        stored_value = my_store.get(("values",), "foo").value["bar"]85        return stored_value + 186 87 88    @entrypoint(store=store)89    def workflow(value: int):90        return my_task(value).result()91 92 93    workflow.invoke(1)94    ```95 96    ```pycon97    398    ```99)rrr�storer
rr�	get_storer s��v�<���0�1�7�7�7rc�J�[
5[[nUR$)
aHAccess LangGraph [`StreamWriter`][langgraph.types.StreamWriter] from inside a graph node or entrypoint task at runtime.100 101Can be called from inside any [`StateGraph`][langgraph.graph.StateGraph] node or102functional API [`task`][langgraph.func.task].103 104!!! warning "Async with Python < 3.11"105 106    If you are using Python < 3.11 and are running LangGraph asynchronously,107    `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)).108 109Example: Using with `StateGraph`110    ```python111    from typing_extensions import TypedDict112    from langgraph.graph import StateGraph, START113    from langgraph.config import get_stream_writer114 115 116    class State(TypedDict):117        foo: int118 119 120    def my_node(state: State):121        my_stream_writer = get_stream_writer()122        my_stream_writer({"custom_data": "Hello!"})123        return {"foo": state["foo"] + 1}124 125 126    graph = (127        StateGraph(State)128        .add_node(my_node)129        .add_edge(START, "my_node")130        .compile(store=store)131    )132 133    for chunk in graph.stream({"foo": 1}, stream_mode="custom"):134        print(chunk)135    ```136 137    ```pycon138    {"custom_data": "Hello!"}139    ```140 141Example: Using with functional API142    ```python143    from langgraph.func import entrypoint, task144    from langgraph.config import get_stream_writer145 146 147    @task148    def my_task(value: int):149        my_stream_writer = get_stream_writer()150        my_stream_writer({"custom_data": "Hello!"})151        return value + 1152 153 154    @entrypoint(store=store)155    def workflow(value: int):156        return my_task(value).result()157 158 159    for chunk in workflow.stream(1, stream_mode="custom"):160        print(chunk)161    ```162 163    ```pycon164    {"custom_data": "Hello!"}165    ```166)rrr�
stream_writer)
�runtimes
 r�get_stream_writerr"~s#��J�l�4� �!3�4�G�� � � r)rr�typingr�langchain_core.runnablesr�langchain_core.runnables.configr�langgraph.store.baser�langgraph._internal._constantsrr�langgraph.typesr	rrrr"r
rr�<module>r)
s[�


��167��3�E�*�C�(�

	�C�

	�D�

	�
N
�N�
N
�[

8�9�[

8�|F

!�<�F

!r