litellm/litellm/rust_bridge/_native.pyi
2026-09-18 15:45:08 -07:00

120 lines
3.9 KiB
Python

from asyncio import Future
from collections.abc import AsyncIterator, Coroutine, Iterator, Mapping, Sequence
from typing import Never, final
from litellm.llms.base_llm.ocr.transformation import OCRResponse
from litellm.rust_bridge.messages.entrypoints import LiteLLMMessagesRequest
from litellm.rust_bridge.ocr.entrypoints import LiteLLMOcrRequest
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
class RustBridgeDeclined(Exception): ...
class RustUpstreamError(Exception): ...
def ocr(
request: LiteLLMOcrRequest,
args: tuple[object, ...],
kwargs: dict[str, object],
) -> OCRResponse: ...
def aocr(
request: LiteLLMOcrRequest,
args: tuple[object, ...],
kwargs: dict[str, object],
) -> Coroutine[object, object, OCRResponse]: ...
def transcription(
model: str,
audio: object,
api_key: str | None = None,
api_base: str | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, object] | None = None,
optional_params: Mapping[str, object] | None = None,
timeout_seconds: float | None = None,
) -> dict[str, object]: ...
def atranscription(
model: str,
audio: object,
api_key: str | None = None,
api_base: str | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, object] | None = None,
optional_params: Mapping[str, object] | None = None,
timeout_seconds: float | None = None,
) -> Future[dict[str, object]]: ...
def messages(
request: LiteLLMMessagesRequest,
args: tuple[object, ...],
kwargs: dict[str, object],
) -> AnthropicMessagesResponse | Iterator[bytes]: ...
def amessages(
request: LiteLLMMessagesRequest,
args: tuple[object, ...],
kwargs: dict[str, object],
) -> Coroutine[object, object, AnthropicMessagesResponse | AsyncIterator[bytes]]: ...
def chat_completions_decline(
model: str,
messages: Sequence[object],
optional_params: Mapping[str, object] | None = None,
custom_llm_provider: str | None = None,
) -> str | None: ...
def chat_completions(
model: str,
messages: Sequence[object],
optional_params: Mapping[str, object] | None = None,
api_key: str | None = None,
api_base: str | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, object] | None = None,
timeout_seconds: float | None = None,
) -> dict[str, object]: ...
def achat_completions(
model: str,
messages: Sequence[object],
optional_params: Mapping[str, object] | None = None,
api_key: str | None = None,
api_base: str | None = None,
custom_llm_provider: str | None = None,
extra_headers: Mapping[str, object] | None = None,
timeout_seconds: float | None = None,
) -> Future[dict[str, object]]: ...
@final
class ResponsesWebSocketConnection:
def __new__(cls, _uninstantiable: Never, /) -> Never: ...
@classmethod
def connect(
cls,
url: str,
headers: Mapping[str, str] | None = None,
timeout_seconds: float | None = None,
) -> Future[ResponsesWebSocketConnection]: ...
def send_text(self, text: str) -> Future[None]: ...
def recv_text(self) -> Future[str | None]: ...
def close(self) -> Future[None]: ...
@final
class TokenCounter:
def __new__(cls, tokenizer_json: str) -> TokenCounter: ...
@staticmethod
def from_cl100k_ranks(rank_file: str) -> TokenCounter: ...
@staticmethod
def from_o200k_ranks(rank_file: str) -> TokenCounter: ...
def acount_request(self, body: bytes) -> Future[dict[str, object]]: ...
def gil_stats() -> dict[str, int]: ...
__all__ = [
"ResponsesWebSocketConnection",
"RustBridgeDeclined",
"RustUpstreamError",
"TokenCounter",
"achat_completions",
"amessages",
"aocr",
"atranscription",
"chat_completions",
"chat_completions_decline",
"gil_stats",
"messages",
"ocr",
"transcription",
]