base

langroid/language_models/base.py

LLMConfig

Bases: BaseSettings

Common configuration for all language models.

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.

OpenAIToolCall

Bases: BaseModel

Represents a single tool call in a list of tool calls generated by OpenAI LLM API. See https://platform.openai.com/docs/api-reference/chat/create

Attributes:

Name	Type	Description
id	str | None	The id of the tool call.
type	ToolTypes	The type of the tool call; only "function" is currently possible (7/26/24).
function	LLMFunctionCall | None	The function call.

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]) -> "OpenAIToolCall":
    """
    Initialize from dictionary.
    Args:
        d: dictionary containing fields to initialize
    """
    id = message["id"]
    type = message["type"]
    function = LLMFunctionCall.from_dict(message["function"])
    return OpenAIToolCall(id=id, type=type, function=function)

LLMTokenUsage

Bases: BaseModel

Usage of tokens by an LLM.

Role

Bases: str, Enum

Possible roles for a message in a chat.

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

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

LanguageModel(config=LLMConfig())

Bases: ABC

Abstract base class for language models.

Source code in langroid/language_models/base.py

def __init__(self, config: LLMConfig = LLMConfig()):
    self.config = config

create(config) staticmethod

Create a language model. Args: config: configuration for language model Returns: instance of language model

Source code in langroid/language_models/base.py

@staticmethod
def create(config: Optional[LLMConfig]) -> Optional["LanguageModel"]:
    """
    Create a language model.
    Args:
        config: configuration for language model
    Returns: instance of language model
    """
    if type(config) is LLMConfig:
        raise ValueError(
            """
            Cannot create a Language Model object from LLMConfig. 
            Please specify a specific subclass of LLMConfig e.g., 
            OpenAIGPTConfig. If you are creating a ChatAgent from 
            a ChatAgentConfig, please specify the `llm` field of this config
            as a specific subclass of LLMConfig, e.g., OpenAIGPTConfig.
            """
        )
    from langroid.language_models.azure_openai import AzureGPT
    from langroid.language_models.mock_lm import MockLM, MockLMConfig
    from langroid.language_models.openai_gpt import OpenAIGPT

    if config is None or config.type is None:
        return None

    if config.type == "mock":
        return MockLM(cast(MockLMConfig, config))

    openai: Union[Type[AzureGPT], Type[OpenAIGPT]]

    if config.type == "azure":
        openai = AzureGPT
    else:
        openai = OpenAIGPT
    cls = dict(
        openai=openai,
    ).get(config.type, openai)
    return cls(config)  # type: ignore

user_assistant_pairs(lst) staticmethod

Given an even-length sequence of strings, split into a sequence of pairs

Parameters:

Name	Type	Description	Default
lst	List[str]	sequence of strings	required

Returns:

Type	Description
List[Tuple[str, str]]	List[Tuple[str,str]]: sequence of pairs of strings

Source code in langroid/language_models/base.py

@staticmethod
def user_assistant_pairs(lst: List[str]) -> List[Tuple[str, str]]:
    """
    Given an even-length sequence of strings, split into a sequence of pairs

    Args:
        lst (List[str]): sequence of strings

    Returns:
        List[Tuple[str,str]]: sequence of pairs of strings
    """
    evens = lst[::2]
    odds = lst[1::2]
    return list(zip(evens, odds))

get_chat_history_components(messages) staticmethod

From the chat history, extract system prompt, user-assistant turns, and final user msg.

Parameters:

Name	Type	Description	Default
messages	List[LLMMessage]	List of messages in the chat history	required

Returns:

Type	Description
Tuple[str, List[Tuple[str, str]], str]	Tuple[str, List[Tuple[str,str]], str]: system prompt, user-assistant turns, final user msg

Source code in langroid/language_models/base.py

@staticmethod
def get_chat_history_components(
    messages: List[LLMMessage],
) -> Tuple[str, List[Tuple[str, str]], str]:
    """
    From the chat history, extract system prompt, user-assistant turns, and
    final user msg.

    Args:
        messages (List[LLMMessage]): List of messages in the chat history

    Returns:
        Tuple[str, List[Tuple[str,str]], str]:
            system prompt, user-assistant turns, final user msg

    """
    # Handle various degenerate cases
    messages = [m for m in messages]  # copy
    DUMMY_SYS_PROMPT = "You are a helpful assistant."
    DUMMY_USER_PROMPT = "Follow the instructions above."
    if len(messages) == 0 or messages[0].role != Role.SYSTEM:
        logger.warning("No system msg, creating dummy system prompt")
        messages.insert(0, LLMMessage(content=DUMMY_SYS_PROMPT, role=Role.SYSTEM))
    system_prompt = messages[0].content

    # now we have messages = [Sys,...]
    if len(messages) == 1:
        logger.warning(
            "Got only system message in chat history, creating dummy user prompt"
        )
        messages.append(LLMMessage(content=DUMMY_USER_PROMPT, role=Role.USER))

    # now we have messages = [Sys, msg, ...]

    if messages[1].role != Role.USER:
        messages.insert(1, LLMMessage(content=DUMMY_USER_PROMPT, role=Role.USER))

    # now we have messages = [Sys, user, ...]
    if messages[-1].role != Role.USER:
        logger.warning(
            "Last message in chat history is not a user message,"
            " creating dummy user prompt"
        )
        messages.append(LLMMessage(content=DUMMY_USER_PROMPT, role=Role.USER))

    # now we have messages = [Sys, user, ..., user]
    # so we omit the first and last elements and make pairs of user-asst messages
    conversation = [m.content for m in messages[1:-1]]
    user_prompt = messages[-1].content
    pairs = LanguageModel.user_assistant_pairs(conversation)
    return system_prompt, pairs, user_prompt

set_stream(stream) abstractmethod

Enable or disable streaming output from API. Return previous value of stream.

Source code in langroid/language_models/base.py

@abstractmethod
def set_stream(self, stream: bool) -> bool:
    """Enable or disable streaming output from API.
    Return previous value of stream."""
    pass

get_stream() abstractmethod

Get streaming status

Source code in langroid/language_models/base.py

@abstractmethod
def get_stream(self) -> bool:
    """Get streaming status"""
    pass

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

Get chat-completion response from LLM.

Parameters:

Name	Type	Description	Default
messages	Union[str, List[LLMMessage]]	message-history to send to the LLM	required
max_tokens	int	max tokens to generate	200
tools	Optional[List[OpenAIToolSpec]]	tools available for the LLM to use in its response	None
tool_choice	ToolChoiceTypes | Dict[str, str | Dict[str, str]]	tool call mode, one of "none", "auto", "required", or a dict specifying a specific tool.	'auto'
functions	Optional[List[LLMFunctionSpec]]	functions available for LLM to call (deprecated)	None
function_call	str | Dict[str, str]	function calling mode, "auto", "none", or a specific fn (deprecated)	'auto'

Source code in langroid/language_models/base.py

@abstractmethod
def chat(
    self,
    messages: Union[str, List[LLMMessage]],
    max_tokens: int = 200,
    tools: Optional[List[OpenAIToolSpec]] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
    functions: Optional[List[LLMFunctionSpec]] = None,
    function_call: str | Dict[str, str] = "auto",
    response_format: Optional[OpenAIJsonSchemaSpec] = None,
) -> LLMResponse:
    """
    Get chat-completion response from LLM.

    Args:
        messages: message-history to send to the LLM
        max_tokens: max tokens to generate
        tools: tools available for the LLM to use in its response
        tool_choice: tool call mode, one of "none", "auto", "required",
            or a dict specifying a specific tool.
        functions: functions available for LLM to call (deprecated)
        function_call: function calling mode, "auto", "none", or a specific fn
                (deprecated)
    """

    pass

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

Async version of chat. See chat for details.

Source code in langroid/language_models/base.py

@abstractmethod
async def achat(
    self,
    messages: Union[str, List[LLMMessage]],
    max_tokens: int = 200,
    tools: Optional[List[OpenAIToolSpec]] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
    functions: Optional[List[LLMFunctionSpec]] = None,
    function_call: str | Dict[str, str] = "auto",
    response_format: Optional[OpenAIJsonSchemaSpec] = None,
) -> LLMResponse:
    """Async version of `chat`. See `chat` for details."""
    pass

update_usage_cost(chat, prompts, completions, cost)

Update usage cost for this LLM. Args: chat (bool): whether to update for chat or completion model prompts (int): number of tokens used for prompts completions (int): number of tokens used for completions cost (float): total token cost in USD

Source code in langroid/language_models/base.py

def update_usage_cost(
    self, chat: bool, prompts: int, completions: int, cost: float
) -> None:
    """
    Update usage cost for this LLM.
    Args:
        chat (bool): whether to update for chat or completion model
        prompts (int): number of tokens used for prompts
        completions (int): number of tokens used for completions
        cost (float): total token cost in USD
    """
    mdl = self.config.chat_model if chat else self.config.completion_model
    if mdl is None:
        return
    if mdl not in self.usage_cost_dict:
        self.usage_cost_dict[mdl] = LLMTokenUsage()
    counter = self.usage_cost_dict[mdl]
    counter.prompt_tokens += prompts
    counter.completion_tokens += completions
    counter.cost += cost
    counter.calls += 1

tot_tokens_cost() classmethod

Return total tokens used and total cost across all models.

Source code in langroid/language_models/base.py

@classmethod
def tot_tokens_cost(cls) -> Tuple[int, float]:
    """
    Return total tokens used and total cost across all models.
    """
    total_tokens = 0
    total_cost = 0.0
    for counter in cls.usage_cost_dict.values():
        total_tokens += counter.total_tokens
        total_cost += counter.cost
    return total_tokens, total_cost

followup_to_standalone(chat_history, question)

Given a chat history and a question, convert it to a standalone question. Args: chat_history: list of tuples of (question, answer) query: follow-up question

Returns: standalone version of the question

Source code in langroid/language_models/base.py

def followup_to_standalone(
    self, chat_history: List[Tuple[str, str]], question: str
) -> str:
    """
    Given a chat history and a question, convert it to a standalone question.
    Args:
        chat_history: list of tuples of (question, answer)
        query: follow-up question

    Returns: standalone version of the question
    """
    history = collate_chat_history(chat_history)

    prompt = f"""
    Given the CHAT HISTORY below, and a follow-up QUESTION or SEARCH PHRASE,
    rephrase the follow-up question/phrase as a STANDALONE QUESTION that
    can be understood without the context of the chat history.

    Chat history: {history}

    Follow-up question: {question} 
    """.strip()
    show_if_debug(prompt, "FOLLOWUP->STANDALONE-PROMPT= ")
    standalone = self.generate(prompt=prompt, max_tokens=1024).message.strip()
    show_if_debug(prompt, "FOLLOWUP->STANDALONE-RESPONSE= ")
    return standalone

StreamingIfAllowed(llm, stream=True)

Context to temporarily enable or disable streaming, if allowed globally via settings.stream

Source code in langroid/language_models/base.py

def __init__(self, llm: LanguageModel, stream: bool = True):
    self.llm = llm
    self.stream = stream