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

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

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.openai_gpt import OpenAIGPT

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

    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 conversationn below, and a follow-up question, rephrase the follow-up 
    question as a standalone question.

    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

get_verbatim_extract_async(question, passage) async

Asynchronously, get verbatim extract from passage that is relevant to a question. Asynch allows parallel calls to the LLM API.

Source code in langroid/language_models/base.py

async def get_verbatim_extract_async(self, question: str, passage: Document) -> str:
    """
    Asynchronously, get verbatim extract from passage
    that is relevant to a question.
    Asynch allows parallel calls to the LLM API.
    """
    async with aiohttp.ClientSession():
        templatized_prompt = EXTRACTION_PROMPT_GPT4
        final_prompt = templatized_prompt.format(
            question=question, content=passage.content
        )
        show_if_debug(final_prompt, "EXTRACT-PROMPT= ")
        final_extract = await self.agenerate(prompt=final_prompt, max_tokens=1024)
        show_if_debug(final_extract.message.strip(), "EXTRACT-RESPONSE= ")
    return final_extract.message.strip()

get_verbatim_extracts(question, passages)

From each passage, extract verbatim text that is relevant to a question, using concurrent API calls to the LLM. Args: question: question to be answered passages: list of passages from which to extract relevant verbatim text LLM: LanguageModel to use for generating the prompt and extract Returns: list of verbatim extracts from passages that are relevant to question

Source code in langroid/language_models/base.py

def get_verbatim_extracts(
    self, question: str, passages: List[Document]
) -> List[Document]:
    """
    From each passage, extract verbatim text that is relevant to a question,
    using concurrent API calls to the LLM.
    Args:
        question: question to be answered
        passages: list of passages from which to extract relevant verbatim text
        LLM: LanguageModel to use for generating the prompt and extract
    Returns:
        list of verbatim extracts from passages that are relevant to question
    """
    docs = asyncio.run(self._get_verbatim_extracts(question, passages))
    return docs

get_summary_answer(question, passages)

Given a question and a list of (possibly) doc snippets, generate an answer if possible Args: question: question to answer passages: list of Document objects each containing a possibly relevant snippet, and metadata Returns: a Document object containing the answer, and metadata containing source citations

Source code in langroid/language_models/base.py

def get_summary_answer(self, question: str, passages: List[Document]) -> Document:
    """
    Given a question and a list of (possibly) doc snippets,
    generate an answer if possible
    Args:
        question: question to answer
        passages: list of `Document` objects each containing a possibly relevant
            snippet, and metadata
    Returns:
        a `Document` object containing the answer,
        and metadata containing source citations

    """

    # Define an auxiliary function to transform the list of
    # passages into a single string
    def stringify_passages(passages: List[Document]) -> str:
        return "\n".join(
            [
                f"""
            Extract: {p.content}
            Source: {p.metadata.source}
            """
                for p in passages
            ]
        )

    passages_str = stringify_passages(passages)
    # Substitute Q and P into the templatized prompt

    final_prompt = SUMMARY_ANSWER_PROMPT_GPT4.format(
        question=f"Question:{question}", extracts=passages_str
    )
    show_if_debug(final_prompt, "SUMMARIZE_PROMPT= ")
    # Generate the final verbatim extract based on the final prompt
    llm_response = self.generate(prompt=final_prompt, max_tokens=1024)
    final_answer = llm_response.message.strip()
    show_if_debug(final_answer, "SUMMARIZE_RESPONSE= ")
    parts = final_answer.split("SOURCE:", maxsplit=1)
    if len(parts) > 1:
        content = parts[0].strip()
        sources = parts[1].strip()
    else:
        content = final_answer
        sources = ""
    return Document(
        content=content,
        metadata={
            "source": "SOURCE: " + sources,
            "cached": llm_response.cached,
        },
    )

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