langgenius/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-22 12:36:14 -04:00
Code Issues 775 Packages Projects Releases 10 Wiki Activity

[PR #22771] refactor(workflow): Rearchitect Streaming Dependency Logic for Complex Graph Topologies #30050

New Issue
Open
opened 2026-02-21 20:46:45 -05:00 by yindo · 0 comments
yindo commented 2026-02-21 20:46:45 -05:00
Owner
Copy Link
Copy Source

Original Pull Request: https://github.com/langgenius/dify/pull/22771

State: open
Merged: No

Important

  1. Make sure you have read our contribution guidelines
  2. Ensure there is an associated issue and you have been assigned to it
  3. Use the correct syntax to link this PR: Fixes #<issue number>.

Summary

Fixes #22686 (old issue in 1.6.1, one month ago which has been fixed).

Fix streaming output failure in complex branch merge workflows. (new issue in 1.8.1)

🐛 Problem

In complex workflows with multiple conditional branches converging to a common node (e.g., multiple If/Else branches → merge point → LLM → Answer), streaming output fails. Answer nodes wait for complete generation before outputting, instead of real-time streaming.

🔧 Root Cause

  1. Aggressive branch pruning: Answer nodes incorrectly removed from rest_node_ids
  2. Static dependency analysis limitation: Cannot handle runtime state of mutually exclusive conditional branches

✨ Solution

  • Smart topology detection: Auto-detect complex branch merge scenarios, enable advanced streaming on-demand
  • Hybrid dependency strategy: Static check first + dynamic path tracing, balancing performance and accuracy
  • Answer node smart recovery: Prevent incorrect pruning in complex scenarios
  • Backward compatibility: Simple workflows maintain original performance, no runtime state overhead

🎯 Core Files

  • graph_engine.py - Complex topology detection logic
  • answer_stream_processor.py - Hybrid dependency check + node recovery mechanism
  • test_branch_merge_streaming.py - Complete streaming output functionality tests
graph TD
    subgraph "Complex Branch Merge Workflow (Fixed)"
        Start([Start]) --> Input[Code: Input]
        Input --> IfElse1{If/Else 1}
        
        %% Multiple conditional branches
        IfElse1 -->|true| IfElse2{If/Else 2}
        IfElse1 -->|false| IfElse3{If/Else 3}
        IfElse1 -->|other| DoSomething[Code: Action]
        
        %% Sub-branches
        IfElse2 -->|true| JoinPoint[Code: Join Point]
        IfElse2 -->|false| Dummy1[Dummy 1]
        IfElse3 -->|true| JoinPoint
        IfElse3 -->|false| Dummy2[Dummy 2]
        DoSomething --> JoinPoint
        
        %% Critical merge point
        JoinPoint --> LLM[LLM Node]
        LLM --> Answer[Answer Node]
        
        %% Styling for the merge point
        style JoinPoint fill:#FFE4B5,stroke:#FF8C00,stroke-width:3px
        style LLM fill:#E6F3FF,stroke:#0066CC,stroke-width:2px
        style Answer fill:#E6FFE6,stroke:#009900,stroke-width:2px
    end

    subgraph "Smart Detection & Processing"
        GraphEngine[Graph Engine] --> Detection{Complex Topology<br/>Detection}
        
        Detection -->|Simple Workflow| StaticMode[Static Mode<br/>No Runtime State]
        Detection -->|Complex Merge| DynamicMode[Dynamic Mode<br/>With Runtime State]
        
        StaticMode --> OriginalLogic[Original Static<br/>Dependency Check]
        DynamicMode --> HybridLogic[Hybrid Logic<br/>Static + Dynamic]
        
        style Detection fill:#FFF2CC,stroke:#D6B656,stroke-width:2px
        style DynamicMode fill:#E1F5FE,stroke:#0277BD,stroke-width:2px
    end

    subgraph "Streaming Flow (After Fix)"
        LLMStream[LLM Streaming] --> StreamEvent[NodeRunStreamChunkEvent]
        StreamEvent --> DependencyCheck{Dependencies Met?}
        
        DependencyCheck -->|Static Pass| StreamOutput[Enable Streaming]
        DependencyCheck -->|Static Fail<br/>+ Branch Merge| DynamicCheck[Dynamic Path<br/>Tracing]
        DynamicCheck -->|Valid Path| StreamOutput
        DynamicCheck -->|Invalid Path| BlockStream[Block Streaming]
        
        StreamOutput --> AnswerRecover[Answer Node<br/>Recovery Check]
        AnswerRecover --> RealTimeOutput[Real-time<br/>Answer Output] 
        
        style StreamOutput fill:#C8E6C9,stroke:#388E3C,stroke-width:2px
        style RealTimeOutput fill:#A5D6A7,stroke:#2E7D32,stroke-width:3px
    end

    %% Connections between subgraphs
    GraphEngine -.-> Start
    DynamicMode -.-> DependencyCheck
    Answer -.-> LLMStream

Screenshots

Before(1.8.1 cloud) After(local)

Checklist

  • This change requires a documentation update, included: Dify Document
  • I understand that this PR may be closed in case there was no previous discussion or issues. (This doesn't apply to typos!)
  • [] I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change.
  • [] I've updated the documentation accordingly.
  • [] I ran dev/reformat(backend) and cd web && npx lint-staged(frontend) to appease the lint gods
**Original Pull Request:** https://github.com/langgenius/dify/pull/22771 **State:** open **Merged:** No --- > [!IMPORTANT] > > 1. Make sure you have read our [contribution guidelines](https://github.com/langgenius/dify/blob/main/CONTRIBUTING.md) > 2. Ensure there is an associated issue and you have been assigned to it > 3. Use the correct syntax to link this PR: `Fixes #<issue number>`. ## Summary Fixes #22686 (old issue in 1.6.1, one month ago which has been fixed). Fix streaming output failure in complex branch merge workflows. (new issue in 1.8.1) ### 🐛 **Problem** In complex workflows with multiple conditional branches converging to a common node (e.g., multiple If/Else branches → merge point → LLM → Answer), streaming output fails. Answer nodes wait for complete generation before outputting, instead of real-time streaming. ### 🔧 **Root Cause** 1. **Aggressive branch pruning**: Answer nodes incorrectly removed from `rest_node_ids` 2. **Static dependency analysis limitation**: Cannot handle runtime state of mutually exclusive conditional branches ### ✨ **Solution** - **Smart topology detection**: Auto-detect complex branch merge scenarios, enable advanced streaming on-demand - **Hybrid dependency strategy**: Static check first + dynamic path tracing, balancing performance and accuracy - **Answer node smart recovery**: Prevent incorrect pruning in complex scenarios - **Backward compatibility**: Simple workflows maintain original performance, no runtime state overhead ### 🎯 **Core Files** - `graph_engine.py` - Complex topology detection logic - `answer_stream_processor.py` - Hybrid dependency check + node recovery mechanism - `test_branch_merge_streaming.py` - Complete streaming output functionality tests ```mermaid graph TD subgraph "Complex Branch Merge Workflow (Fixed)" Start([Start]) --> Input[Code: Input] Input --> IfElse1{If/Else 1} %% Multiple conditional branches IfElse1 -->|true| IfElse2{If/Else 2} IfElse1 -->|false| IfElse3{If/Else 3} IfElse1 -->|other| DoSomething[Code: Action] %% Sub-branches IfElse2 -->|true| JoinPoint[Code: Join Point] IfElse2 -->|false| Dummy1[Dummy 1] IfElse3 -->|true| JoinPoint IfElse3 -->|false| Dummy2[Dummy 2] DoSomething --> JoinPoint %% Critical merge point JoinPoint --> LLM[LLM Node] LLM --> Answer[Answer Node] %% Styling for the merge point style JoinPoint fill:#FFE4B5,stroke:#FF8C00,stroke-width:3px style LLM fill:#E6F3FF,stroke:#0066CC,stroke-width:2px style Answer fill:#E6FFE6,stroke:#009900,stroke-width:2px end subgraph "Smart Detection & Processing" GraphEngine[Graph Engine] --> Detection{Complex Topology<br/>Detection} Detection -->|Simple Workflow| StaticMode[Static Mode<br/>No Runtime State] Detection -->|Complex Merge| DynamicMode[Dynamic Mode<br/>With Runtime State] StaticMode --> OriginalLogic[Original Static<br/>Dependency Check] DynamicMode --> HybridLogic[Hybrid Logic<br/>Static + Dynamic] style Detection fill:#FFF2CC,stroke:#D6B656,stroke-width:2px style DynamicMode fill:#E1F5FE,stroke:#0277BD,stroke-width:2px end subgraph "Streaming Flow (After Fix)" LLMStream[LLM Streaming] --> StreamEvent[NodeRunStreamChunkEvent] StreamEvent --> DependencyCheck{Dependencies Met?} DependencyCheck -->|Static Pass| StreamOutput[Enable Streaming] DependencyCheck -->|Static Fail<br/>+ Branch Merge| DynamicCheck[Dynamic Path<br/>Tracing] DynamicCheck -->|Valid Path| StreamOutput DynamicCheck -->|Invalid Path| BlockStream[Block Streaming] StreamOutput --> AnswerRecover[Answer Node<br/>Recovery Check] AnswerRecover --> RealTimeOutput[Real-time<br/>Answer Output] style StreamOutput fill:#C8E6C9,stroke:#388E3C,stroke-width:2px style RealTimeOutput fill:#A5D6A7,stroke:#2E7D32,stroke-width:3px end %% Connections between subgraphs GraphEngine -.-> Start DynamicMode -.-> DependencyCheck Answer -.-> LLMStream ``` ## Screenshots | Before(1.8.1 cloud) | After(local) | |----------------------|---------------------| | <video src="https://github.com/user-attachments/assets/e484482c-0b5a-4d70-b0d5-5929917a9aa4" controls width="300"> <a href="https://github.com/user-attachments/assets/e484482c-0b5a-4d70-b0d5-5929917a9aa4" target="_blank"></a> </video> | <video src="https://github.com/user-attachments/assets/6aff1fdc-e91e-4ea4-96b9-bee1edaa2bfe" controls width="300"> <a href="https://github.com/user-attachments/assets/6aff1fdc-e91e-4ea4-96b9-bee1edaa2bfe" target="_blank"></a> </video> | ## Checklist - [ ] This change requires a documentation update, included: [Dify Document](https://github.com/langgenius/dify-docs) - [x] I understand that this PR may be closed in case there was no previous discussion or issues. (This doesn't apply to typos!) - [] I've added a test for each change that was introduced, and I tried as much as possible to make a single atomic change. - [] I've updated the documentation accordingly. - [] I ran `dev/reformat`(backend) and `cd web && npx lint-staged`(frontend) to appease the lint gods
yindo added the pull-request label 2026-02-21 20:46:45 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
Knowledge Pipeline
Proposal
WIP
cloud
dependencies
feat:webapp
good first issue
outdated
performance
pull-request
Mirrored from GitHub Pull Request
 python:uv
refactor
size:L
stale
status: accepting prs
status: blocked
status: in discussion
stt & tts
vulnerablity
web
wont-fix
🌊 feat:workflow
🌚 invalid
🐞 bug
👻 feat:rag
💪 enhancement
📚 documentation
🔨 feat:tools
🙋‍♂️ question
🙏 help wanted
🤔 cant-reproduce
🤖 feat:agent
No Label pull-request
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langgenius/dify#30050
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?