feat: Implement two-phase ingestion flow with Docling polling#1572
Merged
Conversation
- Added configuration settings for Docling polling intervals and retries. - Refactored LangflowFileProcessor to support two-phase ingestion with Docling. - Introduced DoclingPollingService to handle backend polling for Docling task completion. - Updated LangflowFileService to manage the new two-phase ingestion process. - Enhanced FileTask model to track Docling task ID and status. - Created unit tests for DoclingPollingService and LangflowFileService to ensure correct behavior during two-phase ingestion. - Updated task service to include new fields for tracking ingestion phases and statuses. Co-authored-by: Copilot <copilot@github.com>
Contributor
|
Caution Review failedThe pull request is closed. ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (5)
WalkthroughAdds a backend-coordinated two-phase ingestion: submit files to Docling Serve, optionally poll Docling to completion (backoff/timeout/transient handling), then run Langflow ingestion; includes config, per-file state, Docling APIs, a polling orchestrator, DI wiring, and unit tests. ChangesTwo-Phase Docling + Langflow Ingestion
Sequence Diagram(s)sequenceDiagram
participant Client
participant LangflowFileService
participant DoclingPollingService
participant DoclingService
participant Langflow
Client->>LangflowFileService: upload_and_ingest_file(filename, content)
LangflowFileService->>DoclingService: submit file -> returns task_id
LangflowFileService->>DoclingPollingService: poll_until_ready(task_id, config)
DoclingPollingService->>DoclingService: check_task_status(task_id) [repeated]
DoclingService-->>DoclingPollingService: status snapshot
DoclingPollingService-->>LangflowFileService: DoclingPollResult (SUCCESS/FAILED/EXPIRED/TIMEOUT)
alt SUCCESS
LangflowFileService->>Langflow: run_ingestion_flow(docling_task_id)
Langflow-->>LangflowFileService: ingestion result
else non-success
LangflowFileService-->>Client: raise / return error
end
🎯 4 (Complex) | ⏱️ ~45 minutes Possibly related PRs:
Suggested reviewers:
Suggested labels: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Tip 💬 Introducing Slack Agent: The best way for teams to turn conversations into code.Slack Agent is built on CodeRabbit's deep understanding of your code, so your team can collaborate across the entire SDLC without losing context.
Built for teams:
One agent for your entire SDLC. Right inside Slack. Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
backend
github-actions
Bot
added
enhancement
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/models/processors.py (1)
1-1:⚠️ Potential issue | 🟠 Major | ⚡ Quick winReplace deprecated
typing.Dictwith built-indictgeneric to clear the Ruff UP035 lint failure.CI is blocking on UP035 for this file. Remove
Dictfrom the import on line 1 and replace all three usages with built-in generics:
- Line 490:
Optional[Dict[str, Any]]→Optional[dict[str, Any]]- Line 553:
Dict[str, Any]→dict[str, Any]- Line 616:
Optional[Dict[str, Any]]→Optional[dict[str, Any]]Python 3.13 (your minimum version) fully supports built-in generics.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/models/processors.py` at line 1, Remove the deprecated typing.Dict import and switch to built-in generics: delete Dict from the import line (keep Any and Optional) and replace the three type annotations using Dict with built-in dict generics — change Optional[Dict[str, Any]] to Optional[dict[str, Any]] (both occurrences) and change Dict[str, Any] to dict[str, Any]; update the import line to "from typing import Any, Optional" so annotations compile and satisfy UP035.
🧹 Nitpick comments (2)
src/services/docling_polling_service.py (1)
138-151: 💤 Low valuePotential extra status check after deadline.
After sleeping on line 151, the loop continues to line 82 and calls
check_task_statusbefore checking the deadline again on line 139. If the remaining time was very small (e.g., 0.1s), the code will sleep for 0.1s, wake up, make one more status check, and then discover the deadline has passed. This extra check is harmless but slightly wasteful.Consider moving the deadline check to the top of the loop (before
check_task_status) to avoid the extra call.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/docling_polling_service.py` around lines 138 - 151, Move the deadline check so it runs at the start of the polling loop before calling check_task_status to avoid making an extra status request after sleeping; specifically, in the polling routine that uses variables deadline, remaining, start, interval and returns a DoclingPollResult with PollOutcome.TIMEOUT, compute remaining = deadline - time.monotonic() and if remaining <= 0 log the timeout and return the timeout DoclingPollResult before invoking check_task_status (and keep the existing await asyncio.sleep(min(interval, remaining)) behavior unchanged).tests/unit/test_docling_service_status_check.py (1)
53-61: 💤 Low valueConsider extracting error message instead of stringifying entire payload.
The test assertion passes as written because
"boom"is present in the string representation of the full payload dict. However, the implementation (line 225 incheck_task_status) setsdetail=str(payload), which stores the entire dict as a string like"{'task_status': 'failure', 'error': 'boom'}". This is fragile and includes unnecessary data. The detail should extract just the error field:detail=payload.get("error", "Unknown error").🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/unit/test_docling_service_status_check.py` around lines 53 - 61, The test indicates check_task_status currently sets detail=str(payload) which stores the whole response dict; update the implementation in the check_task_status function to extract only the error message (e.g., detail = payload.get("error", "Unknown error")) when task_status == "failure" and keep the state mapping to DoclingTaskState.FAILED; this ensures snap.detail contains the specific error string instead of the full payload.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/models/processors.py`:
- Around line 841-849: The code unconditionally instantiates
DoclingPollingService when langflow_file_service.docling_service exists,
breaking the legacy non-polling path; change the creation logic in the processor
so that DoclingPollingService is only constructed when both
langflow_file_service.docling_service is set and the async-ingestion feature
flag/config is enabled (i.e., guard the DoclingPollingService(...) call with the
async-ingestion config), leave docling_polling_service as None otherwise, and
ensure self.docling_polling_service retains that None so downstream callers
receive the non-polling path when the flag is off.
In `@src/services/docling_polling_service.py`:
- Around line 40-44: DoclingPollingService is being instantiated directly in
processors (remove that direct new usage) — instead register and construct it in
the app lifespan and expose it via a dependency provider: add a factory (e.g.,
get_docling_polling_service or create_docling_polling_service) in
dependencies.py that constructs DoclingPollingService using the existing
DoclingService dependency, call that factory from the lifespan block in main.py
to instantiate and store the singleton, and update the processors code to accept
DoclingPollingService via dependency injection (use the new provider) rather
than instantiating DoclingPollingService inline.
- Around line 74-88: There is an indentation inconsistency in the logger.debug
call inside the polling loop: align the first argument of the
logger.debug("Docling polling", task_id=task_id) call to match the other
logger.debug calls in the block (same indentation level as the logger.debug
above and the logger.debug that logs "Snapshot received"); update the
logger.debug invocation in the loop (the one immediately before calling
self.docling_service.check_task_status) so its arguments are indented
consistently with the other logger.debug calls in the method.
In `@src/services/langflow_file_service.py`:
- Around line 538-565: The current flow always calls submit_to_docling even when
docling_polling_service is None; change the control flow in the methods that
handle ingestion (where docling_polling_service, submit_to_docling, and
file_task are used) to branch before calling submit_to_docling: if
docling_polling_service is None, run the legacy Langflow-managed ingestion path
instead of uploading to Docling, and after that legacy path completes set
file_task.docling_status = "SUCCESS" (and persist/update as other branches do)
so tasks don't remain stuck in PROCESSING; ensure the existing polling branch
still calls submit_to_docling and leaves status updates as-is.
---
Outside diff comments:
In `@src/models/processors.py`:
- Line 1: Remove the deprecated typing.Dict import and switch to built-in
generics: delete Dict from the import line (keep Any and Optional) and replace
the three type annotations using Dict with built-in dict generics — change
Optional[Dict[str, Any]] to Optional[dict[str, Any]] (both occurrences) and
change Dict[str, Any] to dict[str, Any]; update the import line to "from typing
import Any, Optional" so annotations compile and satisfy UP035.
---
Nitpick comments:
In `@src/services/docling_polling_service.py`:
- Around line 138-151: Move the deadline check so it runs at the start of the
polling loop before calling check_task_status to avoid making an extra status
request after sleeping; specifically, in the polling routine that uses variables
deadline, remaining, start, interval and returns a DoclingPollResult with
PollOutcome.TIMEOUT, compute remaining = deadline - time.monotonic() and if
remaining <= 0 log the timeout and return the timeout DoclingPollResult before
invoking check_task_status (and keep the existing await
asyncio.sleep(min(interval, remaining)) behavior unchanged).
In `@tests/unit/test_docling_service_status_check.py`:
- Around line 53-61: The test indicates check_task_status currently sets
detail=str(payload) which stores the whole response dict; update the
implementation in the check_task_status function to extract only the error
message (e.g., detail = payload.get("error", "Unknown error")) when task_status
== "failure" and keep the state mapping to DoclingTaskState.FAILED; this ensures
snap.detail contains the specific error string instead of the full payload.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 1b23665b-a1b5-4725-bff1-9c1e706e47be
📒 Files selected for processing (10)
src/config/settings.pysrc/models/processors.pysrc/models/tasks.pysrc/services/docling_polling_service.pysrc/services/docling_service.pysrc/services/langflow_file_service.pysrc/services/task_service.pytests/unit/test_docling_polling_service.pytests/unit/test_docling_service_status_check.pytests/unit/test_langflow_file_service_two_phase.py
Comment on lines
+538
to
+565
| async def submit_to_docling(self, filename: str, content: bytes) -> str: | ||
| """Upload a file to Docling Serve and return the task_id immediately. | ||
|
|
||
| Phase 1 of the two-phase ingestion model. The caller is responsible | ||
| for polling Docling (typically via DoclingPollingService) and only | ||
| invoking Langflow once Docling reports SUCCESS. | ||
| """ | ||
| if self.docling_service is None: | ||
| raise RuntimeError( | ||
| "DoclingService is not configured. Ensure DOCLING_SERVE_URL is set " | ||
| "and the service was injected correctly." | ||
| ) | ||
| try: | ||
| task_id = await self.docling_service.upload_to_docling_direct_async( | ||
| filename, content | ||
| ) | ||
| logger.debug( | ||
| "[LF] Docling submission accepted", | ||
| extra={"task_id": task_id, "filename": filename}, | ||
| ) | ||
| return task_id | ||
| except Exception as e: | ||
| logger.error( | ||
| "[LF] Docling submission failed", | ||
| extra={"error": str(e), "filename": filename}, | ||
| ) | ||
| raise Exception(f"Docling upload failed: {str(e)}") | ||
|
|
Contributor
There was a problem hiding this comment.
The advertised non-polling fallback is still broken.
The docstring says docling_polling_service=None preserves the old Langflow-managed flow, but Line 614 still submits to Docling unconditionally, so deployments without an injected docling_service will now fail before Langflow starts. On top of that, file_task.docling_status is only set to SUCCESS inside the polling branch, so a successful non-polling run would leave the task stuck in PROCESSING. Please branch the legacy path before submit_to_docling() and finalize the Docling status on successful completion.
Also applies to: 593-615, 661-662, 709-710
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/services/langflow_file_service.py` around lines 538 - 565, The current
flow always calls submit_to_docling even when docling_polling_service is None;
change the control flow in the methods that handle ingestion (where
docling_polling_service, submit_to_docling, and file_task are used) to branch
before calling submit_to_docling: if docling_polling_service is None, run the
legacy Langflow-managed ingestion path instead of uploading to Docling, and
after that legacy path completes set file_task.docling_status = "SUCCESS" (and
persist/update as other branches do) so tasks don't remain stuck in PROCESSING;
ensure the existing polling branch still calls submit_to_docling and leaves
status updates as-is.
Collaborator
Author
There was a problem hiding this comment.
Summary:
Fix applied in src/services/langflow_file_service.py — after run_ingestion_flow returns successfully, advance docling_status to SUCCESS alongside the existing phase = COMPLETE. Idempotent for the polling path; corrects the legacy path which previously left tasks stuck at PROCESSING.
Test strengthened in tests/unit/test_langflow_file_service_two_phase.py — test_legacy_path_without_polling_service_calls_langflow_directly now passes a file_task and asserts both docling_status == SUCCESS and phase == COMPLETE on completion. This test would have failed before the fix.
Skipped the suggestion to remove the submit_to_docling call in the legacy path. Reason: that call is intentional from PR #1526 from @lucaseduoli it gives Langflow a task_id so its DoclingRemote component skips re-uploading. Removing it would regress that optimization.
Contributor
There was a problem hiding this comment.
Seems like the humans are having a chat. I'll hop back into my burrow for now. If you need me again, just tag @coderabbitai in a new comment, and I'll come hopping out!
github-actions
Bot
added
enhancement
Added ENABLE_BACKEND_DOCLING_POLLING env flag in src/config/settings.py (defaults to true). Guarded the auto-construction in src/models/processors.py:841-857 so the DoclingPollingService is only built when the flag is on; otherwise self.docling_polling_service stays None and upload_and_ingest_file() falls through to the legacy single-call path. Added two tests asserting both directions of the flag in tests/unit/test_langflow_file_service_two_phase.py.
github-actions
Bot
added
enhancement
…n, wired in the same place and same way as every other service. The processor is back to a passive consumer with no knowledge of feature flags or service construction.
github-actions
Bot
added
enhancement
Contributor
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/dependencies.py (1)
20-20:⚠️ Potential issue | 🟠 Major | ⚡ Quick winFix import source for
AsyncIteratorandSequenceto unblock Ruff UP035.Line 20 currently imports
AsyncIteratorandSequencefromtyping, which violates Ruff UP035. These should come fromcollections.abcinstead.Proposed fix
-from typing import AsyncIterator, Optional, Sequence +from collections.abc import AsyncIterator, Sequence +from typing import Optional🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/dependencies.py` at line 20, Update the import line so AsyncIterator and Sequence are imported from collections.abc instead of typing: remove AsyncIterator and Sequence from typing import and add them to imports from collections.abc while keeping Optional from typing; adjust the import statement(s) in src/dependencies.py accordingly so references to AsyncIterator and Sequence in this module resolve from collections.abc and satisfy Ruff UP035.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/dependencies.py`:
- Line 20: Update the import line so AsyncIterator and Sequence are imported
from collections.abc instead of typing: remove AsyncIterator and Sequence from
typing import and add them to imports from collections.abc while keeping
Optional from typing; adjust the import statement(s) in src/dependencies.py
accordingly so references to AsyncIterator and Sequence in this module resolve
from collections.abc and satisfy Ruff UP035.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 7c2af3eb-d393-498a-ac98-b53dd3dddc92
📒 Files selected for processing (5)
src/app/container.pysrc/dependencies.pysrc/models/processors.pysrc/services/task_service.pytests/unit/test_langflow_file_service_two_phase.py
…ent status fields
github-actions
Bot
added
enhancement
lucaseduoli
approved these changes
May 14, 2026
lucaseduoli
left a comment
lucaseduoli
left a comment
Collaborator
There was a problem hiding this comment.
LGTM! great work. Tested functionally and it works too.
ricofurtado
enabled auto-merge (squash)
May 14, 2026 16:34
github-actions
Bot
added
enhancement
github-actions
Bot
added
enhancement
mpawlow
removed their request for review
May 14, 2026 18:49
This was referenced May 15, 2026
Merged
ricofurtado
added a commit
that referenced
this pull request
May 23, 2026
* feat: Implement two-phase ingestion flow with Docling polling - Added configuration settings for Docling polling intervals and retries. - Refactored LangflowFileProcessor to support two-phase ingestion with Docling. - Introduced DoclingPollingService to handle backend polling for Docling task completion. - Updated LangflowFileService to manage the new two-phase ingestion process. - Enhanced FileTask model to track Docling task ID and status. - Created unit tests for DoclingPollingService and LangflowFileService to ensure correct behavior during two-phase ingestion. - Updated task service to include new fields for tracking ingestion phases and statuses. Co-authored-by: Copilot <copilot@github.com> * fix: Fix summary: Added ENABLE_BACKEND_DOCLING_POLLING env flag in src/config/settings.py (defaults to true). Guarded the auto-construction in src/models/processors.py:841-857 so the DoclingPollingService is only built when the flag is on; otherwise self.docling_polling_service stays None and upload_and_ingest_file() falls through to the legacy single-call path. Added two tests asserting both directions of the flag in tests/unit/test_langflow_file_service_two_phase.py. * Architectural improvement: The polling service is now a true singleton, wired in the same place and same way as every other service. The processor is back to a passive consumer with no knowledge of feature flags or service construction. * fix: Ensure docling_status is set to SUCCESS in legacy path for coherent status fields * style: ruff format (auto) --------- Co-authored-by: Copilot <copilot@github.com> Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
This was referenced May 27, 2026
This was referenced Jun 8, 2026
This was referenced Jul 6, 2026
This was referenced Jul 16, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
This PR updates the ingestion flow so Langflow is triggered only after Docling processing has completed successfully.
The goal is to avoid using Langflow as a long-running coordinator while Docling is still processing documents. Instead, OpenRAG should submit the document to Docling, track the returned
task_id, wait for Docling completion, and only then invoke the Langflow ingestion flow.This improves ingestion reliability and reduces unnecessary Langflow resource usage, especially for large documents, OCR-heavy files, table-heavy PDFs, or cases where Docling processing fails before ingestion can begin.
Problem
Today, Langflow may be started before Docling processing has completed. When this happens, Langflow can remain active while waiting or polling for Docling results.
This creates a few issues:
404 Not Foundare harder to separate from Langflow ingestion failures.Since Docling processing and Langflow ingestion are separate phases, Langflow should only run once the document is ready to be ingested.
Proposed Solution
Introduce a Docling-first ingestion flow:
task_id.task_id.404, OpenRAG marks the ingestion as failed and does not trigger Langflow.The main design principle is:
Changes
This PR may include changes in the following areas:
task_idreturned by Docling Serve.Expected Benefits
Backward Compatibility
The existing ingestion behavior should remain available behind a feature flag or configuration option.
Suggested flag: