agent-workflow-designer

name	agent-workflow-designer
description	Design and implement multi-agent orchestration systems with workflow DAGs, agent routing, handoff protocols, state management, and cost optimization. Use when building AI pipelines with multiple specialized agents, designing fan-out/fan-in patterns, or implementing fault-tolerant agent workflows.
license	MIT + Commons Clause
metadata	{"version":"1.0.0","author":"borghei","category":"engineering","domain":"ai-orchestration","tier":"POWERFUL","updated":"2026-03-09T00:00:00.000Z","frameworks":"langgraph, crewai, autogen, claude-agent-teams"}

Agent Workflow Designer

The agent designs multi-agent orchestration systems using five core patterns: sequential pipeline, parallel fan-out/fan-in, hierarchical delegation, event-driven reactor, and consensus validation. It implements agent routing strategies, circuit breaker reliability patterns, context window budgeting, and cost optimization across LangGraph, CrewAI, AutoGen, and Claude Code agent teams.

Core Capabilities

1. Pattern Selection and Design

• Sequential pipelines with typed handoffs
• Parallel fan-out/fan-in with merge strategies
• Hierarchical delegation with dynamic subtask discovery
• Event-driven reactors with pub/sub agent triggers
• Consensus validation with voting and arbitration

2. Agent Routing

• Intent-based routing with classifier agents
• Skill-based routing using capability matching
• Cost-aware routing (cheap models for simple tasks)
• Load-balanced routing across agent pools
• Fallback chains with graceful degradation

3. State and Context Management

• Persistent workflow state across agent hops
• Context window budgeting and summarization
• Checkpoint/resume for long-running workflows
• Conflict resolution for parallel state updates

4. Reliability Engineering

• Circuit breakers for failing agents
• Retry with exponential backoff and model fallback
• Dead letter queues for unprocessable tasks
• Timeout enforcement at every agent boundary
• Idempotent operations for safe retries

When to Use

• Building multi-step AI pipelines that exceed one agent's capability
• Parallelizing research, analysis, or generation tasks
• Creating specialist agent teams with defined roles and contracts
• Designing fault-tolerant AI workflows for production deployment
• Optimizing cost across workflows with mixed model tiers

Pattern Selection Decision Tree

What does the workflow look like?
│
├─ Linear: step A feeds step B feeds step C
│  └─ SEQUENTIAL PIPELINE
│     Best for: content pipelines, code review chains, data transformation
│
├─ Parallel: N independent tasks, then combine
│  └─ FAN-OUT / FAN-IN
│     Best for: competitive research, multi-source analysis, parallel code gen
│
├─ Tree: orchestrator breaks work into subtasks dynamically
│  └─ HIERARCHICAL DELEGATION
│     Best for: complex projects, open-ended research, code generation with planning
│
├─ Reactive: agents respond to events/triggers
│  └─ EVENT-DRIVEN REACTOR
│     Best for: monitoring, alerting, continuous integration, chat workflows
│
└─ Verification: multiple agents must agree on output
   └─ CONSENSUS VALIDATION
      Best for: high-stakes decisions, code review, fact checking, safety-critical output

Pattern 1: Sequential Pipeline

Each stage transforms input and passes structured output to the next. Type-safe handoffs prevent data loss between stages.

LangGraph Implementation

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
from langchain_anthropic import ChatAnthropic

class PipelineState(TypedDict):
    topic: str
    research: str
    draft: str
    final: str
    stage_costs: Annotated[list[dict], "append"]  # accumulates cost per stage

def research_stage(state: PipelineState) -> dict:
    model = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=2048)
    result = model.invoke(
        f"Research the following topic thoroughly. Provide key facts, statistics, "
        f"and expert perspectives:\n\n{state['topic']}"
    )
    return {
        "research": result.content,
        "stage_costs": [{"stage": "research", "tokens": result.usage_metadata["total_tokens"]}],
    }

def writing_stage(state: PipelineState) -> dict:
    model = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=4096)
    result = model.invoke(
        f"Using this research, write a compelling 800-word blog post with a hook, "
        f"3 main sections, and a CTA:\n\n{state['research']}"
    )
     {
        : result.content,
        : [{: , : result.usage_metadata[]}],
    }

 () -> :
    model = ChatAnthropic(model=, max_tokens=)
    result = model.invoke(
        
        
    )
     {
        : result.content,
        : [{: , : result.usage_metadata[]}],
    }


graph = StateGraph(PipelineState)
graph.add_node(, research_stage)
graph.add_node(, writing_stage)
graph.add_node(, editing_stage)
graph.add_edge(, )
graph.add_edge(, )
graph.add_edge(, END)
graph.set_entry_point()

pipeline = graph.()


result = pipeline.invoke({: })
()

Pattern 2: Parallel Fan-Out / Fan-In

Independent tasks run concurrently. A merge function combines results.

import asyncio
from dataclasses import dataclass

@dataclass
class FanOutTask:
    name: str
    system_prompt: str
    user_message: str
    model: str = "claude-sonnet-4-20250514"

@dataclass
class FanOutResult:
    task_name: str
    output: str
    tokens_used: int
    success: bool
    error: str | None = None

async def fan_out_fan_in(
    tasks: list[FanOutTask],
    merge_prompt: str,
    max_concurrent: int = 5,
    timeout_seconds: float = 60.0,
) -> dict:
    """Execute tasks in parallel with concurrency limit and timeout."""
    import anthropic

    client = anthropic.AsyncAnthropic()
    semaphore = asyncio.Semaphore(max_concurrent)

    async def run_one(task: FanOutTask) -> FanOutResult:
        async with semaphore:
            try:
                response = await asyncio.wait_for(
                    client.messages.create(
                        model=task.model,
                        max_tokens=2048,
                        system=task.system_prompt,
                        messages=[{"role": , : task.user_message}],
                    ),
                    timeout=timeout_seconds,
                )
                 FanOutResult(
                    task_name=task.name,
                    output=response.content[].text,
                    tokens_used=response.usage.input_tokens + response.usage.output_tokens,
                    success=,
                )
             Exception  e:
                 FanOutResult(
                    task_name=task.name, output=, tokens_used=,
                    success=, error=(e),
                )

    
    results =  asyncio.gather(*[run_one(t)  t  tasks])
    successful = [r  r  results  r.success]
    failed = [r  r  results   r.success]

      successful:
         RuntimeError()

    
    combined = .join(
          r  successful
    )

    merge_response =  client.messages.create(
        model=,
        max_tokens=,
        system=,
        messages=[{: , : }],
    )

     {
        : merge_response.content[].text,
        : successful,
        : failed,
        : (r.tokens_used  r  results) + merge_response.usage.input_tokens + merge_response.usage.output_tokens,
    }

Pattern 3: Hierarchical Delegation

An orchestrator agent dynamically decomposes work and delegates to specialists.

from typing import Literal

SPECIALISTS = {
    "researcher": "Find accurate information with sources. Be thorough and cite evidence.",
    "coder": "Write clean, tested code. Include error handling and type hints.",
    "writer": "Create clear, engaging content. Match the requested tone and format.",
    "analyst": "Analyze data and produce evidence-backed conclusions with visualizations.",
    "reviewer": "Review work product for quality, accuracy, and completeness.",
}

@dataclass
class SubTask:
    id: str
    agent: Literal["researcher", "coder", "writer", "analyst", "reviewer"]
    task: str
    depends_on: list[str]
    priority: int = 0  # higher = run first when deps are equal

class HierarchicalOrchestrator:
    def __init__(self, client):
        self.client = client

    async def plan(self, request: str) -> list[SubTask]:
        """Orchestrator creates an execution plan with dependencies."""
        response = await self.client.messages.create(
            model=,
            max_tokens=,
            system=,
            messages=[{: , : request}],
        )
         json
        plan = json.loads(response.content[].text)
         [SubTask(**st)  st  plan[]]

      () -> :
        
        subtasks =  .plan(request)
        results = {}

        
         batch  ._batch_by_dependencies(subtasks):
            batch_results =  asyncio.gather(*[
                ._run_specialist(st, results)  st  batch
            ])
             st, result  (batch, batch_results):
                results[st.] = result

        
        all_outputs = .join(  k, v  results.items())
        synthesis =  .client.messages.create(
            model=,
            max_tokens=,
            system=,
            messages=[{: , : }],
        )
         synthesis.content[].text

     () -> [[SubTask]]:
        
        completed = ()
        remaining = (subtasks)
        batches = []
         remaining:
            batch = [t  t  remaining  (d  completed  d  t.depends_on)]
              batch:
                 ValueError()
            batches.append((batch, key= t: -t.priority))
            completed.update(t.  t  batch)
            remaining = [t  t  remaining  t.   completed]
         batches

Pattern 4: Event-Driven Reactor

Agents react to events from a message bus. Decoupled and scalable.

from collections import defaultdict
from typing import Callable, Any

class AgentEventBus:
    """Simple event bus for agent-to-agent communication."""

    def __init__(self):
        self._handlers: dict[str, list[Callable]] = defaultdict(list)
        self._history: list[dict] = []

    def subscribe(self, event_type: str, handler: Callable):
        self._handlers[event_type].append(handler)

    async def publish(self, event_type: str, payload: Any, source: str):
        event = {"type": event_type, "payload": payload, "source": source}
        self._history.append(event)
        handlers = self._handlers.get(event_type, [])
        results = await asyncio.gather(
            *[h(event) for h in handlers],
            return_exceptions=True,
        )
        errors = [(h, r) for h, r in zip(handlers, results) if isinstance(r, Exception)]
         errors:
             handler, error  errors:
                ()
         results


bus = AgentEventBus()

  ():
    
    diff = event[][]
    
     bus.publish(, {: findings}, )

  ():
    
    

bus.subscribe(, on_pr_opened)
bus.subscribe(, on_security_complete)

Pattern 5: Consensus Validation

Multiple agents independently evaluate the same input. A quorum determines the final output.

@dataclass
class Vote:
    agent: str
    verdict: str  # "approve" | "reject" | "revise"
    confidence: float  # 0.0 - 1.0
    reasoning: str

async def consensus_validate(
    content: str,
    validators: list[dict],  # [{"name": "...", "system": "..."}]
    quorum: float = 0.66,
    confidence_threshold: float = 0.7,
) -> dict:
    """Run content through multiple validators and determine consensus."""
    votes: list[Vote] = []

    # Collect independent votes (no agent sees another's vote)
    vote_tasks = []
    for v in validators:
        vote_tasks.append(get_agent_vote(v["name"], v["system"], content))
    raw_votes = await asyncio.gather(*vote_tasks)
    votes = [v for v in raw_votes if v is not None]

    # Calculate consensus
    approvals = [v for v in votes if v.verdict == "approve"]
    approval_rate = len(approvals) / len(votes) if votes else 0
    avg_confidence = sum(v.confidence  v  votes) / (votes)  votes  

     approval_rate >= quorum  avg_confidence >= confidence_threshold:
         {: , : approval_rate, : votes}
     (v.verdict ==   v  votes):
        rejections = [v  v  votes  v.verdict == ]
         {: , : [r.reasoning  r  rejections], : votes}
    :
         {: , : [v.reasoning  v  votes], : votes}

Agent Routing Strategies

Intent-Based Router

class IntentRouter:
    """Route requests to specialized agents based on intent classification."""

    ROUTING_TABLE = {
        "code_generation": {"agent": "coder", "model": "claude-sonnet-4-20250514"},
        "code_review": {"agent": "reviewer", "model": "claude-sonnet-4-20250514"},
        "research": {"agent": "researcher", "model": "claude-sonnet-4-20250514"},
        "simple_question": {"agent": "assistant", "model": "claude-haiku-4-20250514"},
        "creative_writing": {"agent": "writer", "model": "claude-sonnet-4-20250514"},
        "complex_analysis": {"agent": "analyst", "model": "claude-sonnet-4-20250514"},
    }

    async def route(self, message: str) -> dict:
        # Use a fast, cheap model for classification
        classification = await self.client.messages.create(
            model="claude-haiku-4-20250514",
            max_tokens=50,
            system="Classify the user intent. Respond with ONLY one of: code_generation, code_review, research, simple_question, creative_writing, complex_analysis",
            messages=[{: , : message}],
        )
        intent = classification.content[].text.strip().lower()
         .ROUTING_TABLE.get(intent, .ROUTING_TABLE[])

Context Window Budgeting

MODEL_LIMITS = {
    "claude-sonnet-4-20250514": 200_000,
    "claude-haiku-4-20250514": 200_000,
    "claude-opus-4-20250514": 200_000,
    "gpt-4o": 128_000,
}

class ContextBudget:
    def __init__(self, model: str, pipeline_stages: int, reserve_pct: float = 0.15):
        self.total = MODEL_LIMITS.get(model, 128_000)
        self.reserve = int(self.total * reserve_pct)
        self.per_stage = (self.total - self.reserve) // pipeline_stages
        self.used = 0

    def allocate(self, stage: str) -> int:
        available = self.total - self.reserve - self.used
        allocation = min(self.per_stage, int(available * 0.6))
        return max(allocation, 1000)  # minimum 1000 tokens per stage

    def consume(self, tokens: int):
        self.used += tokens

     () -> :
        estimated_tokens = (text) // 
         estimated_tokens <= budget:
             text
        
        char_limit = budget * 
         text[:char_limit] + 

Cost Optimization Matrix

Strategy	Cost Reduction	Quality Impact	When to Use
Haiku for routing/classification	85-90%	Minimal	Always for intent routing
Haiku for editing/formatting	60-70%	Low	Mechanical tasks
Sonnet for most stages	Baseline	Baseline	Default choice
Opus only for final synthesis	+50% on that stage	Higher quality	High-stakes output
Prompt caching (system prompts)	50-90% per call	None	Repeated system prompts
Truncate intermediate outputs	20-40%	May lose detail	Long pipelines
Parallel + early termination	30-50%	None if threshold met	Search/validation tasks
Batch similar requests	Up to 50%	Increased latency	Non-real-time workloads

Reliability Patterns

Circuit Breaker

import time

class CircuitBreaker:
    """Prevent cascading failures when an agent/model is down."""

    def __init__(self, failure_threshold: int = 5, recovery_time: float = 60.0):
        self.failure_threshold = failure_threshold
        self.recovery_time = recovery_time
        self.failures = 0
        self.state = "closed"  # closed = healthy, open = failing, half-open = testing
        self.last_failure_time = 0.0

    def can_execute(self) -> bool:
        if self.state == "closed":
            return True
        if self.state == "open":
            if time.time() - self.last_failure_time > self.recovery_time:
                self.state = "half-open"
                return True
            return False
        return True  # half-open: allow one test request

    def record_success(self):
        .failures = 
        .state = 

     ():
        .failures += 
        .last_failure_time = time.time()
         .failures >= .failure_threshold:
            .state = 

Common Pitfalls

• Over-orchestration — if a single prompt can handle it, adding agents adds cost and latency, not value
• Circular dependencies in subtask graphs causing infinite loops; always validate DAG structure before execution
• Context bleed — passing entire previous outputs to every stage; summarize or extract only what is needed
• No timeout enforcement — a stuck agent blocks the entire pipeline; set wall-clock timeouts at every boundary
• Silent failures — agent returns plausible but incorrect output; add validation stages for critical paths
• Ignoring cost — 10 parallel Opus calls is expensive; model selection is a cost decision, not just a quality one
• Stateless retries on stateful operations — ensure idempotency before enabling automatic retries
• Single point of failure in orchestrator — if the orchestrator agent fails, the entire workflow fails

Best Practices

1. Start with a single prompt — only add agents when you prove one cannot handle the task
2. Type your handoffs — use dataclasses or TypedDicts for inter-agent data, not raw strings
3. Budget context upfront — calculate token allocations before running the pipeline
4. Use cheap models for routing — Haiku for classification costs 10x less than Sonnet
5. Validate DAG structure at build time, not runtime
6. Log every agent call with input hash, output hash, tokens, latency, and cost
7. Set SLAs per stage — if research takes >30s, timeout and use cached results
8. Test with production-scale inputs — a pipeline that works on 100 words may fail on 10,000

Troubleshooting

Problem	Cause	Solution
Pipeline hangs indefinitely	Missing timeout enforcement on one or more agent stages	Add asyncio.wait_for() with explicit timeout_seconds at every agent boundary; use the Circuit Breaker pattern to fail fast
Circular dependency error at runtime	Subtask graph contains a cycle (e.g., task A depends on B which depends on A)	Validate DAG structure at build time with topological sort; the _batch_by_dependencies method catches this but validation should happen earlier
Context window exceeded mid-pipeline	Intermediate outputs grow beyond the model's token limit	Use the ContextBudget class to allocate tokens per stage; summarize or truncate outputs before passing to the next stage
Fan-out tasks return inconsistent formats	Each parallel agent interprets the output schema differently	Define a shared TypedDict or dataclass for all fan-out results; add a validation step before the merge function
Orchestrator plan creates too many subtasks	The planning prompt does not constrain subtask count, leading to over-decomposition	Add explicit constraints in the planner system prompt (e.g., "maximum 5 subtasks"); review and approve plans before execution in high-stakes workflows
Consensus never reaches quorum	Validators disagree consistently or confidence scores are too low	Lower the quorum threshold, add a tiebreaker agent, or revise validator prompts to align on evaluation criteria
Cost spikes on parallel workflows	Expensive models (Opus) used for all fan-out branches instead of routing by complexity	Apply cost-aware routing: use Haiku for classification and simple tasks, Sonnet for most work, Opus only for final synthesis or high-stakes decisions

Success Criteria

• Pipeline end-to-end latency stays within the defined SLA (e.g., under 60 seconds for a 5-stage workflow) with no stage exceeding its individual timeout
• Agent routing accuracy exceeds 90% when measured against a labeled test set of at least 100 representative requests
• Fan-out/fan-in workflows complete with fewer than 5% task failures across all parallel branches under normal operating conditions
• Total token cost per workflow run decreases by at least 40% after applying model tiering (Haiku for routing, Sonnet for core work, Opus for synthesis)
• Circuit breakers trigger correctly within 5 consecutive failures and recover automatically after the defined recovery window
• Context window utilization stays below 85% of model limits at every pipeline stage, with no truncation-related quality degradation
• All inter-agent handoffs pass schema validation with zero type errors across 100 consecutive workflow executions

Scope & Limitations

This skill covers:

• Design and implementation of five core multi-agent orchestration patterns (sequential, parallel, hierarchical, event-driven, consensus)
• Agent routing strategies including intent-based, skill-based, and cost-aware routing
• Reliability engineering patterns: circuit breakers, retries, timeouts, and dead letter queues
• Context window budgeting, cost optimization, and framework-specific implementations (LangGraph, CrewAI, AutoGen)

This skill does NOT cover:

• Training or fine-tuning the underlying LLMs used by agents (see engineering/ml-pipeline-architect for ML training workflows)
• Infrastructure provisioning, container orchestration, or deployment pipelines (see engineering/cloud-infrastructure-designer for cloud architecture)
• Human-in-the-loop approval workflows or UI design for agent dashboards (see product-team/ux-researcher for user-facing workflow design)
• Long-term agent memory, vector database setup, or RAG pipeline construction (see engineering/rag-pipeline-architect for retrieval-augmented generation)

Integration Points

Skill	Integration	Data Flow
engineering/ml-pipeline-architect	Agent workflows that include ML inference stages use ML Pipeline Architect for model serving and batch prediction design	Workflow DAG exports stage specs to ML pipeline; ML pipeline returns inference endpoints for agent consumption
engineering/rag-pipeline-architect	Research and retrieval agents within workflows rely on RAG pipelines for grounded knowledge access	Agent sends queries to RAG pipeline; RAG returns ranked document chunks with citations for agent context
engineering/cloud-infrastructure-designer	Production deployment of agent workflows requires infrastructure design for scaling, queuing, and monitoring	Workflow resource requirements feed into infrastructure specs; infra returns endpoint URLs, queue ARNs, and scaling policies
engineering/api-design-architect	Inter-agent communication contracts and external API boundaries follow API design standards	Agent handoff schemas are validated against API design specs; API architect provides OpenAPI definitions for external integrations
engineering/system-design-architect	Overall system architecture decisions (sync vs async, monolith vs distributed) shape workflow topology choices	System design constraints (latency budgets, availability targets) inform pattern selection; workflow requirements feed back into system capacity planning
project-management/technical-project-planning	Complex multi-agent projects require structured planning for phased rollout, risk management, and milestone tracking	Workflow complexity estimates feed into project plans; PM skill provides sprint boundaries and dependency timelines for staged deployment