agent-workflow-designer

Design production-grade multi-agent orchestration systems. Covers five core patterns (sequential pipeline, parallel fan-out/fan-in, hierarchical delegation, event-driven, consensus), platform-specific implementations, handoff protocols, state management, error recovery, context window budgeting, and cost optimization.

Zur Installation springen

Repository: seaworld008/Commonly-used-high-value-skills
Letzte Quellaktivität: 27. Juli 2026 um 07:13
Erkannte Sprache von SKILL.md: Englisch
Sterne: 64
Forks: 10

Installationsoptionen

Standardmäßig ist der Prompt ausgewählt, der zuerst die Quelle prüft. Sie können zu einem direkten Befehl wechseln oder eine lokale Kopie herunterladen.

Quelldateien prüfen

Lesen Sie SKILL.md und alle von SkillsMP angezeigten Begleitdateien, bevor Sie sich für eine Installation entscheiden.

Zurück zur Suche Zurück zur Installation

name	agent-workflow-designer
description	Design production-grade multi-agent orchestration systems. Covers five core patterns (sequential pipeline, parallel fan-out/fan-in, hierarchical delegation, event-driven, consensus), platform-specific implementations, handoff protocols, state management, error recovery, context window budgeting, and cost optimization.
zh_description	用于Agent、工作流、设计，支持任务规划、执行、评审和验证。
version	1.1.0
author	seaworld008
source	in-house
source_url
tags	["agent", "designer", "planning", "workflow"]
created_at	2026-03-04
updated_at	2026-07-27
quality	5
complexity	intermediate

Agent Workflow Designer

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

Overview

Design production-grade multi-agent orchestration systems. Covers five core patterns (sequential pipeline, parallel fan-out/fan-in, hierarchical delegation, event-driven, consensus), platform-specific implementations, handoff protocols, state management, error recovery, context window budgeting, and cost optimization.

Core Capabilities

Pattern selection guide for any orchestration requirement
Handoff protocol templates (structured context passing)
State management patterns for multi-agent workflows
Error recovery and retry strategies
Context window budget management
Cost optimization strategies per platform
Platform-specific configs: Claude Code Agent Teams, OpenClaw, CrewAI, AutoGen

When to Use

Building a multi-step AI pipeline that exceeds one agent's context capacity
Parallelizing research, generation, or analysis tasks for speed
Creating specialist agents with defined roles and handoff contracts
Designing fault-tolerant AI workflows for production

Pattern Selection Guide

Is the task sequential (each step needs previous output)?
  YES → Sequential Pipeline
  NO  → Can tasks run in parallel?
          YES → Parallel Fan-out/Fan-in
          NO  → Is there a hierarchy of decisions?
                  YES → Hierarchical Delegation
                  NO  → Is it event-triggered?
                          YES → Event-Driven
                          NO  → Need consensus/validation?
                                  YES → Consensus Pattern

Pattern 1: Sequential Pipeline

Use when: Each step depends on the previous output. Research → Draft → Review → Polish.

# sequential_pipeline.py
from dataclasses import dataclass, field
from typing import Callable, Any
import os
import anthropic

DEFAULT_MODEL = os.environ["ANTHROPIC_MODEL"]

@dataclass
class :
    name: 
    system_prompt: 
    input_key:       
    output_key:      
    model:  = field(default_factory=: DEFAULT_MODEL)
    max_tokens:  = 

 :
     ():
        .stages = stages
        .client = anthropic.Anthropic()
    
     () -> :
        state = {: initial_input}
        
         stage  .stages:
            ()
            
            stage_input = state.get(stage.input_key, )
            
            response = .client.messages.create(
                model=stage.model,
                max_tokens=stage.max_tokens,
                system=stage.system_prompt,
                messages=[{: , : stage_input}],
            )
            
            state[stage.output_key] = response.content[].text
            state[] = response.usage.input_tokens + response.usage.output_tokens
            
            ()
        
         state


pipeline = SequentialPipeline([
    PipelineStage(
        name=,
        system_prompt=,
        input_key=,
        output_key=,
    ),
    PipelineStage(
        name=,
        system_prompt=,
        input_key=,
        output_key=,
    ),
    PipelineStage(
        name=,
        system_prompt=,
        input_key=,
        output_key=,
    ),
])

# parallel_fanout.py
import asyncio
import os
import anthropic
from typing import Any

async def run_agent(client, task_name: str, system: str, user: str, model: str | None = None) -> dict:
    """Single async agent call"""
    model = model or os.environ["ANTHROPIC_MODEL"]
    loop = asyncio.get_event_loop()
    
    def _call():
        return client.messages.create(
            model=model,
            max_tokens=2048,
            system=system,
            messages=[{"role": "user", "content": user}],
        )
    
    response = await loop.run_in_executor(None, _call)
    return {
        "task": task_name,
        "output": response.content[0].text,
        "tokens": response.usage.input_tokens + response.usage.output_tokens,
    }

async def parallel_research(competitors: list[str], research_type: str) -> dict:
    """Fan-out: research all competitors in parallel. Fan-in: synthesize results."""
    client = anthropic.Anthropic()
    
    # FAN-OUT: spawn parallel agent calls
    tasks = [
        run_agent(
            client,
            task_name=competitor,
            system=f"You are a competitive intelligence analyst. Research {competitor} and provide: pricing, key features, target market, and known weaknesses.",
            user=f"Analyze {competitor} for comparison with our product in the {research_type} market.",
        )
        for competitor in competitors
    ]
    
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # Handle failures gracefully
    successful = [r for r in results if not isinstance(r, Exception)]
    failed = [r for r in results if isinstance(r, Exception)]
    
    if failed:
        print(f"Warning: {len(failed)} research tasks failed: {failed}")
    
    # FAN-IN: synthesize
    combined_research = "\n\n".join([
        f"## {r['task']}\n{r['output']}" for r in successful
    ])
    
    synthesis = await run_agent(
        client,
        task_name="synthesizer",
        system="You are a strategic analyst. Synthesize competitor research into a concise comparison matrix and strategic recommendations.",
        user=f"Synthesize these competitor analyses:\n\n{combined_research}",
        model=os.environ["ANTHROPIC_MODEL"],
    )
    
    return {
        "individual_analyses": successful,
        "synthesis": synthesis["output"],
        "total_tokens": sum(r["tokens"] for r in successful) + synthesis["tokens"],
    }

# hierarchical_delegation.py
import json
import os
import anthropic

ORCHESTRATOR_SYSTEM = """You are an orchestration agent. Your job is to:
1. Analyze the user's request
2. Break it into subtasks
3. Assign each to the appropriate specialist agent
4. Collect results and synthesize

Available specialists:
- researcher: finds facts, data, and information
- writer: creates content and documents  
- coder: writes and reviews code
- analyst: analyzes data and produces insights

Respond with a JSON plan:
{
  "subtasks": [
    {"id": "1", "agent": "researcher", "task": "...", "depends_on": []},
    {"id": "2", "agent": "writer", "task": "...", "depends_on": ["1"]}
  ]
}"""

SPECIALIST_SYSTEMS = {
    "researcher": "You are a research specialist. Find accurate, relevant information and cite sources when possible.",
    "writer": "You are a professional writer. Create clear, engaging content in the requested format.",
    "coder": "You are a senior software engineer. Write clean, well-commented code with error handling.",
    "analyst": "You are a data analyst. Provide structured analysis with evidence-backed conclusions.",
}

class HierarchicalOrchestrator:
    def __init__(self):
        self.client = anthropic.Anthropic()
    
    def run(self, user_request: str) -> str:
        # 1. Orchestrator creates plan
        plan_response = self.client.messages.create(
            model=os.environ["ANTHROPIC_MODEL"],
            max_tokens=1024,
            system=ORCHESTRATOR_SYSTEM,
            messages=[{"role": "user", "content": user_request}],
        )
        
        plan = json.loads(plan_response.content[0].text)
        results = {}
        
        # 2. Execute subtasks respecting dependencies
        for subtask in self._topological_sort(plan["subtasks"]):
            context = self._build_context(subtask, results)
            specialist = SPECIALIST_SYSTEMS[subtask["agent"]]
            
            result = self.client.messages.create(
                model=os.environ["ANTHROPIC_MODEL"],
                max_tokens=2048,
                system=specialist,
                messages=[{"role": "user", "content": f"{context}\n\nTask: {subtask['task']}"}],
            )
            results[subtask["id"]] = result.content[0].text
        
        # 3. Final synthesis
        all_results = "\n\n".join([f"### {k}\n{v}" for k, v in results.items()])
        synthesis = self.client.messages.create(
            model=os.environ["ANTHROPIC_MODEL"],
            max_tokens=2048,
            system="Synthesize the specialist outputs into a coherent final response.",
            messages=[{"role": "user", "content": f"Original request: {user_request}\n\nSpecialist outputs:\n{all_results}"}],
        )
        return synthesis.content[0].text
    
    def _build_context(self, subtask: dict, results: dict) -> str:
        if not subtask.get("depends_on"):
            return ""
        deps = [f"Output from task {dep}:\n{results[dep]}" for dep in subtask["depends_on"] if dep in results]
        return "Previous results:\n" + "\n\n".join(deps) if deps else ""
    
    def _topological_sort(self, subtasks: list) -> list:
        # Simple ordered execution respecting depends_on
        ordered, remaining = [], list(subtasks)
        completed = set()
        while remaining:
            for task in remaining:
                if all(dep in completed for dep in task.get("depends_on", [])):
                    ordered.append(task)
                    completed.add(task["id"])
                    remaining.remove(task)
                    break
        return ordered

# Standard handoff context format — use between all agents
@dataclass
class AgentHandoff:
    """Structured context passed between agents in a workflow."""
    task_id: str
    workflow_id: str
    step_number: int
    total_steps: int
    
    # What was done
    previous_agent: str
    previous_output: str
    artifacts: dict  # {"filename": "content"} for any files produced
    
    # What to do next
    current_agent: str
    current_task: str
    constraints: list[str]  # hard rules for this step
    
    # Metadata
    context_budget_remaining: int  # tokens left for this agent
    cost_so_far_usd: float
    
    def to_prompt(self) -> str:
        return f"""
# Agent Handoff — Step {self.step_number}/{self.total_steps}

## Your Task
{self.current_task}

## Constraints
{chr(10).join(f'- {c}' for c in self.constraints)}

## Context from Previous Step ({self.previous_agent})
{self.previous_output[:2000]}{"... [truncated]" if len(self.previous_output) > 2000 else ""}

## Context Budget
You have approximately {self.context_budget_remaining} tokens remaining. Be concise.
"""

import os
import time
from functools import wraps

def with_retry(max_attempts=3, backoff_seconds=2, fallback_model=None):
    """Decorator for agent calls with exponential backoff and model fallback."""
    def decorator(fn):
        @wraps(fn)
        def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return fn(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_attempts - 1:
                        wait = backoff_seconds * (2 ** attempt)
                        print(f"Attempt {attempt+1} failed: {e}. Retrying in {wait}s...")
                        time.sleep(wait)
                        
                        # Fall back to cheaper/faster model on rate limit
                        if fallback_model and "rate_limit" in str(e).lower():
                            kwargs["model"] = fallback_model
            raise last_error
        return wrapper
    return decorator

@with_retry(max_attempts=3, fallback_model=os.environ.get("ANTHROPIC_FALLBACK_MODEL"))
def call_agent(model, system, user):
    ...

# Budget context across a multi-step pipeline
# Rule: never let any step consume more than 60% of remaining budget

class ContextBudget:
    def __init__(self, total_context_tokens: int, reserve_pct: float = 0.2):
        # Read the current limit from the selected provider's official model
        # documentation or API metadata; do not hard-code model generations here.
        total = total_context_tokens
        self.total = total
        self.reserve = int(total * reserve_pct)  # keep 20% as buffer
        self.used = 0
    
    @property
    def remaining(self):
        return self.total - self.reserve - self.used
    
    def allocate(self, step_name: str, requested: int) -> int:
        allocated = min(requested, int(self.remaining * 0.6))  # max 60% of remaining
        print(f"[Budget] {step_name}: allocated {allocated:,} tokens (remaining: {self.remaining:,})")
        return allocated
    
    def consume(self, tokens_used: int):
        self.used += tokens_used

def truncate_to_budget(text: str, token_budget: int, chars_per_token: float = 4.0) -> str:
    """Rough truncation — use tiktoken for precision."""
    char_budget = int(token_budget * chars_per_token)
    if len(text) <= char_budget:
        return text
    return text[:char_budget] + "\n\n[... truncated to fit context budget ...]"

Strategy	Savings	Tradeoff
Use Haiku for routing/classification	85-90%	Slightly less nuanced judgment
Cache repeated system prompts	50-90%	Requires prompt caching setup
Truncate intermediate outputs	20-40%	May lose detail in handoffs
Batch similar tasks	50%	Latency increases
Use Sonnet for most, Opus for final step only	60-70%	Final quality may improve
Short-circuit on confidence threshold	30-50%	Need confidence scoring