2 Architecture Tool System
Blightbow edited this page 2025-12-09 04:34:35 -05:00

Architecture: Tool System

Layer 2 - Tool Categories, Registry, Validation, and Token Advisory


Overview

The tool system provides structured actions for the AI assistant:

  • tools/base.py (846 lines) - Core abstractions and registry
  • tools/init.py (217 lines) - Tool imports and auto-registration
  • 17 tool files - Domain-specific tool implementations

1. Tool Categories

Categories determine execution behavior in ReAct loops.

Category Summary

Category Behavior Loop Impact Examples
SAFE_CHAIN Read-only, low-risk Chain freely inspect_location, recall_memories
TERMINAL Communication Ends loop say, page, whisper
DANGEROUS State modification Single per tick execute_command, set_attribute
ASYNC_REQUIRED External calls Special handling store_memory, delegate_task

Category Enum

class ToolCategory(Enum):
    SAFE_CHAIN = "safe_chain"
    """Fast, read-only or low-risk. Can be chained freely within loops."""

    TERMINAL = "terminal"
    """Communication tools. Ends loop to await player response."""

    DANGEROUS = "dangerous"
    """Modifies game state significantly. Single execution, may require confirmation."""

    ASYNC_REQUIRED = "async_required"
    """External calls, I/O, deferred operations. Requires special async handling."""

2. Tool Base Class

All tools inherit from the Tool ABC.

Required Attributes

class Tool(ABC):
    name: str = ""              # Unique identifier
    description: str = ""       # For LLM understanding
    parameters: Dict = {}       # JSON Schema
    timeout: int = 30           # Max execution seconds
    category: ToolCategory = ToolCategory.SAFE_CHAIN

    # Caching (for read-only tools)
    cacheable: bool = False
    cache_scope: str = "tick"   # "tick" or "session"
    cache_ttl: Optional[float] = None  # TTL in seconds

    @abstractmethod
    def execute(self, character, **kwargs) -> Dict[str, Any]:
        """Execute the tool with validated parameters."""
        raise NotImplementedError

Tool Definition Example

class InspectLocationTool(Tool):
    name = "inspect_location"
    description = "Examine the current location and its contents"
    category = ToolCategory.SAFE_CHAIN
    cacheable = True
    cache_scope = "tick"

    parameters = {
        "type": "object",
        "properties": {
            "detail_level": {
                "type": "string",
                "enum": ["brief", "normal", "detailed"],
                "description": "How much detail to include"
            }
        },
        "required": []
    }

    def execute(self, character, **kwargs):
        detail_level = kwargs.get("detail_level", "normal")
        # ... implementation
        return {"location": {...}, "contents": [...]}

3. Tool Registry

Singleton registry managing all available tools.

Registry API

# Global instance
from evennia.contrib.base_systems.ai.tools import TOOL_REGISTRY

# Register a tool
TOOL_REGISTRY.register(MyToolClass)

# Get a tool by name
tool = TOOL_REGISTRY.get_tool("my_tool")

# List all tools
names = TOOL_REGISTRY.list_tools()

# Get schemas for LLM
schemas = TOOL_REGISTRY.get_tool_schemas(categories=[ToolCategory.SAFE_CHAIN])

# Filter by multiple criteria
tools = TOOL_REGISTRY.filter_tools(
    include_categories={ToolCategory.SAFE_CHAIN, ToolCategory.TERMINAL},
    exclude_categories={ToolCategory.DANGEROUS},
    tool_names={"inspect_location", "say"}
)

Auto-Registration

Tools are automatically registered on module import:

# In tools/__init__.py
def register_default_tools():
    """Register all default tools with the global registry."""
    TOOL_REGISTRY.register(CommandTool)
    TOOL_REGISTRY.register(SayTool)
    # ... 40+ tools

# Auto-register when module is imported
register_default_tools()

4. Parameter Validation

Structured validation with LLM-friendly error feedback.

ValidationResult

@dataclass
class ValidationResult:
    valid: bool = True
    errors: List[Dict[str, Any]] = field(default_factory=list)
    warnings: List[Dict[str, Any]] = field(default_factory=list)

    def add_error(self, param, message, constraint=None, got=None):
        """Add a validation error with context."""

    def format_for_llm(self, tool_name: str) -> str:
        """Format errors for LLM feedback."""

Validation Flow

# Tool validates its parameters
result = tool.validate_params({"name": 123})

if not result.valid:
    error_msg = result.format_for_llm(tool.name)
    # "Parameter validation failed for 'my_tool':
    #    - name: Expected string (expected: type: string) (got: 123)"

JSON Schema Validation

Uses jsonschema library when available, falls back to basic validation:

# Full validation (jsonschema available)
result = validate_params_jsonschema(params, schema)

# Basic validation (fallback)
result = validate_params_basic(params, schema)

5. Token Advisory System

Dual-trigger system for context management during ReAct loops.

Thresholds

Level Threshold Message
Normal < 60% "Sufficient context available"
Warning 60% "Consider wrapping up current operation"
Critical 80% "Recommend immediate response"

Advisory Injection

def inject_token_advisory(result: Dict, usage_percentage: float) -> Dict:
    """
    Inject advisory into tool result if usage is elevated.
    Only adds overhead at WARNING or CRITICAL levels.
    """
    level = TokenAdvisory.get_level(usage_percentage)

    if level == TokenAdvisory.NORMAL:
        return result

    result["_token_advisory"] = TokenAdvisory.get_advisory(usage_percentage)
    return result

Loop Continuation Check

def check_loop_continuation(usage_pct, iteration, max_iterations):
    """
    Returns: {
        "should_continue": bool,
        "reason": str,  # "normal", "token_warning", "token_critical", "max_iterations"
        "message": str,
        "advisory": {...},
        "iterations_remaining": int
    }
    """

6. Tool Organization

Tools are organized by functional domain across 17 files.

File Structure

tools/
├── __init__.py          # Registry, auto-registration
├── base.py              # Tool ABC, ToolCategory, ValidationResult
├── attributes.py        # get_attributes, set_attribute, list_attributes
├── building.py          # execute_batchcode, spawn_prototype
├── channel.py           # channel_send
├── command.py           # execute_command
├── communication.py     # say, pose, emit, page, whisper
├── context.py           # session_memory, projects, system_status
├── delegation.py        # delegate_task, get_delegation_status
├── documentation.py     # doc_lookup
├── entity.py            # entity profiles, relationships, conversations
├── goals.py             # add_goal, update_goal, decompose_goal
├── journal.py           # add_journal_entry, review_journal, search_journal
├── location.py          # inspect_location
├── memory.py            # recall_memories, store_memory, search_episodic
├── search.py            # search_object
├── sleep.py             # go_to_sleep, wait
└── tags.py              # get_tags, add_tag, remove_tag, search_by_tag

Category Distribution

Category Tools
SAFE_CHAIN inspect_location, recall_memories, search_object, get_attributes, list_attributes, get_tags, search_by_tag, review_journal, search_journal, recall_entity_profile, recall_conversation_context, list_projects, get_system_status, doc_lookup
TERMINAL say, pose, emit, page, whisper, channel_send
DANGEROUS execute_command, set_attribute, add_tag, remove_tag, execute_batchcode, spawn_prototype, update_system_prompt
ASYNC_REQUIRED store_memory, delegate_task, go_to_sleep

7. Tool Caching

Read-only tools can cache results to reduce redundant operations.

Cache Configuration

class InspectLocationTool(Tool):
    cacheable = True
    cache_scope = "tick"     # Cleared each tick
    # cache_scope = "session"  # Persists until mode change
    cache_ttl = 300.0        # Optional: TTL in seconds

Cache Lookup

# In tool_execution.py
if tool.cacheable:
    cache_key = (tool_name, frozenset(params.items()))
    cached = tool_cache.get(cache_key, scope=tool.cache_scope)
    if cached:
        return cached

# Execute tool
result = yield tool.execute(character, **params)

# Cache result
if tool.cacheable:
    tool_cache.set(cache_key, result, scope=tool.cache_scope, ttl=tool.cache_ttl)

8. Script-Character Lookup

Tools need to find their associated script for context.

def get_assistant_script_for_character(character):
    """
    Canonical method for tools to find their script.

    1. Tag-based lookup (preferred): ai_assistant:{key}
    2. Fallback: Match by character dbref
    """
    assistant_key = getattr(character.db, "assistant_key", None)
    if assistant_key:
        tag = f"ai_assistant:{assistant_key}"
        scripts = search.search_script_tag(key=tag, category="system")
        if scripts:
            return scripts[0]

    # Fallback...

Key Files

File Lines Purpose
tools/base.py 314-335 ToolCategory enum
tools/base.py 337-460 Tool ABC
tools/base.py 462-634 ToolRegistry class
tools/base.py 637-708 TokenAdvisory system
tools/base.py 41-108 ValidationResult
tools/__init__.py 153-216 register_default_tools()

See also: Architecture-Overview | Architecture-Core-Engine | Architecture-Commands-and-API