​
AgenticFleet Overview

AgenticFleet is a powerful framework for orchestrating multi-agent systems using different fleet patterns. The framework is built around the concept of specialized fleets that coordinate multiple agents for specific tasks.

​
Core Components

​
1. Fleet Patterns

AgenticFleet provides several coordination patterns to organize agent interactions:

​
Swarm Pattern (SwarmCoordinator)

  • Enables parallel processing with multiple agents
  • Weights agents based on expertise
  • Best for tasks requiring diverse perspectives
  • Example: Code review with multiple specialists

​
Sequential Pattern (SequentialCoordinator)

  • Fixed sequence of agent interactions
  • Clear workflow progression
  • Ideal for step-by-step processes
  • Example: Software development lifecycle

​
Routed Pattern (RoutedCoordinator)

  • Content-based routing between agents
  • Conditional branching support
  • Perfect for complex workflows
  • Example: Code review with feedback loops

​
Round Robin Pattern (RoundRobinCoordinator)

  • Fair distribution of tasks
  • Equal participation
  • Simple but effective
  • Example: Load balancing tasks

​
2. Specialized Fleets

AgenticFleet includes pre-built specialized fleets for common use cases:

​
CodeFleet

fleet = CodeFleet(
    agents={
        "architect": architect_agent,
        "developer": developer_agent,
        "reviewer": reviewer_agent
    },
    project_name="MyProject",
    coordination_pattern=CoordinationPattern.SEQUENTIAL
)
  • Specialized for software development
  • Built-in code review workflow
  • Integrated with version control
  • Multiple coordination patterns

​
ResearchFleet

fleet = ResearchFleet(
    agents={
        "researcher": researcher_agent,
        "analyst": analyst_agent,
        "writer": writer_agent
    },
    research_topic="AI Safety",
    research_objective="Literature Review"
)
  • Document analysis and research
  • Citation management
  • Collaborative writing
  • Knowledge synthesis

​
ExecutionFleet

fleet = ExecutionFleet(
    name="TestRunner",
    code_execution_config={
        "work_dir": "./tests",
        "use_docker": True
    }
)
  • Code execution and testing
  • Environment management
  • Result validation
  • Error handling

​
3. Agent Components

​
Base Agents

  • BaseAgent: Foundation for all agents
  • ConversableAgent: Enhanced conversation capabilities
  • FleetAssistant: Fleet-specific assistant features

​
Agent Roles

class AgentRole(str, Enum):
    USER = "user"
    ASSISTANT = "assistant"
    SYSTEM = "system"
    FUNCTION = "function"
    CRITIC = "critic"
    EXECUTOR = "executor"

​
Agent States

class AgentState(str, Enum):
    INITIALIZING = "initializing"
    IDLE = "idle"
    BUSY = "busy"
    PAUSED = "paused"
    STOPPED = "stopped"
    ERROR = "error"

​
4. Communication

​
Message Types

class MessageType(str, Enum):
    TEXT = "text"
    CODE = "code"
    FUNCTION_CALL = "function_call"
    TOOL_CALL = "tool_call"
    ERROR = "error"
    RESULT = "result"

​
Agent Messages

message = AgentMessage(
    source="developer",
    target="reviewer",
    content="Code review request",
    type=MessageType.TEXT,
    priority=5
)

​
5. Workflow Management

workflow = Workflow(
    name="Feature Development",
    tasks={
        "design": Task(assignee="architect"),
        "implement": Task(assignee="developer"),
        "review": Task(assignee="reviewer")
    }
)

​
Best Practices

​
1. Fleet Pattern Selection

Choose patterns based on your use case:

  • Swarm: When parallel processing is beneficial
  • Sequential: For step-by-step workflows
  • Routed: For complex, conditional flows
  • Round Robin: For fair task distribution

​
2. Agent Design

  • Keep agents focused and specialized
  • Use appropriate system messages
  • Implement proper error handling
  • Monitor agent performance metrics

​
3. Fleet Composition

  • Balance agent capabilities
  • Define clear agent roles
  • Set up proper coordination patterns
  • Monitor fleet performance

​
4. Workflow Management

  • Break down complex tasks
  • Set up proper dependencies
  • Monitor progress and metrics
  • Handle errors gracefully

​
Extensions

AgenticFleet can be extended through:

​
1. Custom Agents

class CustomAgent(FleetAssistant):
    def __init__(self, name: str, **kwargs):
        super().__init__(
            name=name,
            system_message=CUSTOM_PROMPT,
            **kwargs
        )

​
2. Custom Fleets

class CustomFleet(FleetChat):
    def __init__(
        self,
        agents: Dict[str, BaseAgent],
        custom_config: Dict[str, Any],
        **kwargs
    ):
        super().__init__(
            agents=agents,
            coordination_pattern=CoordinationPattern.ROUTED,
            **kwargs
        )

​
3. Custom Coordinators

class CustomCoordinator(FleetCoordinator):
    def __init__(
        self,
        fleet: 'FleetChat',
        custom_rules: Dict[str, Any],
        **kwargs
    ):
        super().__init__(fleet, **kwargs)
        self.rules = custom_rules

​
Getting Started

  1. Install AgenticFleet:
pip install agenticfleet
  1. Create a fleet:
from agentic_fleet import CodeFleet, FleetAssistant

# Create agents
architect = FleetAssistant(name="architect")
developer = FleetAssistant(name="developer")
reviewer = FleetAssistant(name="reviewer")

# Create fleet
fleet = CodeFleet(
    agents={
        "architect": architect,
        "developer": developer,
        "reviewer": reviewer
    },
    project_name="MyProject"
)

# Start task
fleet.start_task("Implement new feature")

​
Advanced Usage

​
1. Custom Workflows

from agentic_fleet import Workflow, Task

workflow = Workflow(
    name="Custom Development",
    tasks={
        "design": Task(
            assignee="architect",
            dependencies=[]
        ),
        "implement": Task(
            assignee="developer",
            dependencies=["design"]
        ),
        "review": Task(
            assignee="reviewer",
            dependencies=["implement"]
        )
    }
)

fleet.execute_workflow(workflow)

​
2. Advanced Routing

from agentic_fleet import RoutedCoordinator

coordinator = RoutedCoordinator(
    fleet=fleet,
    routes=[
        ("architect", "developer"),
        ("developer", "reviewer"),
        ("reviewer", "tester")
    ],
    routing_rules={
        ("architect", "developer"): ["implement", "code"],
        ("developer", "reviewer"): ["review", "check"],
        ("reviewer", "tester"): ["test", "verify"]
    }
)

​
3. Custom Metrics

from agentic_fleet import AgentMetrics

metrics = AgentMetrics(
    total_messages=0,
    response_times=[],
    error_count=0,
    tool_usage={},
    token_usage={},
    resource_usage={}
)

agent.update_metrics(metrics)