Agenticx Workflow Designer

AgenticX Workflow Designer

Guide for building workflows that orchestrate agents, tasks, and execution paths.

Core Components

Basic Workflow

pythonCopy code
from agenticx import Workflow, WorkflowNode, WorkflowEdge
from agenticx.core import WorkflowEngine

# Define nodes
research_node = WorkflowNode(
    id="research",
    agent=researcher_agent,
    task=research_task
)

analysis_node = WorkflowNode(
    id="analysis",
    agent=analyst_agent,
    task=analysis_task
)

# Define edges (sequential flow)
edge = WorkflowEdge(source="research", target="analysis")

# Build workflow
workflow = Workflow(
    id="research-pipeline",
    nodes=[research_node, analysis_node],
    edges=[edge]
)

# Execute
engine = WorkflowEngine()
result = engine.run(workflow)

CLI Workflow Creation

bashCopy code
# Create workflow scaffold
agx workflow create research-pipeline --agents "researcher,analyst"

# List workflows
agx workflow list

# Run a workflow file
agx run workflows/research-pipeline.py --verbose

Workflow Patterns

Sequential Pipeline

Nodes execute one after another:

Copy code
[Research] → [Analysis] → [Report]

pythonCopy code
edges = [
    WorkflowEdge(source="research", target="analysis"),
    WorkflowEdge(source="analysis", target="report"),
]

Parallel Execution

Multiple nodes execute concurrently:

Copy code
         ┌→ [Web Search] ─┐
[Start] ─┤                 ├→ [Merge]
         └→ [DB Query]   ─┘

pythonCopy code
edges = [
    WorkflowEdge(source="start", target="web-search"),
    WorkflowEdge(source="start", target="db-query"),
    WorkflowEdge(source="web-search", target="merge"),
    WorkflowEdge(source="db-query", target="merge"),
]

Conditional Routing

Route execution based on output:

pythonCopy code
edge = WorkflowEdge(
    source="classifier",
    target="handler-a",
    condition=lambda result: result.get("category") == "A"
)

Graph-Based Orchestration

For complex DAGs with multiple paths and merge points, use WorkflowGraph:

pythonCopy code
from agenticx.core import WorkflowGraph

graph = WorkflowGraph()
graph.add_node(research_node)
graph.add_node(analysis_node)
graph.add_node(report_node)
graph.add_edge("research", "analysis")
graph.add_edge("analysis", "report")

Triggers

Scheduled Trigger

pythonCopy code
from agenticx.core import TriggerService, ScheduledTrigger

trigger = ScheduledTrigger(
    cron="0 9 * * 1",  # Every Monday at 9am
    workflow_id="weekly-report"
)
service = TriggerService()
service.register(trigger)

Event-Driven Trigger

pythonCopy code
from agenticx.core import EventDrivenTrigger

trigger = EventDrivenTrigger(
    event_type="new_data_available",
    workflow_id="data-pipeline"
)

Execution Context

Track workflow state during execution:

pythonCopy code
from agenticx.core import ExecutionContext, WorkflowStatus

context = ExecutionContext(workflow_id="research-pipeline")
# context.status → WorkflowStatus.RUNNING / COMPLETED / FAILED
# context.node_results → dict of node_id → result

Running Workflow Files

bashCopy code
# Simple run
agx run my_workflow.py

# With config file
agx run my_workflow.py --config config.yaml --verbose

# Debug mode
agx run my_workflow.py --debug

Best Practices

1. Start simple — begin with sequential, add complexity as needed
2. Name nodes clearly — they appear in logs and monitoring
3. Set timeouts — prevent infinite loops in conditional workflows
4. Use validation — validate outputs at each node boundary
5. Monitor execution — enable observability for production workflows