chatomon.orbitals.principled ¶

def __init__(
    self,
    *,
    guard: ScopeGuardClient | ApiBackend,
    description: str | AIServiceDescription | AIServiceDescriptionV2,
    mapping: Mapping[str | Enum, ScopeRule] | None = None,
    headers: HeadersProvider | None = None,
    history_window: int = 6,
    name: str | None = None,
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block",
) -> None:
    super().__init__(name=name, timeout=timeout, on_error=on_error)
    if isinstance(guard, ApiBackend):
        client: ScopeGuardClient = AsyncScopeGuardV2(
            backend="api", api_url=guard.api_url, model=guard.model, api_key=_api_key(guard)
        )
    elif isinstance(guard, AsyncScopeGuardV2 | ScopeGuardV2 | AsyncScopeGuard | ScopeGuard):
        client = guard
    else:
        raise ChatomonConfigError(
            f"guard must be a ScopeGuard client from the orbitals library or an ApiBackend, "
            f"got {type(guard).__name__}"
        )
    self.is_v2 = isinstance(client, AsyncScopeGuardV2 | ScopeGuardV2)
    if self.is_v2 and isinstance(description, AIServiceDescription):
        raise ChatomonConfigError(
            "ScopeGuard V2 needs an AIServiceDescriptionV2 (or a string) as description"
        )
    if not self.is_v2 and isinstance(description, AIServiceDescriptionV2):
        raise ChatomonConfigError(
            "ScopeGuard V1 needs an AIServiceDescription (or a string) as description"
        )
    _check_headers(client, headers)
    if history_window < 0:
        raise ChatomonConfigError(f"history_window must be >= 0, got {history_window!r}")
    # What was given, a client or a backend; `_client` is what the orbital calls.
    self.guard = guard
    self._client = client
    self.description = description
    self.headers = headers
    self.history_window = history_window
    # The overrides as they were given; `rules` is what the orbital applies.
    self.mapping = mapping
    self.rules: dict[str, ScopeRule] = dict(DEFAULT_SCOPE_RULES)
    for key, rule in (mapping or {}).items():
        self.rules[_scope_key(key)] = rule

def __init__(
    self,
    *,
    extractor: AsyncClaimExtractor | ClaimExtractor | ApiBackend,
    description: str | AIServiceDescription | None = None,
    intents_only: bool = False,
    headers: HeadersProvider | None = None,
    history_window: int = 6,
    name: str | None = None,
    reads: Sequence[str] = (),
    timeout: float = 30.0,
    on_error: ErrorPolicy = "block",
) -> None:
    super().__init__(name=name, reads=reads, timeout=timeout, on_error=on_error)
    client: AsyncClaimExtractor | ClaimExtractor
    if isinstance(extractor, ApiBackend):
        url, key = extractor.api_url, _api_key(extractor)
        if extractor.model is None:
            # No model is given, so the library keeps its default.
            client = AsyncClaimExtractor(backend="api", api_url=url, api_key=key)
        else:
            client = AsyncClaimExtractor(
                backend="api", api_url=url, api_key=key, model=extractor.model
            )
    elif isinstance(extractor, AsyncClaimExtractor | ClaimExtractor):
        client = extractor
    else:
        raise ChatomonConfigError(
            f"extractor must be a ClaimExtractor client from the orbitals library "
            f"or an ApiBackend, got {type(extractor).__name__}"
        )
    _check_headers(client, headers)
    if history_window < 0:
        raise ChatomonConfigError(f"history_window must be >= 0, got {history_window!r}")
    # What was given, a client or a backend; `_client` is what the orbital calls.
    self.extractor = extractor
    self._client = client
    self.description = description
    self.intents_only = intents_only
    self.headers = headers
    self.history_window = history_window