atomicactors 0.1.0b1

Ruthlessly simple framework for building agentic systems

Atomic Actor Framework

Note* for openai compatible tool calling in mlx with lfm2 Update(~/anaconda3/lib/python3.11/site-packages/mlx_lm/tokenizer_utils.py) Updated ../../anaconda3/lib/python3.11/site-packages/mlx_lm/tokenizer_utils.py with 4 additions and 1 removal 274 self._tool_call_end = None 275
276 THINK_TOKENS = [("", "")] 277 - TOOL_CALL_TOKENS = [("<tool_call>", "</tool_call>")] 277 + TOOL_CALL_TOKENS = [ 278 + ("<|tool_call_start|>", "<|tool_call_end|>"), 279 + ("<tool_call>", "</tool_call>"), 280 + ] 281
282 vocab w

A ruthlessly simple framework for building agentic systems through composition of atomic actors.

Design Principles

1. Ruthlessly Simple: Core abstraction is minimal and focused
2. Plug, Config, and Play: All behavior driven by configuration
3. Strong OOP: Clean separation of concerns
4. Self-Orchestrated Recursion: Complexity emerges from composition
5. No Library Pollution: NO tool/model-specific code in library source

Core Concepts

AtomicActor

The fundamental building block. An actor executes a simple cycle:

observe() → perceive() → act() → execute

• observe(): Pull observations from environment (configurable)
• perceive(): Build context from observations + history (configurable)
• act(): LLM call → parse response → return actions
• execute: Run actions, update environment

Environment

Free-form key-value store where actors push results and pull observations. Can be shared by multiple actors.

EventBus

A true publish/subscribe event system providing comprehensive observability and orchestration for actors.

Key Features:

• Hierarchical event types: lifecycle.*, action.*, llm.*, etc.
• Pattern matching: Subscribe with wildcards (*) for flexible filtering
• Metadata + Data: Both structural signals and detailed content
• Actor orchestration: Coordinate actors via event subscriptions
• External integration: Connect GUIs, APIs, databases, metrics collectors

📚 See docs/EVENTBUS.md for complete documentation

ToolRegistry

Generic interface for registering and invoking tools. NO tool-specific code.

Architecture

┌─────────────────┐
│  AtomicActor    │
│                 │
│  observe()      │
│  perceive()     │
│  act()          │
│  execute()      │
└────────┬────────┘
         │
         ├─── EventBus ───┐
         │                │
         └─── Environment ◄┘
                  ▲
                  │
         ┌────────┴────────┐
         │   ToolRegistry  │
         │   (tools here)  │
         └─────────────────┘

Usage

Basic Setup

from atomic_actor import (
    AtomicActor,
    Environment,
    EventBus,
    ColoredLoggingHandler,
    ToolRegistry,
    ActorConfig,
    ToolSchema
)

# 1. Create environment
environment = Environment()

# 2. Create event bus with logging subscriber
event_bus = EventBus()

# Subscribe a colored logging handler to all events
logger = ColoredLoggingHandler(stream_to="stdout", color_enabled=True)
event_bus.subscribe("*", "*", logger.handle)

# Optional: Filter to specific event types
# result_logger = ColoredLoggingHandler(event_filter=["action.result", "cycle.data"])
# event_bus.subscribe("*", "*", result_logger.handle)

# 3. Create and configure tools
tools = ToolRegistry()
tools.register(
    name="search",
    func=search_function,
    schema=ToolSchema(...)
)

# 4. Create actor config
config = ActorConfig(
    system_prompt="You are a helpful assistant.",
    observation=ObservationConfig(mode="selective", window=5),
    perception=PerceptionConfig(include_history=True)
)

# 5. Create actor
actor = AtomicActor(
    name="MyActor",
    tools=tools,
    environment=environment,
    event_bus=event_bus,
    config=config
)

# 6. Inject LLM interface (SDK-specific)
actor.set_llm_interface(your_llm_interface)

# 7. Run
result = await actor.run()

Configuration

All actor behavior is driven by configuration:

config = ActorConfig(
    system_prompt="...",

    observation=ObservationConfig(
        mode="selective",  # "delta", "full", "selective"
        fields=["tool_results", "user_messages"],
        window=5
    ),

    perception=PerceptionConfig(
        include_history=True,
        observation_window=3,
        history_rules=[
            HistoryRule(
                tool_type="search",
                include_previous=2
            )
        ]
    )
)

LLM Interface

The framework is LLM-agnostic. Implement a simple interface:

class YourLLMInterface:
    async def complete(self, context, system_prompt, tools, config):
        # Call your LLM SDK here
        response = await your_sdk.call(...)
        return response

    def parse_actions(self, response):
        # Parse response into Action objects
        actions = []
        # ... parsing logic based on config.tool_format
        return actions

Actor Composition

Complex behavior emerges from composing actors:

# Orchestrator actor delegates to task actors
orchestrator = AtomicActor(
    name="Orchestrator",
    tools=orchestrator_tools,  # Includes delegation tool
    ...
)

# Delegation tool spawns child actors
async def delegate_task(task: str):
    child_actor = AtomicActor(
        name=f"TaskActor_{task}",
        environment=shared_environment,  # Shared!
        ...
    )
    return await child_actor.run()

State Management

Actor state is indexed by cycle:

actor.actions[cycle_num]       # Actions for specific cycle
actor.observations[cycle_num]  # Observations for specific cycle
actor.data[cycle_num]          # Data/result for specific cycle

# Latest values
actor.actions[-1]              # Most recent actions
actor.data[-1]                 # Most recent data

Generator Pattern

The clock() method is an async generator that yields state each cycle:

async for state in actor.clock():
    print(f"Cycle {state.cycle_num}: {state.status}")
    # Can inspect, log, or coordinate with other actors

Or use run() to consume the generator and get final result:

result = await actor.run()

Directory Structure

atomic_actor/
├── __init__.py          # Package exports
├── actor.py             # AtomicActor class
├── environment.py       # Environment class
├── event_bus.py         # EventBus class
├── tool_registry.py     # ToolRegistry class
├── config.py            # Configuration classes
└── types.py             # Core types (Action, Observation, etc.)

examples/
└── basic_usage.py       # Usage examples

tests/
└── ...                  # Tests (TBD)

Key Features

✅ Clean separation of concerns: Each component has single responsibility ✅ Configuration-driven: No hardcoded tool/model logic in library ✅ Async by default: Supports parallel operations ✅ Generator pattern: Real-time state inspection ✅ Composable: Actors can spawn child actors ✅ Observable: EventBus reflects state changes ✅ LLM-agnostic: Bring your own SDK

Running the Examples

Local LLM Example (Ready to Run)

examples/local_llm_example.py is a complete working example using a local LLM at localhost:8080:

# Install dependencies
pip install -r requirements.txt

# Run the example (make sure your local LLM is running on port 8080)
python examples/local_llm_example.py

This example demonstrates:

• LFM2_PYTHON tool format parsing
• Local model integration
• Real tool execution
• Complete actor lifecycle

Basic Usage (Template)

See examples/basic_usage.py for a simpler template showing the structure.

Next Steps

To use with your LLM:

1. Implement the LLM interface (complete + parse_actions) - see local_llm_example.py for reference
2. Configure tool schemas
3. Set up your tools in ToolRegistry
4. Configure actor behavior via ActorConfig
5. Run!