Specialized Node Types

8. Specialized Node Types

8.1 State Processor Node

Process and aggregate outputs from multiple states.

custom_nodes:
  - id: summarizer
    custom_node_id: state_processor_node
    model: gpt-4.1-mini
    config:
      state_id: branch-processor # Filter by state ID
      states_status_filter:
        - SUCCEEDED
        - FAILED
      output_template: |
        # Processing Results
        {% for item in items %}
        ## {{ item.file_name }}
        Status: {{ item.status }}
        Result: {{ item.result }}
        {% endfor %}

Use Cases:

• Aggregate results from parallel branches
• Generate summary reports
• Filter and format outputs

8.2 Bedrock Flow Node

Integrate AWS Bedrock Flows for specialized AI workflows.

custom_nodes:
  - id: bedrock-processor
    custom_node_id: bedrock_flow_node
    config:
      flow_id: arn:aws:bedrock:region:account:flow/id
      flow_alias_id: alias-id
      enable_trace: false

Features:

• AWS Bedrock integration
• Flow execution and tracing
• Input/output transformation

8.3 Document Tree Generator

Generate structured document representations.

custom_nodes:
  - id: doc-tree
    custom_node_id: generate_documents_tree
    config:
      datasource_ids:
        - datasource-1
      documents_filtering_pattern: '**/*.md'
      documents_filter:
        - id1
        - id2

Configuration Options:

• datasource_ids: Data sources to process
• documents_filtering_pattern: Glob pattern for filtering
• documents_filter: Specific document IDs
• max_depth: Tree depth limit
• include_metadata: Include document metadata

8.4 Transform Node

Transform and map data without LLM calls. Ideal for processing webhooks, extracting fields from complex JSON, and applying transformations.

custom_nodes:
  - id: transform_pr
    custom_node_id: transform_node
    name: Transform GitHub PR Event
    config:
      input_source: 'context_store'
      input_key: 'github_event'

      mappings:
        # Extract label names from array
        - output_field: 'label_names'
          type: 'array_map'
          source_path: 'pull_request.labels'
          item_field: 'name'

        # Check if WS label exists (references label_names from above)
        - output_field: 'has_ws_label'
          type: 'condition'
          condition: "'WS' in label_names"
          then_value: true
          else_value: false

        # Extract PR number
        - output_field: 'pr_number'
          type: 'extract'
          source_path: 'number'

      # Optional: validate output structure
      output_schema:
        type: 'object'
        properties:
          label_names:
            type: 'array'
          has_ws_label:
            type: 'boolean'
          pr_number:
            type: 'integer'
        required: ['has_ws_label']

      on_error: 'fail'

Configuration Parameters

Parameter	Type	Required	Default	Description
input_source	string	No	context_store	Source of input data: context_store, messages, user_input, state_schema
input_key	string	No	-	Specific key to extract from input source
mappings	array	Yes	-	List of transformation mappings
output_schema	object	No	-	JSON Schema for output validation
on_error	string	No	fail	Error strategy: fail, skip, default, partial
default_output	object	No	{}	Default output when on_error: default

Mapping Types

Transform Node supports six mapping types:

1. Extract - Field Extraction

Extract fields using dot notation for nested objects.

- output_field: 'user_name'
  type: 'extract'
  source_path: 'pull_request.user.login'
  default: 'unknown' # Optional

2. Array Map - Array Processing

Extract fields from arrays of objects, with optional filtering.

# Extract all label names
- output_field: 'label_names'
  type: 'array_map'
  source_path: 'pull_request.labels'
  item_field: 'name'

# Extract only labels starting with "target-"
- output_field: 'target_labels'
  type: 'array_map'
  source_path: 'pull_request.labels'
  item_field: 'name'
  filter_condition: "item.get('name', '').startswith('target-')"

3. Condition - Conditional Logic

Apply boolean logic to determine output values.

- output_field: 'is_actionable'
  type: 'condition'
  condition: "action in ['opened', 'updated', 'reopened']"
  then_value: true
  else_value: false

4. Template - Jinja2 Rendering

Use Jinja2 templates for complex string formatting.

- output_field: 'summary'
  type: 'template'
  template: 'PR #{{ number }}: {{ title }} by @{{ author }}'

5. Constant - Static Values

Assign static values.

- output_field: 'source'
  type: 'constant'
  value: 'github_webhook'

6. Script - Python Expressions

Execute Python expressions with restricted namespace.

- output_field: 'priority_score'
  type: 'script'
  script: 'urgency * 2 + importance'

Sequential Processing

Important: Mappings are processed sequentially. Later mappings can reference fields created by earlier mappings.

mappings:
  # Step 1: Extract array
  - output_field: 'tags'
    type: 'array_map'
    source_path: 'items'
    item_field: 'tag'

  # Step 2: Count tags (uses 'tags' from Step 1)
  - output_field: 'tag_count'
    type: 'script'
    script: 'len(tags)'

  # Step 3: Check count (uses 'tag_count' from Step 2)
  - output_field: 'has_many_tags'
    type: 'condition'
    condition: 'tag_count > 5'
    then_value: true
    else_value: false

Complete Example: GitHub PR Label Router

execution_config:
  custom_nodes:
    - id: check_pr_labels
      custom_node_id: transform_node
      name: Extract and Check PR Labels
      config:
        input_source: 'context_store'
        input_key: 'github_event'

        mappings:
          # Extract all label names
          - output_field: 'label_names'
            type: 'array_map'
            source_path: 'pull_request.labels'
            item_field: 'name'

          # Extract only target labels
          - output_field: 'target_labels'
            type: 'array_map'
            source_path: 'pull_request.labels'
            item_field: 'name'
            filter_condition: "item.get('name', '').startswith('target-')"

          # Check for WS label
          - output_field: 'has_ws'
            type: 'condition'
            condition: "'WS' in label_names"
            then_value: true
            else_value: false

          # Check for any target label
          - output_field: 'has_target'
            type: 'condition'
            condition: 'len(target_labels) > 0'
            then_value: true
            else_value: false

          # Extract PR info
          - output_field: 'pr_number'
            type: 'extract'
            source_path: 'number'

          - output_field: 'pr_title'
            type: 'extract'
            source_path: 'pull_request.title'

          # Create summary
          - output_field: 'summary'
            type: 'template'
            template: "PR #{{ pr_number }}: {{ pr_title }} - Labels: {{ label_names | join(', ') }}"

        on_error: 'fail'

  states:
    - id: extract_labels
      custom_node_id: check_pr_labels
      task: 'Extract and process PR labels'
      next:
        state_id: route_by_labels
        output_key: 'pr_info'
        store_in_context: true

    - id: route_by_labels
      assistant_id: processor
      task: 'Route based on labels'
      next:
        switch:
          cases:
            - condition: 'has_ws == True'
              state_id: ws_workflow
            - condition: 'has_target == True'
              state_id: target_workflow
          default: default_workflow

Use Cases

• Webhook Processing: Transform GitHub, GitLab, Jira webhooks
• Array Operations: Extract fields from collections (labels, assignees, tags)
• Data Validation: Enforce structure with JSON schema
• Conditional Routing: Route workflows based on transformed data
• API Response Mapping: Transform external API responses
• Data Aggregation: Combine data from multiple sources

Benefits

• Fast: No LLM calls - pure data transformation
• Deterministic: Same input always produces same output
• Type-Safe: Built-in validation and type coercion
• Sequential: Reference previous mapping results
• Secure: Restricted Python namespace prevents dangerous operations
• Flexible: Six transformation types cover most use cases

Error Handling

Transform Node supports four error strategies:

# Fail on any error (default)
on_error: "fail"

# Skip transformation and return empty dict
on_error: "skip"

# Return configured default output
on_error: "default"
default_output:
  status: "error"
  message: "Transformation failed"

# Return partial results (skip failed mappings)
on_error: "partial"

---