Architecture: Sub-Agent Delegation

Layer 2 - Task Delegation to Specialized Assistants

---

Overview

The sub-agent system enables a main assistant to delegate tasks to specialized sub-agents:

• Tag-based discovery - Find delegates by assistant tag
• Personality insulation - Control context sharing modes
• Budget management - Limit concurrent sub-agents
• Result summarization - Compress delegate output

---

1. Personality Insulation Modes

Three modes control how context is shared between main and delegate:

Mode	Main Assistant	Delegate	Use Case
none	Full context shared	Inherits goals, memory, state	Deep collaboration
orchestrator	Preserves own personality	Minimal task-focused context	Main directs specialists
delegate	Orchestrates and aggregates	Uses own personality	Independent specialists

Mode: none

Delegate receives:

• Shared goals from main
• Recent conversation memory
• System state (mode, tokens)

Mode: orchestrator

Delegate receives:

• Task description only
• Generic specialist instructions
• No main assistant context

Mode: delegate

Delegate receives:

• Task description
• Top 3 requester goal descriptions
• Instructions to use own expertise

---

2. Configuration

Delegation is configured via execution_config:

script.db.execution_config = {
    "sub_agents_enabled": True,           # Enable delegation
    "sub_agent_budget": 3,                # Max concurrent sub-agents
    "delegate_assistant_tag": "specialist",  # Tag to find delegates
    "personality_insulation": "orchestrator",  # Default mode
    "result_summarization": True,         # Summarize delegate output
}

Context Config Override

Context types can disable delegation:

@dataclass
class ContextConfig:
    allows_sub_agents: bool = True  # Can be False to block delegation

---

3. Delegate Discovery

Delegates are found by tag pattern:

from evennia.contrib.base_systems.ai.sub_agents import (
    find_delegate_assistants,
    get_available_delegates,
)

# Direct tag search (auto-prefixes with "ai_assistant:")
delegates = find_delegate_assistants("specialist", exclude_script=main_script)

# Use configured delegate_assistant_tag
delegates = get_available_delegates(main_script)

---

4. Delegation API

Main Entry Point

from evennia.contrib.base_systems.ai.sub_agents import delegate_task

@inlineCallbacks
def example():
    result = yield delegate_task(
        main_script=script,
        delegate_tag="specialist",
        task_description="Analyze the combat system",
        personality_mode="orchestrator",  # Optional override
    )

    if result["success"]:
        print(f"Delegate {result['delegate_key']}: {result['result']}")
    else:
        print(f"Delegation failed: {result['error']}")

Result Schema

{
    "success": bool,
    "delegate_key": str,      # Which delegate handled it
    "result": str,            # Summarized output
    "iterations": int,        # Actions taken by delegate
    "error": str,             # Error message if failed
}

Check Delegation Allowed

from evennia.contrib.base_systems.ai.sub_agents import check_delegation_allowed

allowed, reason = check_delegation_allowed(main_script, context_type="awake")
# (True, "Delegation allowed")
# (False, "Sub-agent budget exceeded (3/3)")

---

5. Delegation Flow

delegate_task(main_script, delegate_tag, task_description, personality_mode)
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  1. Check if delegation allowed                                   │
│       - sub_agents_enabled?                                       │
│       - Budget available?                                         │
│       - Context allows sub-agents?                                │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  2. Find delegate                                                 │
│       find_delegate_assistants(tag, exclude=main)                │
│       → Use first available                                       │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  3. Increment active sub-agent count                              │
│       main_script.ndb.active_sub_agents += 1                      │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  4. Build delegate context                                        │
│       build_delegate_context(main, task, mode)                    │
│       → Returns context dict per personality mode                 │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  5. Execute on delegate                                           │
│       _execute_delegate_task(delegate, context)                   │
│           │                                                       │
│           ├── Safety: delegate has character?                     │
│           ├── Safety: not in emergency stop?                      │
│           ├── Safety: wait if already ticking (5s max)            │
│           ├── Add delegation event to delegate queue              │
│           └── yield delegate.at_tick()                            │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  6. Extract and summarize result                                  │
│       _extract_delegate_output(delegate)                          │
│       if result_summarization:                                    │
│           _summarize_result(main, result)                         │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
┌──────────────────────────────────────────────────────────────────┐
│  7. Decrement active count (finally block)                        │
│       main_script.ndb.active_sub_agents -= 1                      │
└──────────────────────────────────────────────────────────────────┘
    │
    ▼
Return result dict

---

6. Safety Mechanisms

Budget Enforcement

budget = execution_config.get("sub_agent_budget", 3)
active = main_script.ndb.active_sub_agents or 0

if active >= budget:
    return {"success": False, "error": "Sub-agent budget exceeded"}

Tick Collision Prevention

Before executing on delegate:

# Wait up to 5 seconds for delegate to finish current tick
max_wait_attempts = 10
for attempt in range(max_wait_attempts):
    if not delegate_script.db.is_ticking:
        break
    yield deferLater(reactor, 0.5, lambda: None)
else:
    return {"error": "tick_collision"}

Emergency Stop Check

if delegate_script.db.emergency_stop:
    return {"error": "emergency_stop"}

Character Validation

character = delegate_script.get_character()
if not character:
    return {"error": "no_character"}

---

7. Delegation Status

Query current delegation state:

from evennia.contrib.base_systems.ai.sub_agents import get_delegation_status

status = get_delegation_status(main_script)
# {
#     "enabled": True,
#     "budget": 3,
#     "active": 1,
#     "personality_mode": "orchestrator",
#     "delegate_tag": "specialist",
#     "available_delegates": ["combat_expert", "building_expert"],
# }

---

8. Delegate Task Tool

The LLM uses the delegate_task tool to invoke delegation:

class DelegateTaskTool(Tool):
    name = "delegate_task"
    category = ToolCategory.ASYNC_REQUIRED

    parameters = {
        "delegate_tag": str,       # Required: tag to find delegate
        "task_description": str,   # Required: what to delegate
        "personality_mode": str,   # Optional: override mode
    }

---

Key Files

File	Lines	Purpose
sub_agents.py	42-94	Delegate discovery
sub_agents.py	96-132	Delegation permission checks
sub_agents.py	134-196	Context building per mode
sub_agents.py	198-294	delegate_task() main function
sub_agents.py	296-395	_execute_delegate_task()
sub_agents.py	433-479	Result summarization
sub_agents.py	481-507	get_delegation_status()

---

See also: Architecture-Core-Engine | Architecture-Task-Assessment | Data-Flow-02-ReAct-Loop