diff --git a/libs/langgraph/langgraph/stream/_mux.py b/libs/langgraph/langgraph/stream/_mux.py index 28f559e19..cd2b7d353 100644 --- a/libs/langgraph/langgraph/stream/_mux.py +++ b/libs/langgraph/langgraph/stream/_mux.py @@ -32,7 +32,6 @@ class StreamMux: def __init__(self, transformers: list[StreamTransformer] | None = None) -> None: self._event_log: list[ProtocolEvent] = [] self._transformers: list[StreamTransformer] = list(transformers or []) - self._channels: list[StreamChannel[Any]] = [] self._current_namespace: list[str] = [] self._next_emit_seq: int = 0 @@ -115,14 +114,9 @@ class StreamMux: return self._closed = True - # Finalize transformers for transformer in self._transformers: transformer.finalize() - # Close wired channels - for channel in self._channels: - channel._close() - def fail(self, error: BaseException) -> None: """Fail the mux and propagate the error to all consumers.""" if self._closed: @@ -130,14 +124,9 @@ class StreamMux: self._closed = True self._error = error - # Fail transformers for transformer in self._transformers: transformer.fail(error) - # Fail wired channels - for channel in self._channels: - channel._fail(error) - # -- Inspection --------------------------------------------------------- @property @@ -202,8 +191,6 @@ class StreamMux: for _key, value in items.items(): if is_stream_channel(value): channel: StreamChannel[Any] = value - self._channels.append(channel) - def _make_forwarder(ch: StreamChannel[Any]) -> Any: def _forward(item: Any) -> None: if self._closed: diff --git a/libs/langgraph/langgraph/stream/stream_channel.py b/libs/langgraph/langgraph/stream/stream_channel.py index 97849b8eb..b5fe0542a 100644 --- a/libs/langgraph/langgraph/stream/stream_channel.py +++ b/libs/langgraph/langgraph/stream/stream_channel.py @@ -5,10 +5,6 @@ When the :class:`StreamMux` detects a ``StreamChannel`` in a transformer's ``init()`` return, it wires every ``push()`` call to inject a :class:`ProtocolEvent` into the main event stream using the channel's name as the ``method``. - -In-process consumers iterate the channel directly. Remote SDK clients -subscribe via ``session.subscribe("custom:")``. - """ from __future__ import annotations @@ -24,8 +20,7 @@ class StreamChannel(Generic[T]): Transformer authors create a ``StreamChannel`` in ``init()`` and call ``push()`` inside ``process()`` to emit domain objects. The - mux auto-wires pushes to protocol events and auto-closes/fails the - channel on run completion. + mux auto-wires pushes to protocol events. """ __slots__ = ("channel_name", "_items", "_on_push") @@ -41,20 +36,10 @@ class StreamChannel(Generic[T]): if self._on_push is not None: self._on_push(item) - # -- Internal (called by the mux) --------------------------------------- - def _wire(self, fn: Callable[[Any], None]) -> None: """Wire a callback invoked on every ``push()``. Called by the mux.""" self._on_push = fn - def _close(self) -> None: - """No-op for compatibility. Called by the mux on normal completion.""" - pass - - def _fail(self, err: BaseException) -> None: - """No-op for compatibility. Called by the mux on failure.""" - pass - def is_stream_channel(value: object) -> bool: """Check if *value* is a :class:`StreamChannel` instance."""