Multi Agent

# multi-agent Configuration
# Category: general
# Generated: 2025-12-30

skill:
  name: multi-agent
  version: "1.0.0"
  category: general

settings:
  # Default settings for multi-agent
  enabled: true
  log_level: info

  # Category-specific defaults
  validation:
    strict_mode: false
    auto_fix: false

  output:
    format: markdown
    include_examples: true

# Environment-specific overrides
environments:
  development:
    log_level: debug
    validation:
      strict_mode: false

  production:
    log_level: warn
    validation:
      strict_mode: true

# Integration settings
integrations:
  # Enable/disable integrations
  git: true
  linter: true
  formatter: true

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "multi-agent Configuration Schema",
  "type": "object",
  "properties": {
    "skill": {
      "type": "object",
      "properties": {
        "name": {
          "type": "string"
        },
        "version": {
          "type": "string",
          "pattern": "^\\d+\\.\\d+\\.\\d+$"
        },
        "category": {
          "type": "string",
          "enum": [
            "api",
            "testing",
            "devops",
            "security",
            "database",
            "frontend",
            "algorithms",
            "machine-learning",
            "cloud",
            "containers",
            "general"
          ]
        }
      },
      "required": [
        "name",
        "version"
      ]
    },
    "settings": {
      "type": "object",
      "properties": {
        "enabled": {
          "type": "boolean",
          "default": true
        },
        "log_level": {
          "type": "string",
          "enum": [
            "debug",
            "info",
            "warn",
            "error"
          ]
        }
      }
    }
  },
  "required": [
    "skill"
  ]
}

#!/usr/bin/env python3
"""
Orchestrator-Worker Multi-Agent System
=======================================

Production implementation of the Orchestrator-Worker pattern.
Based on Anthropic's "Building Effective Agents" recommendations.

Requirements:
    pip install langgraph langchain-anthropic

Usage:
    python orchestrator_worker.py "Research AI safety and write a summary report"
"""

import sys
import os
from typing import Annotated, TypedDict, List, Literal, Optional
from dataclasses import dataclass
from enum import Enum
import json

from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage


# =============================================================================
# CONFIGURATION
# =============================================================================

class WorkerType(Enum):
    """Available worker types."""
    RESEARCHER = "researcher"
    WRITER = "writer"
    ANALYST = "analyst"
    CODER = "coder"
    REVIEWER = "reviewer"


@dataclass
class WorkerConfig:
    """Configuration for a worker agent."""
    type: WorkerType
    system_prompt: str
    capabilities: List[str]


WORKER_CONFIGS = {
    WorkerType.RESEARCHER: WorkerConfig(
        type=WorkerType.RESEARCHER,
        system_prompt="""You are a Research Specialist.
Your role is to gather information, find relevant sources, and compile research notes.
Be thorough and cite your sources. Focus on accuracy over speed.""",
        capabilities=["web_search", "document_analysis", "source_verification"]
    ),
    WorkerType.WRITER: WorkerConfig(
        type=WorkerType.WRITER,
        system_prompt="""You are a Content Writer.
Your role is to create clear, well-structured written content.
Focus on readability, proper formatting, and engaging prose.""",
        capabilities=["content_creation", "editing", "formatting"]
    ),
    WorkerType.ANALYST: WorkerConfig(
        type=WorkerType.ANALYST,
        system_prompt="""You are a Data Analyst.
Your role is to analyze information, identify patterns, and provide insights.
Use data-driven reasoning and present findings clearly.""",
        capabilities=["data_analysis", "pattern_recognition", "visualization"]
    ),
    WorkerType.CODER: WorkerConfig(
        type=WorkerType.CODER,
        system_prompt="""You are a Software Developer.
Your role is to write, debug, and optimize code.
Follow best practices, write clean code, and include documentation.""",
        capabilities=["code_generation", "debugging", "optimization"]
    ),
    WorkerType.REVIEWER: WorkerConfig(
        type=WorkerType.REVIEWER,
        system_prompt="""You are a Quality Reviewer.
Your role is to review work, check for errors, and suggest improvements.
Be thorough but constructive in your feedback.""",
        capabilities=["quality_assurance", "fact_checking", "proofreading"]
    ),
}


# =============================================================================
# STATE DEFINITION
# =============================================================================

@dataclass
class Task:
    """A task assigned to a worker."""
    id: str
    description: str
    worker_type: WorkerType
    status: str = "pending"  # pending, in_progress, completed, failed
    result: Optional[str] = None
    dependencies: List[str] = None

    def __post_init__(self):
        if self.dependencies is None:
            self.dependencies = []


class OrchestratorState(TypedDict):
    """State for the orchestrator-worker system."""
    messages: Annotated[list, add_messages]
    original_request: str
    plan: List[dict]
    tasks: List[dict]
    current_task_idx: int
    results: dict
    final_output: str
    status: str  # planning, executing, synthesizing, complete


# =============================================================================
# ORCHESTRATOR
# =============================================================================

def create_plan(state: OrchestratorState) -> OrchestratorState:
    """
    Orchestrator creates execution plan.

    Analyzes the request and breaks it into tasks.
    """
    from langchain_anthropic import ChatAnthropic

    llm = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=2048)

    planning_prompt = f"""Analyze this request and create an execution plan.

REQUEST: {state['original_request']}

Available worker types and their capabilities:
{json.dumps({w.value: {"capabilities": c.capabilities} for w, c in WORKER_CONFIGS.items()}, indent=2)}

Create a plan with ordered tasks. Each task should specify:
1. task_id: Unique identifier (e.g., "task_1")
2. description: What needs to be done
3. worker_type: Which worker should handle it
4. dependencies: List of task_ids that must complete first

Respond in JSON format:
{{
    "plan_summary": "Brief description of the approach",
    "tasks": [
        {{
            "task_id": "task_1",
            "description": "...",
            "worker_type": "researcher|writer|analyst|coder|reviewer",
            "dependencies": []
        }}
    ]
}}"""

    response = llm.invoke([
        SystemMessage(content="You are an orchestrator that breaks down complex requests into tasks."),
        HumanMessage(content=planning_prompt)
    ])

    # Parse plan
    try:
        # Extract JSON from response
        content = response.content
        start = content.find("{")
        end = content.rfind("}") + 1
        plan_data = json.loads(content[start:end])

        tasks = []
        for t in plan_data["tasks"]:
            tasks.append({
                "id": t["task_id"],
                "description": t["description"],
                "worker_type": t["worker_type"],
                "status": "pending",
                "result": None,
                "dependencies": t.get("dependencies", [])
            })

        print(f"\n📋 Plan created with {len(tasks)} tasks")
        for t in tasks:
            print(f"   - {t['id']}: {t['description'][:50]}... ({t['worker_type']})")

    except Exception as e:
        print(f"⚠️ Plan parsing failed: {e}")
        tasks = [{
            "id": "task_1",
            "description": state["original_request"],
            "worker_type": "researcher",
            "status": "pending",
            "result": None,
            "dependencies": []
        }]

    return {
        **state,
        "plan": plan_data.get("tasks", []) if 'plan_data' in dir() else [],
        "tasks": tasks,
        "current_task_idx": 0,
        "status": "executing"
    }


def execute_task(state: OrchestratorState) -> OrchestratorState:
    """
    Execute current task using appropriate worker.
    """
    from langchain_anthropic import ChatAnthropic

    tasks = state["tasks"]
    idx = state["current_task_idx"]

    if idx >= len(tasks):
        return {**state, "status": "synthesizing"}

    task = tasks[idx]

    # Check dependencies
    for dep_id in task["dependencies"]:
        dep_task = next((t for t in tasks if t["id"] == dep_id), None)
        if dep_task and dep_task["status"] != "completed":
            print(f"⏳ Waiting for dependency: {dep_id}")
            return state

    print(f"\n🔧 Executing {task['id']}: {task['description'][:50]}...")

    # Get worker config
    worker_type = WorkerType(task["worker_type"])
    config = WORKER_CONFIGS[worker_type]

    # Build context from completed tasks
    context = ""
    for t in tasks:
        if t["status"] == "completed" and t["result"]:
            context += f"\n[{t['id']} Result]:\n{t['result']}\n"

    # Execute with worker
    llm = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=2048)

    worker_prompt = f"""Complete this task:

TASK: {task['description']}

ORIGINAL REQUEST: {state['original_request']}

{f"CONTEXT FROM PREVIOUS TASKS:{context}" if context else ""}

Provide your output directly. Be thorough and focused on the task."""

    response = llm.invoke([
        SystemMessage(content=config.system_prompt),
        HumanMessage(content=worker_prompt)
    ])

    # Update task
    task["status"] = "completed"
    task["result"] = response.content

    print(f"   ✅ Completed: {response.content[:100]}...")

    # Update results
    results = state.get("results", {})
    results[task["id"]] = response.content

    return {
        **state,
        "tasks": tasks,
        "current_task_idx": idx + 1,
        "results": results
    }


def synthesize_results(state: OrchestratorState) -> OrchestratorState:
    """
    Orchestrator synthesizes all results into final output.
    """
    from langchain_anthropic import ChatAnthropic

    print("\n📊 Synthesizing results...")

    llm = ChatAnthropic(model="claude-sonnet-4-20250514", max_tokens=4096)

    # Compile all results
    results_text = ""
    for task in state["tasks"]:
        if task["result"]:
            results_text += f"\n## {task['id']}: {task['description']}\n{task['result']}\n"

    synthesis_prompt = f"""Synthesize these task results into a final, cohesive response.

ORIGINAL REQUEST: {state['original_request']}

TASK RESULTS:
{results_text}

Create a well-organized final response that:
1. Addresses the original request completely
2. Integrates insights from all tasks
3. Is clear and actionable
4. Highlights key findings or recommendations"""

    response = llm.invoke([
        SystemMessage(content="You are an orchestrator synthesizing work from multiple specialists."),
        HumanMessage(content=synthesis_prompt)
    ])

    return {
        **state,
        "final_output": response.content,
        "status": "complete"
    }


# =============================================================================
# ROUTING
# =============================================================================

def route_orchestrator(state: OrchestratorState) -> Literal["execute", "synthesize", "end"]:
    """Route based on current state."""
    status = state["status"]

    if status == "executing":
        if state["current_task_idx"] < len(state["tasks"]):
            return "execute"
        return "synthesize"
    elif status == "synthesizing":
        return "synthesize"
    else:
        return "end"


# =============================================================================
# GRAPH CONSTRUCTION
# =============================================================================

def build_orchestrator_system() -> StateGraph:
    """Build the orchestrator-worker graph."""

    graph = StateGraph(OrchestratorState)

    # Add nodes
    graph.add_node("plan", create_plan)
    graph.add_node("execute", execute_task)
    graph.add_node("synthesize", synthesize_results)

    # Set entry point
    graph.set_entry_point("plan")

    # Add edges
    graph.add_conditional_edges(
        "plan",
        route_orchestrator,
        {"execute": "execute", "synthesize": "synthesize", "end": END}
    )
    graph.add_conditional_edges(
        "execute",
        route_orchestrator,
        {"execute": "execute", "synthesize": "synthesize", "end": END}
    )
    graph.add_edge("synthesize", END)

    return graph.compile()


# =============================================================================
# MAIN
# =============================================================================

def main():
    """Run orchestrator-worker demo."""
    if len(sys.argv) < 2:
        print("Usage: python orchestrator_worker.py '<complex request>'")
        print("\nExample requests:")
        print("  - 'Research AI safety trends and write a summary'")
        print("  - 'Analyze our sales data and create a report with recommendations'")
        print("  - 'Review this code, fix bugs, and add documentation'")
        sys.exit(1)

    request = sys.argv[1]

    print("\n🎭 Orchestrator-Worker Multi-Agent System")
    print(f"📝 Request: {request}")
    print("=" * 60)

    # Check for API key
    if not os.environ.get("ANTHROPIC_API_KEY"):
        print("❌ Error: ANTHROPIC_API_KEY environment variable not set")
        sys.exit(1)

    # Build system
    system = build_orchestrator_system()

    # Initial state
    initial_state = {
        "messages": [HumanMessage(content=request)],
        "original_request": request,
        "plan": [],
        "tasks": [],
        "current_task_idx": 0,
        "results": {},
        "final_output": "",
        "status": "planning"
    }

    # Run system
    final_state = None
    for event in system.stream(initial_state):
        for node, state in event.items():
            final_state = state

    print("\n" + "=" * 60)
    print("📤 FINAL OUTPUT")
    print("=" * 60)

    if final_state and final_state.get("final_output"):
        print(final_state["final_output"])
    else:
        print("No output generated")

    print("\n📊 Execution Summary:")
    if final_state:
        print(f"   - Tasks executed: {len(final_state.get('tasks', []))}")
        completed = sum(1 for t in final_state.get('tasks', []) if t['status'] == 'completed')
        print(f"   - Tasks completed: {completed}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Validation script for multi-agent skill.
Category: general
"""

import os
import sys
import yaml
import json
from pathlib import Path


def validate_config(config_path: str) -> dict:
    """
    Validate skill configuration file.

    Args:
        config_path: Path to config.yaml

    Returns:
        dict: Validation result with 'valid' and 'errors' keys
    """
    errors = []

    if not os.path.exists(config_path):
        return {"valid": False, "errors": ["Config file not found"]}

    try:
        with open(config_path, 'r') as f:
            config = yaml.safe_load(f)
    except yaml.YAMLError as e:
        return {"valid": False, "errors": [f"YAML parse error: {e}"]}

    # Validate required fields
    if 'skill' not in config:
        errors.append("Missing 'skill' section")
    else:
        if 'name' not in config['skill']:
            errors.append("Missing skill.name")
        if 'version' not in config['skill']:
            errors.append("Missing skill.version")

    # Validate settings
    if 'settings' in config:
        settings = config['settings']
        if 'log_level' in settings:
            valid_levels = ['debug', 'info', 'warn', 'error']
            if settings['log_level'] not in valid_levels:
                errors.append(f"Invalid log_level: {settings['log_level']}")

    return {
        "valid": len(errors) == 0,
        "errors": errors,
        "config": config if not errors else None
    }


def validate_skill_structure(skill_path: str) -> dict:
    """
    Validate skill directory structure.

    Args:
        skill_path: Path to skill directory

    Returns:
        dict: Structure validation result
    """
    required_dirs = ['assets', 'scripts', 'references']
    required_files = ['SKILL.md']

    errors = []

    # Check required files
    for file in required_files:
        if not os.path.exists(os.path.join(skill_path, file)):
            errors.append(f"Missing required file: {file}")

    # Check required directories
    for dir in required_dirs:
        dir_path = os.path.join(skill_path, dir)
        if not os.path.isdir(dir_path):
            errors.append(f"Missing required directory: {dir}/")
        else:
            # Check for real content (not just .gitkeep)
            files = [f for f in os.listdir(dir_path) if f != '.gitkeep']
            if not files:
                errors.append(f"Directory {dir}/ has no real content")

    return {
        "valid": len(errors) == 0,
        "errors": errors,
        "skill_name": os.path.basename(skill_path)
    }


def main():
    """Main validation entry point."""
    skill_path = Path(__file__).parent.parent

    print(f"Validating multi-agent skill...")
    print(f"Path: {skill_path}")

    # Validate structure
    structure_result = validate_skill_structure(str(skill_path))
    print(f"\nStructure validation: {'PASS' if structure_result['valid'] else 'FAIL'}")
    if structure_result['errors']:
        for error in structure_result['errors']:
            print(f"  - {error}")

    # Validate config
    config_path = skill_path / 'assets' / 'config.yaml'
    if config_path.exists():
        config_result = validate_config(str(config_path))
        print(f"\nConfig validation: {'PASS' if config_result['valid'] else 'FAIL'}")
        if config_result['errors']:
            for error in config_result['errors']:
                print(f"  - {error}")
    else:
        print("\nConfig validation: SKIPPED (no config.yaml)")

    # Summary
    all_valid = structure_result['valid']
    print(f"\n==================================================")
    print(f"Overall: {'VALID' if all_valid else 'INVALID'}")

    return 0 if all_valid else 1


if __name__ == "__main__":
    sys.exit(main())