diff --git a/libs/langgraph/langgraph/pregel/remote.py b/libs/langgraph/langgraph/pregel/remote.py index 1233cd63f..f0001659b 100644 --- a/libs/langgraph/langgraph/pregel/remote.py +++ b/libs/langgraph/langgraph/pregel/remote.py @@ -19,7 +19,6 @@ from langchain_core.runnables.graph import ( from langchain_core.runnables.graph import ( Node as DrawableNode, ) -from langchain_core.runnables.schema import StandardStreamEvent, StreamEvent from langgraph_sdk.client import ( LangGraphClient, SyncLangGraphClient, @@ -27,25 +26,34 @@ from langgraph_sdk.client import ( get_sync_client, ) from langgraph_sdk.schema import Checkpoint, ThreadState +from langgraph_sdk.schema import StreamMode as StreamModeSDK from typing_extensions import Self from langgraph.checkpoint.base import CheckpointMetadata +from langgraph.constants import INTERRUPT +from langgraph.errors import GraphInterrupt from langgraph.pregel.protocol import PregelProtocol from langgraph.pregel.types import All, PregelTask, StateSnapshot, StreamMode from langgraph.types import Interrupt from langgraph.utils.config import merge_configs +class RemoteException(Exception): + """Exception raised when an error occurs in the remote graph.""" + + pass + + class RemoteGraph(PregelProtocol, Runnable): def __init__( self, graph_id: str, - config: Optional[RunnableConfig] = None, url: Optional[str] = None, api_key: Optional[str] = None, headers: Optional[dict[str, str]] = None, client: Optional[LangGraphClient] = None, sync_client: Optional[SyncLangGraphClient] = None, + config: Optional[RunnableConfig] = None, ): """Specify `url`, `api_key`, and/or `headers` to create default sync and async clients. @@ -348,6 +356,37 @@ class RemoteGraph(PregelProtocol, Runnable): ) return self._get_config(response["checkpoint"]) + def _get_stream_modes( + self, + stream_mode: Optional[Union[StreamMode, list[StreamMode]]], + default: StreamMode = "updates", + ) -> tuple[list[StreamModeSDK], bool, bool]: + """Return a tuple of the final list of stream modes sent to the + remote graph and a boolean flag indicating if stream mode 'updates' + was present in the original list of stream modes. + + 'updates' mode is added to the list of stream modes so that interrupts + can be detected in the remote graph. + """ + updated_stream_modes: list[StreamMode] = [] + req_updates = False + req_single = True + # coerce to list, or add default stream mode + if stream_mode: + if isinstance(stream_mode, str): + updated_stream_modes.append(stream_mode) + else: + req_single = False + updated_stream_modes.extend(stream_mode) + else: + updated_stream_modes.append(default) + # add 'updates' mode if not present + if "updates" in updated_stream_modes: + req_updates = True + else: + updated_stream_modes.append("updates") + return (updated_stream_modes, req_updates, req_single) + def stream( self, input: Union[dict[str, Any], Any], @@ -360,18 +399,40 @@ class RemoteGraph(PregelProtocol, Runnable): ) -> Iterator[Union[dict[str, Any], Any]]: merged_config = merge_configs(self.config, config) sanitized_config = self._sanitize_config(merged_config) + stream_modes, req_updates, req_single = self._get_stream_modes(stream_mode) for chunk in self.sync_client.runs.stream( - thread_id=sanitized_config["configurable"]["thread_id"], + thread_id=cast(str, sanitized_config["configurable"]["thread_id"]), assistant_id=self.graph_id, input=input, config=sanitized_config, - stream_mode=stream_mode, # type: ignore - interrupt_before=interrupt_before, # type: ignore - interrupt_after=interrupt_after, # type: ignore + stream_mode=stream_modes, + interrupt_before=interrupt_before, + interrupt_after=interrupt_after, stream_subgraphs=subgraphs, + if_not_exists="create", ): - yield chunk + if chunk.event.startswith("updates"): + if isinstance(chunk.data, dict) and INTERRUPT in chunk.data: + raise GraphInterrupt(chunk.data[INTERRUPT]) + if not req_updates: + continue + elif chunk.event.startswith("error"): + raise RemoteException(chunk.data) + if subgraphs: + if "|" in chunk.event: + mode, ns_ = chunk.event.split("|", 1) + ns = tuple(ns_.split("|")) + else: + mode, ns = chunk.event, () + if req_single: + yield ns, chunk.data + else: + yield ns, mode, chunk.data + elif req_single: + yield chunk.data + else: + yield chunk async def astream( self, @@ -385,47 +446,40 @@ class RemoteGraph(PregelProtocol, Runnable): ) -> AsyncIterator[Union[dict[str, Any], Any]]: merged_config = merge_configs(self.config, config) sanitized_config = self._sanitize_config(merged_config) + stream_modes, req_updates, req_single = self._get_stream_modes(stream_mode) async for chunk in self.client.runs.stream( thread_id=sanitized_config["configurable"]["thread_id"], assistant_id=self.graph_id, input=input, config=sanitized_config, - stream_mode=stream_mode if stream_mode else "values", # type: ignore - interrupt_before=interrupt_before, # type: ignore - interrupt_after=interrupt_after, # type: ignore + stream_mode=stream_modes, + interrupt_before=interrupt_before, + interrupt_after=interrupt_after, stream_subgraphs=subgraphs, + if_not_exists="create", ): - yield chunk - - async def astream_events( - self, - input: Any, - config: Optional[RunnableConfig] = None, - **kwargs: Any, - ) -> AsyncIterator[StreamEvent]: - merged_config = merge_configs(self.config, config) - sanitized_config = self._sanitize_config(merged_config) - - # manually add 'events' to stream modes list - stream_mode: list[str] = kwargs.get("stream_mode", []) - if "events" not in stream_mode: - stream_mode.append("events") - - async for chunk in self.client.runs.stream( - thread_id=sanitized_config["configurable"]["thread_id"], - assistant_id=self.graph_id, - input=input, - config=sanitized_config, - stream_mode=stream_mode, # type: ignore - interrupt_before=kwargs.get("interrupt_before"), - interrupt_after=kwargs.get("interrupt_after"), - stream_subgraphs=kwargs.get("subgraphs", False), - ): - yield StandardStreamEvent( - event=chunk.event, - data=chunk.data, - ) + if chunk.event.startswith("updates"): + if isinstance(chunk.data, dict) and INTERRUPT in chunk.data: + raise GraphInterrupt(chunk.data[INTERRUPT]) + if not req_updates: + continue + elif chunk.event.startswith("error"): + raise RemoteException(chunk.data) + if subgraphs: + if "|" in chunk.event: + mode, ns_ = chunk.event.split("|", 1) + ns = tuple(ns_.split("|")) + else: + mode, ns = chunk.event, () + if req_single: + yield ns, chunk.data + else: + yield ns, mode, chunk.data + elif req_single: + yield chunk.data + else: + yield chunk def invoke( self, @@ -443,8 +497,9 @@ class RemoteGraph(PregelProtocol, Runnable): assistant_id=self.graph_id, input=input, config=sanitized_config, - interrupt_before=interrupt_before, # type: ignore - interrupt_after=interrupt_after, # type: ignore + interrupt_before=interrupt_before, + interrupt_after=interrupt_after, + if_not_exists="create", ) async def ainvoke( @@ -463,6 +518,7 @@ class RemoteGraph(PregelProtocol, Runnable): assistant_id=self.graph_id, input=input, config=sanitized_config, - interrupt_before=interrupt_before, # type: ignore - interrupt_after=interrupt_after, # type: ignore + interrupt_before=interrupt_before, + interrupt_after=interrupt_after, + if_not_exists="create", ) diff --git a/libs/langgraph/tests/test_remote_pregel.py b/libs/langgraph/tests/test_remote_graph.py similarity index 72% rename from libs/langgraph/tests/test_remote_pregel.py rename to libs/langgraph/tests/test_remote_graph.py index f2ca53583..46b3ab9a4 100644 --- a/libs/langgraph/tests/test_remote_pregel.py +++ b/libs/langgraph/tests/test_remote_graph.py @@ -7,7 +7,9 @@ from langchain_core.runnables.graph import ( from langchain_core.runnables.graph import ( Node as DrawableNode, ) +from langgraph_sdk.schema import StreamPart +from langgraph.errors import GraphInterrupt from langgraph.pregel.remote import RemoteGraph from langgraph.pregel.types import StateSnapshot @@ -473,17 +475,97 @@ def test_stream(): # set up test mock_sync_client = MagicMock() mock_sync_client.runs.stream.return_value = [ - {"chunk": "data1"}, - {"chunk": "data2"}, - {"chunk": "data3"}, + StreamPart(event="values", data={"chunk": "data1"}), + StreamPart(event="values", data={"chunk": "data2"}), + StreamPart(event="values", data={"chunk": "data3"}), + StreamPart(event="updates", data={"chunk": "data4"}), + StreamPart(event="updates", data={"__interrupt__": ()}), ] # call method / assertions remote_pregel = RemoteGraph(sync_client=mock_sync_client, graph_id="test_graph_id") - config = {"configurable": {"thread_id": "thread_1"}} - result = list(remote_pregel.stream({"input": "data"}, config)) - assert result == [{"chunk": "data1"}, {"chunk": "data2"}, {"chunk": "data3"}] + # stream modes doesn't include 'updates' + stream_parts = [] + with pytest.raises(GraphInterrupt): + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode="values", + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + {"chunk": "data1"}, + {"chunk": "data2"}, + {"chunk": "data3"}, + ] + + mock_sync_client.runs.stream.return_value = [ + StreamPart(event="updates", data={"chunk": "data3"}), + StreamPart(event="updates", data={"chunk": "data4"}), + StreamPart(event="updates", data={"__interrupt__": ()}), + ] + + # default stream_mode is updates + stream_parts = [] + with pytest.raises(GraphInterrupt): + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + {"chunk": "data3"}, + {"chunk": "data4"}, + ] + + # list stream_mode includes mode names + stream_parts = [] + with pytest.raises(GraphInterrupt): + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ("updates", {"chunk": "data3"}), + ("updates", {"chunk": "data4"}), + ] + + # subgraphs + list modes + stream_parts = [] + with pytest.raises(GraphInterrupt): + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ((), "updates", {"chunk": "data3"}), + ((), "updates", {"chunk": "data4"}), + ] + + # subgraphs + single mode + stream_parts = [] + with pytest.raises(GraphInterrupt): + for stream_part in remote_pregel.stream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ((), {"chunk": "data3"}), + ((), {"chunk": "data4"}), + ] @pytest.mark.anyio @@ -492,20 +574,139 @@ async def test_astream(): mock_async_client = MagicMock() async_iter = MagicMock() async_iter.__aiter__.return_value = [ - {"chunk": "data1"}, - {"chunk": "data2"}, - {"chunk": "data3"}, + StreamPart(event="values", data={"chunk": "data1"}), + StreamPart(event="values", data={"chunk": "data2"}), + StreamPart(event="values", data={"chunk": "data3"}), + StreamPart(event="updates", data={"chunk": "data4"}), + StreamPart(event="updates", data={"__interrupt__": ()}), ] mock_async_client.runs.stream.return_value = async_iter # call method / assertions remote_pregel = RemoteGraph(client=mock_async_client, graph_id="test_graph_id") - config = {"configurable": {"thread_id": "thread_1"}} - chunks = [] - async for chunk in remote_pregel.astream({"input": "data"}, config): - chunks.append(chunk) - assert chunks == [{"chunk": "data1"}, {"chunk": "data2"}, {"chunk": "data3"}] + # stream modes doesn't include 'updates' + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode="values", + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + {"chunk": "data1"}, + {"chunk": "data2"}, + {"chunk": "data3"}, + ] + + async_iter = MagicMock() + async_iter.__aiter__.return_value = [ + StreamPart(event="updates", data={"chunk": "data3"}), + StreamPart(event="updates", data={"chunk": "data4"}), + StreamPart(event="updates", data={"__interrupt__": ()}), + ] + mock_async_client.runs.stream.return_value = async_iter + + # default stream_mode is updates + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + {"chunk": "data3"}, + {"chunk": "data4"}, + ] + + # list stream_mode includes mode names + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ("updates", {"chunk": "data3"}), + ("updates", {"chunk": "data4"}), + ] + + # subgraphs + list modes + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ((), "updates", {"chunk": "data3"}), + ((), "updates", {"chunk": "data4"}), + ] + + # subgraphs + single mode + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + ((), {"chunk": "data3"}), + ((), {"chunk": "data4"}), + ] + + async_iter = MagicMock() + async_iter.__aiter__.return_value = [ + StreamPart(event="updates|my|subgraph", data={"chunk": "data3"}), + StreamPart(event="updates|hello|subgraph", data={"chunk": "data4"}), + StreamPart(event="updates|bye|subgraph", data={"__interrupt__": ()}), + ] + mock_async_client.runs.stream.return_value = async_iter + + # subgraphs + list modes + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + stream_mode=["updates"], + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + (("my", "subgraph"), "updates", {"chunk": "data3"}), + (("hello", "subgraph"), "updates", {"chunk": "data4"}), + ] + + # subgraphs + single mode + stream_parts = [] + with pytest.raises(GraphInterrupt): + async for stream_part in remote_pregel.astream( + {"input": "data"}, + config={"configurable": {"thread_id": "thread_1"}}, + subgraphs=True, + ): + stream_parts.append(stream_part) + + assert stream_parts == [ + (("my", "subgraph"), {"chunk": "data3"}), + (("hello", "subgraph"), {"chunk": "data4"}), + ] def test_invoke(): @@ -572,7 +773,7 @@ async def test_langgraph_cloud_integration(): "messages": [ { "role": "human", - "content": "Hello world!", + "content": "What's the weather in SF?", } ] } @@ -580,7 +781,8 @@ async def test_langgraph_cloud_integration(): # test invoke response = app.invoke( input, - config={"configurable": {"thread_id": "2dc3e3e7-39ac-4597-aa57-4404b944e82a"}}, + config={"configurable": {"thread_id": "39a6104a-34e7-4f83-929c-d9eb163003c9"}}, + interrupt_before=["agent"], ) print("response:", response["messages"][-1].content) diff --git a/libs/sdk-py/langgraph_sdk/client.py b/libs/sdk-py/langgraph_sdk/client.py index f89a0f6a5..239f4cf2a 100644 --- a/libs/sdk-py/langgraph_sdk/client.py +++ b/libs/sdk-py/langgraph_sdk/client.py @@ -39,6 +39,7 @@ from langgraph_sdk.schema import ( Cron, DisconnectMode, GraphSchema, + IfNotExists, Item, Json, ListNamespaceResponse, @@ -1169,18 +1170,19 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> AsyncIterator[StreamPart]: ... @@ -1191,13 +1193,13 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, @@ -1210,19 +1212,20 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> AsyncIterator[StreamPart]: """Create a run and stream the results. @@ -1248,6 +1251,8 @@ class RunsClient: webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -1293,6 +1298,7 @@ class RunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_disconnect": on_disconnect, "on_completion": on_completion, "after_seconds": after_seconds, @@ -1313,12 +1319,12 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, after_seconds: Optional[int] = None, @@ -1331,16 +1337,17 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Run: ... @@ -1350,16 +1357,17 @@ class RunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, on_completion: Optional[OnCompletionBehavior] = None, after_seconds: Optional[int] = None, ) -> Run: @@ -1383,6 +1391,8 @@ class RunsClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -1466,6 +1476,7 @@ class RunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_completion": on_completion, "after_seconds": after_seconds, } @@ -1495,11 +1506,12 @@ class RunsClient: config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: ... @@ -1512,8 +1524,8 @@ class RunsClient: input: Optional[dict] = None, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, @@ -1530,12 +1542,13 @@ class RunsClient: config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: """Create a run, wait until it finishes and return the final state. @@ -1558,6 +1571,8 @@ class RunsClient: Must be one of 'delete' or 'keep'. multitask_strategy: Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -1619,6 +1634,7 @@ class RunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_disconnect": on_disconnect, "on_completion": on_completion, "after_seconds": after_seconds, @@ -3232,18 +3248,19 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Iterator[StreamPart]: ... @@ -3254,13 +3271,13 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, @@ -3273,19 +3290,20 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, - feedback_keys: Optional[list[str]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, + feedback_keys: Optional[Sequence[str]] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Iterator[StreamPart]: """Create a run and stream the results. @@ -3311,6 +3329,8 @@ class SyncRunsClient: webhook: Webhook to call after LangGraph API call is done. multitask_strategy: Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -3356,6 +3376,7 @@ class SyncRunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_disconnect": on_disconnect, "on_completion": on_completion, "after_seconds": after_seconds, @@ -3376,12 +3397,12 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_completion: Optional[OnCompletionBehavior] = None, after_seconds: Optional[int] = None, @@ -3394,16 +3415,17 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Run: ... @@ -3413,17 +3435,18 @@ class SyncRunsClient: assistant_id: str, *, input: Optional[dict] = None, - stream_mode: Union[StreamMode, list[StreamMode]] = "values", + stream_mode: Union[StreamMode, Sequence[StreamMode]] = "values", stream_subgraphs: bool = False, metadata: Optional[dict] = None, config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, multitask_strategy: Optional[MultitaskStrategy] = None, on_completion: Optional[OnCompletionBehavior] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Run: """Create a background run. @@ -3446,6 +3469,8 @@ class SyncRunsClient: Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. on_completion: Whether to delete or keep the thread created for a stateless run. Must be one of 'delete' or 'keep'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -3529,6 +3554,7 @@ class SyncRunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_completion": on_completion, "after_seconds": after_seconds, } @@ -3558,11 +3584,12 @@ class SyncRunsClient: config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: ... @@ -3575,8 +3602,8 @@ class SyncRunsClient: input: Optional[dict] = None, metadata: Optional[dict] = None, config: Optional[Config] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, @@ -3593,12 +3620,13 @@ class SyncRunsClient: config: Optional[Config] = None, checkpoint: Optional[Checkpoint] = None, checkpoint_id: Optional[str] = None, - interrupt_before: Optional[Union[All, list[str]]] = None, - interrupt_after: Optional[Union[All, list[str]]] = None, + interrupt_before: Optional[Union[All, Sequence[str]]] = None, + interrupt_after: Optional[Union[All, Sequence[str]]] = None, webhook: Optional[str] = None, on_disconnect: Optional[DisconnectMode] = None, on_completion: Optional[OnCompletionBehavior] = None, multitask_strategy: Optional[MultitaskStrategy] = None, + if_not_exists: Optional[IfNotExists] = None, after_seconds: Optional[int] = None, ) -> Union[list[dict], dict[str, Any]]: """Create a run, wait until it finishes and return the final state. @@ -3621,6 +3649,8 @@ class SyncRunsClient: Must be one of 'delete' or 'keep'. multitask_strategy: Multitask strategy to use. Must be one of 'reject', 'interrupt', 'rollback', or 'enqueue'. + if_not_exists: How to handle missing thread. Defaults to 'reject'. + Must be either 'reject' (raise error if missing), or 'create' (create new thread). after_seconds: The number of seconds to wait before starting the run. Use to schedule future runs. @@ -3682,6 +3712,7 @@ class SyncRunsClient: "checkpoint": checkpoint, "checkpoint_id": checkpoint_id, "multitask_strategy": multitask_strategy, + "if_not_exists": if_not_exists, "on_disconnect": on_disconnect, "on_completion": on_completion, "after_seconds": after_seconds, diff --git a/libs/sdk-py/langgraph_sdk/schema.py b/libs/sdk-py/langgraph_sdk/schema.py index 4e5314692..827c2e33a 100644 --- a/libs/sdk-py/langgraph_sdk/schema.py +++ b/libs/sdk-py/langgraph_sdk/schema.py @@ -69,6 +69,13 @@ Defines action after completion: All = Literal["*"] """Represents a wildcard or 'all' selector.""" +IfNotExists = Literal["create", "reject"] +""" +Specifies behavior if the thread doesn't exist: +- "create": Create a new thread if it doesn't exist. +- "reject": Reject the operation if the thread doesn't exist. +""" + class Config(TypedDict, total=False): """Configuration options for a call."""