Lint

libs/langgraph/langgraph/pregel/__init__.py

@@ -553,20 +553,22 @@ class Pregel(PregelProtocol):
             self.validate()
 
     def get_graph(
-        self, config: RunnableConfig | None = None, *, xray: int | bool = False
+        self, config: Optional[RunnableConfig] = None, *, xray: Union[int, bool] = False
     ) -> Graph:
         raise NotImplementedError
 
     async def aget_graph(
-        self, config: RunnableConfig | None = None, *, xray: int | bool = False
+        self, config: Optional[RunnableConfig] = None, *, xray: Union[int, bool] = False
     ) -> Graph:
         raise NotImplementedError
 
-    def copy(self, update: dict[str, Any] | None = None) -> Self:
+    def copy(self, update: Optional[dict[str, Any]] = None) -> Self:
         attrs = {**self.__dict__, **(update or {})}
         return self.__class__(**attrs)
 
-    def with_config(self, config: RunnableConfig | None = None, **kwargs: Any) -> Self:
+    def with_config(
+        self, config: Optional[RunnableConfig] = None, **kwargs: Any
+    ) -> Self:
         return self.copy(
             {"config": merge_configs(self.config, config, cast(RunnableConfig, kwargs))}
         )

libs/langgraph/langgraph/pregel/loop.py

@@ -582,7 +582,7 @@ class PregelLoop(LoopProtocol):
                 else:
                     task.writes.append((k, v))
 
-    def _first(self, *, input_keys: Union[str, Sequence[str]]) -> set[str] | None:
+    def _first(self, *, input_keys: Union[str, Sequence[str]]) -> Union[set[str]]:
         # resuming from previous checkpoint requires
         # - finding a previous checkpoint
         # - receiving None input (outer graph) or RESUMING flag (subgraph)