Introduce "tasks" and "checkpoints" stream modes

- These are split out of "debug" stream mode, which is now an alias for ["tasks", "checkpoints"]

libs/langgraph/langgraph/pregel/__init__.py

@@ -2236,6 +2236,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
             stream_mode = ["values"]
         elif stream_mode is None:
             stream_mode = self.stream_mode
+        elif stream_mode == "debug":
+            stream_mode = ["checkpoints", "tasks"]
         if not isinstance(stream_mode, list):
             stream_mode = [stream_mode]
         if self.checkpointer is False:
@@ -2298,7 +2300,8 @@ class Pregel(PregelProtocol[StateT, InputT, OutputT], Generic[StateT, InputT, Ou
                 - `"custom"`: Emit custom data from inside nodes or tasks using `StreamWriter`.
                 - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
                     Will be emitted as 2-tuples `(LLM token, metadata)`.
-                - `"debug"`: Emit debug events with as much information as possible for each step.
+                - `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
+                - `"tasks"`: Emit events when tasks start and finish, including their results and errors.
 
                 You can pass a list as the `stream_mode` parameter to stream multiple modes at once.
                 The streamed outputs will be tuples of `(mode, data)`.

libs/langgraph/langgraph/pregel/loop.py

@@ -119,6 +119,7 @@ from langgraph.types import (
     PregelScratchpad,
     RetryPolicy,
     StreamChunk,
+    StreamMode,
     StreamProtocol,
 )
 from langgraph.utils.config import patch_configurable
@@ -422,7 +423,7 @@ class PregelLoop:
             ),
         ):
             # produce debug output
-            self._emit("debug", map_debug_tasks, self.step, [pushed])
+            self._emit("tasks", map_debug_tasks, self.step, [pushed])
             # debug flag
             if self.debug:
                 print_step_tasks(self.step, [pushed])
@@ -472,7 +473,7 @@ class PregelLoop:
         # produce debug output
         if self._checkpointer_put_after_previous is not None:
             self._emit(
-                "debug",
+                "checkpoints",
                 map_debug_checkpoint,
                 self.step - 1,  # printing checkpoint for previous step
                 {
@@ -509,7 +510,7 @@ class PregelLoop:
             raise GraphInterrupt()
 
         # produce debug output
-        self._emit("debug", map_debug_tasks, self.step, self.tasks.values())
+        self._emit("tasks", map_debug_tasks, self.step, self.tasks.values())
 
         # debug flag
         if self.debug:
@@ -834,7 +835,7 @@ class PregelLoop:
 
     def _emit(
         self,
-        mode: str,
+        mode: StreamMode,
         values: Callable[P, Iterator[Any]],
         *args: P.args,
         **kwargs: P.kwargs,
@@ -885,7 +886,7 @@ class PregelLoop:
                 )
             if not cached:
                 self._emit(
-                    "debug",
+                    "tasks",
                     map_debug_task_results,
                     self.step,
                     (task, writes),

libs/langgraph/langgraph/types.py

@@ -46,7 +46,7 @@ Checkpointer = Union[None, bool, BaseCheckpointSaver]
 - False disables checkpointing, even if the parent graph has a checkpointer.
 - None inherits checkpointer from the parent graph."""
 
-StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
+StreamMode = Literal["values", "updates", "checkpoints", "tasks", "messages", "custom"]
 """How the stream method should emit outputs.
 
 - `"values"`: Emit all values in the state after each step, including interrupts.
@@ -55,7 +55,8 @@ StreamMode = Literal["values", "updates", "debug", "messages", "custom"]
     If multiple updates are made in the same step (e.g. multiple nodes are run) then those updates are emitted separately.
 - `"custom"`: Emit custom data using from inside nodes or tasks using `StreamWriter`.
 - `"messages"`: Emit LLM messages token-by-token together with metadata for any LLM invocations inside nodes or tasks.
-- `"debug"`: Emit debug events with as much information as possible for each step.
+- `"checkpoints"`: Emit an event when a checkpoint is created, in the same format as returned by get_state().
+- `"tasks"`: Emit events when tasks start and finish, including their results and errors.
 """
 
 StreamWriter = Callable[[Any], None]