Add streaming callbacks: @callback(..., stream=True) - #3888
Conversation
A callback registered with stream=True is a generator (or async
generator) whose yields are pushed to the browser as they are produced.
Each yielded value has the same shape as a regular return value and
replaces the outputs; yielding dash.Patch gives incremental updates
(e.g. LLM token streaming). The last yield is the final value.
Transport follows the callback's normal selection: over the WebSocket
callback transport, frames ride the open connection as callback_response
messages with stream: true; otherwise the HTTP response streams NDJSON
(application/x-ndjson), one frame per line with a terminal {"done": true}
frame. Works on Flask, Quart, and FastAPI; sync generators warn at
registration since they occupy a server worker for the whole stream.
- dash/_streaming.py: StreamedCallbackResponse marker, context-safe
iteration helpers (callback context travels in a contextvars snapshot
so set_props/ctx work after dispatch returns), NDJSON serialization,
and a thread bridge for async generators on Flask.
- dash/_callback.py: stream wrappers building one frame per yield via
_prepare_response; per-yield no_update/PreventUpdate handling,
on_error support, error frames; registration validation (mutually
exclusive with background/clientside/mcp_enabled/api_endpoint).
- backends: streaming response branches in each serve_callback;
ws.py frame emitter and stream consumers for both the event-loop and
threadpool dispatch paths.
- renderer: applyStreamFrame applies frames on arrival through the
sideUpdate path (Patch applies exactly once); NDJSON reader in
handleServerside; stream-aware callback_response handling keeps the
request pending until the terminal frame; loading states span the
whole stream.
fastapi.testclient imports starlette.testclient, which requires httpx. Skip the protocol-level stream tests when httpx is not installed and add httpx to the CI requirements so the websocket job actually runs them.
KoolADE85
left a comment
KoolADE85
left a comment
There was a problem hiding this comment.
The implementation here will easily starve the server of resources.
Run the sample app with gunicorn -w 1 and observe:
- It will terminate streams that take longer than 30s (by default)
- The number of simultaneous requests is limited to the number of workers you spawn. After that limit, the server becomes unresponsive and/or terminates worker threads.
| stream = _kwargs.get("stream", False) | ||
| is_gen_func = inspect.isgeneratorfunction(func) | ||
| is_async_gen_func = inspect.isasyncgenfunction(func) | ||
| if stream: | ||
| if not (is_gen_func or is_async_gen_func): | ||
| raise StreamCallbackError( | ||
| f"stream=True callback '{callback_id}' must be a generator " | ||
| "function (or async generator function) that yields output " | ||
| "updates." | ||
| ) |
There was a problem hiding this comment.
This looks like two "sources of truth" that must agree with each other (stream=True and async/generator).
What do you think about removing the stream flag and just detect streaming automatically based on isasyncgenfunction/ isgeneratorfunction? That would allow devs to just start yielding values without needing to manage flags or understand our API (because there would be no API at all).
There was a problem hiding this comment.
Good catch, only need a yield, stream keyword is redundant.
A stream=True callback that goes quiet between yields sends no bytes, so every proxy in a typical deployment eventually closes the connection on its own idle timeout (nginx's proxy_read_timeout defaults to 60s). The renderer treats a stream that ends without a terminal frame as a dropped connection, so this looked like silent truncation. The NDJSON transports now emit a blank line every stream_keepalive_interval milliseconds (new Dash argument, default 15000, None or 0 disables) that the callback spends between yields. The renderer already skips blank lines, so no client change is needed. The WebSocket transport is unaffected -- it has websocket_heartbeat_interval. The two paths need different mechanics. The async path holds the pending __anext__ across timeouts; asyncio.wait_for would cancel the user generator mid-step every time a keepalive came due. The sync path drives the frame generator on a pump thread, since a blocking next() cannot be interrupted on a timer -- one consequence is that a client disconnect no longer raises GeneratorExit into the user generator at its current yield, which is one more reason sync generators warn at registration. Note this does not address gunicorn's worker timeout, which kills the worker rather than the connection and is only fixable via --worker-class.
With a sync flask backend, it's not really possible to fix at the app level. Need to run with There is also a warning recommending a different backend when streaming, but maybe we should not allow it? |
|
@T4rk1n I solved that issue with my event streaming package by limiting the event stream to the normal 30 sec server timeout. With that, the stream stays in the scope of a normal http request while still providing its advantages. |
This is too limiting for LLM purposes, conversations can last a long time. It's entirely out of the application scope since we now support multiple backends and there is a big difference in how uvicorn and gunicorn handle this by default:
Then there is also reverse proxy (nginx & such) settings that might get in the way. We'll configure plotly-cloud/enterprise to accept the streaming properly without user configuration and have some documentation for flask apps and gunicorn. |
A generator callback streams by definition, so the opt-in keyword was redundant: the only two states it could express beyond the function's own shape were both already errors (stream=True on a non-generator, and a generator function registered without stream=True). Decorating a generator or async generator function is now enough. register_callback detects the generator and flips callback_map[...]["stream"], and the combination checks move with it: _validate_stream_callback rejects background=True, mcp_enabled=True and api_endpoint at registration, keyed off the detected function shape rather than the keyword. The clientside guard goes away with the keyword it rejected. Also drops `stream` from the renderer's ICallbackDefinition, which was never read -- the client keys off the NDJSON content type on HTTP and the stream / done markers on WebSocket responses, both of which live on the response types.
|
| warnings.warn( | ||
| f"Streaming callback '{callback_id}' is a synchronous " | ||
| "generator; it will occupy a server worker for the whole " | ||
| "stream. Define it with 'async def' so it runs on the " | ||
| "event loop instead.", |
There was a problem hiding this comment.
I think this warning understates the potential problem.
| warnings.warn( | |
| f"Streaming callback '{callback_id}' is a synchronous " | |
| "generator; it will occupy a server worker for the whole " | |
| "stream. Define it with 'async def' so it runs on the " | |
| "event loop instead.", | |
| warnings.warn( | |
| f"Callback '{callback_id}' is a synchronous generator; " | |
| "it can slow down or stall the entire app under heavy load. " | |
| "Consider 'async def' instead for better performance.", |
| if ws_callback.is_shutdown: | ||
| return {"status": "prevent_update"} |
There was a problem hiding this comment.
2 bugs around a similar root cause here (also applies to the async version below):
- hot-reloading is broken while any stream is alive
- streams are left running orphaned after a server shutdown
Here we handle client disconnect - maybe it's also a good place to handle server shutdowns?
3 participants
A callback registered with
stream=Trueis a generator (or async generator) whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yieldingdash.Patchgives incremental updates (e.g. LLM token streaming). The last yield is the final value.Transport follows the callback's normal selection: over the WebSocket callback transport, frames ride the open connection as callback_response messages with stream: true; otherwise the HTTP response streams NDJSON (application/x-ndjson), one frame per line with a terminal
{"done": true}frame. Works on Flask, Quart, and FastAPI; sync generators warn at registration since they occupy a server worker for the whole stream.dash/_streaming.py: StreamedCallbackResponse marker, context-safe iteration helpers (callback context travels in a contextvars snapshot soset_props/ctxwork after dispatch returns), NDJSON serialization, and a thread bridge for async generators on Flask.dash/_callback.py: stream wrappers building one frame per yield via _prepare_response; per-yieldno_update/PreventUpdatehandling, on_error support, error frames; registration validation (mutually exclusive with background/clientside/mcp_enabled/api_endpoint).backends: streaming response branches in each serve_callback;ws.pyframe emitter and stream consumers for both the event-loop and threadpool dispatch paths.renderer: applyStreamFrame applies frames on arrival through the sideUpdate path (Patch applies exactly once); NDJSON reader in handleServerside; stream-aware callback_response handling keeps the request pending until the terminal frame; loading states span the whole stream.Demo
Two examples:
appends to the output while the callback keeps running.
bar width/label, and set_props pushes a side update along the way.