From 5b73e38c3888321c97c050c18ff76ffcb2aaf8e6 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 8 Apr 2025 10:44:12 -0700 Subject: [PATCH 1/3] Make compatible with langchain-core 0.1 by conditionally importing _StreamingCallbackHandler --- libs/langgraph/langgraph/pregel/__init__.py | 24 ++++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 545c8faf1..1b28260d1 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -39,7 +39,11 @@ from langchain_core.runnables.utils import ( ConfigurableFieldSpec, get_unique_config_specs, ) -from langchain_core.tracers._streaming import _StreamingCallbackHandler + +try: + from langchain_core.tracers._streaming import _StreamingCallbackHandler +except ImportError: + _StreamingCallbackHandler = None from pydantic import BaseModel from typing_extensions import Self @@ -2529,13 +2533,17 @@ class Pregel(PregelProtocol): run_id=config.get("run_id"), ) # if running from astream_log() run each proc with streaming - do_stream = next( - ( - cast(_StreamingCallbackHandler, h) - for h in run_manager.handlers - if isinstance(h, _StreamingCallbackHandler) - ), - None, + do_stream = ( + next( + ( + cast(_StreamingCallbackHandler, h) # type: ignore + for h in run_manager.handlers + if isinstance(h, _StreamingCallbackHandler) + ), + None, + ) + if _StreamingCallbackHandler is not None + else False ) try: # assign defaults From 0b3bf37a558457f3e166f8a7227b446244e99efa Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 8 Apr 2025 10:47:55 -0700 Subject: [PATCH 2/3] Fix the rest --- libs/langgraph/langgraph/pregel/__init__.py | 10 +++--- libs/langgraph/langgraph/pregel/messages.py | 8 ++++- libs/langgraph/langgraph/utils/runnable.py | 38 +++++++++++++-------- 3 files changed, 35 insertions(+), 21 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index 1b28260d1..cb0a89a79 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -39,11 +39,6 @@ from langchain_core.runnables.utils import ( ConfigurableFieldSpec, get_unique_config_specs, ) - -try: - from langchain_core.tracers._streaming import _StreamingCallbackHandler -except ImportError: - _StreamingCallbackHandler = None from pydantic import BaseModel from typing_extensions import Self @@ -129,6 +124,11 @@ from langgraph.utils.fields import get_enhanced_type_hints from langgraph.utils.pydantic import create_model, is_supported_by_pydantic from langgraph.utils.queue import AsyncQueue, SyncQueue # type: ignore[attr-defined] +try: + from langchain_core.tracers._streaming import _StreamingCallbackHandler +except ImportError: + _StreamingCallbackHandler = None + WriteValue = Union[Callable[[Input], Output], Any] diff --git a/libs/langgraph/langgraph/pregel/messages.py b/libs/langgraph/langgraph/pregel/messages.py index 867012fa6..d53598220 100644 --- a/libs/langgraph/langgraph/pregel/messages.py +++ b/libs/langgraph/langgraph/pregel/messages.py @@ -7,6 +7,7 @@ from typing import ( List, Optional, Sequence, + TypeVar, Union, cast, ) @@ -15,11 +16,16 @@ from uuid import UUID, uuid4 from langchain_core.callbacks import BaseCallbackHandler from langchain_core.messages import BaseMessage from langchain_core.outputs import ChatGenerationChunk, LLMResult -from langchain_core.tracers._streaming import T, _StreamingCallbackHandler from langgraph.constants import NS_SEP, TAG_HIDDEN, TAG_NOSTREAM from langgraph.types import StreamChunk +try: + from langchain_core.tracers._streaming import _StreamingCallbackHandler +except ImportError: + _StreamingCallbackHandler = object + +T = TypeVar("T") Meta = tuple[tuple[str, ...], dict[str, Any]] diff --git a/libs/langgraph/langgraph/utils/runnable.py b/libs/langgraph/langgraph/utils/runnable.py index f21420ad0..f51da12d9 100644 --- a/libs/langgraph/langgraph/utils/runnable.py +++ b/libs/langgraph/langgraph/utils/runnable.py @@ -36,7 +36,6 @@ from langchain_core.runnables.config import ( var_child_runnable_config, ) from langchain_core.runnables.utils import Input, Output -from langchain_core.tracers._streaming import _StreamingCallbackHandler from typing_extensions import TypeGuard from langgraph.constants import ( @@ -54,6 +53,11 @@ from langgraph.utils.config import ( patch_config, ) +try: + from langchain_core.tracers._streaming import _StreamingCallbackHandler +except ImportError: + _StreamingCallbackHandler = None + def _set_config_context( config: RunnableConfig, @@ -683,13 +687,15 @@ class RunnableSeq(Runnable): iterator = step.stream(input, config, **kwargs) else: iterator = step.transform(iterator, config) - if stream_handler := next( - ( - cast(_StreamingCallbackHandler, h) - for h in run_manager.handlers - if isinstance(h, _StreamingCallbackHandler) - ), - None, + if _StreamingCallbackHandler is not None and ( + stream_handler := next( + ( + cast(_StreamingCallbackHandler, h) # type: ignore + for h in run_manager.handlers + if isinstance(h, _StreamingCallbackHandler) + ), + None, + ) ): # populates streamed_output in astream_log() output if needed iterator = stream_handler.tap_output_iter(run_manager.run_id, iterator) @@ -749,13 +755,15 @@ class RunnableSeq(Runnable): aiterator = step.atransform(aiterator, config) if hasattr(aiterator, "aclose"): stack.push_async_callback(aiterator.aclose) - if stream_handler := next( - ( - cast(_StreamingCallbackHandler, h) - for h in run_manager.handlers - if isinstance(h, _StreamingCallbackHandler) - ), - None, + if _StreamingCallbackHandler is not None and ( + stream_handler := next( + ( + cast(_StreamingCallbackHandler, h) # type: ignore + for h in run_manager.handlers + if isinstance(h, _StreamingCallbackHandler) + ), + None, + ) ): # populates streamed_output in astream_log() output if needed aiterator = stream_handler.tap_output_aiter( From cee6a450dc9bd4fcecd32bee14a7ba279939fdd8 Mon Sep 17 00:00:00 2001 From: Nuno Campos Date: Tue, 8 Apr 2025 10:52:28 -0700 Subject: [PATCH 3/3] Lint --- libs/langgraph/langgraph/pregel/__init__.py | 4 ++-- libs/langgraph/langgraph/pregel/messages.py | 2 +- libs/langgraph/langgraph/utils/runnable.py | 6 +++--- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/libs/langgraph/langgraph/pregel/__init__.py b/libs/langgraph/langgraph/pregel/__init__.py index cb0a89a79..4711965f6 100644 --- a/libs/langgraph/langgraph/pregel/__init__.py +++ b/libs/langgraph/langgraph/pregel/__init__.py @@ -127,7 +127,7 @@ from langgraph.utils.queue import AsyncQueue, SyncQueue # type: ignore[attr-def try: from langchain_core.tracers._streaming import _StreamingCallbackHandler except ImportError: - _StreamingCallbackHandler = None + _StreamingCallbackHandler = None # type: ignore WriteValue = Union[Callable[[Input], Output], Any] @@ -2536,7 +2536,7 @@ class Pregel(PregelProtocol): do_stream = ( next( ( - cast(_StreamingCallbackHandler, h) # type: ignore + cast(_StreamingCallbackHandler, h) for h in run_manager.handlers if isinstance(h, _StreamingCallbackHandler) ), diff --git a/libs/langgraph/langgraph/pregel/messages.py b/libs/langgraph/langgraph/pregel/messages.py index d53598220..5766c2b63 100644 --- a/libs/langgraph/langgraph/pregel/messages.py +++ b/libs/langgraph/langgraph/pregel/messages.py @@ -23,7 +23,7 @@ from langgraph.types import StreamChunk try: from langchain_core.tracers._streaming import _StreamingCallbackHandler except ImportError: - _StreamingCallbackHandler = object + _StreamingCallbackHandler = object # type: ignore T = TypeVar("T") Meta = tuple[tuple[str, ...], dict[str, Any]] diff --git a/libs/langgraph/langgraph/utils/runnable.py b/libs/langgraph/langgraph/utils/runnable.py index f51da12d9..4dab44023 100644 --- a/libs/langgraph/langgraph/utils/runnable.py +++ b/libs/langgraph/langgraph/utils/runnable.py @@ -56,7 +56,7 @@ from langgraph.utils.config import ( try: from langchain_core.tracers._streaming import _StreamingCallbackHandler except ImportError: - _StreamingCallbackHandler = None + _StreamingCallbackHandler = None # type: ignore def _set_config_context( @@ -690,7 +690,7 @@ class RunnableSeq(Runnable): if _StreamingCallbackHandler is not None and ( stream_handler := next( ( - cast(_StreamingCallbackHandler, h) # type: ignore + cast(_StreamingCallbackHandler, h) for h in run_manager.handlers if isinstance(h, _StreamingCallbackHandler) ), @@ -758,7 +758,7 @@ class RunnableSeq(Runnable): if _StreamingCallbackHandler is not None and ( stream_handler := next( ( - cast(_StreamingCallbackHandler, h) # type: ignore + cast(_StreamingCallbackHandler, h) for h in run_manager.handlers if isinstance(h, _StreamingCallbackHandler) ),