Metadata-Version: 2.4
Name: nano-swarm
Version: 1.1.0
Summary: Swarm intelligence coordination protocol — nano-agents self-organize to execute business processes
Project-URL: Repository, https://github.com/nano-ai/nano-swarm
Project-URL: Changelog, https://github.com/nano-ai/nano-swarm/blob/main/CHANGELOG.md
License-Expression: MIT
License-File: LICENSE
Keywords: agents,coordination,dag,orchestration,sop,swarm,workflow
Classifier: Development Status :: 5 - Production/Stable
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Software Development :: Libraries :: Application Frameworks
Classifier: Typing :: Typed
Requires-Python: >=3.12
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Requires-Dist: rich>=13.0
Requires-Dist: simpleeval>=1.0
Provides-Extra: bot
Requires-Dist: slack-bolt>=1.18; extra == 'bot'
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Provides-Extra: integrations
Requires-Dist: httpx>=0.27; extra == 'integrations'
Provides-Extra: llm
Requires-Dist: anthropic>=0.30; extra == 'llm'
Requires-Dist: openai>=1.0; extra == 'llm'
Description-Content-Type: text/markdown

# 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

```bash
# 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

```yaml
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.

```yaml
# 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'"
```

```yaml
# 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

```python
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

```python
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`:

```python
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

```python
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:

```python
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

```python
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).

```python
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:

```bash
# 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

```bash
pip install nano-swarm[bot]   # adds slack_bolt
```

### Slack app setup

Create a Slack app at [api.slack.com/apps](https://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

```bash
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:

```python
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:

```python
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:

```python
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](LICENSE).
