language_models

langroid/language_models/init.py

LLMConfig

Bases: BaseSettings

Common configuration for all language models.

LLMMessage

Bases: BaseModel

Class representing an entry in the msg-history sent to the LLM API. It could be one of these: - a user message - an LLM ("Assistant") response - a fn-call or tool-call-list from an OpenAI-compatible LLM API response - a result or results from executing a fn or tool-call(s)

api_dict(model, has_system_role=True)

Convert to dictionary for API request, keeping ONLY the fields that are expected in an API call! E.g., DROP the tool_id, since it was only for use in the Assistants API, not the completion API.

Parameters:

Name	Type	Description	Default
has_system_role	bool	whether the message has a system role (if not, set to "user" role)	True

Returns: dict: dictionary representation of LLM message

Source code in langroid/language_models/base.py

def api_dict(self, model: str, has_system_role: bool = True) -> Dict[str, Any]:
    """
    Convert to dictionary for API request, keeping ONLY
    the fields that are expected in an API call!
    E.g., DROP the tool_id, since it was only for use in the Assistants API,
        not the completion API.

    Args:
        has_system_role: whether the message has a system role (if not,
            set to "user" role)
    Returns:
        dict: dictionary representation of LLM message
    """
    d = self.model_dump()
    files: List[FileAttachment] = d.pop("files")
    if len(files) > 0 and self.role == Role.USER:
        # In there are files, then content is an array of
        # different content-parts
        d["content"] = [
            dict(
                type="text",
                text=self.content or "",
            )
        ] + [f.to_dict(model) for f in self.files]

    # if there is a key k = "role" with value "system", change to "user"
    # in case has_system_role is False
    if not has_system_role and "role" in d and d["role"] == "system":
        d["role"] = "user"
        if d.get("content"):
            d["content"] = "[ADDITIONAL SYSTEM MESSAGE:]\n\n" + d["content"]
    # drop None values since API doesn't accept them (this omits `content`
    # entirely when it is None, i.e. "no content")
    dict_no_none = {k: v for k, v in d.items() if v is not None}
    if "name" in dict_no_none and dict_no_none["name"] == "":
        # OpenAI API does not like empty name
        del dict_no_none["name"]
    if "function_call" in dict_no_none:
        # arguments must be a string
        if "arguments" in dict_no_none["function_call"]:
            dict_no_none["function_call"]["arguments"] = json.dumps(
                dict_no_none["function_call"]["arguments"]
            )
    if dict_no_none.get("tool_calls"):
        # convert tool calls to API format
        for tc in dict_no_none["tool_calls"]:
            if "arguments" in tc["function"]:
                # arguments must be a string
                if tc["function"]["arguments"] is None:
                    tc["function"]["arguments"] = "{}"
                else:
                    tc["function"]["arguments"] = json.dumps(
                        tc["function"]["arguments"]
                    )
            if "extra_content" in tc and tc["extra_content"] is None:
                del tc["extra_content"]
    else:
        # An empty tool-call list is not a call and conveys no useful API
        # information. Omitting it also lets the empty-message fallback
        # below produce a valid message.
        dict_no_none.pop("tool_calls", None)
    # A message with empty content (None was dropped above, or "") AND no
    # tool/function call would be an entirely empty message, which some APIs
    # (e.g. Gemini) reject; fall back to a single space in that case only.
    # When there IS a tool/function call, empty content is fine (and Gemini
    # 3.x in fact REQUIRES it — it rejects an assistant turn that carries
    # both a call and non-empty text content).
    has_call = bool(dict_no_none.get("tool_calls")) or (
        dict_no_none.get("function_call") is not None
    )
    if not has_call and not dict_no_none.get("content"):
        dict_no_none["content"] = " "
    # IMPORTANT! drop fields that are not expected in API call
    dict_no_none.pop("tool_id", None)
    dict_no_none.pop("timestamp", None)
    dict_no_none.pop("chat_document_id", None)
    return dict_no_none

LLMFunctionCall

Bases: BaseModel

Structure of LLM response indicating it "wants" to call a function. Modeled after OpenAI spec for function_call field in ChatCompletion API.

from_dict(message) staticmethod

Initialize from dictionary. Args: d: dictionary containing fields to initialize

Source code in langroid/language_models/base.py

@staticmethod
def from_dict(message: Dict[str, Any]) -> "LLMFunctionCall":
    """
    Initialize from dictionary.
    Args:
        d: dictionary containing fields to initialize
    """
    fun_call = LLMFunctionCall(name=message["name"])
    fun_args_str = message["arguments"]
    # sometimes may be malformed with invalid indents,
    # so we try to be safe by removing newlines.
    if fun_args_str is not None:
        fun_args_str = fun_args_str.replace("\n", "").strip()
        dict_or_list = parse_imperfect_json(fun_args_str)

        if not isinstance(dict_or_list, dict):
            raise ValueError(
                f"""
                    Invalid function args: {fun_args_str}
                    parsed as {dict_or_list},
                    which is not a valid dict.
                    """
            )
        fun_args = dict_or_list
    else:
        fun_args = None
    fun_call.arguments = fun_args

    return fun_call

LLMFunctionSpec

Bases: BaseModel

Description of a function available for the LLM to use. To be used when calling the LLM chat() method with the functions parameter. Modeled after OpenAI spec for functions fields in ChatCompletion API.

Role

Bases: str, Enum

Possible roles for a message in a chat.

LLMTokenUsage

Bases: BaseModel

Usage of tokens by an LLM.

LLMResponse

Bases: BaseModel

Class representing response from LLM.

to_LLMMessage()

Convert LLM response to an LLMMessage, to be included in the message-list sent to the API. This is currently NOT used in any significant way in the library, and is only provided as a utility to construct a message list for the API when directly working with an LLM object.

In a ChatAgent, an LLM response is first converted to a ChatDocument, which is in turn converted to an LLMMessage via ChatDocument.to_LLMMessage() See ChatAgent._prep_llm_messages() and ChatAgent.llm_response_messages

Source code in langroid/language_models/base.py

def to_LLMMessage(self) -> LLMMessage:
    """Convert LLM response to an LLMMessage, to be included in the
    message-list sent to the API.
    This is currently NOT used in any significant way in the library, and is only
    provided as a utility to construct a message list for the API when directly
    working with an LLM object.

    In a `ChatAgent`, an LLM response is first converted to a ChatDocument,
    which is in turn converted to an LLMMessage via `ChatDocument.to_LLMMessage()`
    See `ChatAgent._prep_llm_messages()` and `ChatAgent.llm_response_messages`
    """
    return LLMMessage(
        role=Role.ASSISTANT,
        content=self.message,
        name=None if self.function_call is None else self.function_call.name,
        function_call=self.function_call,
        tool_calls=self.oai_tool_calls,
    )

get_recipient_and_message(recognize_recipient_in_content=True)

If message or function_call of an LLM response contains an explicit recipient name, return this recipient name and message stripped of the recipient name if specified.

Two cases: (a) message contains addressing string TO[<name>]:<content>, or (b) message is empty and function_call/tool_call with explicit recipient

Parameters:

Name	Type	Description	Default
recognize_recipient_in_content	bool	When True (default), parses message text for TO[<recipient>]:<content> patterns and top-level JSON {"recipient": "..."} fields. When False, only function_call/tool_call recipient fields are checked.	True

Returns:

Type	Description
str	name of recipient, which may be empty string if no recipient
str	content of message

Source code in langroid/language_models/base.py

def get_recipient_and_message(
    self,
    recognize_recipient_in_content: bool = True,
) -> Tuple[str, str]:
    """
    If `message` or `function_call` of an LLM response contains an explicit
    recipient name, return this recipient name and `message` stripped
    of the recipient name if specified.

    Two cases:
    (a) `message` contains addressing string ``TO[<name>]:<content>``, or
    (b) `message` is empty and function_call/tool_call with explicit `recipient`

    Args:
        recognize_recipient_in_content (bool): When True (default), parses
            message text for ``TO[<recipient>]:<content>`` patterns and
            top-level JSON ``{"recipient": "..."}`` fields. When False,
            only function_call/tool_call ``recipient`` fields are checked.

    Returns:
        (str): name of recipient, which may be empty string if no recipient
        (str): content of message

    """

    if self.function_call is not None:
        # in this case we ignore message, since all information is in function_call
        msg = ""
        args = self.function_call.arguments
        recipient = ""
        if isinstance(args, dict):
            recipient = args.get("recipient", "")
        return recipient, msg
    else:
        msg = self.message or ""
        if self.oai_tool_calls is not None:
            # get the first tool that has a recipient field, if any
            for tc in self.oai_tool_calls:
                if tc.function is not None and tc.function.arguments is not None:
                    recipient = tc.function.arguments.get(
                        "recipient"
                    )  # type: ignore
                    if recipient is not None and recipient != "":
                        return recipient, ""

    if not recognize_recipient_in_content:
        return "", msg

    # It's not a function or tool call, so continue looking to see
    # if a recipient is specified in the message.

    # First check if message contains "TO: <recipient> <content>"
    recipient_name, content = parse_message(msg) if msg is not None else ("", "")
    # check if there is a top level json that specifies 'recipient',
    # and retain the entire message as content.
    if recipient_name == "":
        recipient_name = top_level_json_field(msg, "recipient") if msg else ""
        content = msg
    return recipient_name, content

OpenAIChatModel

Bases: ModelName

Enum for OpenAI Chat models

AnthropicModel

Bases: ModelName

Enum for Anthropic models

GeminiModel

Bases: ModelName

Enum for Gemini models

MiniMaxModel

Bases: ModelName

Enum for MiniMax models

OpenAICompletionModel

Bases: str, Enum

Enum for OpenAI Completion models

OpenAIGPTConfig(**kwargs)

Bases: LLMConfig

Class for any LLM with an OpenAI-like API: besides the OpenAI models this includes: (a) locally-served models behind an OpenAI-compatible API (b) non-local models, using a proxy adaptor lib like litellm that provides an OpenAI-compatible API. (We could rename this class to OpenAILikeConfig, but we keep it as-is for now)

Important Note: Due to the env_prefix = "OPENAI_" defined below, all of the fields below can be set AND OVERRIDDEN via env vars,

by upper-casing the name and prefixing with OPENAI_, e.g.

OPENAI_MAX_OUTPUT_TOKENS=1000.

If any of these is defined in this way in the environment

(either via explicit setenv or export or via .env file + load_dotenv()),

the environment variable takes precedence over the value in the config.

Source code in langroid/language_models/openai_gpt.py

def __init__(self, **kwargs) -> None:  # type: ignore
    api_base_was_supplied = "api_base" in kwargs
    local_model = "api_base" in kwargs and kwargs["api_base"] is not None

    chat_model = kwargs.get("chat_model", "")
    local_prefixes = ["local/", "litellm/", "ollama/", "vllm/", "llamacpp/"]
    if any(chat_model.startswith(prefix) for prefix in local_prefixes):
        local_model = True

    warn_gpt_3_5 = (
        "chat_model" not in kwargs.keys()
        and not local_model
        and default_openai_chat_model == OpenAIChatModel.GPT3_5_TURBO
    )

    if warn_gpt_3_5:
        existing_hook = kwargs.get("run_on_first_use", noop)

        def with_warning() -> None:
            existing_hook()
            gpt_3_5_warning()

        kwargs["run_on_first_use"] = with_warning

    super().__init__(**kwargs)
    env_prefix = self.model_config.get("env_prefix")
    env_api_base_name = f"{env_prefix}API_BASE"
    env_api_base = os.getenv(env_api_base_name)
    if not self.model_config.get("case_sensitive"):
        case_insensitive_env = {
            name.lower(): value for name, value in os.environ.items()
        }
        env_api_base = case_insensitive_env.get(env_api_base_name.lower())
    api_base_was_supplied = api_base_was_supplied or (
        self.api_base is not None
        and (env_prefix != "OPENAI_" or self.api_base != env_api_base)
    )
    self._api_base_was_supplied = api_base_was_supplied

model_copy(*, update=None, deep=False)

Copy config while preserving nested model instances and subclasses.

Important: Avoid reconstructing via model_dump as that coerces nested models to their annotated base types (dropping subclass-only fields). Instead, defer to Pydantic's native model_copy, which keeps nested BaseModel instances (and their concrete subclasses) intact.

An api_base supplied through update is caller configuration, so the copy must retain that provenance for provider-specific routing.

Source code in langroid/language_models/openai_gpt.py

def model_copy(
    self, *, update: Mapping[str, Any] | None = None, deep: bool = False
) -> "OpenAIGPTConfig":
    """
    Copy config while preserving nested model instances and subclasses.

    Important: Avoid reconstructing via `model_dump` as that coerces nested
    models to their annotated base types (dropping subclass-only fields).
    Instead, defer to Pydantic's native `model_copy`, which keeps nested
    `BaseModel` instances (and their concrete subclasses) intact.

    An `api_base` supplied through `update` is caller configuration, so the
    copy must retain that provenance for provider-specific routing.
    """
    # Delegate to BaseSettings/BaseModel implementation to preserve types
    copied = super().model_copy(update=update, deep=deep)
    if update is not None and "api_base" in update:
        copied._api_base_was_supplied = True
    return copied  # type: ignore[return-value]

create(prefix) classmethod

Create a config class whose params can be set via a desired prefix from the .env file or env vars. E.g., using

OllamaConfig = OpenAIGPTConfig.create("ollama")
ollama_config = OllamaConfig()

you can have a group of params prefixed by "OLLAMA_", to be used with models served via ollama. This way, you can maintain several setting-groups in your .env file, one per model type.

Source code in langroid/language_models/openai_gpt.py

@classmethod
def create(cls, prefix: str) -> Type["OpenAIGPTConfig"]:
    """Create a config class whose params can be set via a desired
    prefix from the .env file or env vars.
    E.g., using
    ```python
    OllamaConfig = OpenAIGPTConfig.create("ollama")
    ollama_config = OllamaConfig()
    ```
    you can have a group of params prefixed by "OLLAMA_", to be used
    with models served via `ollama`.
    This way, you can maintain several setting-groups in your .env file,
    one per model type.
    """

    class DynamicConfig(OpenAIGPTConfig):
        pass

    DynamicConfig.model_config = SettingsConfigDict(env_prefix=prefix.upper() + "_")
    return DynamicConfig

OpenAIGPT(config=OpenAIGPTConfig())

Bases: LanguageModel

Class for OpenAI LLMs

Source code in langroid/language_models/openai_gpt.py

def __init__(self, config: OpenAIGPTConfig = OpenAIGPTConfig()):
    """
    Args:
        config: configuration for openai-gpt model
    """
    # copy the config to avoid modifying the original; deep to decouple
    # nested models while preserving their concrete subclasses
    # The api_key_provider must NOT be deep-copied: the client cache is
    # keyed on its identity (so a copy would defeat cache sharing), and it
    # may hold non-copyable state (e.g. a threading.Lock). Clear it in a
    # shallow copy first, then restore the original callable afterwards.
    api_key_provider = config.api_key_provider
    if api_key_provider is not None:
        config = config.model_copy(update={"api_key_provider": None})
    config = config.model_copy(deep=True)
    if api_key_provider is not None:
        config.api_key_provider = api_key_provider
    super().__init__(config)
    self.config: OpenAIGPTConfig = config
    # save original model name such as `provider/model` before
    # we strip out the `provider` - we retain the original in
    # case some params are specific to a provider.
    self.chat_model_orig = self.config.chat_model_orig or self.config.chat_model

    # Run the first time the model is used
    self.run_on_first_use = cache(self.config.run_on_first_use)

    # global override of chat_model,
    # to allow quick testing with other models
    if settings.chat_model != "":
        self.config.chat_model = settings.chat_model
        self.chat_model_orig = settings.chat_model
        self.config.completion_model = settings.chat_model

    # Whether the caller asked for a vertexai/ route, recorded BEFORE the
    # `model//formatter` split below can eat the prefix.
    asked_for_vertexai = self.config.chat_model.startswith(VERTEXAI_PREFIX)

    if len(parts := self.config.chat_model.split("//")) > 1:
        # there is a formatter specified, e.g.
        # "litellm/ollama/mistral//hf" or
        # "local/localhost:8000/v1//mistral-instruct-v0.2"
        formatter = parts[1]
        self.config.chat_model = parts[0]
        if formatter == "hf":
            # e.g. "litellm/ollama/mistral//hf" -> "litellm/ollama/mistral"
            formatter = find_hf_formatter(self.config.chat_model)
            if formatter != "":
                # e.g. "mistral"
                self.config.formatter = formatter
                logging.warning(
                    f"""
                    Using completions (not chat) endpoint with HuggingFace
                    chat_template for {formatter} for
                    model {self.config.chat_model}
                    """
                )
        else:
            # e.g. "local/localhost:8000/v1//mistral-instruct-v0.2"
            self.config.formatter = formatter

    if asked_for_vertexai and not self.config.chat_model.startswith(
        VERTEXAI_PREFIX
    ):
        # The `//formatter` suffix consumed the route itself, e.g.
        # "vertexai//hf" -> the bare model "vertexai". Left alone this
        # falls through to the generic branch and quietly talks to
        # api.openai.com, which is never what someone typing `vertexai`
        # meant. (On a VertexAIConfig it would also take a Google
        # credential there -- the refusal below catches that case too, but
        # this one fires for a plain OpenAIGPTConfig as well.)
        raise ValueError(
            f"chat_model {self.chat_model_orig!r} is not a usable "
            "vertexai/ route: the //formatter suffix left no model behind. "
            "Write vertexai/<publisher>/<model>//<formatter>, e.g. "
            "vertexai/google/gemini-2.5-flash//hf."
        )

    # A vertexai/ route must not inherit OPENAI_-prefixed settings, so
    # rebuild the config as a clean VertexAIConfig.
    #
    # Placed here deliberately, and the position is load-bearing twice
    # over: AFTER the settings.chat_model override above, so the direct
    # route and the override path get the same clean config; and AFTER the
    # `model//formatter` split just above, so both branches below see the
    # final chat_model. Before the split, "vertexai//hf" still carried the
    # prefix and slipped past both branches -- then the split turned it
    # into the bare model "vertexai", which routes to api.openai.com,
    # taking a VERTEXAI_API_KEY with it.
    if self.config.chat_model.startswith(VERTEXAI_PREFIX) and not isinstance(
        self.config, VertexAIConfig
    ):
        self.config = _as_vertexai_config(self.config)
        # The conversion drops chat_model_orig (OPENAI_CHAT_MODEL_ORIG
        # could otherwise decide which provider branch claims this
        # request), so recompute it from the model actually in effect.
        self.chat_model_orig = self.config.chat_model
    elif isinstance(self.config, VertexAIConfig) and not (
        self.config.chat_model.startswith(VERTEXAI_PREFIX)
    ):
        # The converse of the above, and the same leak running backwards: a
        # VertexAIConfig carries a credential from VERTEXAI_API_KEY, and
        # without the vertexai/ prefix this would fall through to the
        # generic branch -- api_base None, i.e. api.openai.com -- and send
        # that credential there. Reachable via the settings.chat_model
        # override alone (`-m gpt-4o`), so refuse rather than guess.
        raise ValueError(
            f"chat_model {self.config.chat_model!r} is not a vertexai/ "
            "route, but the config is a VertexAIConfig, whose credential "
            "is meant for Google. Use an OpenAIGPTConfig for non-Vertex "
            "models. (If this came from a global chat_model override such "
            "as `-m <model>`, that override cannot be applied to a "
            "VertexAIConfig. If you passed an existing llm.config back in, "
            "note that its vertexai/ prefix was already stripped.)"
        )

    if self.config.formatter is not None:
        self.config.hf_formatter = HFFormatter(
            HFPromptFormatterConfig(model_name=self.config.formatter)
        )

    self.supports_json_schema: bool = self.config.supports_json_schema or False
    self.supports_strict_tools: bool = self.config.supports_strict_tools or False

    OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", DUMMY_API_KEY)
    # read from self.config, not the `config` argument: a vertexai/ route
    # replaced self.config above, and the original still holds the
    # OPENAI_-prefixed key we must not use.
    self.api_key = self.config.api_key

    # if model name starts with "litellm",
    # set the actual model name by stripping the "litellm/" prefix
    # and set the litellm flag to True
    if self.config.chat_model.startswith("litellm/") or self.config.litellm:
        # e.g. litellm/ollama/mistral
        self.config.litellm = True
        self.api_base = self.config.api_base
        if self.config.chat_model.startswith("litellm/"):
            # strip the "litellm/" prefix
            # e.g. litellm/ollama/llama2 => ollama/llama2
            self.config.chat_model = self.config.chat_model.split("/", 1)[1]
    elif self.config.chat_model.startswith("local/"):
        # expect this to be of the form "local/localhost:8000/v1",
        # depending on how the model is launched locally.
        # In this case the model served locally behind an OpenAI-compatible API
        # so we can just use `openai.*` methods directly,
        # and don't need a adaptor library like litellm
        self.config.litellm = False
        self.config.seed = None  # some models raise an error when seed is set
        # Extract the api_base from the model name after the "local/" prefix
        self.api_base = self.config.chat_model.split("/", 1)[1]
        if not self.api_base.startswith("http"):
            self.api_base = "http://" + self.api_base
    elif self.config.chat_model.startswith("ollama/"):
        self.config.ollama = True

        # use api_base from config if set, else fall back on OLLAMA_BASE_URL
        self.api_base = self.config.api_base or OLLAMA_BASE_URL
        if self.api_key == OPENAI_API_KEY:
            self.api_key = OLLAMA_API_KEY
        self.config.chat_model = self.config.chat_model.replace("ollama/", "")
    elif self.config.chat_model.startswith("vllm/"):
        self.supports_json_schema = True
        self.config.chat_model = self.config.chat_model.replace("vllm/", "")
        if self.api_key == OPENAI_API_KEY:
            self.api_key = os.environ.get("VLLM_API_KEY", DUMMY_API_KEY)
        self.api_base = self.config.api_base or "http://localhost:8000/v1"
        if not self.api_base.startswith("http"):
            self.api_base = "http://" + self.api_base
        if not self.api_base.endswith("/v1"):
            self.api_base = self.api_base + "/v1"
    elif self.config.chat_model.startswith("llamacpp/"):
        self.supports_json_schema = True
        self.api_base = self.config.chat_model.split("/", 1)[1]
        if not self.api_base.startswith("http"):
            self.api_base = "http://" + self.api_base
        if self.api_key == OPENAI_API_KEY:
            self.api_key = os.environ.get("LLAMA_API_KEY", DUMMY_API_KEY)
    else:
        self.api_base = self.config.api_base
        # If api_base is unset we use OpenAI's endpoint, which supports
        # these features (with JSON schema restricted to a limited set of models)
        self.supports_strict_tools = self.api_base is None
        self.supports_json_schema = (
            self.api_base is None and self.info().has_structured_output
        )

    if settings.chat_model != "":
        # if we're overriding chat model globally, set completion model to same
        self.config.completion_model = self.config.chat_model

    if self.config.formatter is not None:
        # we want to format chats -> completions using this specific formatter
        self.config.use_completion_for_chat = True
        self.config.completion_model = self.config.chat_model

    if self.config.use_completion_for_chat:
        self.config.use_chat_for_completion = False

    self.is_groq = self.config.chat_model.startswith("groq/")
    self.is_cerebras = self.config.chat_model.startswith("cerebras/")
    self.is_gemini = self.is_gemini_model()
    self.is_deepseek = self.is_deepseek_model()
    self.is_minimax = self.is_minimax_model()
    self.is_glhf = self.config.chat_model.startswith("glhf/")
    self.is_openrouter = self.config.chat_model.startswith("openrouter/")
    self.is_langdb = self.config.chat_model.startswith("langdb/")
    self.is_portkey = self.config.chat_model.startswith("portkey/")
    self.is_vertexai = self.config.chat_model.startswith(VERTEXAI_PREFIX)
    self.is_litellm_proxy = self.config.chat_model.startswith("litellm-proxy/")

    if self.config.api_key_provider is not None and (
        self.is_groq or self.is_cerebras or self.config.litellm
    ):
        raise ValueError(
            "api_key_provider is only supported for models served via an "
            "OpenAI-compatible endpoint (using the OpenAI client); it "
            "cannot be used with Groq, Cerebras, or the litellm adapter."
        )

    if self.is_groq:
        # use groq-specific client
        self.config.chat_model = self.config.chat_model.replace("groq/", "")
        if self.api_key == OPENAI_API_KEY:
            self.api_key = os.getenv("GROQ_API_KEY", DUMMY_API_KEY)
        if self.config.use_cached_client:
            self.client = get_groq_client(api_key=self.api_key)
            self.async_client = get_async_groq_client(api_key=self.api_key)
        else:
            # Create new clients without caching
            self.client = Groq(api_key=self.api_key)
            self.async_client = AsyncGroq(api_key=self.api_key)
    elif self.is_cerebras:
        # use cerebras-specific client
        self.config.chat_model = self.config.chat_model.replace("cerebras/", "")
        if self.api_key == OPENAI_API_KEY:
            self.api_key = os.getenv("CEREBRAS_API_KEY", DUMMY_API_KEY)
        if self.config.use_cached_client:
            self.client = get_cerebras_client(api_key=self.api_key)
            # TODO there is not async client, so should we do anything here?
            self.async_client = get_async_cerebras_client(api_key=self.api_key)
        else:
            # Create new clients without caching
            self.client = Cerebras(api_key=self.api_key)
            self.async_client = AsyncCerebras(api_key=self.api_key)
    else:
        # in these cases, there's no specific client: OpenAI python client suffices
        #
        # vertexai/ is tested FIRST, deliberately: the other branches key
        # off `chat_model_orig`, and if one of them claimed a vertexai/
        # model the route would silently become that provider's -- a public
        # endpoint, with no ADC credential and no regional URL.
        if self.is_vertexai:
            # self.config was rebuilt as a VertexAIConfig above, so this
            # branch cannot see an OPENAI_-prefixed api_key/headers.
            vertex_config = cast(VertexAIConfig, self.config)
            # "vertexai/google/gemini-2.5-flash" -> "google/gemini-2.5-flash",
            # which is the model id the endpoint expects.
            vertex_model = vertex_config.chat_model[len(VERTEXAI_PREFIX) :]
            publisher, _, bare_model = vertex_model.partition("/")
            if not publisher or not bare_model:
                raise ValueError(
                    f"Invalid vertexai/ model {vertex_config.chat_model!r}: "
                    "expected vertexai/<publisher>/<model>, e.g. "
                    "vertexai/google/gemini-2.5-flash."
                )
            vertex_config.chat_model = vertex_model
            if vertex_config.completion_model.startswith(VERTEXAI_PREFIX):
                vertex_config.completion_model = vertex_config.completion_model[
                    len(VERTEXAI_PREFIX) :
                ]
            if vertex_config.api_base:
                # An explicit api_base (VERTEXAI_API_BASE, or api_base on a
                # directly-constructed VertexAIConfig) targets a private/PSC
                # endpoint; honor it rather than overwriting it. The
                # conversion never carries api_base over from an
                # OPENAI_-prefixed config, so a value here is deliberate.
                self.api_base = vertex_config.api_base
            else:
                vertex_project = (
                    vertex_config.project_id
                    or os.getenv("GOOGLE_CLOUD_PROJECT")
                    or os.getenv("GCP_PROJECT")
                )
                if not vertex_project:
                    raise ValueError(
                        "A GCP project is required for a vertexai/ model: "
                        "set VertexAIConfig(project_id=...), or "
                        "VERTEXAI_PROJECT_ID, GOOGLE_CLOUD_PROJECT or "
                        "GCP_PROJECT in the environment."
                    )
                vertex_location = (
                    vertex_config.location
                    or os.getenv("GOOGLE_CLOUD_LOCATION")
                    or DEFAULT_VERTEXAI_LOCATION
                )
                # Re-validate: the values may have come from
                # GOOGLE_CLOUD_*/GCP_* above, which the field validators
                # never see, and both are interpolated into the URL below.
                vertex_config.project_id = _validate_vertexai_id(
                    "project_id", vertex_project
                )
                vertex_config.location = _validate_vertexai_id(
                    "location", vertex_location
                )
                self.api_base = (
                    f"https://{vertex_config.location}-aiplatform."
                    f"googleapis.com/v1beta1"
                    f"/projects/{vertex_config.project_id}"
                    f"/locations/{vertex_config.location}/endpoints/openapi"
                )
            no_explicit_key = vertex_config.api_key in ("", DUMMY_API_KEY)
            if vertex_config.api_key_provider is None and no_explicit_key:
                # No explicit Vertex credential: fall back to Google ADC,
                # resolved per request so its ~1h lifetime is handled here
                # rather than by the caller.
                vertex_config.api_key_provider = _adc_access_token
        elif self.is_litellm_proxy:
            self.config.chat_model = self.config.chat_model.replace(
                "litellm-proxy/", ""
            )
            if self.api_key == OPENAI_API_KEY:
                self.api_key = self.config.litellm_proxy.api_key or self.api_key
            self.api_base = self.config.litellm_proxy.api_base or self.api_base
        elif self.is_gemini:
            if self.api_key == OPENAI_API_KEY:
                self.api_key = os.getenv("GEMINI_API_KEY", DUMMY_API_KEY)
            # Prefer caller config, then Gemini env, then the default.
            gemini_api_base = os.getenv("GEMINI_API_BASE", "")
            explicit_api_base = (
                self.config.api_base if self.config._api_base_was_supplied else None
            )
            self.api_base = explicit_api_base or gemini_api_base or GEMINI_BASE_URL
            if (
                self.config.chat_model.startswith("gemini/")
                or self.api_base == GEMINI_BASE_URL
            ):
                self.config.chat_model = self.config.chat_model.split("/", 1)[1]
        elif self.is_glhf:
            self.config.chat_model = self.config.chat_model.replace("glhf/", "")
            if self.api_key == OPENAI_API_KEY:
                self.api_key = os.getenv("GLHF_API_KEY", DUMMY_API_KEY)
            self.api_base = GLHF_BASE_URL
        elif self.is_openrouter:
            self.config.chat_model = self.config.chat_model.replace(
                "openrouter/", ""
            )
            if self.api_key == OPENAI_API_KEY:
                self.api_key = os.getenv("OPENROUTER_API_KEY", DUMMY_API_KEY)
            self.api_base = OPENROUTER_BASE_URL
        elif self.is_deepseek:
            self.config.chat_model = self.config.chat_model.replace("deepseek/", "")
            self.api_base = DEEPSEEK_BASE_URL
            if self.api_key == OPENAI_API_KEY:
                self.api_key = os.getenv("DEEPSEEK_API_KEY", DUMMY_API_KEY)
        elif self.is_minimax:
            self.config.chat_model = self.config.chat_model.replace("minimax/", "")
            # Honor caller-supplied base URL (e.g. regional endpoints,
            # proxies) instead of always forcing the default.
            openai_api_base = os.getenv("OPENAI_API_BASE")
            explicit_api_base = (
                self.config.api_base
                if self.config.api_base and self.config.api_base != openai_api_base
                else None
            )
            self.api_base = explicit_api_base or MINIMAX_BASE_URL
            if self.api_key == OPENAI_API_KEY:
                # Only overwrite with MINIMAX_API_KEY when it is actually
                # set, so users who intentionally put their MiniMax key in
                # OPENAI_API_KEY are not silently downgraded to a dummy key.
                minimax_key = os.getenv("MINIMAX_API_KEY", "")
                if minimax_key:
                    self.api_key = minimax_key
            # Recompute capabilities now that the prefix has been stripped
            # and self.info() can find the model in MODEL_INFO.
            self.supports_strict_tools = True
            self.supports_json_schema = self.info().has_structured_output
        elif self.is_langdb:
            self.config.chat_model = self.config.chat_model.replace("langdb/", "")
            self.api_base = self.config.langdb_params.base_url
            project_id = self.config.langdb_params.project_id
            if project_id:
                self.api_base += "/" + project_id + "/v1"
            if self.api_key == OPENAI_API_KEY:
                self.api_key = self.config.langdb_params.api_key or DUMMY_API_KEY

            if self.config.langdb_params:
                params = self.config.langdb_params
                if params.project_id:
                    self.config.headers["x-project-id"] = params.project_id
                if params.label:
                    self.config.headers["x-label"] = params.label
                if params.run_id:
                    self.config.headers["x-run-id"] = params.run_id
                if params.thread_id:
                    self.config.headers["x-thread-id"] = params.thread_id
        elif self.is_portkey:
            # Parse the model string and extract provider/model
            provider, model = self.config.portkey_params.parse_model_string(
                self.config.chat_model
            )
            self.config.chat_model = model
            if provider:
                self.config.portkey_params.provider = provider

            # Set Portkey base URL
            self.api_base = self.config.portkey_params.base_url + "/v1"

            # Set API key - use provider's API key from env if available
            if self.api_key == OPENAI_API_KEY:
                self.api_key = self.config.portkey_params.get_provider_api_key(
                    self.config.portkey_params.provider, DUMMY_API_KEY
                )

            # Add Portkey-specific headers
            self.config.headers.update(self.config.portkey_params.get_headers())

        # Sanitize the API key: strip leading/trailing whitespace
        # (including stray newlines from .env files or CI secrets).
        self.api_key = self.api_key.strip()

        # Create http_client if needed - Priority order:
        # 1. http_client_factory (most flexibility, not cacheable)
        # 2. http_client_config (cacheable, moderate flexibility)
        # 3. http_verify_ssl=False (cacheable, simple SSL bypass)
        http_client = None
        async_http_client = None
        http_client_config_used = None

        if self.config.http_client_factory is not None:
            # Use the factory to create http_client (not cacheable)
            http_client = self.config.http_client_factory()
            if isinstance(http_client, (list, tuple)):
                if len(http_client) != 2:
                    raise ValueError(
                        "http_client_factory must return either a single "
                        "httpx.Client or a tuple of "
                        "(httpx.Client, httpx.AsyncClient)"
                    )
                http_client, async_http_client = http_client
            else:
                # set async_http_client to None - so that it will
                # be created later
                async_http_client = None
        elif self.config.http_client_config is not None:
            # Use config dict (cacheable)
            http_client_config_used = self.config.http_client_config
        elif not self.config.http_verify_ssl:
            # Simple SSL bypass (cacheable)
            http_client_config_used = {"verify": False}
            logging.warning(
                "SSL verification has been disabled. This is insecure and "
                "should only be used in trusted environments (e.g., "
                "corporate networks with self-signed certificates)."
            )

        # With an api_key_provider, hand the callable itself to the
        # OpenAI client, which resolves a fresh token on each request.
        openai_api_key: Union[str, Callable[[], str]] = (
            self.config.api_key_provider
            if self.config.api_key_provider is not None
            else self.api_key
        )

        # On a vertexai/ route, hide the SDK's own OPENAI_* channels
        # for the duration of client construction -- see
        # _without_openai_sdk_env. A nullcontext elsewhere, so no
        # other route changes behaviour.
        with _without_openai_sdk_env() if self.is_vertexai else nullcontext():
            if self.config.use_cached_client:
                self.client = get_openai_client(
                    api_key=openai_api_key,
                    base_url=self.api_base,
                    organization=self.config.organization,
                    timeout=Timeout(self.config.timeout),
                    default_headers=self.config.headers,
                    http_client=http_client,
                    http_client_config=http_client_config_used,
                    sdk_env_scrubbed=self.is_vertexai,
                )
                self.async_client = get_async_openai_client(
                    api_key=openai_api_key,
                    base_url=self.api_base,
                    organization=self.config.organization,
                    timeout=Timeout(self.config.timeout),
                    default_headers=self.config.headers,
                    http_client=async_http_client,
                    http_client_config=http_client_config_used,
                    sdk_env_scrubbed=self.is_vertexai,
                )
            else:
                # Create new clients without caching
                client_kwargs: Dict[str, Any] = dict(
                    api_key=openai_api_key,
                    base_url=self.api_base,
                    organization=self.config.organization,
                    timeout=Timeout(self.config.timeout),
                    default_headers=self.config.headers,
                )
                if http_client is not None:
                    client_kwargs["http_client"] = http_client
                elif http_client_config_used is not None:
                    # Create http_client from config for non-cached scenario
                    try:
                        httpx = import_httpx()
                    except ImportError:
                        raise ValueError(missing_httpx_message())
                    client_kwargs["http_client"] = httpx.Client(
                        **http_client_config_used
                    )
                self.client = OpenAI(**client_kwargs)

                async_client_kwargs: Dict[str, Any] = dict(
                    api_key=(
                        # AsyncOpenAI awaits its api_key callable
                        wrap_api_key_provider_async(openai_api_key)
                        if callable(openai_api_key)
                        else openai_api_key
                    ),
                    base_url=self.api_base,
                    organization=self.config.organization,
                    timeout=Timeout(self.config.timeout),
                    default_headers=self.config.headers,
                )
                if async_http_client is not None:
                    async_client_kwargs["http_client"] = async_http_client
                elif http_client_config_used is not None:
                    # Create async http_client from config for non-cached scenario
                    try:
                        httpx = import_httpx()
                    except ImportError:
                        raise ValueError(missing_httpx_message())
                    async_client_kwargs["http_client"] = httpx.AsyncClient(
                        **http_client_config_used
                    )
                self.async_client = AsyncOpenAI(**async_client_kwargs)

    self.cache: CacheDB | None = None
    use_cache = self.config.cache_config is not None
    if "redis" in settings.cache_type and use_cache:
        # read and write self.config, not the `config` argument: a
        # vertexai/ route replaced self.config above, and mutating the
        # original would leave self.config.cache_config disagreeing with
        # the cache actually built here.
        if self.config.cache_config is None or not isinstance(
            self.config.cache_config,
            RedisCacheConfig,
        ):
            # switch to fresh redis config if needed
            self.config.cache_config = RedisCacheConfig(
                fake="fake" in settings.cache_type
            )
        if "fake" in settings.cache_type:
            # force use of fake redis if global cache_type is "fakeredis"
            self.config.cache_config.fake = True
        self.cache = RedisCache(self.config.cache_config)
    elif settings.cache_type != "none" and use_cache:
        raise ValueError(
            f"Invalid cache type {settings.cache_type}. "
            "Valid types are redis, fakeredis, none"
        )

    self.config._validate_litellm()

is_gemini_model()

Are we using the gemini OpenAI-compatible API?

Source code in langroid/language_models/openai_gpt.py

def is_gemini_model(self) -> bool:
    """Are we using the gemini OpenAI-compatible API?"""
    return self.chat_model_orig.startswith(GEMINI_MODEL_PREFIXES)

is_minimax_model()

Are we using the MiniMax OpenAI-compatible API?

Source code in langroid/language_models/openai_gpt.py

def is_minimax_model(self) -> bool:
    """Are we using the MiniMax OpenAI-compatible API?"""
    minimax_models = [e.value for e in MiniMaxModel]
    return (
        self.chat_model_orig in minimax_models
        or self.chat_model_orig.startswith("minimax/")
    )

unsupported_params()

List of params that are not supported by the current model

Source code in langroid/language_models/openai_gpt.py

def unsupported_params(self) -> List[str]:
    """
    List of params that are not supported by the current model
    """
    unsupported = set(self.info().unsupported_params)
    return list(unsupported)

rename_params()

Map of param name -> new name for specific models. Currently main troublemaker is o1* series.

Source code in langroid/language_models/openai_gpt.py

def rename_params(self) -> Dict[str, str]:
    """
    Map of param name -> new name for specific models.
    Currently main troublemaker is o1* series.
    """
    return self.info().rename_params

chat_context_length()

Context-length for chat-completion models/endpoints. Get it from the config if explicitly given, otherwise use model_info based on model name, and fall back to generic model_info if there's no match.

Source code in langroid/language_models/openai_gpt.py

def chat_context_length(self) -> int:
    """
    Context-length for chat-completion models/endpoints.
    Get it from the config if explicitly given,
     otherwise use model_info based on model name, and fall back to
     generic model_info if there's no match.
    """
    return self.config.chat_context_length or self.info().context_length

completion_context_length()

Context-length for completion models/endpoints. Get it from the config if explicitly given, otherwise use model_info based on model name, and fall back to generic model_info if there's no match.

Source code in langroid/language_models/openai_gpt.py

def completion_context_length(self) -> int:
    """
    Context-length for completion models/endpoints.
    Get it from the config if explicitly given,
     otherwise use model_info based on model name, and fall back to
     generic model_info if there's no match.
    """
    return (
        self.config.completion_context_length
        or self.completion_info().context_length
    )

chat_cost()

(Prompt, Cached, Generation) cost per 1000 tokens, for chat-completion models/endpoints. Get it from the dict, otherwise fail-over to general method

Source code in langroid/language_models/openai_gpt.py

def chat_cost(self) -> Tuple[float, float, float]:
    """
    (Prompt, Cached, Generation) cost per 1000 tokens, for chat-completion
    models/endpoints.
    Get it from the dict, otherwise fail-over to general method
    """
    info = self.info()
    cached_cost_per_million = info.cached_cost_per_million
    if not cached_cost_per_million:
        cached_cost_per_million = info.input_cost_per_million
    return (
        info.input_cost_per_million / 1000,
        cached_cost_per_million / 1000,
        info.output_cost_per_million / 1000,
    )

set_stream(stream)

Enable or disable streaming output from API. Args: stream: enable streaming output from API Returns: previous value of stream

Source code in langroid/language_models/openai_gpt.py

def set_stream(self, stream: bool) -> bool:
    """Enable or disable streaming output from API.
    Args:
        stream: enable streaming output from API
    Returns: previous value of stream
    """
    tmp = self.config.stream
    self.config.stream = stream
    return tmp

get_stream()

Get streaming status.

Source code in langroid/language_models/openai_gpt.py

def get_stream(self) -> bool:
    """Get streaming status."""
    return self.config.stream and settings.stream and self.info().allows_streaming

tool_deltas_to_tools(tools) staticmethod

Convert accumulated tool-call deltas to OpenAIToolCall objects. Adapted from this excellent code: https://community.openai.com/t/help-for-function-calls-with-streaming/627170/2

Parameters:

Name	Type	Description	Default
tools	List[Dict[str, Any]]	list of tool deltas received from streaming API	required

Returns:

Name	Type	Description
str	str	plain text corresponding to tool calls that failed to parse
	List[OpenAIToolCall]	List[OpenAIToolCall]: list of OpenAIToolCall objects
	List[Dict[str, Any]]	List[Dict[str, Any]]: list of tool dicts (to reconstruct OpenAI API response, so it can be cached)

Source code in langroid/language_models/openai_gpt.py

@staticmethod
def tool_deltas_to_tools(
    tools: List[Dict[str, Any]],
) -> Tuple[
    str,
    List[OpenAIToolCall],
    List[Dict[str, Any]],
]:
    """
    Convert accumulated tool-call deltas to OpenAIToolCall objects.
    Adapted from this excellent code:
     https://community.openai.com/t/help-for-function-calls-with-streaming/627170/2

    Args:
        tools: list of tool deltas received from streaming API

    Returns:
        str: plain text corresponding to tool calls that failed to parse
        List[OpenAIToolCall]: list of OpenAIToolCall objects
        List[Dict[str, Any]]: list of tool dicts
            (to reconstruct OpenAI API response, so it can be cached)
    """
    # Initialize a dictionary with default values

    # idx -> dict repr of tool
    # (used to simulate OpenAIResponse object later, and also to
    # accumulate function args as strings)
    idx2tool_dict: Dict[str, Dict[str, Any]] = defaultdict(
        lambda: {
            "id": None,
            "function": {"arguments": "", "name": None},
            "type": None,
            "extra_content": None,
        }
    )

    for tool_delta in tools:
        if tool_delta["id"] is not None:
            idx2tool_dict[tool_delta["index"]]["id"] = tool_delta["id"]

        if tool_delta["function"]["name"] is not None:
            idx2tool_dict[tool_delta["index"]]["function"]["name"] = tool_delta[
                "function"
            ]["name"]

        idx2tool_dict[tool_delta["index"]]["function"]["arguments"] += tool_delta[
            "function"
        ]["arguments"]

        if tool_delta["type"] is not None:
            idx2tool_dict[tool_delta["index"]]["type"] = tool_delta["type"]

        if tool_delta.get("extra_content") is not None:
            idx2tool_dict[tool_delta["index"]]["extra_content"] = tool_delta[
                "extra_content"
            ]

    # (try to) parse the fn args of each tool
    contents: List[str] = []
    good_indices = []
    id2args: Dict[str, None | Dict[str, Any]] = {}
    for idx, tool_dict in idx2tool_dict.items():
        failed_content, args_dict = OpenAIGPT._parse_function_args(
            tool_dict["function"]["arguments"]
        )
        # used to build tool_calls_list below
        id2args[tool_dict["id"]] = args_dict or None  # if {}, store as None
        if failed_content != "":
            contents.append(failed_content)
        else:
            good_indices.append(idx)

    # remove the failed tool calls
    idx2tool_dict = {
        idx: tool_dict
        for idx, tool_dict in idx2tool_dict.items()
        if idx in good_indices
    }

    # create OpenAIToolCall list
    tool_calls_list = [
        OpenAIToolCall(
            id=tool_dict["id"],
            function=LLMFunctionCall(
                name=tool_dict["function"]["name"],
                arguments=id2args.get(tool_dict["id"]),
            ),
            type=tool_dict["type"],
            extra_content=tool_dict.get("extra_content"),
        )
        for tool_dict in idx2tool_dict.values()
    ]
    return "\n".join(contents), tool_calls_list, list(idx2tool_dict.values())

OpenAICallParams

Bases: BaseModel

Various params that can be sent to an OpenAI API chat-completion call. When specified, any param here overrides the one with same name in the OpenAIGPTConfig. See OpenAI API Reference for details on the params: https://platform.openai.com/docs/api-reference/chat

VertexAIConfig(**kwargs)

Bases: OpenAIGPTConfig

Config for Google Vertex AI models served at its OpenAI-compatible endpoint, i.e. chat_model="vertexai/<publisher>/<model>".

This is a separate config class, rather than extra fields on OpenAIGPTConfig, for one reason: OpenAIGPTConfig is a pydantic BaseSettings with env_prefix="OPENAI_", so a config built in a process that has OPENAI_API_KEY, OPENAI_HEADERS, OPENAI_ORGANIZATION or OPENAI_API_BASE set inherits every one of them -- and would then send them to Google. Overriding env_prefix to VERTEXAI_ means those variables are not an env source for this class at all, so there is nothing to inherit: the fix is structural rather than a field-by-field scrub.

Credentials, in precedence order:

• api_key_provider, called per request (so a rotating token never goes stale);
• api_key (or VERTEXAI_API_KEY);
• otherwise Google Application Default Credentials, installed as an api_key_provider, which requires gcloud auth application-default login or GOOGLE_APPLICATION_CREDENTIALS.

See docs/notes/gemini.md.

Source code in langroid/language_models/openai_gpt.py

def __init__(self, **kwargs) -> None:  # type: ignore
    api_base_was_supplied = "api_base" in kwargs
    local_model = "api_base" in kwargs and kwargs["api_base"] is not None

    chat_model = kwargs.get("chat_model", "")
    local_prefixes = ["local/", "litellm/", "ollama/", "vllm/", "llamacpp/"]
    if any(chat_model.startswith(prefix) for prefix in local_prefixes):
        local_model = True

    warn_gpt_3_5 = (
        "chat_model" not in kwargs.keys()
        and not local_model
        and default_openai_chat_model == OpenAIChatModel.GPT3_5_TURBO
    )

    if warn_gpt_3_5:
        existing_hook = kwargs.get("run_on_first_use", noop)

        def with_warning() -> None:
            existing_hook()
            gpt_3_5_warning()

        kwargs["run_on_first_use"] = with_warning

    super().__init__(**kwargs)
    env_prefix = self.model_config.get("env_prefix")
    env_api_base_name = f"{env_prefix}API_BASE"
    env_api_base = os.getenv(env_api_base_name)
    if not self.model_config.get("case_sensitive"):
        case_insensitive_env = {
            name.lower(): value for name, value in os.environ.items()
        }
        env_api_base = case_insensitive_env.get(env_api_base_name.lower())
    api_base_was_supplied = api_base_was_supplied or (
        self.api_base is not None
        and (env_prefix != "OPENAI_" or self.api_base != env_api_base)
    )
    self._api_base_was_supplied = api_base_was_supplied

RateLimitConfig

Bases: BaseSettings

Settings for the pro-active rate limiter.

Every field can be overridden by an env var with the LANGROID_RATE_LIMIT_ prefix, e.g. LANGROID_RATE_LIMIT_ENABLED=1.

RateLimiter(config=None, name='')

Paces sends so provider rate limits are approached, not exceeded.

Thread-safe and asyncio-safe: state is guarded by a threading.Lock held only for the (non-blocking) bookkeeping, while the wait itself happens outside the lock via time.sleep or asyncio.sleep.

Slots are handed out as reservations off a shared monotonic clock, so N concurrent callers queue rather than all retrying at once.

Source code in langroid/language_models/rate_limiter.py

def __init__(
    self, config: Optional[RateLimitConfig] = None, name: str = ""
) -> None:
    self.config = config or RateLimitConfig()
    self.name = name
    self._lock = threading.Lock()
    self._next_send_at = 0.0  # monotonic time of the next free slot
    # Hard cooldown floor: no caller may send before this, including one
    # that is already asleep on an earlier reservation.
    self._gate_until = 0.0
    self._request_rate: Optional[float] = None  # requests/sec, from headers
    self._token_rate: Optional[float] = None  # tokens/sec, from headers
    self._avg_tokens: Optional[float] = None  # EWMA tokens per request
    self._fallback_interval = 0.0  # header-free AIMD interval
    self._seen_headers = False
    self._sends = 0
    self._waits = 0
    self._total_wait = 0.0
    self._rate_limit_errors = 0
    self._capped_waits = 0

acquire()

Block until it is this caller's turn to send. Returns seconds waited.

Source code in langroid/language_models/rate_limiter.py

def acquire(self) -> float:
    """Block until it is this caller's turn to send. Returns seconds waited."""
    waited = 0.0
    for wait in self._wait_steps():
        time.sleep(wait)
        waited += wait
    return waited

acquire_async() async

Async variant of acquire. Returns seconds waited.

Source code in langroid/language_models/rate_limiter.py

async def acquire_async(self) -> float:
    """Async variant of `acquire`. Returns seconds waited."""
    waited = 0.0
    for wait in self._wait_steps():
        await asyncio.sleep(wait)
        waited += wait
    return waited

observe_response(headers=None, tokens_used=None)

Take in what a response reveals about the budget.

Information only: callers may call this more than once per request (the headers arrive with the response, a streaming request's token usage only later), so it must NOT be where the AIMD recovery is applied -- see observe_success, which is called exactly once per request.

Parameters:

Name	Type	Description	Default
headers	Optional[Mapping[str, Any]]	Response headers, if the provider/transport exposes them.	None
tokens_used	Optional[int]	Total tokens billed for this request, if reported. Used as an EWMA estimate of the per-request token cost, which is what makes token-budget pacing possible. It is an estimate, not a guarantee of provider quota compliance.	None

Source code in langroid/language_models/rate_limiter.py

def observe_response(
    self,
    headers: Optional[Mapping[str, Any]] = None,
    tokens_used: Optional[int] = None,
) -> None:
    """Take in what a response reveals about the budget.

    Information only: callers may call this more than once per request (the
    headers arrive with the response, a streaming request's token usage
    only later), so it must NOT be where the AIMD recovery is applied --
    see `observe_success`, which is called exactly once per request.

    Args:
        headers: Response headers, if the provider/transport exposes them.
        tokens_used: Total tokens billed for this request, if reported.
            Used as an EWMA estimate of the per-request token cost, which
            is what makes token-budget pacing possible. It is an estimate,
            not a guarantee of provider quota compliance.
    """
    snap = RateLimitSnapshot.from_headers(headers) if headers is not None else None
    with self._lock:
        if snap is not None:
            self._apply_snapshot_locked(snap)
        if tokens_used is not None and tokens_used > 0:
            if self._avg_tokens is None:
                self._avg_tokens = float(tokens_used)
            else:
                self._avg_tokens = 0.7 * self._avg_tokens + 0.3 * tokens_used

observe_success(tokens_used=None)

Record that one request was accepted; recover the send rate.

Call this EXACTLY once per request that the provider accepted. It is the only place the header-free AIMD interval decays, so calling it twice for one request would halve the backoff twice over.

Source code in langroid/language_models/rate_limiter.py

def observe_success(self, tokens_used: Optional[int] = None) -> None:
    """Record that one request was accepted; recover the send rate.

    Call this EXACTLY once per request that the provider accepted. It is
    the only place the header-free AIMD interval decays, so calling it
    twice for one request would halve the backoff twice over.
    """
    self.observe_response(tokens_used=tokens_used)
    with self._lock:
        if self._fallback_interval > 0:
            decayed = self._fallback_interval * self.config.recovery_factor
            self._fallback_interval = 0.0 if decayed < 1e-3 else decayed

observe_rate_limit_error(headers=None)

Learn from a 429: back off, and stall until the budget recovers.

Source code in langroid/language_models/rate_limiter.py

def observe_rate_limit_error(
    self, headers: Optional[Mapping[str, Any]] = None
) -> None:
    """Learn from a 429: back off, and stall until the budget recovers."""
    snap = RateLimitSnapshot.from_headers(headers) if headers is not None else None
    with self._lock:
        self._rate_limit_errors += 1
        hold = 0.0
        if snap is not None:
            self._apply_snapshot_locked(snap)
            hold = max(
                snap.retry_after or 0.0,
                self._refill_wait(snap, tokens=False),
                self._refill_wait(snap, tokens=True),
            )
        if hold <= 0:
            # The response says nothing actionable: no headers, too few to
            # derive a rate from, or a 429 whose budgets both read healthy.
            # Multiplicatively reduce the send rate instead, and recover it
            # on each success.
            self._fallback_interval = min(
                self.config.max_interval,
                max(
                    self.config.error_interval,
                    self._fallback_interval * self.config.backoff_factor,
                ),
            )
        self._stall_locked(
            min(max(hold, self._fallback_interval), self.config.max_wait)
        )
        logger.debug(
            "rate limiter %s: 429 observed, interval now %.4fs",
            self.name,
            self._interval_locked(),
        )

stats()

Counters and discovered state, for tests and diagnostics.

sends is the "did it run at all" signal: zero means the limiter was never consulted, which is different from "it was consulted and never needed to wait" (sends > 0, waits == 0).

Source code in langroid/language_models/rate_limiter.py

def stats(self) -> Dict[str, Any]:
    """Counters and discovered state, for tests and diagnostics.

    `sends` is the "did it run at all" signal: zero means the limiter was
    never consulted, which is different from "it was consulted and never
    needed to wait" (`sends > 0, waits == 0`).
    """
    with self._lock:
        return dict(
            name=self.name,
            sends=self._sends,
            waits=self._waits,
            total_wait=self._total_wait,
            capped_waits=self._capped_waits,
            rate_limit_errors=self._rate_limit_errors,
            seen_headers=self._seen_headers,
            request_rate=self._request_rate,
            token_rate=self._token_rate,
            avg_tokens=self._avg_tokens,
            fallback_interval=self._fallback_interval,
            interval=self._interval_locked(),
            gate_remaining=max(0.0, self._gate_until - time.monotonic()),
        )

MockLM(config=MockLMConfig())

Bases: LanguageModel

Source code in langroid/language_models/mock_lm.py

def __init__(self, config: MockLMConfig = MockLMConfig()):
    super().__init__(config)
    self.config: MockLMConfig = config

chat(messages, max_tokens=200, tools=None, tool_choice='auto', functions=None, function_call='auto', response_format=None)

Mock chat function for testing

Source code in langroid/language_models/mock_lm.py

def chat(
    self,
    messages: Union[str, List[lm.LLMMessage]],
    max_tokens: int = 200,
    tools: Optional[List[OpenAIToolSpec]] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
    functions: Optional[List[lm.LLMFunctionSpec]] = None,
    function_call: str | Dict[str, str] = "auto",
    response_format: Optional[OpenAIJsonSchemaSpec] = None,
) -> lm.LLMResponse:
    """
    Mock chat function for testing
    """
    last_msg = (
        messages[-1].content or "" if isinstance(messages, list) else messages
    )
    return self._response(last_msg)

achat(messages, max_tokens=200, tools=None, tool_choice='auto', functions=None, function_call='auto', response_format=None) async

Mock chat function for testing

Source code in langroid/language_models/mock_lm.py

async def achat(
    self,
    messages: Union[str, List[lm.LLMMessage]],
    max_tokens: int = 200,
    tools: Optional[List[OpenAIToolSpec]] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
    functions: Optional[List[lm.LLMFunctionSpec]] = None,
    function_call: str | Dict[str, str] = "auto",
    response_format: Optional[OpenAIJsonSchemaSpec] = None,
) -> lm.LLMResponse:
    """
    Mock chat function for testing
    """
    last_msg = (
        messages[-1].content or "" if isinstance(messages, list) else messages
    )
    return await self._response_async(last_msg)

generate(prompt, max_tokens=200)

Mock generate function for testing

Source code in langroid/language_models/mock_lm.py

def generate(self, prompt: str, max_tokens: int = 200) -> lm.LLMResponse:
    """
    Mock generate function for testing
    """
    return self._response(prompt)

agenerate(prompt, max_tokens=200) async

Mock generate function for testing

Source code in langroid/language_models/mock_lm.py

async def agenerate(self, prompt: str, max_tokens: int = 200) -> LLMResponse:
    """
    Mock generate function for testing
    """
    return await self._response_async(prompt)

MockLMConfig

Bases: LLMConfig

Mock Language Model Configuration.

Attributes:

Name	Type	Description
response_dict	Dict[str, str]	A "response rule-book", in the form of a dictionary; if last msg in dialog is x,then respond with response_dict[x]

AzureConfig(**kwargs)

Bases: OpenAIGPTConfig

Configuration for Azure OpenAI GPT.

Attributes:

Name	Type	Description
type	str	should be azure.
api_version	str	can be set in the .env file as AZURE_OPENAI_API_VERSION.
deployment_name	str | None	can be optionally set in the .env file as AZURE_OPENAI_DEPLOYMENT_NAME and should be based the custom name you chose for your deployment when you deployed a model.
model_name	str	[DEPRECATED] can be set in the .env file as AZURE_OPENAI_MODEL_NAME and should be based on the model name chosen during setup.
chat_model	str	the chat model name to use. Can be set via the env variable AZURE_OPENAI_CHAT_MODEL. Recommended to use this instead of model_name.

Source code in langroid/language_models/azure_openai.py

def __init__(self, **kwargs) -> None:  # type: ignore
    if "model_name" in kwargs and "chat_model" not in kwargs:
        kwargs["chat_model"] = kwargs["model_name"]
    super().__init__(**kwargs)

AzureGPT(config)

Bases: OpenAIGPT

Class to access OpenAI LLMs via Azure. These env variables can be obtained from the file .azure_env. Azure OpenAI doesn't support completion

Source code in langroid/language_models/azure_openai.py

def __init__(self, config: AzureConfig):
    # This will auto-populate config values from .env file
    load_dotenv()
    super().__init__(config)
    self.config: AzureConfig = config

    if (
        self.config.azure_openai_client_provider
        or self.config.azure_openai_async_client_provider
    ):
        if not self.config.azure_openai_client_provider:
            self.client = None
            logger.warning(
                "Using user-provided Azure OpenAI client, but only async "
                "client has been provided. Synchronous calls will fail."
            )
        if not self.config.azure_openai_async_client_provider:
            self.async_client = None
            logger.warning(
                "Using user-provided Azure OpenAI client, but no async "
                "client has been provided. Asynchronous calls will fail."
            )

        if self.config.azure_openai_client_provider:
            self.client = self.config.azure_openai_client_provider()
        if self.config.azure_openai_async_client_provider:
            self.async_client = self.config.azure_openai_async_client_provider()
            self.async_client.timeout = Timeout(self.config.timeout)
    else:
        if self.config.api_key == "":
            raise ValueError(
                """
                AZURE_OPENAI_API_KEY not set in .env file,
                please set it to your Azure API key."""
            )

        if self.config.api_base == "":
            raise ValueError(
                """
                AZURE_OPENAI_API_BASE not set in .env file,
                please set it to your Azure API key."""
            )

        self.client = AzureOpenAI(
            api_key=self.config.api_key,
            azure_endpoint=self.config.api_base,
            api_version=self.config.api_version,
            azure_deployment=self.config.deployment_name,
        )
        self.async_client = AsyncAzureOpenAI(
            api_key=self.config.api_key,
            azure_endpoint=self.config.api_base,
            api_version=self.config.api_version,
            azure_deployment=self.config.deployment_name,
            timeout=Timeout(self.config.timeout),
        )

    self.supports_json_schema = (
        self.config.api_version >= azureStructuredOutputAPIMin
        and self.config.api_version in azureStructuredOutputList
    )