Plugins

Plugins

Plugins are concrete implementations of the engine contracts. Built-in plugins live in src/plugins/builtin/ and the external-world plugin lives in src/plugins/external/. All plugins are referenced by their fully-qualified Python dotted path in YAML configs or register_agent calls.

Built-in plugins

OllamaLLMBrain

Source: src/plugins/builtin/ollama/brain.py

Calls a local Ollama model to decide actions. Owns personality, system prompt, prompt assembly, LLM call, and response parsing. The brain emits a BRAIN_DECIDED event containing the full prompt and raw LLM output for observability. Uses an asyncio semaphore (default: 2 concurrent requests) to prevent mass timeouts when many agents share one Ollama instance.

python
from src.plugins.builtin.ollama.brain import OllamaLLMBrain, Personality

brain = OllamaLLMBrain(
    personality=Personality(
        traits    = ["strategic", "ambitious"],
        goals     = ["accumulate resources", "gain influence"],
        backstory = "A former merchant who lost everything to a corrupt guild.",
    ),
    llm_config={
        "model":          "qwen2.5:14b",
        "base_url":       "http://localhost:11434",
        "temperature":    0.8,
        "max_concurrent": 2,   # max simultaneous Ollama requests
    },
)

YAML config:

yaml
brain:
  class: src.plugins.builtin.ollama.brain.OllamaLLMBrain
  personality:
    traits: [strategic, ambitious, distrustful]
    goals: [accumulate gold, become faction leader]
    backstory: A former merchant who lost everything to a corrupt guild.
  llm_config:
    model: qwen2.5:14b
    base_url: http://localhost:11434
    temperature: 0.8

The brain builds a two-part prompt: a system prompt listing available actions with their schemas, and a user prompt with the agent's current state, world perception, nearby agents, and memory. The LLM must respond with a single JSON object matching the Intent schema.

Ollama required

Requires Ollama running at http://localhost:11434. The default model is qwen2.5:14b. Run ollama pull qwen2.5:14b before first use.

IdleBrain

Source: src/plugins/builtin/idle_brain/brain.py

Always returns the same Intent. No LLM, no I/O, no state. Useful for:

  • Integration tests that need registered agents but not Ollama
  • Smoke tests to verify the engine pipeline
  • Placeholder while building a new action handler
python
from src.plugins.builtin.idle_brain.brain import IdleBrain

# Always returns action="idle"
brain = IdleBrain()

# Override the returned intent
brain = IdleBrain(
    action     = "move",
    parameters = {"direction": "north"},
    reasoning  = "patrol route",
)

SimpleMemory

Source: src/plugins/builtin/simple_memory/memory.py

Rolling deque of formatted strings. Default capacity is 20 entries. The recall format is: [t{tick}] {location} {state_ext...} → {action}: {outcome[:70]}

python
from src.plugins.builtin.simple_memory.memory import SimpleMemory

memory = SimpleMemory(capacity=20)
memory.store(tick=1, observation="forest health=90", intent=intent, outcome="gathered 3 food")
recent = memory.recall(n=6)   # last 6 entries as a formatted string

# Checkpointing
data = memory.serialize()       # bytes
memory.restore(data)

WeightedGraphSocial

Source: src/plugins/builtin/simple_social/social.py

Directed weighted graph where edge weights represent relationship strength in [-1.0, +1.0]. Positive weights are friendly relationships; negative weights are hostile. The describe(agent_id) method returns a human-readable summary used in brain context.

Updated automatically from ActionResult.side_effects via SocialEffectSubscriber. Handlers never call the social system directly.

yaml
social:
  class: src.plugins.builtin.simple_social.social.WeightedGraphSocial

ReplayBrain

Source: src/plugins/builtin/replay_brain/brain.py

Dual-mode brain for deterministic research and testing. Supports both recording (wraps an inner brain, writes intents to JSONL) and replay (reads intents from JSONL deterministically). Implements Closeable and Snapshotable.

yaml
# Record mode — wraps OllamaLLMBrain; saves all intents to a JSONL file
brain:
  class: src.plugins.builtin.replay_brain.brain.ReplayBrain
  mode: record
  path: runs/run1.jsonl
  inner:
    class: src.plugins.builtin.ollama.brain.OllamaLLMBrain
    personality:
      traits: [strategic]
      goals: [survive]

# Replay mode — replays from JSONL; no LLM called; 100% deterministic
brain:
  class: src.plugins.builtin.replay_brain.brain.ReplayBrain
  mode: replay
  path: runs/run1.jsonl

When the replay file is exhausted the brain returns Intent(action="idle", reasoning="(replay exhausted)") rather than raising an exception.

JSONReplayRecorderSubscriber

Source: src/plugins/builtin/replay_recorder/recorder.py

EventBus subscriber that writes one JSON record per event to a JSONL file. Subscribe it to specific event types or the "*" wildcard. Call close() when the simulation ends (or use as a context manager).

python
from src.plugins.builtin.replay_recorder.recorder import JSONReplayRecorderSubscriber
from src.engine.event_bus import ACTION_COMPLETED, BRAIN_DECIDED

recorder = JSONReplayRecorderSubscriber("runs/run1.jsonl")
bus.subscribe(ACTION_COMPLETED, recorder)
bus.subscribe(BRAIN_DECIDED,    recorder)   # full prompt + raw output
# later:
recorder.close()

HostedWorld

Source: src/plugins/external/world.py

Reference implementation of the ExternalWorld protocol. The host owns all world state; Python owns cognition. Satisfies World, WorldContext, ExternalWorld, and Snapshotable via structural duck-typing.

python
from src.plugins.external.world import HostedWorld
from src.engine.simulation import Simulation

world = HostedWorld()
sim   = Simulation(agents=[...], world=world, registry=registry)

# Per-tick cycle
world.push_metadata({"time_of_day": "noon", "weather": "clear"})
world.push_observation("agent_001", {
    "location": "marketplace",
    "nearby_agents": [
        {"id": "agent_002", "name": "Bob", "inventory": {}, "ext": {}}
    ],
})
summary = await sim.run_tick()

# Retrieve commands to relay to the host engine
commands = world.collect_commands()
# e.g. [{"agent_id": "agent_001", "type": "navigate", "direction": "north"}]

Handler pattern for host-authoritative actions:

python
class NavigateHandler:
    def execute(self, agent, intent, context) -> ActionResult:
        direction = intent.parameters.get("direction", "north")
        return ActionResult(
            success=True,
            outcome_text=f"moving {direction}",
            engine_commands=[{"type": "navigate", "direction": direction}],
            state_mutations={"energy_delta": -5},
        )

Custom plugins

Custom Brain walkthrough

Here is a minimal OpenAI-based brain that follows the same structure as OllamaLLMBrain:

python
"""my_project/brains/openai_brain.py"""
from __future__ import annotations
import json
from openai import AsyncOpenAI
from src.contracts.action import Intent, ActionSchema, parse_intent
from src.contracts.brain import BrainContext
from src.contracts.world import AgentView


class OpenAIBrain:
    def __init__(
        self,
        model: str = "gpt-4o-mini",
        api_key: str | None = None,
        personality: dict | None = None,
    ):
        self.model = model
        self.client = AsyncOpenAI(api_key=api_key)
        self.personality = personality or {}

    async def decide(
        self,
        agent: AgentView,
        observation: dict,
        actions: list[ActionSchema],
        context: BrainContext,
    ) -> Intent:
        system = self._build_system(actions)
        prompt = self._build_prompt(agent, observation, context)

        response = await self.client.chat.completions.create(
            model    = self.model,
            messages = [
                {"role": "system", "content": system},
                {"role": "user",   "content": prompt},
            ],
            response_format={"type": "json_object"},
        )
        raw = response.choices[0].message.content
        return parse_intent(raw, valid_actions={s.name for s in actions})

    def _build_system(self, actions):
        lines = ["AVAILABLE ACTIONS (use exactly these names):"]
        for s in actions:
            lines.append(s.prompt_block())
        lines.append('\nRespond ONLY with JSON: {"action": "...", "target": null, "parameters": {}, "reasoning": "..."}')
        return "\n".join(lines)

    def _build_prompt(self, agent, obs, context):
        return f"""
Name: {agent.name}
Traits: {self.personality.get('traits', [])}
Goals: {self.personality.get('goals', [])}
Location: {obs.get('location', 'unknown')}
Inventory: {agent.inventory}
Memory: {context.memory}
""".strip()

Reference it in YAML:

yaml
brain:
  class: my_project.brains.openai_brain.OpenAIBrain
  model: gpt-4o-mini
  personality:
    traits: [curious, resourceful]
    goals: [explore the world, make allies]

Custom Memory

python
"""my_project/memory/file_memory.py"""
import json
import pickle
from pathlib import Path
from src.contracts.action import Intent


class FileMemory:
    """Persists memory to a JSON file between runs."""

    def __init__(self, path: str, capacity: int = 50):
        self._path     = Path(path)
        self._capacity = capacity
        self._log: list[str] = []
        if self._path.exists():
            self._log = json.loads(self._path.read_text())

    def store(self, tick: int, observation: str, intent: Intent, outcome: str) -> None:
        entry = f"[t{tick}] {observation} → {intent.action}: {outcome[:70]}"
        self._log.append(entry)
        if len(self._log) > self._capacity:
            self._log = self._log[-self._capacity:]
        self._path.write_text(json.dumps(self._log))

    def recall(self, n: int = 6) -> str:
        return "\n".join(self._log[-n:]) or "No memories yet."

    def serialize(self) -> bytes:
        return pickle.dumps(self._log)

    def restore(self, data: bytes) -> None:
        self._log = pickle.loads(data)