pengjunjie/langflow
1
0
Fork 1
mirror of https://github.com/langflow-ai/langflow.git synced 2026-07-23 17:06:46 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(execution): route Loop subgraph through coordinator (seam #4)

Browse Source
This commit is contained in:
ogabrielluiz
2026-05-04 17:49:09 -03:00
parent 154172451a
commit b2f1bef1d0
2 changed files with 8 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
src/lfx/src/lfx/base/flow_controls/loop_utils.py
View File

@ -265,8 +265,13 @@ async def execute_loop_body(
# Execute subgraph and collect results
# Pass event_manager so UI receives events from subgraph execution
from lfx.execution import StepResult, get_default_coordinator
results = []
async for result in iteration_subgraph.async_start(event_manager=event_manager):
async for step in get_default_coordinator().run(iteration_subgraph, inputs=[], event_manager=event_manager):
if not isinstance(step, StepResult):
continue
result = step.payload
results.append(result)
# Stop all on error (as per design decision)
if hasattr(result, "valid") and not result.valid:

4
src/lfx/tests/unit/components/flow_controls/test_loop_events.py
View File

@ -82,7 +82,7 @@ class TestEventManagerPropagation:
received_event_manager = None
# Create a mock subgraph that captures the event_manager
async def mock_async_start(event_manager=None):
async def mock_async_start(event_manager=None, **kwargs): # noqa: ARG001
nonlocal received_event_manager
received_event_manager = event_manager
yield MagicMock(valid=True, result_dict=MagicMock(outputs={}))
@ -119,7 +119,7 @@ class TestEventManagerPropagation:
mock_event_manager = MagicMock()
event_manager_calls = []
async def mock_async_start(event_manager=None):
async def mock_async_start(event_manager=None, **kwargs): # noqa: ARG001
event_manager_calls.append(event_manager)
yield MagicMock(valid=True, result_dict=MagicMock(outputs={}))