Graph Streaming Infrastructure#

Overview#

Graph streaming in Dify controls when and how Answer nodes emit token chunks to the client during workflow execution. The infrastructure sits between the graphon engine and Dify's SSE pipeline, coordinating edge traversal order so that Answer nodes stream only after all their upstream dependencies have resolved.

Entry point: iter_dify_graph_engine_events() in api/core/workflow/workflow_entry.py wraps GraphEngine.run() with a two-filter chain: HumanInputFormEventFilter (from api/core/workflow/nodes/human_input/boundary.py) followed by ResponseStreamFilter (from graphon.filters) 1. HumanInputFormEventFilter transforms generic Human Input node completion events into specific NodeRunHumanInputFormFilledEvent or NodeRunHumanInputFormTimeoutEvent before ResponseStreamFilter gates token streaming. The WorkflowEntry class stores the ResponseStreamFilter instance and passes it through to the generator 2.

Event pipeline: graphon emits raw events → HumanInputFormEventFilter converts Human Input completion into form-filled/timeout events → ResponseStreamFilter gates token streaming and passes variable chunks as QueueTextChunkEvent → _handle_text_chunk_event() in AdvancedChatAppGenerateTaskPipeline appends tokens to WorkflowTaskState.answer → MessageCycleManager.message_to_stream_response() wraps each token as a MessageStreamResponse (SSE message event) → delivered to the client 3.

ResponseStreamFilter and Blocking Edge Coordination#

ResponseStreamFilter (from graphon.filters) gates Answer node streaming by maintaining a paths_map: a set of blocking edges that must be traversed before each downstream Answer node is unlocked. As edges are taken, they are removed from the map; when a path becomes empty, the node may stream.

Blocking node types: The coordinator's find_paths() function treats edges originating from nodes with execution types BRANCH, CONTAINER, and RESPONSE as blocking 4. PR #26364 (merged Sep 2025) added NodeExecutionType.RESPONSE to this set, preventing partial or duplicated replies when a Response node is still emitting tokens. A regression test validating Answer node emission order ships alongside this fix in PR #26377.

Variable-blocking nodes: Individual node types can declare that they block streaming for specific variable selectors via a blocks_variable_output(variable_selectors) method. PR #30832 (merged Jan 2026) added this override to VariableAssigner v1 — without it, the coordinator treated the assigner as non-blocking and could start streaming Answer nodes before a conversation variable was actually updated 5. The fix is covered by a unit test in api/tests/unit_tests/core/workflow/graph_engine/test_streaming_conversation_variables.py 5.

Complex branch-merge topologies: When multiple if-else branches converge to a shared join node → LLM → Answer, static dependency analysis can fail to model mutually exclusive branches. A hybrid strategy using _has_complex_branch_merge_topology() in graph_engine.py + an AnswerStreamProcessor for dynamic runtime path tracing addresses this 6.

Streaming in Loop Containers#

Each iteration of a Loop node creates a fresh child GraphEngine via _WorkflowChildEngineBuilder.build_child_engine() 7. This means a new ResponseStreamFilter-equivalent streaming coordinator is created for every iteration.

Stale subgraph variable bug: Before PR #30059 (merged Dec 2025), variables produced by nodes inside the loop subgraph persisted in the parent variable_pool across iterations. When a subsequent iteration's streaming coordinator found existing values in the pool, it could fall back to those stale outputs and re-stream prior-iteration content. The fix adds _clear_loop_subgraph_variables(loop_node_ids) to LoopNode._run() in api/core/workflow/nodes/loop/loop_node.py, which removes all variables produced by subgraph nodes from self.graph_runtime_state.variable_pool at the start of each iteration before the child engine is recreated 8.

The same PR standardized loop completion metadata: WorkflowNodeExecutionMetadataKey.COMPLETED_REASON is now set to typed LoopCompletedReason.LOOP_BREAK or LoopCompletedReason.LOOP_COMPLETED values 8.

Assign Variable nodes inside loops: Issue #38246 documents that Assign Variable nodes inside loops could silently fail to execute (zero execution records) due to missing parentId metadata not being set by the frontend for that node type specifically. This is a distinct problem from streaming — it prevents the node from being included in the loop's execution graph at all.

Pause/Resume State Management#

Dify workflows can pause at Human Input (HITL) nodes and resume after user form submission. Two state components must survive a pause boundary 3:

1. GraphRuntimeState — variable pool, LLM usage counters, execution metadata
2. ResponseStreamFilter — streaming gate state (the paths_map and all associated session tracking fields)

Both are serialized by PauseStatePersistenceLayer into a WorkflowResumptionContext (persisted to the database) on every GraphRunPausedEvent. On resume, WorkflowResumptionContext.get_response_stream_filter() calls ResponseStreamFilter.loads() to restore the filter, which is then threaded as an optional parameter through WorkflowEntry.__init__ 9 and on to iter_dify_graph_engine_events() 10.

Before the fix (Dify 1.15.0 / graphon v0.5.x): On resume, iter_dify_graph_engine_events() always instantiated a fresh ResponseStreamFilter() 11. If an upstream if-else or conditional-branch edge was traversed before the pause, that edge would not be re-emitted after resume. The new filter had no record of it, so the downstream Answer node's paths_map entry never emptied and streaming was permanently blocked. The workflow would report succeeded and node_finished/workflow_finished would contain correct output — but the frontend received no token stream and the final message was empty 11.

Serialized filter state includes: paths_map, active_session, waiting_sessions, pending_sessions, stream_buffers, stream_positions, closed_streams, node_execution_ids 3.

Backward compatibility: The persisted serialized_response_stream_filter_state field is optional. Runs paused before the fix resume with fresh-filter behavior for that one stale run 3.

Key Source Files and Notable Bugs#

Source Files#

Notable Bugs and Fixes#