Agent Workflow Designer

Tier: POWERFUL
Category: Engineering
Domain: Multi-Agent Systems / AI Orchestration

---

Overview

Design production-grade multi-agent workflows with clear pattern choice, handoff contracts, failure handling, and cost/context controls.

Core Capabilities

• Workflow pattern selection for multi-step agent systems
• Skeleton config generation for fast workflow bootstrapping
• Context and cost discipline across long-running flows
• Error recovery and retry strategy scaffolding
• Documentation pointers for operational pattern tradeoffs

---

When to Use

• A single prompt is insufficient for task complexity
• You need specialist agents with explicit boundaries
• You want deterministic workflow structure before implementation
• You need validation loops for quality or safety gates

---

Quick Start

# Generate a sequential workflow skeleton
python3 scripts/workflow_scaffolder.py sequential --name content-pipeline

# Generate an orchestrator workflow and save it
python3 scripts/workflow_scaffolder.py orchestrator --name incident-triage --output workflows/incident-triage.json

---

Pattern Map

• sequential: strict step-by-step dependency chain
• parallel: fan-out/fan-in for independent subtasks
• router: dispatch by intent/type with fallback
• orchestrator: planner coordinates specialists with dependencies
• evaluator: generator + quality gate loop

Detailed templates: references/workflow-patterns.md

---

Recommended Workflow

1. Select pattern based on dependency shape and risk profile.
2. Scaffold config via scripts/workflow_scaffolder.py.
3. Define handoff contract fields for every edge.
4. Add retry/timeouts and output validation gates.
5. Dry-run with small context budgets before scaling.

---

Common Pitfalls

• Over-orchestrating tasks solvable by one well-structured prompt
• Missing timeout/retry policies for external-model calls
• Passing full upstream context instead of targeted artifacts
• Ignoring per-step cost accumulation

Best Practices

1. Start with the smallest pattern that can satisfy requirements.
2. Keep handoff payloads explicit and bounded.
3. Validate intermediate outputs before fan-in synthesis.
4. Enforce budget and timeout limits in every step.

---

Referenced Files

The following files are referenced in this skill and included for context.

scripts/workflow_scaffolder.py

#!/usr/bin/env python3
"""Generate workflow skeleton configs from common multi-agent patterns."""

from __future__ import annotations

import argparse
import json
from pathlib import Path
from typing import Dict, List


def sequential_template(name: str) -> Dict:
    return {
        "name": name,
        "pattern": "sequential",
        "steps": [
            {"id": "research", "agent": "researcher", "next": "draft"},
            {"id": "draft", "agent": "writer", "next": "review"},
            {"id": "review", "agent": "reviewer", "next": None},
        ],
        "retry": {"max_attempts": 2, "backoff_seconds": 2},
    }


def parallel_template(name: str) -> Dict:
    return {
        "name": name,
        "pattern": "parallel",
        "fan_out": {
            "tasks": ["research_a", "research_b", "research_c"],
            "agent": "analyst",
        },
        "fan_in": {"agent": "synthesizer", "output": "combined_report"},
        "timeouts": {"per_task_seconds": 180, "fan_in_seconds": 120},
    }


def router_template(name: str) -> Dict:
    return {
        "name": name,
        "pattern": "router",
        "router": {"agent": "router", "routes": ["sales", "support", "engineering"]},
        "handlers": {
            "sales": {"agent": "sales_specialist"},
            "support": {"agent": "support_specialist"},
            "engineering": {"agent": "engineering_specialist"},
        },
        "fallback": {"agent": "generalist"},
    }


def orchestrator_template(name: str) -> Dict:
    return {
        "name": name,
        "pattern": "orchestrator",
        "orchestrator": {"agent": "orchestrator", "planning": "dynamic"},
        "specialists": ["researcher", "coder", "analyst", "writer"],
        "execution": {
            "dependency_mode": "dag",
            "max_parallel": 3,
            "completion_policy": "all_required",
        },
    }


def evaluator_template(name: str) -> Dict:
    return {
        "name": name,
        "pattern": "evaluator",
        "generator": {"agent": "generator"},
        "evaluator": {"agent": "evaluator", "criteria": ["accuracy", "format", "safety"]},
        "loop": {
            "max_iterations": 3,
            "pass_threshold": 0.8,
            "on_fail": "revise_and_retry",
        },
    }


PATTERNS = {
    "sequential": sequential_template,
    "parallel": parallel_template,
    "router": router_template,
    "orchestrator": orchestrator_template,
    "evaluator": evaluator_template,
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Generate a workflow skeleton config from a pattern.")
    parser.add_argument("pattern", choices=sorted(PATTERNS.keys()), help="Workflow pattern")
    parser.add_argument("--name", default="new-workflow", help="Workflow name")
    parser.add_argument("--output", help="Optional output path for JSON config")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    config = PATTERNS[args.pattern](args.name)
    payload = json.dumps(config, indent=2)

    if args.output:
        out = Path(args.output)
        out.parent.mkdir(parents=True, exist_ok=True)
        out.write_text(payload + "\n", encoding="utf-8")
        print(f"Wrote workflow config to {out}")
    else:
        print(payload)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

references/workflow-patterns.md

# Workflow Pattern Templates

## Sequential

Use when each step depends on prior output.

```json
{
  "pattern": "sequential",
  "steps": ["research", "draft", "review"]
}

Parallel

Use when independent tasks can fan out and then fan in.

{
  "pattern": "parallel",
  "fan_out": ["task_a", "task_b", "task_c"],
  "fan_in": "synthesizer"
}

Router

Use when tasks must be routed to specialized handlers by intent.

{
  "pattern": "router",
  "router": "intent_router",
  "routes": ["sales", "support", "engineering"],
  "fallback": "generalist"
}

Orchestrator

Use when dynamic planning and dependency management are required.

{
  "pattern": "orchestrator",
  "orchestrator": "planner",
  "specialists": ["researcher", "analyst", "coder"],
  "dependency_mode": "dag"
}

Evaluator

Use when output quality gates are mandatory before finalization.

{
  "pattern": "evaluator",
  "generator": "content_agent",
  "evaluator": "quality_agent",
  "max_iterations": 3,
  "pass_threshold": 0.8
}

Pattern Selection Heuristics

• Choose sequential for strict linear workflows.
• Choose parallel for throughput and latency reduction.
• Choose router for intent- or type-based branching.
• Choose orchestrator for complex adaptive workflows.
• Choose evaluator when correctness/quality loops are required.

Handoff Minimum Contract

• workflow_id
• step_id
• task
• constraints
• upstream_artifacts
• budget_tokens
• timeout_seconds