refactor(core): Use duck typing for _StreamingCallbackHandler (#32535)

It's used in langgraph and maybe elsewhere, so would be preferable if it
could just be duck-typed
This commit is contained in:
William FH 2025-08-19 05:41:07 -07:00 committed by GitHub
parent d204f0dd55
commit b470c79f1d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -1,15 +1,16 @@
"""Internal tracers used for stream_log and astream events implementations."""
import abc
import typing
from collections.abc import AsyncIterator, Iterator
from typing import TypeVar
from uuid import UUID
T = TypeVar("T")
T = typing.TypeVar("T")
class _StreamingCallbackHandler(abc.ABC):
"""For internal use.
# THIS IS USED IN LANGGRAPH.
@typing.runtime_checkable
class _StreamingCallbackHandler(typing.Protocol[T]):
"""Types for streaming callback handlers.
This is a common mixin that the callback handlers
for both astream events and astream log inherit from.
@ -18,13 +19,11 @@ class _StreamingCallbackHandler(abc.ABC):
to produce callbacks for intermediate results.
"""
@abc.abstractmethod
def tap_output_aiter(
self, run_id: UUID, output: AsyncIterator[T]
) -> AsyncIterator[T]:
"""Used for internal astream_log and astream events implementations."""
@abc.abstractmethod
def tap_output_iter(self, run_id: UUID, output: Iterator[T]) -> Iterator[T]:
"""Used for internal astream_log and astream events implementations."""