LLM¶

def get_driver(slug: str) -> DriverInfo | None:
    """Return the catalog entry for ``slug``, or ``None`` if unknown."""
    return DRIVERS_BY_SLUG.get(slug)

def resolve_base_url(driver: str, base_url: str | None) -> str | None:
    """Return the endpoint a provider should call.

    The provider's own ``base_url`` wins; otherwise the driver default,
    which is ``None`` for drivers that have none.
    """
    if base_url:
        return base_url.rstrip('/')
    info = DRIVERS_BY_SLUG.get(driver)
    return info.default_base_url if info else None

async def list_models(
    driver: str, api_key: str, base_url: str | None, *, bearer: bool = False
) -> list[DiscoveredModel]:
    """Return the models ``driver`` reports for ``api_key``.

    Parameters:
        driver: Provider driver slug.
        api_key: Decrypted credential for the provider.
        base_url: Endpoint to call, already resolved against the
            driver's default.
        bearer: ``api_key`` is a workload identity access token.

    Returns:
        The provider's models, in the order the provider returned them.

    Raises:
        DiscoveryError: The driver cannot be discovered, or the call
            failed. The message is safe to show an admin.

    """
    if driver == 'anthropic':
        return await _list_anthropic(api_key, base_url, bearer=bearer)
    if driver in ('openai', 'openai_compatible'):
        if not base_url:
            raise DiscoveryError(
                f'Driver {driver!r} has no endpoint configured'
            )
        return await _list_openai(api_key, base_url)
    raise DiscoveryError(f'Driver {driver!r} does not support discovery')

async def decide(
    api_key: str,
    base_url: str,
    model: str,
    state: object,
    questions: dict[str, typing.Any],
    transport: httpx.AsyncBaseTransport | None = None,
) -> DecisionResult:
    """Ask ``model`` the ``questions`` about ``state``.

    Parameters:
        api_key: Decrypted credential for the provider.
        base_url: The provider endpoint, for example
            ``https://api.typesafe.ai/v1``.
        model: The model id, for example ``jev-latest``.
        state: A string, object, or array.
        questions: Question id to question, in the TypeSafe shape.
        transport: Test seam for :class:`httpx.MockTransport`.

    Raises:
        DecisionError: The call failed; the message is safe to show.

    """
    url = base_url.rstrip('/') + '/systemone'
    body = {'model': model, 'state': state, 'questions': questions}
    try:
        async with httpx.AsyncClient(
            timeout=TIMEOUT, transport=transport
        ) as client:
            async with client.stream(
                'POST',
                url,
                json=body,
                headers={'Authorization': f'Bearer {api_key}'},
            ) as response:
                if response.status_code >= 400:
                    raise DecisionError(
                        f'TypeSafe returned HTTP {response.status_code}'
                    )
                content = await _read_capped(response)
    except DecisionError:
        raise
    except httpx.TimeoutException as e:
        raise DecisionError('TypeSafe did not answer in time') from e
    except httpx.HTTPError as e:
        raise DecisionError(
            f'Could not reach TypeSafe ({type(e).__name__})'
        ) from e
    return _parse(content)