Class ControlLayer

control_layer.py:829–1076

The orchestrator. Sits between your application logic and the LLM. Composes all eight components: InputGuard → TokenBudget → PromptBuilder → CircuitBreaker → LLMCaller (timeout) → ResponseValidator → RetryEngine (jittered backoff) → FallbackRouter → AuditLogger Swap

# =============================================================================

class ControlLayer:
    """
    The orchestrator. Sits between your application logic and the LLM.

    Composes all eight components:
      InputGuard → TokenBudget → PromptBuilder →
      CircuitBreaker → LLMCaller (timeout) → ResponseValidator →
      RetryEngine (jittered backoff) → FallbackRouter → AuditLogger

    Swap the LLM in one line:
        layer = ControlLayer(llm_fn=your_api_call, ...)

    The ControlPacket returned contains the final response, attempt count,
    total latency, token usage, and a full audit ID for tracing.
    """

    def __init__(
        self,
        llm_fn: Callable[[str], str],
        system_prompt: str,
        schema: Optional[ResponseSchema] = None,
        config: Optional[ControlLayerConfig] = None,
        # Legacy compat: accept total_tokens and max_attempts directly
        total_tokens: Optional[int] = None,
        max_attempts: Optional[int] = None,
    ):
        self.config = config or ControlLayerConfig(
            total_tokens=total_tokens or 800,
            max_attempts=max_attempts or 3,
        )
        self.schema = schema or ResponseSchema()

        self.input_guard     = InputGuard(self.config.max_input_chars)
        self.prompt_builder  = PromptBuilder(system_prompt, self.config)
        self.validator       = ResponseValidator()
        self.retry_engine    = RetryEngine(self.config)
        self.fallback_router = FallbackRouter()
        self.circuit_breaker = CircuitBreaker(
            self.config.cb_failure_threshold,
            self.config.cb_recovery_seconds,
        )
        self.llm_caller = LLMCaller(llm_fn, self.config.timeout_seconds)
        self.audit      = AuditLogger(self.config.audit_log_path)

    def register_fallback(self, name: str, fn: Callable[[str], str]) -> None:
        self.fallback_router.register(name, fn)

    def run(
        self,
        user_input: str,
        constraints: Optional[List[str]] = None,
        context: str = "",
    ) -> ControlPacket:
        constraints = constraints or []
        audit_id = self._make_audit_id(user_input)
        start = time.perf_counter()

        # ── Input Guard ────────────────────────────────────────────────────