feat: add functionality to stream or poll events from the build process #5940

ogabrielluiz · 2025-01-26T15:10:07Z

This PR refactors the build flow process by introducing a job‐based event system that decouples the build execution from the API response and provides flexible event delivery (streaming or polling). Key changes include:

Backend API & Event Handling
- A new module (langflow/api/build.py) implements asynchronous build flow processing. It sets up an event queue per build job (identified by a generated job ID) and streams build events via an EventManager.
- A new DisconnectHandlerStreamingResponse (in langflow/api/disconnect.py) gracefully handles client disconnects.
- The chat build endpoint (in langflow/api/v1/chat.py) has been refactored to start the build process, return a job ID, and expose a separate events endpoint for polling/streaming events.
Queue Service Implementation
- A new QueueService (with accompanying factory in services/queue/factory.py and service in services/queue/service.py) manages per-job asyncio queues, starts build tasks, and performs periodic cleanup of completed or cancelled queues.
- The dependency injection system (in services/deps.py) now provides access to the QueueService.
Configuration & Settings Updates
- The configuration schema now includes an event_delivery option (a literal of "polling" or "streaming"), and backend settings have been updated to support this.
Task Backend Improvements
- The AnyIO task backend has been enhanced with better type annotations and cancellation support, and references to Celery have been removed to default to AnyIO.
Test & Frontend Updates
- New test utilities (in tests/unit/build_utils.py) and updated test cases validate the new build flow process, ensuring that job IDs are returned and events are correctly streamed or polled.
- Frontend code (in src/frontend/src/utils/buildUtils.ts and config queries) has been updated to honor the event_delivery setting. It now falls back to polling if streaming is not available.
- Added frontend test utility to parameterize build tests to run on both modes.

Overall, this refactor decouples build processing from the client response by using a queue-based event system, improves robustness via proper resource cleanup, and provides flexible event delivery options to support various deployment scenarios.

Copilot

Copilot reviewed 6 out of 16 changed files in this pull request and generated 1 comment.

Files not reviewed (10)

src/backend/base/langflow/events/event_manager.py: Evaluated as low risk
src/backend/base/langflow/services/schema.py: Evaluated as low risk
src/backend/base/langflow/services/task/service.py: Evaluated as low risk
src/backend/base/langflow/main.py: Evaluated as low risk
src/backend/base/langflow/graph/utils.py: Evaluated as low risk
src/backend/base/langflow/services/database/models/transactions/crud.py: Evaluated as low risk
src/backend/base/langflow/api/v1/chat.py: Evaluated as low risk
src/backend/tests/unit/components/logic/test_loop.py: Evaluated as low risk
src/backend/base/langflow/services/deps.py: Evaluated as low risk
src/backend/tests/unit/test_chat_endpoint.py: Evaluated as low risk

Comments suppressed due to low confidence (2)

src/backend/base/langflow/services/queue/service.py:37

The create_queue method should validate that the job_id is a valid UUID. Add a validation check for the job_id to ensure it is a valid UUID.

def create_queue(self, job_id: str) -> tuple[asyncio.Queue, EventManager]:

src/frontend/src/utils/buildUtils.ts:2

The variable 'runId' is used but not defined within the provided code snippet. Ensure 'runId' is defined elsewhere in the code to avoid runtime errors.

if (onBuildUpdate) onBuildUpdate(data, status, runId);

src/backend/base/langflow/services/queue/service.py

Co-authored-by: Christophe Bornet <cbornet@hotmail.com>

@cbornet

Co-author: @cbornet

…ce initialization

…d logging (#6312) * feat: implement LimitedBackgroundTasks for controlled vertex build logging * refactor: replace BackgroundTasks with LimitedBackgroundTasks in build_flow endpoint

Copilot reviewed 35 out of 49 changed files in this pull request and generated no comments.

Files not reviewed (14)

src/backend/base/langflow/events/event_manager.py: Evaluated as low risk
src/backend/base/langflow/services/schema.py: Evaluated as low risk
src/backend/base/langflow/api/v1/chat.py: Evaluated as low risk
src/backend/base/langflow/services/database/models/vertex_builds/crud.py: Evaluated as low risk
.github/changes-filter.yaml: Evaluated as low risk
src/backend/base/langflow/main.py: Evaluated as low risk
src/backend/base/langflow/services/settings/base.py: Evaluated as low risk
src/backend/base/langflow/api/v1/monitor.py: Evaluated as low risk
src/backend/base/langflow/services/database/models/transactions/crud.py: Evaluated as low risk
src/backend/base/langflow/graph/utils.py: Evaluated as low risk
src/backend/base/langflow/api/v1/schemas.py: Evaluated as low risk
src/backend/base/langflow/base/agents/agent.py: Evaluated as low risk
src/backend/base/langflow/services/deps.py: Evaluated as low risk
src/backend/base/langflow/api/v1/endpoints.py: Evaluated as low risk

Comments suppressed due to low confidence (3)

src/backend/base/langflow/services/job_queue/service.py:116

The error message should include the job_id for better clarity. Suggestion: f"Queue for job_id {job_id} already exists".

msg = f"Queue for job_id {job_id} already exists"

src/backend/base/langflow/services/job_queue/service.py:152

The error message is repeated. Ensure consistency in error handling.

msg = "Queue service is closed"

src/backend/base/langflow/services/job_queue/service.py:212

Handle exceptions during task cancellation more gracefully. Log the error and continue with the cleanup.

await asyncio.wait([task])

anovazzi1

would be nice to have tests for both modes in frontend

anovazzi1 · 2025-02-13T14:32:54Z

src/backend/base/langflow/api/limited_background_tasks.py

+
+    If more than max_vertex_builds_per_vertex tasks are added for a given vertex_id,
+    the oldest task is removed so that only the most recent remain.
+    This only applies to log_vertex_build tasks.


If we are going to apply this only to the log_vertex_build we should make it explicity on the class name

anovazzi1 · 2025-02-13T15:30:21Z

src/backend/base/langflow/api/v1/chat.py

why there are two places were we create a job_id, can't we center it on one place only?

anovazzi1 · 2025-02-13T21:10:47Z

src/frontend/src/CustomNodes/GenericNode/components/NodeStatus/index.tsx

+  const eventDelivery = useGetConfig((state) => state.data?.event_delivery);
+  const shouldStreamEvents = () => {
+    // Get from useGetConfig store
+    return eventDelivery === "streaming";


We should use an enum here instead of a plain string

anovazzi1 · 2025-02-13T21:13:43Z

src/frontend/src/utils/buildUtils.ts

  } catch (e: any) {
-    if (e.message === "Endpoint not available") {
-      return await buildVertices(params);
+    if (


would be good to put that inside some consts

dosubot bot added the size:XXL This PR changes 1000+ lines, ignoring generated files. label Jan 26, 2025

ogabrielluiz marked this pull request as draft January 26, 2025 15:10

dosubot bot added the enhancement New feature or request label Jan 26, 2025

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Jan 26, 2025

ogabrielluiz force-pushed the refactor-build-flow branch from 8d930e3 to 3b9cc03 Compare January 26, 2025 15:14

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Jan 26, 2025

ogabrielluiz requested a review from NadirJ January 26, 2025 15:15

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Jan 26, 2025

ogabrielluiz requested review from italojohnny, anovazzi1 and Copilot January 27, 2025 11:47

Copilot AI reviewed Jan 27, 2025

View reviewed changes

src/backend/base/langflow/services/queue/service.py Outdated Show resolved Hide resolved

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Jan 27, 2025

ogabrielluiz changed the title ~~feat: integrate QueueService for job handling and event streaming~~ feat: add functionality to stream or poll events from the build process Jan 27, 2025

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Jan 27, 2025

ogabrielluiz added 2 commits January 27, 2025 11:14

refactor: simplify TaskService by removing Celery integration

af9738f

refactor: enhance AnyIO backend task management and error handling

f5aac07

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 12, 2025

remove noqa comment and change argument name

da2b60a

Co-authored-by: Christophe Bornet <cbornet@hotmail.com>

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 12, 2025

ogabrielluiz added 5 commits February 12, 2025 08:29

refactor: streamline event polling logic in get_flow_events_response

b3abceb

refactor: use getattr for safer session attribute access

751ade4

Co-author: @cbornet

feat: add is_started method to JobQueueService

b3e3a60

refactor: modify JobQueueService start method and main.py queue servi…

0345edc

…ce initialization

feat: add ready state and teardown method to JobQueueService

c035a1a

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 12, 2025

ogabrielluiz added 2 commits February 12, 2025 08:39

refactor: simplify job queue cleanup logic in JobQueueService

38f151a

refactor: improve error logging in active_user fixture

5c00060

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 12, 2025

refactor: improve AnyIO task management with TaskGroup and CancelScope

c45ffdb

ogabrielluiz removed the lgtm This PR has been approved by a maintainer label Feb 12, 2025

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 12, 2025

refactor: Implement LimitedBackgroundTasks for controlled vertex buil…

cf5961a

…d logging (#6312) * feat: implement LimitedBackgroundTasks for controlled vertex build logging * refactor: replace BackgroundTasks with LimitedBackgroundTasks in build_flow endpoint

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 13, 2025

ogabrielluiz requested a review from Copilot February 13, 2025 11:44

Copilot AI reviewed Feb 13, 2025

View reviewed changes

refactor: improve task cancellation error handling in JobQueueService

d8a5ee1

github-actions bot added enhancement New feature or request and removed enhancement New feature or request labels Feb 13, 2025

anovazzi1 requested changes Feb 13, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: add functionality to stream or poll events from the build process #5940

feat: add functionality to stream or poll events from the build process #5940

ogabrielluiz commented Jan 26, 2025 •

edited

Loading

Copilot AI left a comment

anovazzi1 left a comment

anovazzi1 Feb 13, 2025

anovazzi1 Feb 13, 2025

anovazzi1 Feb 13, 2025

anovazzi1 Feb 13, 2025

feat: add functionality to stream or poll events from the build process #5940

Are you sure you want to change the base?

feat: add functionality to stream or poll events from the build process #5940

Conversation

ogabrielluiz commented Jan 26, 2025 • edited Loading

Copilot AI left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

anovazzi1 left a comment

Choose a reason for hiding this comment

anovazzi1 Feb 13, 2025

Choose a reason for hiding this comment

anovazzi1 Feb 13, 2025

Choose a reason for hiding this comment

anovazzi1 Feb 13, 2025

Choose a reason for hiding this comment

anovazzi1 Feb 13, 2025

Choose a reason for hiding this comment

ogabrielluiz commented Jan 26, 2025 •

edited

Loading