nano-swarm 1.1.0

Swarm intelligence coordination protocol — nano-agents self-organize to execute business processes

Nano AI — Swarm Intelligence Coordination Protocol

Nano-agents self-organize to execute business processes. No central controller.

┌──────────────────────────────────────────────────────────────────────────┐
│                            SOP (YAML DAG)                                │
│  collect_info → create_accounts → ┬─ setup_vpn (conditional) ─┐         │
│                                   ├─ assign_training ──────────┤         │
│                                   └─ notify_manager  ──────────┴→ welcome│
└──────────────────────────────────────────────────────────────────────────┘
        │
        ▼
┌──────────────────────────────────────────────────────────────────────────┐
│                       Message Bus (pub/sub)                               │
│    TASK_OFFER → BID → TASK_ASSIGN → TASK_COMPLETE / TASK_FAILED          │
└──────────────────────────────────────────────────────────────────────────┘
        │           │           │           │           │           │
        ▼           ▼           ▼           ▼           ▼           ▼
   ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
   │ Info   │ │Account │ │  VPN   │ │Training│ │ Notify │ │Welcome │
   │Collect │ │Provis. │ │ Setup  │ │ Assign │ │Manager │ │ Send   │
   └────────┘ └────────┘ └────────┘ └────────┘ └────────┘ └────────┘

Features

• Capability-based bidding — agents self-select via confidence scores, no hardcoded routing
• Confidence-based routing — pluggable bid scoring with historical success rate, speed, and load balancing
• DAG execution — parallel branches run automatically, dependencies respected
• SOP composition — compose complex workflows from reusable sub-SOPs via sop_ref
• Conditional steps — skip steps based on runtime context (condition field on SOPStep)
• LLM-powered agents — LLMAgent delegates step execution to any LLM provider with JSON parsing
• CLI runner — nano run and nano validate for one-command SOP execution
• Slack bot — /nano slash command and @mention interface with thread-based progress and on-the-fly SOP generation
• Agent delegation — agents can spawn sub-swarms to break work into smaller workflows
• Agent memory — persistent per-agent state and performance history across executions
• SOP autopsy — rule-based and LLM-powered post-mortem analysis with improvement suggestions
• Execution analytics — step durations, success rates, retry counts, slowest-step detection
• SOP generation — natural language → validated YAML SOP via any LLM provider
• Event persistence — full audit trail of every bus message (JSONL)
• Integration framework — abstract base + Slack webhook integration
• Fault tolerance — configurable retry policy with automatic re-offering
• Safe condition evaluation — AST-based expression parsing via simpleeval (no eval())
• Provider-agnostic LLM — OpenAI, Anthropic, Ollama, or Mock (no API keys for testing)
• Full observability — every state transition emits a structured event

How it works

1. SOP defines a business process as a DAG of steps in YAML
2. Swarm loads the SOP, identifies ready steps, and publishes TASK_OFFER messages
3. Agents with matching capabilities respond with BID (confidence score)
4. Swarm picks the best bid, sends TASK_ASSIGN to the winner
5. Agent executes and publishes TASK_COMPLETE or TASK_FAILED
6. Swarm updates the DAG — newly unblocked steps run in parallel
7. Failed tasks are automatically re-offered (configurable retry policy)

Quickstart

# Install
pip install -e ".[dev]"

# Run tests (no API keys needed)
pytest tests/ -v

# Validate an SOP
nano validate my_sop.yaml

# Execute an SOP (requires LLM provider config)
nano run my_sop.yaml --context '{"name": "Alice"}' --llm-provider openai

# Start the Slack bot (requires SLACK_BOT_TOKEN + SLACK_APP_TOKEN)
nano bot

Project structure

nano/
├── __init__.py           # Public API (33 exports)
├── agent.py              # NanoAgent base class, DelegatingAgent, TaskResult
├── message.py            # Message model + MessageType enum
├── bus.py                # MessageBus ABC + InProcessBus
├── sop.py                # SOP + SOPStep models, YAML loader, DAG validation
├── protocol.py           # BidEntry, RetryPolicy, select_best_bid
├── scoring.py            # BidScorer ABC, ConfidenceScorer, WeightedBidScorer
├── swarm.py              # Swarm orchestrator (coordination loop)
├── llm_agent.py          # LLMAgent — LLM-powered step execution
├── cli.py                # CLI runner (nano run / nano validate)
├── memory.py             # AgentMemory + AgentMemoryStore (persistent state)
├── autopsy.py            # SuggestionAnalyzer + AutopsyLLMAnalyzer
├── analytics.py          # SwarmEventRecorder + ExecutionAnalytics
├── persistence.py        # EventStore (JSONL audit trail)
├── sop_generator.py      # AI-powered SOP generation from natural language
├── bot/
│   ├── app.py            # Slack Bolt application (slash command, @mention, actions)
│   ├── cli.py            # Bot entrypoint (env vars, Socket Mode startup)
│   ├── intent.py         # LLM-based intent parsing from natural language
│   ├── registry.py       # SOP template discovery and lookup
│   ├── runner.py         # High-level SOP executor with live event callbacks
│   └── formatter.py      # Swarm events → Slack Block Kit payloads
├── integrations/
│   ├── base.py           # Integration ABC
│   ├── mock.py           # MockIntegration (testing)
│   └── slack.py          # SlackWebhookIntegration (httpx)
└── llm/
    ├── base.py           # LLMProvider ABC
    ├── mock.py           # MockProvider (testing)
    ├── openai.py         # OpenAI wrapper
    ├── anthropic.py      # Anthropic wrapper
    └── ollama.py         # Ollama wrapper

Define an SOP

name: Employee Onboarding
steps:
  - id: collect_info
    name: Collect Employee Information
    capability: info_collection
    outputs: [employee_name, role, department]

  - id: create_accounts
    name: Provision Accounts
    capability: account_provisioning
    depends_on: [collect_info]
    inputs: [employee_name]
    outputs: [email_created, slack_handle]

  - id: setup_vpn
    name: Setup VPN Access
    capability: vpn_setup
    depends_on: [create_accounts]
    condition: "role == 'Backend Developer' or department == 'Security'"
    outputs: [vpn_configured]

Steps with a condition are evaluated at runtime against the accumulated context and upstream outputs. If the condition is false, the step is skipped (not failed) and downstream dependencies treat it as satisfied.

SOP composition

Build complex workflows from reusable building blocks. A step with sop_ref inlines a sub-SOP at load time — sub-step IDs are namespaced (parent.child), dependencies are rewired automatically, and the Swarm sees a single flat DAG.

# account_provisioning.sop.yaml  (reusable sub-SOP)
name: Account Provisioning
steps:
  - id: create_accounts
    name: Provision Accounts
    capability: account_provisioning
  - id: setup_vpn
    name: Setup VPN
    capability: vpn_setup
    depends_on: [create_accounts]
    condition: "role == 'Engineer'"

# onboarding.yaml  (parent SOP)
name: Employee Onboarding
steps:
  - id: collect_info
    name: Collect Info
    capability: info_collection
  - id: provision
    name: Account Provisioning
    sop_ref: account_provisioning.sop.yaml
    depends_on: [collect_info]
  - id: send_welcome
    name: Welcome
    capability: welcome_delivery
    depends_on: [provision]

After expansion, the Swarm executes: collect_info → provision.create_accounts → provision.setup_vpn → send_welcome. Recursive composition (sub-SOPs referencing other sub-SOPs) is supported with cycle detection and a depth limit of 10.

Create a custom agent

from nano import NanoAgent, TaskResult

class MyAgent(NanoAgent):
    def __init__(self):
        super().__init__(name="MyAgent", capabilities={"my_capability"})

    async def process(self, task_payload):
        inputs = task_payload.get("inputs", {})
        # Your logic here
        return TaskResult(success=True, outputs={"result": "done"})

Run a swarm

import asyncio
from nano import InProcessBus, SOP, Swarm, SwarmEventRecorder, ExecutionAnalytics

async def main():
    bus = InProcessBus()
    sop = SOP.from_yaml("my_sop.yaml")

    recorder = SwarmEventRecorder()
    swarm = Swarm(bus=bus, sop=sop, on_event=recorder)

    swarm.register(MyAgent())
    outputs = await swarm.execute(context={"employee_name": "Alice", "role": "Engineer"})

    # Analyse execution
    analytics = ExecutionAnalytics(recorder.events)
    print(analytics.summary())

asyncio.run(main())

Execution analytics

ExecutionAnalytics operates on swarm events captured by SwarmEventRecorder:

analytics = ExecutionAnalytics(recorder.events)

analytics.step_durations()    # {"step_a": 0.12, "step_b": 0.25}
analytics.step_outcomes()     # {"step_a": "completed", "step_b": "skipped"}
analytics.success_rate()      # 0.80
analytics.slowest_steps(n=3)  # [("step_b", 0.25), ("step_a", 0.12)]
analytics.retry_count()       # {"step_c": 2}
analytics.summary()           # combined dict of all the above

Event persistence

from nano import EventStore, InProcessBus

bus = InProcessBus()
store = EventStore(path="events.jsonl")
store.attach(bus)

# ... run swarm ...

store.detach(bus)
store.close()

# Replay later
events = EventStore.load("events.jsonl")

Confidence-based routing

By default the Swarm picks the agent with the highest confidence. Inject a WeightedBidScorer to factor in historical success rate, speed, and load:

from nano import WeightedBidScorer, AgentHistory, Swarm, InProcessBus, SOP

scorer = WeightedBidScorer(
    confidence_weight=0.4,
    success_rate_weight=0.3,
    speed_weight=0.1,
    load_weight=0.2,
)

# Optionally seed with cross-execution history
history = {
    (agent.id, "account_provisioning"): AgentHistory(success_count=50, failure_count=2, total_duration=25.0),
}

swarm = Swarm(bus=bus, sop=sop, scorer=scorer, history=history)

The Swarm automatically tracks agent performance during execution and uses it for subsequent bid scoring within the same run.

SOP generation

from nano import generate_sop
from nano.llm.mock import MockProvider

llm = MockProvider(...)  # or OpenAIProvider, AnthropicProvider, OllamaProvider
sop = await generate_sop("Onboard a new engineer with accounts and training", llm)
print(sop.name, len(sop.steps), "steps")

LLM-powered agents

LLMAgent delegates step execution to any LLM provider. It builds a prompt from the step description and inputs, calls the LLM, and parses the response as JSON (with a plain-text fallback).

from nano import LLMAgent
from nano.llm.openai import OpenAIProvider

llm = OpenAIProvider()  # uses OPENAI_API_KEY env var

agent = LLMAgent(
    name="Analyst",
    capabilities={"data_analysis", "report_generation"},
    llm=llm,
    temperature=0.3,
)

Override _build_prompt() in a subclass for capability-specific prompt engineering.

CLI

The nano CLI runs SOPs from the command line with auto-generated LLM agents:

# Validate SOP structure (no LLM needed)
nano validate onboarding.yaml

# Execute with OpenAI (auto-creates one LLMAgent per capability)
nano run onboarding.yaml --llm-provider openai --context '{"name": "Alice"}'

# Use a specific model and save outputs
nano run onboarding.yaml --llm-provider openai --llm-model gpt-4o-mini --output results.json

# Verbose mode (shows every swarm event)
nano run onboarding.yaml --llm-provider mock --verbose

Slack bot

Nano includes an interactive Slack bot that receives natural language commands, matches them to SOP templates (or generates one on the fly), executes the swarm, and posts thread-based progress updates.

Install

pip install nano-swarm[bot]   # adds slack_bolt

Slack app setup

Create a Slack app at api.slack.com/apps:

1. Socket Mode — enable under Settings → Socket Mode, generate an app-level token (xapp-...)
2. Bot Token Scopes (under OAuth & Permissions): chat:write, commands, app_mentions:read
3. Slash Command — create /nano pointing to your app
4. Event Subscriptions — subscribe to the app_mention bot event
5. Install the app to your workspace and copy the Bot User OAuth Token (xoxb-...)

Environment variables

Variable	Required	Default	Description
SLACK_BOT_TOKEN	Yes	—	Bot User OAuth Token (xoxb-...)
SLACK_APP_TOKEN	Yes	—	App-Level Token (xapp-...) for Socket Mode
NANO_SOP_DIR	No	./sops	Directory containing SOP YAML templates
NANO_LLM_PROVIDER	No	mock	LLM provider: mock, openai, anthropic, ollama
NANO_LLM_MODEL	No	—	Model name override
NANO_LOG_LEVEL	No	INFO	Logging level

Start the bot

export SLACK_BOT_TOKEN=xoxb-...
export SLACK_APP_TOKEN=xapp-...
export NANO_LLM_PROVIDER=openai
export NANO_SOP_DIR=./sops

nano bot

Usage

• Slash command: /nano onboard Sarah Chen as Backend Developer in Engineering
• Mention: @Nano run the deploy pipeline for staging

If a matching SOP template exists in NANO_SOP_DIR, it executes immediately. If not, the bot generates one on the fly and posts a preview with Approve / Cancel buttons before running.

All progress updates are posted as thread replies — only major events (step completed, skipped, failed, swarm finished) to keep channels clean.

Agent delegation

DelegatingAgent can spawn isolated sub-swarms inside process() to break work into smaller workflows:

from nano import DelegatingAgent, SOP, TaskResult

class OrchestratorAgent(DelegatingAgent):
    async def process(self, task_payload):
        sub_sop = SOP.from_yaml("sub_workflow.yaml")
        sub_agents = [...]  # agents for the sub-workflow
        result = await self.delegate(sub_sop, sub_agents, task_payload.get("inputs", {}))
        return result

Each delegation creates its own bus + swarm (no message collision). Depth limit is configurable (max_delegation_depth, default 3).

Agent memory

Persistent per-agent state survives across executions:

from nano import AgentMemoryStore, Swarm

store = AgentMemoryStore("./agent_memories")

# After execution — save agent performance data
swarm.save_agent_memories(store)

# Next execution — load historical performance for smarter routing
history = Swarm.load_history_from_store([agent_a, agent_b], store)
swarm = Swarm(bus=bus, sop=sop, history=history)

SOP autopsy

Analyse execution history to generate improvement suggestions:

from nano import SuggestionAnalyzer, AutopsyLLMAnalyzer

# Rule-based analysis (no LLM needed)
analyzer = SuggestionAnalyzer()
suggestions = analyzer.analyze(execution_runs)
# Returns: high_retry_rate, always_skipped, very_slow_step, no_bids

# LLM-powered narrative report
llm_analyzer = AutopsyLLMAnalyzer(llm_provider)
report = await llm_analyzer.analyze(execution_runs)
print(report.narrative)

License

MIT — see LICENSE.