Workflow Session Management#

RAGFlow agent workflow sessions persist the complete execution state — DSL, execution path, conversation history, globals, and retrieval results — across API calls. This allows multi-turn interactions and mid-workflow pauses (e.g., waiting for user form input) to survive process boundaries.

Key entities:

A session ID is a 32-char hex UUID (via get_uuid()) that also serves as the Canvas task_id 2.

Session Lifecycle#

New Session#

When no session_id is provided 3:

1. The agent's DSL is fetched from UserCanvas.dsl.
2. A new Canvas is instantiated and canvas.reset() is called to zero out path, globals, history, and memory.
3. An API4Conversation row is inserted via API4ConversationService.save() with the initial DSL and an empty message list.

The REST create_agent_session endpoint 4 follows the same pattern but also seeds a prologue message from canvas.get_prologue().

Session Resumption#

When a session_id is supplied 5:

1. API4ConversationService.get_by_id(session_id) fetches the stored row.
2. Canvas(conv.dsl, ...) is constructed from the persisted DSL — no reset() is called, so the prior path, globals, history, retrieval, and all component output states are fully restored.

Persistence After a Run#

After canvas.run() completes 6:

• conv.dsl = str(canvas) serializes the entire updated canvas via Graph.__str__(), which deep-copies the DSL structure, serializes each component's parameter object via ComponentParamBase.__str__(), and converts any non-serializable callables to None.
• API4ConversationService.append_message(conv_id, conv) writes the updated row — new messages, references, errors, and the refreshed DSL — back to the database.

The _run_workflow_session() REST path follows an identical pattern via its inner persist_workflow_session() coroutine.

DSL State Structure#

The dsl field is the single source of truth for a session's runtime state. Its top-level keys are 7:

Canvas.load() reads all these keys back into live in-memory fields on construction, so a resumed Canvas object is indistinguishable from one that ran continuously.

Workflow Pausing via UserFillUp#

UserFillUp (agent/component/fillup.py) pauses a workflow mid-execution to collect structured user input (form fields, file uploads) before continuing downstream.

Pause Mechanism#

After each execution batch, the canvas checks whether any UserFillUp component appears in the remaining path 8:

1. All UserFillUp nodes in the remaining path are invoked to resolve their current field states.
2. Canvas._is_input_field_satisfied() tests each field. Unsatisfied fields (value is None or empty) are collected into another_inputs.
3. If any fields are unsatisfied, self.path is trimmed to the pending nodes and a user_inputs SSE event is yielded with the field descriptors and optional tips. The generator then returns (not raises), leaving path in a paused state.
4. The paused DSL is saved to API4Conversation.dsl exactly as described above, with path[0] pointing to the UserFillUp node.

Resume Detection#

At the start of every Canvas._run_impl() call, the is_resume flag is set 9:

is_resume = bool(self.path) and self.path[0].lower().find("userfillup") >= 0

When is_resume is True 10 11:

• The workflow_started event is not re-emitted.
• Execution starts at idx = 0 (the UserFillUp node), not at len(self.path) - 1.
• begin is not re-appended to the path.

Input Handling and Loop Safety#

UserFillUp._invoke() receives the inputs dict from canvas.run(**run_kwargs). If inputs is empty — meaning no new user answer was submitted — _clear_form_values() sets every non-optional file field's value to None. This ensures the satisfaction check will pause again rather than silently reusing a stale answer from a previous loop iteration.

When inputs is populated, each value is resolved via _resolve_input_value() (handles file uploads via FileService.get_files(), JSON-parses object-typed fields) and written to component output via set_output().

REST API Surface#

All agent session endpoints are in api/apps/restful_apis/agent_api.py.

Session completion (running the workflow) flows through _run_workflow_session() 12, which handles both streaming (SSE via sse() inner generator) and non-streaming responses. The older completion() function in canvas_service.py is used by the embedded/iframe flow and the OpenAI-compatible endpoint.

The user_inputs SSE event signals a mid-workflow pause. Clients should re-POST to the completion endpoint with the same session_id and the collected inputs dict to resume execution.