Agent Workflow Designer

# 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.

```json
{
  "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.

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

## Orchestrator

Use when dynamic planning and dependency management are required.

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

## Evaluator

Use when output quality gates are mandatory before finalization.

```json
{
  "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`


#!/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,