AI-Agents-Orchestrator

Features Documentation

Table of Contents

Core Features

Multi-Agent Collaboration

The orchestrator coordinates multiple AI coding assistants to work together on complex tasks.

graph LR
    A[Task] --> B[Codex<br/>Implementation]
    B --> C[Gemini<br/>Review]
    C --> D[Claude<br/>Refinement]
    D --> E[Quality Code]

Benefits:

Supported AI Agents:

Configurable Workflows

Define how agents collaborate for different scenarios.

Workflow Agents Iterations Use Case
default Codex → Gemini → Claude 3 Production code with quality assurance
quick Codex only 1 Fast prototyping
thorough Codex → Copilot → Gemini → Claude → Gemini 5 Mission-critical code
review-only Gemini → Claude 2 Existing code improvement
document Claude → Gemini 2 Documentation generation

Custom Workflows:

workflows:
  custom:
    max_iterations: 4
    min_suggestions_threshold: 5
    steps:
      - agent: "codex"
        task: "implement"
      - agent: "gemini"
        task: "security_review"
      - agent: "claude"
        task: "refine"

CLI Features

Interactive Shell

Interactive Shell

A powerful REPL-style interface for natural conversations with AI agents.

Features:

Example Session:

$ ./ai-orchestrator shell

orchestrator (default) > create a REST API for blog posts

✓ Task completed successfully!
📁 Generated Files:
  📄 api/blog.py
  📄 api/models.py
  📄 api/routes.py

Workspace: ./workspace

orchestrator (default) > add authentication with JWT

💡 Detected as follow-up to previous task
✓ Authentication added!
📁 Generated Files:
  📄 api/auth.py
  📄 api/middleware.py

orchestrator (default) > also add rate limiting

💡 Detected as follow-up to previous task
✓ Rate limiting implemented!

orchestrator (default) > /save blog-api-project
✓ Session saved to: sessions/blog-api-project.json

orchestrator (default) > /exit
Goodbye!

CLI Commands

Command Description Example
/help Show all available commands /help
/followup <msg> Explicit follow-up to previous task /followup add tests
/agents List all available agents and status /agents
/workflows List all available workflows /workflows
/switch <agent> Switch to specific agent /switch claude
/workflow <name> Change current workflow /workflow thorough
/history Show conversation history /history
/context Show current context /context
/save [name] Save current session /save my-project
/load <name> Load saved session /load my-project
/reset Clear all context /reset
/info Show system information /info
/clear Clear screen /clear
/exit, /quit Exit the shell /exit

One-Shot Mode

Execute single tasks without interactive shell:

# Basic usage
./ai-orchestrator run "Create a Python calculator"

# With workflow selection
./ai-orchestrator run "Build authentication system" --workflow default

# With options
./ai-orchestrator run "Refactor database layer" \
  --workflow thorough \
  --max-iterations 5 \
  --verbose

# Dry run to see execution plan
./ai-orchestrator run "Add error handling" --dry-run

# Custom output directory
./ai-orchestrator run "Generate CLI tool" --output ./my-output

Smart Follow-Up Detection

The CLI automatically detects when messages should continue from previous tasks.

Auto-Detected Keywords:

Behavior:

graph TD
    A[User Input] --> B{Has Previous Task?}
    B -->|No| C[Execute New Task]
    B -->|Yes| D{Contains Keywords?}
    D -->|Yes| E[Auto Follow-Up]
    D -->|No| F{Message Short?}
    F -->|Yes| G[Prompt User]
    F -->|No| H[Execute New Task]
    G --> I{User Choice}
    I -->|Continue| E
    I -->|New| H

Web UI Features

Web UI

Modern Interface

Built with Vue 3, the Web UI provides a rich visual experience:

Technology Stack:

Main Interface Components

Left Sidebar:

Main Content Area:

Header:

Real-Time Features

sequenceDiagram
    participant U as User Browser
    participant F as Flask Server
    participant O as Orchestrator

    U->>F: Submit Task (HTTP)
    F->>O: Execute Task

    loop Progress Updates
        O->>F: Progress Event
        F->>U: Socket.IO Update
        U->>U: Update UI
    end

    O->>F: Task Complete
    F->>U: Final Result
    U->>U: Display Output

Real-time Updates:

Monaco Code Editor

Full-featured code editor integrated into the UI:

Features:

File Operations:

Conversation Mode Toggle

Enable ChatGPT-like continuous conversations:

stateDiagram-v2
    [*] --> ConversationOff
    ConversationOff --> ConversationOn: Enable Toggle
    ConversationOn --> ConversationOff: Disable Toggle

    ConversationOff: Each message = New Task
    ConversationOn: Messages continue previous task

When Enabled:

When Disabled:

Follow-Up Section

Separate green section for quick additions after task completion:

Features:

Use Cases:

Conversation Mode

How It Works

CLI Implementation:

graph TD
    A[User Message] --> B{Analyze Message}
    B --> C{Short + Keywords?}
    C -->|Yes| D[Auto Follow-Up]
    C -->|No| E{Ambiguous?}
    E -->|Yes| F[Prompt User]
    E -->|No| G[New Task]
    F --> H{User Choice}
    H -->|Continue| D
    H -->|New| G

UI Implementation:

Context Preservation

What gets preserved in conversation mode:

context:
  previous_task: "create a REST API"
  previous_output: "Full AI response..."
  generated_files:
    - "api/routes.py"
    - "api/models.py"
  workspace: "./workspace"
  workflow: "default"
  timestamp: "2024-01-15T10:30:00Z"

Best Practices

Use Conversation Mode For:

Don’t Use For:

Workflow System

Workflow Execution Flow

graph TD
    START[Start] --> LOAD[Load Workflow Config]
    LOAD --> VALIDATE[Validate Configuration]
    VALIDATE --> INIT[Initialize Agents]
    INIT --> ITER{Iteration < Max?}

    ITER -->|Yes| EXEC[Execute Workflow Steps]
    EXEC --> AGENT1[Agent 1]
    AGENT1 --> AGENT2[Agent 2]
    AGENT2 --> AGENT3[Agent 3]

    AGENT3 --> COLLECT[Collect Feedback]
    COLLECT --> CHECK{Sufficient<br/>Suggestions?}

    CHECK -->|Yes| UPDATE[Update Context]
    UPDATE --> ITER

    CHECK -->|No| AGG[Aggregate Results]
    ITER -->|No| AGG

    AGG --> REPORT[Generate Report]
    REPORT --> END[End]

Iteration Control

Workflows can iterate until quality thresholds are met:

Iteration Triggers:

Stop Conditions:

Example:

settings:
  max_iterations: 3
  min_suggestions_threshold: 5
  quality_threshold: 0.8

AI Agent Integration

Agent Status Monitoring

graph TD
    A[Check Agent] --> B{Command Exists?}
    B -->|Yes| C{Authenticated?}
    B -->|No| D[Not Available]
    C -->|Yes| E[Available]
    C -->|No| F[Auth Required]

    E --> G[Test Execution]
    G --> H{Success?}
    H -->|Yes| I[Ready]
    H -->|No| J[Error State]

Status Indicators:

Agent Capabilities

Codex (OpenAI):

Gemini (Google):

Claude (Anthropic):

Copilot (GitHub):

Session Management

Save and Load Sessions

Persist entire conversation contexts for later:

What Gets Saved:

CLI Usage:

# Save session
orchestrator > /save my-project

# Load session
./ai-orchestrator shell --load my-project

# Or within shell
orchestrator > /load my-project

# List saved sessions
./ai-orchestrator sessions

File Format (JSON):

{
  "session_id": "uuid-here",
  "created_at": "2024-01-15T10:00:00Z",
  "workflow": "default",
  "conversation_history": [
    {
      "role": "user",
      "content": "create a REST API",
      "timestamp": "2024-01-15T10:00:00Z"
    },
    {
      "role": "assistant",
      "content": "API created...",
      "files": ["api.py", "models.py"],
      "timestamp": "2024-01-15T10:01:30Z"
    }
  ],
  "workspace": "./workspace",
  "metadata": {}
}

File Management

Workspace Organization

workspace/
├── session-uuid/
│   ├── api/
│   │   ├── routes.py
│   │   ├── models.py
│   │   └── __init__.py
│   ├── tests/
│   │   └── test_api.py
│   └── docs/
│       └── README.md

File Tracking

The orchestrator tracks all generated files:

graph LR
    A[Agent Response] --> B[Extract Code Blocks]
    B --> C[Parse File Paths]
    C --> D[Validate Paths]
    D --> E{Exists?}
    E -->|Yes| F[Create Backup]
    E -->|No| G[Create File]
    F --> G
    G --> H[Register File]
    H --> I[Update UI]

Features:

File Operations

CLI:

# Files listed after task completion
📁 Generated Files:
  📄 api/routes.py
  📄 api/models.py
  📄 tests/test_api.py

Workspace: ./workspace/session-abc123

UI:

Security Features

Input Validation

All user inputs are validated and sanitized:

class SecurityValidator:
    def validate_task(self, task: str) -> bool:
        # Command injection prevention
        if self._contains_shell_metacharacters(task):
            raise SecurityError("Potential command injection")

        # Path traversal prevention
        if self._contains_path_traversal(task):
            raise SecurityError("Path traversal detected")

        # Length validation
        if len(task) > MAX_TASK_LENGTH:
            raise ValidationError("Task too long")

        return True

Protected Against:

Rate Limiting

Token bucket algorithm prevents abuse:

graph LR
    A[Request] --> B{Tokens Available?}
    B -->|Yes| C[Consume Token]
    B -->|No| D[Rate Limit Error]
    C --> E[Process Request]
    E --> F[Refill Tokens Over Time]

Configuration:

rate_limiting:
  enabled: true
  requests_per_minute: 10
  burst_size: 20
  per_user: true

Audit Logging

All security-relevant events are logged:

logger.info(
    "security_event",
    event_type="task_execution",
    user_id="user-123",
    task_hash="abc...",
    timestamp="2024-01-15T10:00:00Z",
    success=True
)

Logged Events:

Monitoring & Metrics

Prometheus Metrics

Comprehensive metrics for production monitoring:

Task Metrics:

orchestrator_tasks_total
orchestrator_task_duration_seconds
orchestrator_task_failures_total
orchestrator_task_success_rate

Agent Metrics:

orchestrator_agent_calls_total
orchestrator_agent_errors_total
orchestrator_agent_response_time_seconds
orchestrator_agent_availability

System Metrics:

orchestrator_cache_hits_total
orchestrator_cache_misses_total
orchestrator_active_sessions
orchestrator_memory_usage_bytes

Health Checks

Endpoints:

Health Check Response:

{
  "status": "healthy",
  "version": "1.0.0",
  "agents": {
    "claude": "available",
    "codex": "available",
    "gemini": "available"
  },
  "uptime_seconds": 3600,
  "active_sessions": 5
}

Advanced Features

Caching System

Multi-layer caching for performance:

Cache Layers:

  1. In-Memory - Fast, 5-minute TTL
  2. File-Based - Persistent, 24-hour TTL
  3. Distributed - Redis (optional)

Cache Strategy:

graph TD
    A[Request] --> B{Memory Cache?}
    B -->|Hit| C[Return Result]
    B -->|Miss| D{File Cache?}
    D -->|Hit| E[Load to Memory]
    E --> C
    D -->|Miss| F[Execute Task]
    F --> G[Cache Result]
    G --> C

Async Execution

Parallel execution where possible:

async def execute_parallel_agents(agents, task):
    results = await asyncio.gather(
        *[agent.execute_async(task) for agent in agents],
        return_exceptions=True
    )
    return [r for r in results if not isinstance(r, Exception)]

Retry Logic

Automatic retry with exponential backoff:

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=2, max=10),
    retry=retry_if_exception_type(TransientError)
)
def execute_with_retry(agent, task):
    return agent.execute(task)

Circuit Breaker

Prevent cascading failures:

stateDiagram-v2
    [*] --> Closed
    Closed --> Open: Failure Threshold
    Open --> HalfOpen: Timeout
    HalfOpen --> Closed: Success
    HalfOpen --> Open: Failure

    Closed: Normal Operation
    Open: Reject Requests
    HalfOpen: Test Recovery

For more information:

This site is open source. Improve this page.