TOON Optimizations

roboco/llm/__init__.py

@@ -0,0 +1,15 @@
+"""
+LLM Communication Layer
+
+Provides utilities for efficient communication with Large Language Models,
+including TOON serialization for token-efficient data transfer.
+"""
+
+from roboco.llm.metrics import ToonMetrics
+from roboco.llm.toon_adapter import ToonAdapter, ToonConfig
+
+__all__ = [
+    "ToonAdapter",
+    "ToonConfig",
+    "ToonMetrics",
+]

roboco/llm/metrics.py

@@ -0,0 +1,86 @@
+"""
+TOON Metrics
+
+Tracks token savings and usage statistics for TOON vs JSON serialization.
+"""
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+
+
+@dataclass
+class ToonMetrics:
+    """
+    Metrics for tracking TOON serialization efficiency.
+
+    Tracks character counts (as proxy for tokens) for JSON vs TOON
+    to measure actual savings in production.
+    """
+
+    json_chars: int = 0
+    toon_chars: int = 0
+    encode_count: int = 0
+    decode_count: int = 0
+    decode_fallback_count: int = 0
+    started_at: datetime = field(default_factory=lambda: datetime.now(UTC))
+
+    @property
+    def savings_percent(self) -> float:
+        """Calculate percentage of characters saved using TOON."""
+        if self.json_chars == 0:
+            return 0.0
+        return (1 - self.toon_chars / self.json_chars) * 100
+
+    @property
+    def fallback_rate(self) -> float:
+        """Calculate rate of fallback to JSON decoding."""
+        if self.decode_count == 0:
+            return 0.0
+        return (self.decode_fallback_count / self.decode_count) * 100
+
+    def record_encode(self, json_chars: int, toon_chars: int) -> None:
+        """Record an encode operation with character counts."""
+        self.json_chars += json_chars
+        self.toon_chars += toon_chars
+        self.encode_count += 1
+
+    def record_decode(self, used_fallback: bool = False) -> None:
+        """Record a decode operation."""
+        self.decode_count += 1
+        if used_fallback:
+            self.decode_fallback_count += 1
+
+    def to_dict(self) -> dict:
+        """Convert metrics to dictionary for logging/reporting."""
+        return {
+            "json_chars": self.json_chars,
+            "toon_chars": self.toon_chars,
+            "savings_percent": round(self.savings_percent, 2),
+            "encode_count": self.encode_count,
+            "decode_count": self.decode_count,
+            "fallback_rate": round(self.fallback_rate, 2),
+            "started_at": self.started_at.isoformat(),
+        }
+
+    def reset(self) -> None:
+        """Reset all metrics."""
+        self.json_chars = 0
+        self.toon_chars = 0
+        self.encode_count = 0
+        self.decode_count = 0
+        self.decode_fallback_count = 0
+        self.started_at = datetime.now(UTC)
+
+
+# Global metrics holder
+class _MetricsHolder:
+    """Holder for singleton ToonMetrics instance."""
+
+    instance: ToonMetrics | None = None
+
+
+def get_toon_metrics() -> ToonMetrics:
+    """Get the global TOON metrics instance."""
+    if _MetricsHolder.instance is None:
+        _MetricsHolder.instance = ToonMetrics()
+    return _MetricsHolder.instance

roboco/llm/toon_adapter.py

@@ -0,0 +1,191 @@
+"""
+TOON Adapter
+
+Provides serialization/deserialization between Python objects and TOON format
+for token-efficient LLM communication. TOON (Token-Oriented Object Notation)
+achieves 30-60% fewer tokens than JSON while maintaining semantic clarity.
+
+Usage:
+    adapter = ToonAdapter()
+
+    # Encode data for LLM prompt
+    toon_str = adapter.encode({"name": "Alice", "age": 30})
+
+    # Decode LLM response (falls back to JSON if TOON fails)
+    data = adapter.decode(response_text)
+
+    # Format for embedding in prompt
+    prompt_section = adapter.format_for_prompt("Task Context", task_data)
+"""
+
+import json
+from dataclasses import dataclass
+from typing import Any
+
+import structlog
+import toon
+from pydantic import BaseModel
+
+logger = structlog.get_logger()
+
+
+@dataclass
+class ToonConfig:
+    """Configuration for TOON encoding."""
+
+    delimiter: str = ","
+    indent: int = 2
+    include_length: bool = True
+
+
+class ToonAdapter:
+    """
+    Adapter for TOON serialization at LLM boundaries.
+
+    Converts Python dicts/Pydantic models to TOON for sending to LLMs,
+    and parses TOON responses back to Python objects. Falls back to
+    JSON parsing if TOON decode fails.
+    """
+
+    def __init__(self, config: ToonConfig | None = None) -> None:
+        """
+        Initialize the TOON adapter.
+
+        Args:
+            config: Optional configuration for TOON encoding.
+        """
+        self.config = config or ToonConfig()
+        self.log = logger.bind(component="toon_adapter")
+
+    def encode(self, data: dict[str, Any] | list[Any] | BaseModel) -> str:
+        """
+        Convert Python object to TOON for LLM consumption.
+
+        Args:
+            data: Dictionary, list, or Pydantic model to encode.
+
+        Returns:
+            TOON-formatted string.
+        """
+        if isinstance(data, BaseModel):
+            data = data.model_dump()
+
+        return toon.encode(data, indent=self.config.indent)
+
+    def decode(self, toon_str: str) -> dict[str, Any] | list[Any]:
+        """
+        Parse TOON response from LLM.
+
+        Falls back to JSON parsing if TOON decode fails, logging a warning.
+
+        Args:
+            toon_str: TOON-formatted string from LLM response.
+
+        Returns:
+            Parsed Python dict or list.
+
+        Raises:
+            ValueError: If neither TOON nor JSON parsing succeeds.
+        """
+        # Try TOON first
+        toon_err_msg = ""
+        try:
+            return toon.decode(toon_str)
+        except Exception as toon_error:
+            toon_err_msg = str(toon_error)
+            self.log.warning(
+                "TOON decode failed, trying JSON fallback",
+                error=toon_err_msg,
+            )
+
+        # Fallback to JSON
+        try:
+            return json.loads(toon_str)
+        except json.JSONDecodeError as json_error:
+            self.log.error(
+                "Both TOON and JSON decode failed",
+                toon_error=toon_err_msg,
+                json_error=str(json_error),
+            )
+            raise ValueError(
+                f"Failed to decode response as TOON or JSON: {toon_str[:100]}..."
+            ) from json_error
+
+    def format_for_prompt(self, label: str, data: dict[str, Any]) -> str:
+        """
+        Format data with label for embedding in LLM prompt.
+
+        Args:
+            label: Section label (e.g., "Task Context", "Requirements").
+            data: Data to encode.
+
+        Returns:
+            Formatted string suitable for prompt inclusion.
+        """
+        encoded = self.encode(data)
+        return f"{label}:\n{encoded}"
+
+    def format_tabular_request(
+        self,
+        fields: list[str],
+        description: str,
+        example_rows: list[list[str]] | None = None,
+    ) -> str:
+        """
+        Format a request for tabular TOON response.
+
+        Args:
+            fields: Column names for the table.
+            description: What the LLM should return.
+            example_rows: Optional example data rows.
+
+        Returns:
+            Formatted instruction for LLM to return TOON tabular data.
+        """
+        fields_str = ",".join(fields)
+        header = f"[N,]{{{fields_str}}}:"
+        instruction = f"{description}\n\nFormat response as TOON tabular:\n{header}"
+
+        if example_rows:
+            instruction += "\n"
+            for row in example_rows:
+                instruction += f"{self.config.delimiter.join(row)}\n"
+
+        return instruction
+
+    def estimate_token_savings(
+        self,
+        data: dict[str, Any] | list[Any],
+    ) -> tuple[int, int, float]:
+        """
+        Estimate token savings of TOON vs JSON for given data.
+
+        Args:
+            data: Data to compare.
+
+        Returns:
+            Tuple of (json_chars, toon_chars, savings_percent).
+        """
+        json_str = json.dumps(data, separators=(",", ":"))
+        toon_str = self.encode(data)
+
+        json_chars = len(json_str)
+        toon_chars = len(toon_str)
+
+        savings = (1 - toon_chars / json_chars) * 100 if json_chars > 0 else 0.0
+
+        return json_chars, toon_chars, savings
+
+
+# Module-level singleton holder
+class _AdapterHolder:
+    """Holder for singleton ToonAdapter instance."""
+
+    instance: ToonAdapter | None = None
+
+
+def get_toon_adapter() -> ToonAdapter:
+    """Get the default TOON adapter singleton."""
+    if _AdapterHolder.instance is None:
+        _AdapterHolder.instance = ToonAdapter()
+    return _AdapterHolder.instance