pip install agentic-fleet

from agentic_fleet import Agent
from agentic_fleet.agent.specialized import (
    CodeAssistant,
    ResearchAssistant
)

# Basic agent
agent = Agent(
    name="assistant",
    model="gpt-4",
    temperature=0.7
)

# Specialized code assistant
code_agent = CodeAssistant(
    role="developer",
    capabilities=["python", "api_design"]
)

# Research assistant
research_agent = ResearchAssistant(
    role="researcher",
    capabilities=["data_analysis", "literature_review"]
)

from agentic_fleet import FleetChat
from agentic_fleet.fleets.coordination import CoordinationPattern

# Create fleet
fleet = FleetChat(
    name="development_team",
    agents={
        "architect": architect,
        "developer": developer
    },
    coordination_pattern=CoordinationPattern.SEQUENTIAL
)

# Start chat
response = await fleet.initiate_chat(
    initiator_id="architect",
    message="Design a REST API"
)

# Send message
response = await fleet.send_message(
    content="Add authentication to the API"
)

from agentic_fleet.memory import Memory, MemoryType

# Create memory
memory = Memory(
    type=MemoryType.VECTOR,
    capacity=1000
)

# Add to memory
await memory.add(
    key="project_requirements",
    value="Build REST API with auth"
)

# Query memory
results = await memory.search(
    query="API requirements",
    limit=5
)

from agentic_fleet.tools import (
    CodeAnalyzer,
    GitTool,
    DatabaseTool
)

# Code analysis tool
analyzer = CodeAnalyzer()
results = await analyzer.analyze_code(
    code="def hello(): pass"
)

# Git integration
git = GitTool(repo_path="./project")
await git.commit("Add new feature")

# Database operations
db = DatabaseTool(connection_string="...")
await db.query("SELECT * FROM users")

from agentic_fleet.agent import BaseAgent

class SecurityAgent(BaseAgent):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.capabilities.extend([
            "security_audit",
            "vulnerability_scan"
        ])
    
    async def audit_code(self, code: str):
        return await self.analyze(
            task="security_audit",
            content=code
        )

from agentic_fleet.fleets.coordination import BasePattern

class PriorityPattern(BasePattern):
    def __init__(self, priorities: dict):
        self.priorities = priorities
    
    async def next_agent(self, context):
        return max(
            self.priorities,
            key=self.priorities.get
        )

from agentic_fleet.middleware import Middleware

class LoggingMiddleware(Middleware):
    async def before_message(self, message):
        print(f"Sending: {message}")
        return message
    
    async def after_message(self, response):
        print(f"Received: {response}")
        return response

# Add to fleet
fleet.add_middleware(LoggingMiddleware())

from agentic_fleet.events import EventHandler

class ChatHandler(EventHandler):
    async def on_message(self, message):
        # Handle new message
        pass
    
    async def on_error(self, error):
        # Handle error
        pass

# Add to fleet
fleet.add_handler(ChatHandler())

try:
    response = await fleet.send_message(content)
except FleetError as e:
    logger.error(f"Fleet error: {e}")
except AgentError as e:
    logger.error(f"Agent error: {e}")

async with FleetChat(agents=agents) as fleet:
    await fleet.initiate_chat(message)

# Configure resource limits
fleet.configure(
    max_tokens=1000,
    timeout=30,
    retry_attempts=3
)

import logging
from agentic_fleet.logging import setup_logging

# Configure logging
setup_logging(
    level=logging.INFO,
    handlers=[
        "console",
        "file"
    ]
)

from agentic_fleet.examples.code import create_dev_fleet

fleet = await create_dev_fleet(
    project_type="web_api",
    team_size=3
)

from agentic_fleet.examples.research import create_research_fleet

fleet = await create_research_fleet(
    topic="AI Safety",
    depth="comprehensive"
)

from agentic_fleet.testing import FleetTestCase

class TestDevFleet(FleetTestCase):
    async def setUp(self):
        self.fleet = await create_dev_fleet()
    
    async def test_code_generation(self):
        response = await self.fleet.generate_code(
            prompt="Create a FastAPI app"
        )
        self.assertValidPython(response.code)

from agentic_fleet.debug import FleetDebugger

debugger = FleetDebugger(fleet)
debugger.enable_tracing()

# Get debug info
info = await debugger.get_state()
print(info.messages)
print(info.memory)
print(info.performance)