RAGFlow/Agent Canvas — DAG Execution Model

Overview & Mental Model

The DAG Execution Model is the core scheduling and execution engine of Agent Canvas Canvas & Execution — Canvas & Execution" class="xref">Canvas. It takes a directed acyclic graph (DAG) of components and executes them in dependency order, resolving variables at runtime and recording a complete execution trace.

Topological Scheduling Algorithm

The foundation of the execution model. Kahn's algorithm iteratively removes nodes with no incoming edges, building a safe execution order.

Algorithm Steps

1. Compute In-Degrees: For each node in the DAG, count incoming edges (dependencies). Result: in_degree[node_id] = count.
2. Initialize Queue: Find all nodes with in_degree 0 (no dependencies). Add to queue. These can execute immediately.
3. Process Queue Loop:
   • Dequeue node N from front of queue
   • Add N to sorted_order list
   • For each outgoing edge (N → M): decrement in_degree[M] by 1
   • If in_degree[M] becomes 0, enqueue M
4. Detect Cycles: After loop, if len(sorted_order) < len(nodes), a cycle exists. Raise ValidationError.
5. Result: sorted_order is a valid topological sort. Nodes appearing earlier have no dependencies on nodes appearing later.

Time & Space Complexity

Example Trace: 5-Node DAG

DAG Structure:
  A (no dependencies)
  B depends on A
  C depends on A
  D depends on B, C
  E depends on D

Initial in-degrees: A:0, B:1, C:1, D:2, E:1

Step 1: in_degree[A]=0 → enqueue A
Step 2: dequeue A → sorted_order=[A]. Process edges: A→B, A→C
        in_degree[B]=0, in_degree[C]=0 → enqueue B, C
Step 3: dequeue B → sorted_order=[A,B]. Process edge: B→D
        in_degree[D]=1
Step 4: dequeue C → sorted_order=[A,B,C]. Process edge: C→D
        in_degree[D]=0 → enqueue D
Step 5: dequeue D → sorted_order=[A,B,C,D]. Process edge: D→E
        in_degree[E]=0 → enqueue E
Step 6: dequeue E → sorted_order=[A,B,C,D,E]. No outgoing edges.
Step 7: queue empty. Return sorted_order.

Result: Valid execution order is A → B,C (parallel) → D → E

Variable Resolution at Runtime

Variables in component parameters are resolved lazily (at execution time), not eagerly (at parse time). This enables workflows to handle non-deterministic LLM outputs without recompilation.

Resolution Algorithm

1. Detect Variable Ref: Parameter value contains "{{component_id@attribute.path}}" or similar syntax.
2. Parse Reference: Extract component_id and attribute path (e.g., "llm_1" and "output.choices[0].text").
3. Look Up Source: Find component_id in Canvas._outputs. If not found, check _variables (global scope). If not found, return None or raise error.
4. Traverse Path: Split path on dots and brackets. Navigate from object to target attribute. At each step:
   • If step is dictionary key (e.g., "output"), access dict[key]
   • If step is list index (e.g., "[0]"), access list[index]
   • If step is attribute, access getattr(obj, attr)
5. Return Value: Final result is passed as parameter value to component.run().

Time Complexity

O(D × S) where D = path depth (number of attribute accesses) and S = scope stack depth (for nested iterations). Practical complexity is usually O(5) or less for typical workflows.

Example: LLM Output Resolution

Parameter: "summary": "{{llm_1@output.choices[0].text}}"

1. Detect: "{{..." → variable reference detected
2. Parse: component_id="llm_1", path="output.choices[0].text"
3. Lookup: Canvas._outputs["llm_1"] = {output: {choices: [{text: "Hello World", ...}]}}
4. Traverse:
   - output → dict["output"] = {choices: [...]}
   - choices → dict["choices"] = [{text: "Hello World", ...}]
   - [0] → list[0] = {text: "Hello World", ...}
   - text → dict["text"] = "Hello World"
5. Result: summary = "Hello World"

Component receives concrete value, not template.

Execution Trace Example

A complete walk-through of a 3-node workflow showing topological order, variable resolution, and history recording.

DAG: llm_1 → agent_1 → format_1

Execute phase:
  1. Topological sort: [llm_1, agent_1, format_1]
  2. Node llm_1 (in_degree=0):
     - Input: prompt="Summarize: {{user_input}}", user_input="Hello"
     - Resolve: prompt="Summarize: Hello"
     - Run: llm_1.run({prompt="Summarize: Hello"}) → {output: {text: "Hi", ...}}
     - Cache: _outputs["llm_1"] = {output: {text: "Hi", ...}}
     - Record: history.append({step_id=1, node_id="llm_1", inputs={...}, outputs={...}, error=None, timestamp=..., duration=245ms})
  
  3. Node agent_1 (dependencies done):
     - Input: context="{{llm_1@output.text}}"
     - Resolve: context="Hi"
     - Run: agent_1.run({context="Hi"}) → {tool_calls: [...], reasoning: "..."}
     - Cache: _outputs["agent_1"] = {...}
     - Record: history.append({step_id=2, node_id="agent_1", ...})
  
  4. Node format_1 (dependencies done):
     - Input: content="{{agent_1@tool_calls}}"
     - Resolve: content=[...tool calls...]
     - Run: format_1.run({content=[...]}) → {formatted: "..."}
     - Cache: _outputs["format_1"] = {formatted: "..."}
     - Record: history.append({step_id=3, node_id="format_1", ...})

Final: Canvas.execute() returns ({llm_1: {...}, agent_1: {...}, format_1: {...}}, history=[...])

Iteration & Sub-DAG Wrapping

When an Iteration node executes, it wraps the sub-DAG in a new scope, clones component instances, and executes the sub-DAG once per list item.

Iteration Execution Phases

1. Receive Input: Iteration node gets list of items (e.g., search results).
2. Loop Start: For i in range(len(items)):
   • Create new scope: _scope[i] = {iteration_var: items[i]}
   • Clone sub-DAG components: re-instantiate each component in the iteration (isolation)
   • Execute sub-DAG: topologically sort sub-DAG, greedily execute with scoped variable resolution
   • Collect results: aggregate output from final node in sub-DAG into results_list[i]
3. Merge Results: Iteration node output = {results: results_list, count: len(items)}

Scoped Variable Resolution During Iteration

When resolving "{{item@field}}" inside an iteration, the resolver first checks the current iteration scope, then parent scopes, then global scope.

Scope Stack:
  Global: {user_input: "hello", ...}
  Parent Iteration (i=0): {items: [...], current_item: {...}}
  Current Iteration (i=1): {current_item: items[1]}

Resolve "{{current_item@text}}":
  1. Check current scope (i=1): current_item = items[1]
  2. Access .text → items[1].text
  3. Return value

This enables loop variables to shadow parent scope variables safely.

Error Handling & Recovery

When a component raises an exception, Canvas applies a tiered recovery strategy.

Annotated Code Walkthrough

Key sections from canvas.py that implement this algorithm:

Topological Sort Implementation

def topological_sort(nodes, edges):
    """Kahn's algorithm: build in-degree map, queue, and process."""
    in_degree = {n: 0 for n in nodes}
    for src, dst in edges:
        in_degree[dst] += 1
    
    queue = [n for n in nodes if in_degree[n] == 0]
    sorted_order = []
    
    while queue:
        node = queue.pop(0)  # FIFO order
        sorted_order.append(node)
        
        for src, dst in edges:
            if src == node:
                in_degree[dst] -= 1
                if in_degree[dst] == 0:
                    queue.append(dst)
    
    if len(sorted_order) != len(nodes):
        raise ValueError("Cycle detected in DAG")
    
    return sorted_order

Variable Resolution Implementation

def get_variable_value(self, path):
    """Resolve {{component_id@attr.path}} at runtime."""
    if '@' not in path:
        return self._variables.get(path)
    
    component_id, attr_path = path.split('@', 1)
    obj = self._outputs.get(component_id)
    
    for step in attr_path.split('.'):
        if obj is None:
            return None
        if step.startswith('[') and step.endswith(']'):
            idx = int(step[1:-1])
            obj = obj[idx]
        else:
            obj = obj.get(step) if isinstance(obj, dict) else getattr(obj, step, None)
    
    return obj

Edge Cases & Boundaries

Configuration & Tuning

Related Topics

• Canvas & Execution Overview — Parent L2 module overview
• Variable Resolution — Deep-dive into attribute path traversal (this page)
• Iteration Sub-DAG Semantics — How loop scope isolation works (see Iteration node in component system)
• Component Lifecycle — How components are instantiated and executed (Component System module)