language_models

langroid/language_models/init.py

LLMMessage

Bases: BaseModel

Class representing message sent to, or received from, LLM.

api_dict()

Convert to dictionary for API request. DROP the tool_id, since it is only for use in the Assistant API, not the completion API. Returns: dict: dictionary representation of LLM message

Source code in langroid/language_models/base.py

def api_dict(self) -> Dict[str, Any]:
    """
    Convert to dictionary for API request.
    DROP the tool_id, since it is only for use in the Assistant API,
    not the completion API.
    Returns:
        dict: dictionary representation of LLM message
    """
    d = self.dict()
    # 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"]
            )
    dict_no_none.pop("tool_id", None)
    dict_no_none.pop("timestamp", None)
    return dict_no_none

LLMFunctionCall

Bases: BaseModel

Structure of LLM response indicate 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()
        fun_args = ast.literal_eval(fun_args_str)
    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.

LLMResponse

Bases: BaseModel

Class representing response from LLM.

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 "TO: ", or (b) message is empty and function_call with to: <name>

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 "TO: <name> <content>", or
    (b) `message` is empty and `function_call` with `to: <name>`

    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
        if isinstance(args, dict):
            recipient = args.get("recipient", "")
        return recipient, msg
    else:
        msg = self.message

    # It's not a function 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

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

    # 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.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)
        )

    # 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
        self.api_base = OLLAMA_BASE_URL
        self.api_key = OLLAMA_API_KEY
        self.config.chat_model = self.config.chat_model.replace("ollama/", "")
    else:
        self.api_base = self.config.api_base

    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/")

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

    self.config._validate_litellm()

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.config.chat_model, 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

Source code in langroid/language_models/openai_gpt.py

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

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_GPT_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/"]
    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.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."""
        )

    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."""
        )

    # set the chat model to be the same as the model_name
    # This corresponds to the gpt model you chose for your deployment
    # when you deployed a model
    self.set_chat_model()

    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_chat_model()

Sets the chat model configuration based on the model name specified in the .env. This function checks the model_name in the configuration and sets the appropriate chat model in the config.chat_model. It supports handling for '35-turbo' and 'gpt-4' models. For 'gpt-4', it further delegates the handling to handle_gpt4_model method. If the model name does not match any predefined models, it defaults to OpenAIChatModel.GPT4.

Source code in langroid/language_models/azure_openai.py

def set_chat_model(self) -> None:
    """
    Sets the chat model configuration based on the model name specified in the
    ``.env``. This function checks the `model_name` in the configuration and sets
    the appropriate chat model in the `config.chat_model`. It supports handling for
    '35-turbo' and 'gpt-4' models. For 'gpt-4', it further delegates the handling
    to `handle_gpt4_model` method. If the model name does not match any predefined
    models, it defaults to `OpenAIChatModel.GPT4`.
    """
    MODEL_35_TURBO = "35-turbo"
    MODEL_GPT4 = "gpt-4"

    if self.config.model_name == MODEL_35_TURBO:
        self.config.chat_model = OpenAIChatModel.GPT3_5_TURBO
    elif self.config.model_name == MODEL_GPT4:
        self.handle_gpt4_model()
    else:
        self.config.chat_model = OpenAIChatModel.GPT4

handle_gpt4_model()

Handles the setting of the GPT-4 model in the configuration. This function checks the model_version in the configuration. If the version is not set, it raises a ValueError indicating that the model version needs to be specified in the .env file. It sets OpenAIChatModel.GPT4_TURBO if the version is '1106-Preview', otherwise, it defaults to setting OpenAIChatModel.GPT4.

Source code in langroid/language_models/azure_openai.py

def handle_gpt4_model(self) -> None:
    """
    Handles the setting of the GPT-4 model in the configuration.
    This function checks the `model_version` in the configuration.
    If the version is not set, it raises a ValueError indicating that the model
    version needs to be specified in the ``.env`` file.
    It sets `OpenAIChatModel.GPT4_TURBO` if the version is
    '1106-Preview', otherwise, it defaults to setting `OpenAIChatModel.GPT4`.
    """
    VERSION_1106_PREVIEW = "1106-Preview"

    if self.config.model_version == "":
        raise ValueError(
            "AZURE_OPENAI_MODEL_VERSION not set in .env file. "
            "Please set it to the chat model version used in your deployment."
        )

    if self.config.model_version == VERSION_1106_PREVIEW:
        self.config.chat_model = OpenAIChatModel.GPT4_TURBO
    else:
        self.config.chat_model = OpenAIChatModel.GPT4