Skip to content

Standalone Interlock Kernel

director_ai.interlock packages the halt/interlock kernel as a lightweight bring-your-own-scorer library surface. It does not load retrieval, NLI, vector, server, or training dependencies. The caller supplies a scorer callable, and the kernel decides whether candidate tokens may be admitted.

Use it when another runtime already owns generation and scoring, but still needs Director-compatible halt decisions and tenant-safe SafetyEvent records.

from director_ai import InterlockKernel, InterlockPolicy

kernel = InterlockKernel(
    InterlockPolicy(
        hard_limit=0.5,
        window_size=4,
        window_threshold=0.55,
        hook_id="gateway.interlock",
        policy_id="policy.gateway.regulated",
    )
)

result = kernel.run(
    token_stream,
    scorer=lambda candidate_text: external_score(candidate_text),
)

if result.decision == "halt":
    audit(result.halt_event)
    return result.output

Guarantees

  • low-score tokens are scored before they are appended to output
  • hard-limit, sliding-window, and optional downward-trend checks are available
  • warn_only=True keeps the stream running while still emitting a warning event
  • scorer results can be floats or objects with a .score attribute
  • scores must be finite and in [0, 1]
  • halt and warning events use evidence references such as interlock://token/4, not rejected token text
  • the module is importable as director_ai.interlock and from the root package

Full API

director_ai.interlock.InterlockPolicy dataclass 

InterlockPolicy(hard_limit: float = 0.5, window_size: int = 4, window_threshold: float = 0.5, trend_window: int = 0, trend_threshold: float = 0.2, warn_only: bool = False, hook_id: str = 'interlock.kernel', hook_scope: str = 'streaming', policy_id: str = 'policy.interlock.default', tenant_safe_explanation: str = 'Interlock policy stopped or flagged the stream.')

Policy thresholds for the standalone interlock kernel.

director_ai.interlock.InterlockDecision dataclass 

InterlockDecision(decision: str, output: str, scores: tuple[float, ...], halt_index: int = -1, halt_reason: str = '', halt_event: SafetyEvent | None = None, evidence_refs: tuple[str, ...] = tuple())

Result of applying the interlock to a token sequence.

to_dict 

to_dict() -> dict[str, Any]

Return a JSON-safe result without rejected token text.

director_ai.interlock.InterlockKernel 

InterlockKernel(policy: InterlockPolicy | None = None)

Dependency-light halt gate for streams scored by caller code.

run 

run(tokens: Iterable[str], *, scorer: Callable[[str], float | Any], request_id: str = '', tenant_id: str = '') -> InterlockDecision

Score candidate output and stop before admitting unsafe tokens.