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(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 is only for use in the Assistant 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, 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 is only for use in the Assistant 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.dict()
    # 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 "content" in d:
            d["content"] = "[ADDITIONAL SYSTEM MESSAGE:]\n\n" + d["content"]
    # drop None values since API doesn't accept them
    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 "tool_calls" in dict_no_none:
        # convert tool calls to API format
        for tc in dict_no_none["tool_calls"]:
            if "arguments" in tc["function"]:
                # arguments must be a string
                tc["function"]["arguments"] = json.dumps(
                    tc["function"]["arguments"]
                )
    # 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()

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: ", or (b) message is empty and function_call/tool_call with explicit recipient

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,
) -> 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`


    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
        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, ""

    # 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: str, Enum

Enum for OpenAI Chat models

AnthropicModel

Bases: str, Enum

Enum for Anthropic models

GeminiModel

Bases: str, Enum

Enum for Gemini 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.

Source code in langroid/language_models/openai_gpt.py

def __init__(self, **kwargs) -> None:  # type: ignore
    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 defaultOpenAIChatModel == 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)

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.Config.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
    config = config.copy()
    super().__init__(config)
    self.config: OpenAIGPTConfig = config
    # save original model name such as `provider/model` before
    # we strip out the `provider`
    self.chat_model_orig = 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

    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 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

    # 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
        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/", "")
        self.api_key = VLLM_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
        self.api_key = LLAMACPP_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.config.chat_model in openAIStructuredOutputList
        )

    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

    # NOTE: The api_key should be set in the .env file, or via
    # an explicit `export OPENAI_API_KEY=xxx` or `setenv OPENAI_API_KEY xxx`
    # Pydantic's BaseSettings will automatically pick it up from the
    # .env file
    # The config.api_key is ignored when not using an OpenAI model
    if self.is_openai_completion_model() or self.is_openai_chat_model():
        self.api_key = config.api_key
        if self.api_key == DUMMY_API_KEY:
            self.api_key = os.getenv("OPENAI_API_KEY", DUMMY_API_KEY)
    else:
        self.api_key = DUMMY_API_KEY

    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_glhf = self.config.chat_model.startswith("glhf/")
    self.is_openrouter = self.config.chat_model.startswith("openrouter/")

    if self.is_groq:
        # use groq-specific client
        self.config.chat_model = self.config.chat_model.replace("groq/", "")
        self.api_key = os.getenv("GROQ_API_KEY", DUMMY_API_KEY)
        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/", "")
        self.api_key = os.getenv("CEREBRAS_API_KEY", DUMMY_API_KEY)
        self.client = Cerebras(
            api_key=self.api_key,
        )
        # TODO there is not async client, so should we do anything here?
        self.async_client = AsyncCerebras(
            api_key=self.api_key,
        )
    else:
        # in these cases, there's no specific client: OpenAI python client suffices
        if self.is_gemini:
            self.config.chat_model = self.config.chat_model.replace("gemini/", "")
            self.api_key = os.getenv("GEMINI_API_KEY", DUMMY_API_KEY)
            self.api_base = GEMINI_BASE_URL
        elif self.is_glhf:
            self.config.chat_model = self.config.chat_model.replace("glhf/", "")
            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/", ""
            )
            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
            self.api_key = os.getenv("DEEPSEEK_API_KEY", DUMMY_API_KEY)

        self.client = OpenAI(
            api_key=self.api_key,
            base_url=self.api_base,
            organization=self.config.organization,
            timeout=Timeout(self.config.timeout),
        )
        self.async_client = AsyncOpenAI(
            api_key=self.api_key,
            organization=self.config.organization,
            base_url=self.api_base,
            timeout=Timeout(self.config.timeout),
        )

    self.cache: CacheDB | None = None
    use_cache = self.config.cache_config is not None
    if settings.cache_type == "momento" and use_cache:
        from langroid.cachedb.momento_cachedb import (
            MomentoCache,
            MomentoCacheConfig,
        )

        if config.cache_config is None or not isinstance(
            config.cache_config,
            MomentoCacheConfig,
        ):
            # switch to fresh momento config if needed
            config.cache_config = MomentoCacheConfig()
        self.cache = MomentoCache(config.cache_config)
    elif "redis" in settings.cache_type and use_cache:
        if config.cache_config is None or not isinstance(
            config.cache_config,
            RedisCacheConfig,
        ):
            # switch to fresh redis config if needed
            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"
            config.cache_config.fake = True
        self.cache = RedisCache(config.cache_config)
    elif settings.cache_type != "none" and use_cache:
        raise ValueError(
            f"Invalid cache type {settings.cache_type}. "
            "Valid types are momento, redis, fakeredis, none"
        )

    self.config._validate_litellm()

requires_first_user_message()

Does the chat_model require a non-empty first user message? TODO: Add other models here; we know gemini requires a non-empty user message, after the system message.

Source code in langroid/language_models/openai_gpt.py

def requires_first_user_message(self) -> bool:
    """
    Does the chat_model require a non-empty first user message?
    TODO: Add other models here; we know gemini requires a non-empty
    user message, after the system message.
    """
    return self.is_gemini_model()

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
    """
    match self.chat_model_orig:
        case OpenAIChatModel.O1_MINI | OpenAIChatModel.O1_PREVIEW:
            return ["temperature", "stream"]
        case _:
            return []

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.
    """
    match self.config.chat_model:
        case (
            OpenAIChatModel.O1_MINI
            | OpenAIChatModel.O1_PREVIEW
            | GeminiModel.GEMINI_1_5_FLASH
            | GeminiModel.GEMINI_1_5_FLASH_8B
            | GeminiModel.GEMINI_1_5_PRO
        ):
            return {"max_tokens": "max_completion_tokens"}
        case _:
            return {}

chat_context_length()

Context-length 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_context_length(self) -> int:
    """
    Context-length for chat-completion models/endpoints
    Get it from the dict, otherwise fail-over to general method
    """
    model = (
        self.config.completion_model
        if self.config.use_completion_for_chat
        else self.config.chat_model
    )
    return _context_length.get(model, super().chat_context_length())

completion_context_length()

Context-length for completion models/endpoints Get it from the dict, otherwise fail-over to general method

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 dict, otherwise fail-over to general method
    """
    model = (
        self.config.chat_model
        if self.config.use_chat_for_completion
        else self.config.completion_model
    )
    return _context_length.get(model, super().completion_context_length())

chat_cost()

(Prompt, 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]:
    """
    (Prompt, Generation) cost per 1000 tokens, for chat-completion
    models/endpoints.
    Get it from the dict, otherwise fail-over to general method
    """
    return _cost_per_1k_tokens.get(self.chat_model_orig, super().chat_cost())

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. Note we disable streaming in quiet mode.

Source code in langroid/language_models/openai_gpt.py

def get_stream(self) -> bool:
    """Get streaming status. Note we disable streaming in quiet mode."""
    return (
        self.config.stream
        and settings.stream
        and self.config.chat_model not in NON_STREAMING_MODELS
        and not settings.quiet
    )

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,
        }
    )

    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"]

    # (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"],
        )
        for tool_dict in idx2tool_dict.values()
    ]
    return "\n".join(contents), tool_calls_list, list(idx2tool_dict.values())

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 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 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	can be 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	can be set in the .env file as AZURE_OPENAI_MODEL_NAME and should be based on the model name chosen during setup.
model_version	str	can be set in the .env file as AZURE_OPENAI_MODEL_VERSION and should be based on the model name chosen during setup.

Source code in langroid/language_models/openai_gpt.py

def __init__(self, **kwargs) -> None:  # type: ignore
    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 defaultOpenAIChatModel == 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)

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 Attributes: config (AzureConfig): AzureConfig object api_key (str): Azure API key api_base (str): Azure API base url api_version (str): Azure API version model_name (str): the name of gpt model in your deployment model_version (str): the version of gpt model in your deployment

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.deployment_name == "":
        raise ValueError(
            """
            AZURE_OPENAI_DEPLOYMENT_NAME not set in .env file,
            please set it to your Azure openai deployment name."""
        )
    self.deployment_name = self.config.deployment_name

    if self.config.model_name == "":
        raise ValueError(
            """
            AZURE_OPENAI_MODEL_NAME not set in .env file,
            please set it to chat model name in your deployment."""
        )

    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),
        )

    # set the chat model to be the same as the model_name
    self.config.chat_model = self.config.model_name

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