AgentSwarm Orchestrator - Python Implementation

This Python implementation of AgentSwarm Orchestrator was inspired by https://kimik25.com. It provides a simple, intuitive API for managing autonomous agent swarms with async support.

Features

• Simple API: Easy to use with Pythonic design patterns
• Async Support: Built on asyncio for efficient concurrent operations
• 256K Context: Inspired by Kimi K2.5's extended context window
• Type Hints: Full type annotations for better IDE support
• Dataclasses: Clean, structured data models

Installation

pip install kimi25-opensource

Usage

import asyncio
from agent_swarm import AgentSwarm, SwarmConfig, TaskMessage
import uuid

async def main():
    # Initialize swarm with 256K context window
    config = SwarmConfig(
        max_agents=100,
        context_window=256000,
        topology='mesh'
    )

    swarm = AgentSwarm(config)

    # Spawn specialized agents
    agent = swarm.spawn_agent('data-processor', 'analysis', 'transform')

    # Dispatch messages
    agent.dispatch_message(TaskMessage(
        id=str(uuid.uuid4()),
        msg_type='process',
        payload={'data': 'example'}
    ))

    # Get swarm statistics
    print(swarm.get_stats())

    # Clean shutdown
    await swarm.shutdown()

asyncio.run(main())

API Reference

Classes

SwarmConfig

Configuration dataclass for swarm initialization.

Parameters:

• max_agents (int): Maximum number of agents (default: 100)
• context_window (int): Token context limit (default: 256000)
• topology (str): Swarm topology (default: 'mesh')

TaskMessage

Message dataclass for inter-agent communication.

Parameters:

• id (str): Unique message identifier
• msg_type (str): Message type
• payload (Dict[str, Any]): Message data
• from_agent (str): Source agent ID (auto-set)
• to_agent (str): Target agent ID

Agent

Represents a single autonomous agent.

Methods:

• dispatch_message(msg: TaskMessage) -> bool: Send a message
• process_messages(handler: Callable) -> Awaitable: Process queued messages

AgentSwarm

Main swarm coordinator class.

Methods:

• spawn_agent(agent_id: str, *capabilities) -> Agent: Create a new agent
• broadcast_message(from_agent, msg_type, payload) -> int: Send to all agents
• get_agent(agent_id: str) -> Agent: Retrieve agent by ID
• get_stats() -> Dict: Get swarm statistics
• shutdown() -> Awaitable: Gracefully shutdown

Enums

TopologyType

Supported swarm topologies:

• HIERARCHICAL: Tree-based coordination
• MESH: Peer-to-peer communication
• HYBRID: Combined approach

Examples

Async Message Processing

import asyncio
from agent_swarm import AgentSwarm, SwarmConfig

async def message_handler(msg):
    print(f"Processing {msg.msg_type}")
    # Handle message asynchronously
    await asyncio.sleep(0.1)

async def main():
    swarm = AgentSwarm(SwarmConfig(max_agents=50))
    agent = swarm.spawn_agent('worker', 'processing')

    await agent.process_messages(message_handler)

asyncio.run(main())

Custom Topology

from agent_swarm import SwarmConfig, TopologyType

config = SwarmConfig(
    max_agents=200,
    context_window=512000,
    topology=TopologyType.HIERARCHICAL.value
)

Development

Running Tests

pip install -e ".[dev]"
pytest

Code Formatting

black agent_swarm/
mypy agent_swarm/

Performance

The Python implementation is optimized for:

• Async I/O: Efficient concurrent operations via asyncio
• Memory efficiency: Minimal overhead per agent
• Developer productivity: Clean, readable code

Links

• Source: https://kimik25.com
• Repository: https://onedao/wddaily/kimi25-opensource
• Documentation: https://onedao/wddaily/kimi25-opensource#readme
• PyPI Package: https://pypi.org/project/kimi25-opensource

License

MIT License - See LICENSE file for details.