---

name: workflow-orchestration description: Design and implement DAG-based workflows with parallel execution, retries, and error handling. Use when building complex multi-step agent workflows. license: Apache-2.0 compatibility: Python 3.10+, AsyncIO metadata: author: paracle-core-team version: "1.0.0" category: automation level: advanced display_name: "Workflow Orchestration" tags: - workflows - dag - orchestration - async capabilities: - workflow_design - dag_execution - parallel_processing - error_handling allowed-tools: Read Write Bash(python:*)

Workflow Orchestration Skill

When to use this skill

Use when:

• Building multi-step agent workflows
• Designing task dependencies (DAG)
• Implementing parallel execution
• Handling workflow errors and retries
• Orchestrating multiple agents

Workflow Structure

Copy# .parac/workflows/data-pipeline.yaml
name: data-pipeline
description: Extract, transform, and load data

steps:
  - id: extract
    agent: data-extractor
    input:
      source: database

  - id: transform
    agent: data-transformer
    depends_on: [extract]
    input:
      rules: transformation_rules.json

  - id: load
    agent: data-loader
    depends_on: [transform]
    input:
      destination: warehouse

  - id: validate
    agent: data-validator
    depends_on: [load]

  - id: notify
    agent: notifier
    depends_on: [validate]
    on_error: continue  # Don't fail workflow if notification fails

retry:
  max_attempts: 3
  backoff: exponential

timeout: 3600  # 1 hour

DAG Construction

Copyfrom paracle_orchestration import DAG, Step

# Create DAG
dag = DAG(name="data-pipeline")

# Add steps
extract = Step(
    id="extract",
    agent_id="data-extractor",
    input={"source": "database"},
)

transform = Step(
    id="transform",
    agent_id="data-transformer",
    depends_on=["extract"],
    input={"rules": "transformation_rules.json"},
)

load = Step(
    id="load",
    agent_id="data-loader",
    depends_on=["transform"],
    input={"destination": "warehouse"},
)

# Add to DAG
dag.add_steps([extract, transform, load])

# Validate DAG (check for cycles)
dag.validate()

Parallel Execution

Copyfrom paracle_orchestration import DAG, Step

dag = DAG(name="parallel-processing")

# Steps that can run in parallel
fetch_data = Step(id="fetch-data", agent_id="fetcher")

# These three steps have no dependencies, so they run in parallel
process_a = Step(id="process-a", agent_id="processor-a", depends_on=["fetch-data"])
process_b = Step(id="process-b", agent_id="processor-b", depends_on=["fetch-data"])
process_c = Step(id="process-c", agent_id="processor-c", depends_on=["fetch-data"])

# Merge results after all processing is done
merge = Step(
    id="merge",
    agent_id="merger",
    depends_on=["process-a", "process-b", "process-c"],
)

dag.add_steps([fetch_data, process_a, process_b, process_c, merge])

# Execution order:
# 1. fetch-data
# 2. process-a, process-b, process-c (parallel)
# 3. merge

Error Handling & Retries

Copyfrom paracle_orchestration import Step, RetryPolicy

# Retry configuration
retry_policy = RetryPolicy(
    max_attempts=3,
    backoff="exponential",  # 1s, 2s, 4s
    retry_on=[TimeoutError, ConnectionError],  # Only retry these
)

step = Step(
    id="api-call",
    agent_id="api-agent",
    retry_policy=retry_policy,
    timeout=30,  # 30 seconds per attempt
)

# Custom error handling
step = Step(
    id="critical-step",
    agent_id="critical-agent",
    on_error="fail",  # fail (default), continue, retry
)

step = Step(
    id="optional-step",
    agent_id="optional-agent",
    on_error="continue",  # Don't fail workflow if this fails
)

Best Practices

1. Design clear DAGs - Visualize dependencies
2. Use parallel execution - For independent steps
3. Handle errors gracefully - Retries and fallbacks
4. Monitor workflows - Track progress and failures
5. Test workflows - Unit and integration tests
6. Version workflows - Track changes over time
7. Implement rollbacks - For critical operations

Common Patterns

Fan-out / Fan-in

Copy# One step spawns multiple parallel steps
fetch = Step(id="fetch", ...)
process_1 = Step(id="p1", depends_on=["fetch"])
process_2 = Step(id="p2", depends_on=["fetch"])
process_3 = Step(id="p3", depends_on=["fetch"])
merge = Step(id="merge", depends_on=["p1", "p2", "p3"])

Sequential Pipeline

Copy# Steps execute one after another
step1 = Step(id="s1", ...)
step2 = Step(id="s2", depends_on=["s1"])
step3 = Step(id="s3", depends_on=["s2"])

Resources

• Orchestration Engine: packages/paracle_orchestration/
• DAG Implementation: packages/paracle_orchestration/dag.py
• Workflow Examples: content/examples/workflows/
• Engine Documentation: content/docs/technical/workflow-orchestration.md