karenina.adapters.langchain.models

models

LangChain model classes for custom endpoints.

This module contains custom LangChain model classes for providers that need special handling (OpenRouter, custom OpenAI-compatible endpoints).

These classes are adapter-internal and should not be imported directly. Use the LLMPort interface via get_llm() instead.

Classes

ChatOpenAIEndpoint

Bases: ChatOpenAI

ChatOpenAI wrapper for user-provided custom endpoints.

Unlike ChatOpenRouter, this does NOT automatically read from environment. API key must be explicitly provided by the user.

This class is used for connecting to OpenAI-compatible endpoints like: - vLLM servers - SGLang servers - Ollama with OpenAI compatibility layer - Text Generation Inference (TGI) - Local LLM servers

Source code in src/karenina/adapters/langchain/models.py

class ChatOpenAIEndpoint(ChatOpenAI):
    """ChatOpenAI wrapper for user-provided custom endpoints.

    Unlike ChatOpenRouter, this does NOT automatically read from environment.
    API key must be explicitly provided by the user.

    This class is used for connecting to OpenAI-compatible endpoints like:
    - vLLM servers
    - SGLang servers
    - Ollama with OpenAI compatibility layer
    - Text Generation Inference (TGI)
    - Local LLM servers
    """

    openai_api_key: SecretStr | None = Field(alias="api_key", default=None)

    @property
    def lc_secrets(self) -> dict[str, str]:
        # Return empty dict - we don't want LangChain trying to read from env
        return {}

    def __init__(
        self,
        base_url: str,
        openai_api_key: str | SecretStr | None = None,
        endpoint_base_url_mode: str = "auto_v1",
        **kwargs: Any,
    ) -> None:
        # Do NOT fallback to environment - require explicit API key
        if openai_api_key is None:
            raise ValueError(
                "API key is required for openai_endpoint interface. "
                "This interface does not automatically read from environment variables."
            )

        if isinstance(openai_api_key, str):
            openai_api_key = SecretStr(openai_api_key)

        # Resolve the base URL. "auto_v1" (default) appends /v1 for the common
        # OpenAI-compatible convention; "raw" uses the URL exactly as given, for
        # endpoints whose OpenAI-compatible path is not at /v1 (for example the
        # z.ai coding endpoint, served at .../v4/chat/completions).
        if endpoint_base_url_mode not in {"auto_v1", "raw"}:
            raise ValueError(f"endpoint_base_url_mode must be 'auto_v1' or 'raw', got {endpoint_base_url_mode!r}")
        if endpoint_base_url_mode == "raw":
            normalized_url = base_url.rstrip("/")
        else:
            normalized_url = _normalize_openai_endpoint_url(base_url)

        # Inject bounded httpx clients unless the caller already supplied them.
        # See _build_httpx_clients docstring and issue 194 for rationale.
        if "http_client" not in kwargs and "http_async_client" not in kwargs:
            sync_client, async_client = _build_httpx_clients()
            kwargs["http_client"] = sync_client
            kwargs["http_async_client"] = async_client
        # Inject the per-request timeout that openai SDK actually honors.
        # Without this, openai's per-request timeout override (~600s) defeats
        # the read timeout we tried to set on the http client.
        if "request_timeout" not in kwargs and "timeout" not in kwargs:
            kwargs["request_timeout"] = _build_default_request_timeout()

        # vLLM and other OpenAI-compatible endpoints drop the `usage` block
        # from streaming responses unless stream_options.include_usage is set.
        # langchain-openai exposes this as `stream_usage=True` on ChatOpenAI;
        # default it on so token accounting works for streaming flows (karenina
        # uses LLMPort.stream_invoke for all openai_endpoint answer generation).
        # Callers can override by passing stream_usage=False explicitly.
        if "stream_usage" not in kwargs:
            kwargs["stream_usage"] = True

        super().__init__(
            base_url=normalized_url,
            api_key=openai_api_key,
            **kwargs,
        )

ChatOpenRouter

Bases: ChatOpenAI

LangChain ChatOpenAI wrapper for OpenRouter API.

OpenRouter provides a unified API for accessing multiple LLM providers. This class automatically configures the correct base URL and API key handling for OpenRouter.

The API key is read from the OPENROUTER_API_KEY environment variable if not explicitly provided.

Source code in src/karenina/adapters/langchain/models.py

class ChatOpenRouter(ChatOpenAI):
    """LangChain ChatOpenAI wrapper for OpenRouter API.

    OpenRouter provides a unified API for accessing multiple LLM providers.
    This class automatically configures the correct base URL and API key
    handling for OpenRouter.

    The API key is read from the OPENROUTER_API_KEY environment variable
    if not explicitly provided.
    """

    openai_api_key: SecretStr | None = Field(alias="api_key", default=None)

    @property
    def lc_secrets(self) -> dict[str, str]:
        return {"openai_api_key": "OPENROUTER_API_KEY"}

    def __init__(self, openai_api_key: str | None = None, **kwargs: Any) -> None:
        openai_api_key = openai_api_key or os.environ.get("OPENROUTER_API_KEY")
        # Inject bounded httpx clients unless the caller already supplied them.
        # See _build_httpx_clients docstring and issue 194 for rationale.
        if "http_client" not in kwargs and "http_async_client" not in kwargs:
            sync_client, async_client = _build_httpx_clients()
            kwargs["http_client"] = sync_client
            kwargs["http_async_client"] = async_client
        # Inject the per-request timeout that openai SDK actually honors.
        # Without this, openai's per-request timeout override (~600s) defeats
        # the read timeout we tried to set on the http client.
        if "request_timeout" not in kwargs and "timeout" not in kwargs:
            kwargs["request_timeout"] = _build_default_request_timeout()
        # OpenRouter proxies OpenAI-compatible providers; same streaming-usage
        # default as ChatOpenAIEndpoint.
        if "stream_usage" not in kwargs:
            kwargs["stream_usage"] = True
        super().__init__(
            base_url="https://openrouter.ai/api/v1",
            api_key=SecretStr(openai_api_key) if openai_api_key else None,
            **kwargs,
        )