Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions haystack/dataclasses/streaming_chunk.py
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,8 @@ def select_streaming_callback(
"""
Picks the correct streaming callback given an optional initial and runtime callback.

The runtime callback takes precedence over the initial callback.
The runtime callback takes precedence over the initial callback. Only the selected callback is checked for
compatibility with the execution context.

In an async context (`requires_async=True`), a sync callback is accepted but emits a warning: it will run inline on
the event loop and may block it. In a sync context (`requires_async=False`), an async callback is rejected because
Expand All @@ -247,7 +248,7 @@ def select_streaming_callback(
:returns:
The selected callback.
"""
if init_callback is not None:
if init_callback is not None and runtime_callback is None:
if requires_async and not _is_callable_async_compatible(init_callback):
logger.warning(
"A sync streaming callback was provided at initialization for use in an async context. "
Expand All @@ -265,7 +266,7 @@ def select_streaming_callback(
if not requires_async and _is_callable_async_compatible(runtime_callback):
raise ValueError("The runtime callback cannot be a coroutine.")

return runtime_callback or init_callback
return runtime_callback if runtime_callback is not None else init_callback


async def _invoke_streaming_callback(callback: StreamingCallbackT, chunk: StreamingChunk) -> None:
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
fixes:
- |
Runtime streaming callbacks now override initialization callbacks during
compatibility checks as well as selection, including callback objects whose
truth value is false. A synchronous runtime callback
can replace an asynchronous initialization callback for synchronous runs
without an error. Unused initialization callbacks no longer emit misleading
warnings about blocking the event loop.
24 changes: 22 additions & 2 deletions test/dataclasses/test_streaming_chunk.py
Original file line number Diff line number Diff line change
Expand Up @@ -363,9 +363,27 @@ async def async_callback(chunk):


class TestSelectStreamingCallback:
def test_runtime_precedes_init(self):
selected = select_streaming_callback(sync_callback, async_callback, requires_async=True)
def test_runtime_precedes_init(self, caplog):
with caplog.at_level(logging.WARNING):
selected = select_streaming_callback(sync_callback, async_callback, requires_async=True)
assert selected is async_callback
assert "sync streaming callback" not in caplog.text

def test_sync_runtime_overrides_async_init(self):
selected = select_streaming_callback(async_callback, sync_callback, requires_async=False)
assert selected is sync_callback

def test_empty_runtime_callback_collection_receives_chunks(self):
class RecordingCallback(list[StreamingChunk]):
def __call__(self, chunk: StreamingChunk) -> None:
self.append(chunk)

runtime_callback = RecordingCallback()
selected = select_streaming_callback(sync_callback, runtime_callback, requires_async=False)
assert selected is not None
chunk = StreamingChunk(content="hello")
selected(chunk)
assert runtime_callback == [chunk]

def test_runtime_only(self):
selected = select_streaming_callback(None, sync_callback, requires_async=False)
Expand All @@ -384,6 +402,8 @@ def test_async_callback_in_sync_context_raises(self):
select_streaming_callback(None, async_callback, requires_async=False)
with pytest.raises(ValueError, match="cannot be a coroutine"):
select_streaming_callback(async_callback, None, requires_async=False)
with pytest.raises(ValueError, match="cannot be a coroutine"):
select_streaming_callback(sync_callback, async_callback, requires_async=False)

def test_sync_callback_in_async_context_warns_but_returns(self, caplog):
with caplog.at_level(logging.WARNING):
Expand Down
Loading