base

langroid/language_models/base.py

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.

LLMMessage

Bases: BaseModel

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

api_dict()

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. 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, 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.
    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"]
            )
    # 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.

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
        recipient = ""
        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

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

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