chat_agent

langroid/agent/chat_agent.py

ChatAgentConfig

Bases: AgentConfig

Configuration for ChatAgent

Attributes:

Name	Type	Description
system_message	str	system message to include in message sequence (typically defines role and task of agent). Used only if task is not specified in the constructor.
user_message	Optional[str]	user message to include in message sequence. Used only if task is not specified in the constructor.
use_tools	bool	whether to use our own ToolMessages mechanism
handle_llm_no_tool	Any	desired agent_response when LLM generates non-tool msg.
use_functions_api	bool	whether to use functions/tools native to the LLM API (e.g. OpenAI's function_call or tool_call mechanism)
use_tools_api	bool	When use_functions_api is True, if this is also True, the OpenAI tool-call API is used, rather than the older/deprecated function-call API. However the tool-call API has some tricky aspects, hence we set this to False by default.
strict_recovery	bool	whether to enable strict schema recovery when there is a tool-generation error.
enable_orchestration_tool_handling	bool	whether to enable handling of orchestration tools, e.g. ForwardTool, DoneTool, PassTool, etc.
output_format	Optional[type]	When supported by the LLM (certain OpenAI LLMs and local LLMs served by providers such as vLLM), ensures that the output is a JSON matching the corresponding schema via grammar-based decoding
handle_output_format	bool	When output_format is a ToolMessage T, controls whether T is "enabled for handling".
use_output_format	bool	When output_format is a ToolMessage T, controls whether T is "enabled for use" (by LLM) and instructions on using T are added to the system message.
instructions_output_format	bool	Controls whether we generate instructions for output_format in the system message.
use_tools_on_output_format	bool	Controls whether to automatically switch to the Langroid-native tools mechanism when output_format is set. Note that LLMs may generate tool calls which do not belong to output_format even when strict JSON mode is enabled, so this should be enabled when such tool calls are not desired.
output_format_include_defaults	bool	Whether to include fields with default arguments in the output schema
full_citations	bool	Whether to show source reference citation + content for each citation, or just the main reference citation.
search_for_tools_everywhere	bool	Whether to search for tools everywhere, or only in specific LLM response elements based on use_tools / use_functions_api / use_tools_api config settings.
recognize_recipient_in_content	bool	Whether to parse LLM response text content for recipient routing patterns, specifically: - TO[<recipient>]:<content> addressing format, and - JSON {"recipient": "<name>"} at the top level of the message. When False, only structured routing via function_call/tool_call recipient fields is recognized. Default is True. Note: this is distinct from TaskConfig.recognize_string_signals, which controls Task-level signals like DONE, PASS, and SEND_TO. To fully disable all text-based routing, set both to False.
context_overflow_strategy	Literal['truncate', 'drop_turns']	Strategy for handling context overflow when message history exceeds model context length. Options: - "truncate": Truncate content of early messages (preserves all messages but with shortened content). This maintains the message sequence. - "drop_turns": Drop complete conversation turns (USER + all responses until next USER). More aggressive but cleaner for voice agents. Default is "truncate" for backward compatibility.

ChatAgent(config=ChatAgentConfig(), task=None)

Bases: Agent

Chat Agent interacting with external env (could be human, or external tools). The agent (the LLM actually) is provided with an optional "Task Spec", which is a sequence of LLMMessages. These are used to initialize the task_messages of the agent. In most applications we will use a ChatAgent rather than a bare Agent. The Agent class mainly exists to hold various common methods and attributes. One difference between ChatAgent and Agent is that ChatAgent's llm_response method uses "chat mode" API (i.e. one that takes a message sequence rather than a single message), whereas the same method in the Agent class uses "completion mode" API (i.e. one that takes a single message).

config: settings for the agent

Source code in langroid/agent/chat_agent.py

def __init__(
    self,
    config: ChatAgentConfig = ChatAgentConfig(),
    task: Optional[List[LLMMessage]] = None,
):
    """
    Chat-mode agent initialized with task spec as the initial message sequence
    Args:
        config: settings for the agent

    """
    super().__init__(config)
    self.config: ChatAgentConfig = config
    self.config._set_fn_or_tools()
    self.message_history: List[LLMMessage] = []
    # emit the attachment-token-estimate warning at most once per agent
    self._attachment_tokens_warned: bool = False
    self.init_state()
    # An agent's "task" is defined by a system msg and an optional user msg;
    # These are "priming" messages that kick off the agent's conversation.
    self.system_message: str = self.config.system_message
    self.user_message: str | None = self.config.user_message

    if task is not None:
        # if task contains a system msg, we override the config system msg
        if len(task) > 0 and task[0].role == Role.SYSTEM:
            self.system_message = task[0].content or ""
        # if task contains a user msg, we override the config user msg
        if len(task) > 1 and task[1].role == Role.USER:
            self.user_message = task[1].content

    # system-level instructions for using tools/functions:
    # We maintain these as tools/functions are enabled/disabled,
    # and whenever an LLM response is sought, these are used to
    # recreate the system message (via `_create_system_and_tools_message`)
    # each time, so it reflects the current set of enabled tools/functions.
    # (a) these are general instructions on using certain tools/functions,
    #   if they are specified in a ToolMessage class as a classmethod `instructions`
    self.system_tool_instructions: str = ""
    # (b) these are only for the builtin in Langroid TOOLS mechanism:
    self.system_tool_format_instructions: str = ""

    self.llm_functions_map: Dict[str, LLMFunctionSpec] = {}
    self.llm_functions_handled: Set[str] = set()
    self.llm_functions_usable: Set[str] = set()
    self.llm_function_force: Optional[Dict[str, str]] = None

    self.output_format: Optional[type[ToolMessage | BaseModel]] = None

    self.saved_requests_and_tool_setings = self._requests_and_tool_settings()
    # This variable is not None and equals a `ToolMessage` T, if and only if:
    # (a) T has been set as the output_format of this agent, AND
    # (b) T has been "enabled for use" ONLY for enforcing this output format, AND
    # (c) T has NOT been explicitly "enabled for use" by this Agent.
    self.enabled_use_output_format: Optional[type[ToolMessage]] = None
    # As above but deals with "enabled for handling" instead of "enabled for use".
    self.enabled_handling_output_format: Optional[type[ToolMessage]] = None
    if config.output_format is not None:
        self.set_output_format(config.output_format)
    # instructions specifically related to enforcing `output_format`
    self.output_format_instructions = ""

    # controls whether to disable strict schemas for this agent if
    # strict mode causes exception
    self.disable_strict = False
    # Tracks whether any strict tool is enabled; used to determine whether to set
    # `self.disable_strict` on an exception
    self.any_strict = False
    # Tracks the set of tools on which we force-disable strict decoding
    self.disable_strict_tools_set: set[str] = set()

    # search for tools according to the agent configuration
    if not config.search_for_tools_everywhere:
        if config.use_functions_api:
            if config.use_tools_api:
                self.search_for_tools = {SearchForTools.TOOLS.value}
            else:
                self.search_for_tools = {SearchForTools.FUNCTIONS.value}
        else:
            self.search_for_tools = {SearchForTools.CONTENT.value}

    if self.config.enable_orchestration_tool_handling:
        # Only enable HANDLING by `agent_response`, NOT LLM generation of these.
        # This is useful where tool-handlers or agent_response generate these
        # tools, and need to be handled.
        # We don't want enable orch tool GENERATION by default, since that
        # might clutter-up the LLM system message unnecessarily.
        from langroid.agent.tools.orchestration import (
            AgentDoneTool,
            AgentSendTool,
            DonePassTool,
            DoneTool,
            ForwardTool,
            PassTool,
            ResultTool,
            SendTool,
        )

        self.enable_message(ForwardTool, use=False, handle=True)
        self.enable_message(DoneTool, use=False, handle=True)
        self.enable_message(AgentDoneTool, use=False, handle=True)
        self.enable_message(PassTool, use=False, handle=True)
        self.enable_message(DonePassTool, use=False, handle=True)
        self.enable_message(SendTool, use=False, handle=True)
        self.enable_message(AgentSendTool, use=False, handle=True)
        self.enable_message(ResultTool, use=False, handle=True)

task_messages property

The task messages are the initial messages that define the task of the agent. There will be at least a system message plus possibly a user msg. Returns: List[LLMMessage]: the task messages

all_llm_tools_known property

All known tools; we include output_format if it is a ToolMessage.

init_state()

Initialize the state of the agent. Just conversation state here, but subclasses can override this to initialize other state.

Source code in langroid/agent/chat_agent.py

def init_state(self) -> None:
    """
    Initialize the state of the agent. Just conversation state here,
    but subclasses can override this to initialize other state.
    """
    super().init_state()
    self.clear_history(0)
    self.clear_dialog()

from_id(id) staticmethod

Get an agent from its ID Args: agent_id (str): ID of the agent Returns: ChatAgent: The agent with the given ID

Source code in langroid/agent/chat_agent.py

@staticmethod
def from_id(id: str) -> "ChatAgent":
    """
    Get an agent from its ID
    Args:
        agent_id (str): ID of the agent
    Returns:
        ChatAgent: The agent with the given ID
    """
    return cast(ChatAgent, Agent.from_id(id))

clone(i=0)

Create i'th clone of this agent, ensuring tool use/handling is cloned. Important: We assume all member variables are in the init method here and in the Agent class. TODO: We are attempting to clone an agent after its state has been changed in possibly many ways. Below is an imperfect solution. Caution advised. Revisit later.

Source code in langroid/agent/chat_agent.py

def clone(self, i: int = 0) -> "ChatAgent":
    """Create i'th clone of this agent, ensuring tool use/handling is cloned.
    Important: We assume all member variables are in the __init__ method here
    and in the Agent class.
    TODO: We are attempting to clone an agent after its state has been
    changed in possibly many ways. Below is an imperfect solution. Caution advised.
    Revisit later.
    """
    agent_cls = type(self)
    # Use model_copy to preserve Pydantic subclass types (like MockLMConfig)
    # instead of deepcopy which loses subclass information
    config_copy = self.config.model_copy(deep=True)
    config_copy.name = f"{config_copy.name}-{i}"
    new_agent = agent_cls(config_copy)
    new_agent.system_tool_instructions = self.system_tool_instructions
    new_agent.system_tool_format_instructions = self.system_tool_format_instructions
    new_agent.llm_tools_map = self.llm_tools_map
    new_agent.llm_tools_known = self.llm_tools_known
    new_agent.llm_tools_handled = self.llm_tools_handled
    new_agent.llm_tools_usable = self.llm_tools_usable
    new_agent.llm_functions_map = self.llm_functions_map
    new_agent.llm_functions_handled = self.llm_functions_handled
    new_agent.llm_functions_usable = self.llm_functions_usable
    new_agent.llm_function_force = self.llm_function_force
    # Ensure each clone gets its own vecdb client when supported.
    new_agent.vecdb = None if self.vecdb is None else self.vecdb.clone()
    self._clone_extra_state(new_agent)
    new_agent.id = ObjectRegistry.new_id()
    if self.config.add_to_registry:
        ObjectRegistry.register_object(new_agent)
    return new_agent

clear_history(start=-2, end=-1)

Clear the message history, deleting messages from index start, up to index end.

Parameters:

Name	Type	Description	Default
start	int	index of first message to delete; default = -2 (i.e. delete last 2 messages, typically these are the last user and assistant messages)	-2
end	int	index of last message to delete; Default = -1 (i.e. delete all messages up to the last one)	-1

Source code in langroid/agent/chat_agent.py

def clear_history(self, start: int = -2, end: int = -1) -> None:
    """
    Clear the message history, deleting  messages from index `start`,
    up to index `end`.

    Args:
        start (int): index of first message to delete; default = -2
                (i.e. delete last 2 messages, typically these
                are the last user and assistant messages)
        end (int): index of last message to delete; Default = -1
                (i.e. delete all messages up to the last one)
    """
    n = len(self.message_history)
    if start < 0:
        start = max(0, n + start)
    end_ = n if end == -1 else end + 1
    dropped = self.message_history[start:end_]
    # consider the dropped msgs in REVERSE order, so we are
    # carefully updating self.oai_tool_calls
    for msg in reversed(dropped):
        self._drop_msg_update_tool_calls(msg)
        # clear out the chat document from the ObjectRegistry
        ChatDocument.delete_id(msg.chat_document_id)
    del self.message_history[start:end_]

update_history(message, response)

Update the message history with the latest user message and LLM response. Args: message (str): user message response: (str): LLM response

Source code in langroid/agent/chat_agent.py

def update_history(self, message: str, response: str) -> None:
    """
    Update the message history with the latest user message and LLM response.
    Args:
        message (str): user message
        response: (str): LLM response
    """
    self.message_history.extend(
        [
            LLMMessage(role=Role.USER, content=message),
            LLMMessage(role=Role.ASSISTANT, content=response),
        ]
    )

export_history()

Export message history as a portable, versioned JSON snapshot.

Returns:

Type	Description
str	A JSON string containing the current message history.

Source code in langroid/agent/chat_agent.py

def export_history(self) -> str:
    """Export message history as a portable, versioned JSON snapshot.

    Returns:
        A JSON string containing the current message history.
    """
    messages: List[Dict[str, Any]] = []
    for message in self.message_history:
        json.dumps(
            message.model_dump(mode="python", exclude={"files"}),
            allow_nan=False,
            default=str,
        )
        data = message.model_dump(mode="json", exclude={"files"})
        data["chat_document_id"] = ""
        data["files"] = [
            {
                "content_base64": base64.b64encode(file.content).decode("ascii"),
                "filename": file.filename,
                "mime_type": file.mime_type,
                "url": file.url,
                "detail": file.detail,
            }
            for file in message.files
        ]
        messages.append(data)
    return json.dumps({"version": 1, "messages": messages}, allow_nan=False)

import_history(snapshot, max_size_bytes=100 * 1024 * 1024)

Restore message history from :meth:export_history output.

Parameters:

Name	Type	Description	Default
snapshot	str	Versioned JSON snapshot to restore.	required
max_size_bytes	int	Maximum total decoded attachment size. Defaults to 100 MiB.	100 * 1024 * 1024

Raises:

Type	Description
ValueError	If the snapshot is malformed, has an unknown version, starts with a non-system message, or exceeds max_size_bytes.

Source code in langroid/agent/chat_agent.py

def import_history(
    self,
    snapshot: str,
    max_size_bytes: int = 100 * 1024 * 1024,
) -> None:
    """Restore message history from :meth:`export_history` output.

    Args:
        snapshot: Versioned JSON snapshot to restore.
        max_size_bytes: Maximum total decoded attachment size. Defaults to
            100 MiB.

    Raises:
        ValueError: If the snapshot is malformed, has an unknown version,
            starts with a non-system message, or exceeds ``max_size_bytes``.
    """
    try:
        if max_size_bytes < 0:
            raise ValueError("max_size_bytes must be non-negative")
        payload = json.loads(
            snapshot,
            parse_constant=_reject_json_constant,
            parse_float=_parse_json_float,
        )
        if (
            not isinstance(payload, dict)
            or type(payload.get("version")) is not int
            or payload["version"] != 1
        ):
            raise ValueError("unsupported version")
        raw_messages = payload.get("messages")
        if not isinstance(raw_messages, list):
            raise ValueError("messages must be a list")
        if not raw_messages:
            raise ValueError("messages must not be empty")

        messages: List[LLMMessage] = []
        total_attachment_bytes = 0
        for raw_message in raw_messages:
            if not isinstance(raw_message, dict):
                raise ValueError("message must be an object")
            message_data = dict(raw_message)
            raw_files = message_data.get("files", [])
            if not isinstance(raw_files, list):
                raise ValueError("files must be a list")
            files = []
            for raw_file in raw_files:
                if not isinstance(raw_file, dict):
                    raise ValueError("file must be an object")
                file_data = dict(raw_file)
                encoded_content = file_data.pop("content_base64")
                if not isinstance(encoded_content, str):
                    raise ValueError("content_base64 must be a string")
                encoded_bytes = encoded_content.encode("ascii")
                padding = len(encoded_bytes) - len(encoded_bytes.rstrip(b"="))
                decoded_size = max(
                    0,
                    len(encoded_bytes) // 4 * 3 - min(padding, 2),
                )
                if total_attachment_bytes + decoded_size > max_size_bytes:
                    raise ValueError(
                        "decoded attachments exceed "
                        f"max_size_bytes={max_size_bytes}"
                    )
                content = base64.b64decode(encoded_bytes, validate=True)
                total_attachment_bytes += len(content)
                if total_attachment_bytes > max_size_bytes:
                    raise ValueError(
                        "decoded attachments exceed "
                        f"max_size_bytes={max_size_bytes}"
                    )
                files.append(FileAttachment(content=content, **file_data))
            message_data["files"] = files
            message_data["chat_document_id"] = ""
            messages.append(LLMMessage.model_validate(message_data))
        if messages[0].role != Role.SYSTEM:
            raise ValueError("first message must have role 'system'")
        seen_call_ids: Set[str] = set()
        for message in messages:
            if message.role in (Role.TOOL, Role.FUNCTION) and (
                message.function_call is not None or message.tool_calls is not None
            ):
                raise ValueError("result messages must not contain call payloads")
            if message.function_call is not None and message.role != Role.ASSISTANT:
                raise ValueError("function calls must have role 'assistant'")
            if message.function_call is not None:
                if not message.function_call.name.strip():
                    raise ValueError("function call names must be nonblank")
            if message.tool_call_id is not None and message.role != Role.TOOL:
                raise ValueError("tool_call_id is only valid on tool results")
            if message.role == Role.FUNCTION:
                name = message.name
                if name is None or not name or name != name.strip():
                    raise ValueError(
                        "function result name must be nonblank and normalized"
                    )
            if message.tool_calls is not None:
                if message.role != Role.ASSISTANT:
                    raise ValueError("tool calls must have role 'assistant'")
                for call in message.tool_calls:
                    if call.function is None:
                        raise ValueError("tool calls must include a function")
                    if not call.function.name.strip():
                        raise ValueError("function call names must be nonblank")
                    call_id = call.id
                    if call_id is None or not call_id or call_id != call_id.strip():
                        raise ValueError(
                            "tool call IDs must be nonblank and normalized"
                        )
                    if call_id in seen_call_ids:
                        raise ValueError("tool call IDs must be unique")
                    seen_call_ids.add(call_id)
                if not message.tool_calls:
                    message.tool_calls = None
    except (
        KeyError,
        RecursionError,
        TypeError,
        UnicodeEncodeError,
        ValueError,
    ) as exc:
        raise ValueError(f"Invalid history snapshot: {exc}") from exc

    all_calls = {
        call.id: call
        for message in messages
        for call in (message.tool_calls or [])
        if call.id is not None
    }
    completed_call_ids = {
        message.tool_call_id
        for message in messages
        if message.role == Role.TOOL and message.tool_call_id is not None
    }
    old_chat_document_ids = {
        message.chat_document_id
        for message in self.message_history
        if message.chat_document_id
    }
    for chat_document_id in old_chat_document_ids:
        ChatDocument.delete_id(chat_document_id)
    self.message_history = messages
    self.oai_tool_id2call = all_calls
    self.oai_tool_calls = [
        call
        for call_id, call in all_calls.items()
        if call_id not in completed_call_ids
    ]

tool_format_rules()

Specification of tool formatting rules (typically JSON-based but can be non-JSON, e.g. XMLToolMessage), based on the currently enabled usable ToolMessages

Returns:

Name	Type	Description
str	str	formatting rules

Source code in langroid/agent/chat_agent.py

def tool_format_rules(self) -> str:
    """
    Specification of tool formatting rules
    (typically JSON-based but can be non-JSON, e.g. XMLToolMessage),
    based on the currently enabled usable `ToolMessage`s

    Returns:
        str: formatting rules
    """
    # ONLY Usable tools (i.e. LLM-generation allowed),
    usable_tool_classes: List[Type[ToolMessage]] = [
        t
        for t in list(self.llm_tools_map.values())
        if t.default_value("request") in self.llm_tools_usable
    ]

    if len(usable_tool_classes) == 0:
        return ""
    format_instructions = "\n\n".join(
        [
            msg_cls.format_instructions(tool=self.config.use_tools)
            for msg_cls in usable_tool_classes
        ]
    )
    # if any of the enabled classes has json_group_instructions, then use that,
    # else fall back to ToolMessage.json_group_instructions
    for msg_cls in usable_tool_classes:
        if hasattr(msg_cls, "json_group_instructions") and callable(
            getattr(msg_cls, "json_group_instructions")
        ):
            return msg_cls.group_format_instructions().format(
                format_instructions=format_instructions
            )
    return ToolMessage.group_format_instructions().format(
        format_instructions=format_instructions
    )

tool_instructions()

Instructions for tools or function-calls, for enabled and usable Tools. These are inserted into system prompt regardless of whether we are using our own ToolMessage mechanism or the LLM's function-call mechanism.

Returns:

Name	Type	Description
str	str	concatenation of instructions for all usable tools

Source code in langroid/agent/chat_agent.py

def tool_instructions(self) -> str:
    """
    Instructions for tools or function-calls, for enabled and usable Tools.
    These are inserted into system prompt regardless of whether we are using
    our own ToolMessage mechanism or the LLM's function-call mechanism.

    Returns:
        str: concatenation of instructions for all usable tools
    """
    enabled_classes: List[Type[ToolMessage]] = list(self.llm_tools_map.values())
    if len(enabled_classes) == 0:
        return ""
    instructions = []
    for msg_cls in enabled_classes:
        if msg_cls.default_value("request") in self.llm_tools_usable:
            class_instructions = ""
            if hasattr(msg_cls, "instructions") and inspect.ismethod(
                msg_cls.instructions
            ):
                class_instructions = msg_cls.instructions()
            if (
                self.config.use_tools
                and hasattr(msg_cls, "langroid_tools_instructions")
                and inspect.ismethod(msg_cls.langroid_tools_instructions)
            ):
                class_instructions += msg_cls.langroid_tools_instructions()
            # example will be shown in tool_format_rules() when using TOOLs,
            # so we don't need to show it here.
            example = "" if self.config.use_tools else (msg_cls.usage_examples())
            if example != "":
                example = "EXAMPLES:\n" + example
            guidance = (
                ""
                if class_instructions == ""
                else ("GUIDANCE: " + class_instructions)
            )
            if guidance == "" and example == "":
                continue
            instructions.append(
                textwrap.dedent(
                    f"""
                    TOOL: {msg_cls.default_value("request")}:
                    {guidance}
                    {example}
                    """.lstrip()
                )
            )
    if len(instructions) == 0:
        return ""
    instructions_str = "\n\n".join(instructions)
    return textwrap.dedent(
        f"""
        === GUIDELINES ON SOME TOOLS/FUNCTIONS USAGE ===
        {instructions_str}
        """.lstrip()
    )

augment_system_message(message)

Augment the system message with the given message. Args: message (str): system message

Source code in langroid/agent/chat_agent.py

def augment_system_message(self, message: str) -> None:
    """
    Augment the system message with the given message.
    Args:
        message (str): system message
    """
    self.system_message += "\n\n" + message

last_message_with_role(role)

from message_history, return the last message with role role

Source code in langroid/agent/chat_agent.py

def last_message_with_role(self, role: Role) -> LLMMessage | None:
    """from `message_history`, return the last message with role `role`"""
    n_role_msgs = len([m for m in self.message_history if m.role == role])
    if n_role_msgs == 0:
        return None
    idx = self.nth_message_idx_with_role(role, n_role_msgs)
    return self.message_history[idx]

last_message_idx_with_role(role)

Index of last message in message_history, with specified role. Return -1 if not found. Index = 0 is the first message in the history.

Source code in langroid/agent/chat_agent.py

def last_message_idx_with_role(self, role: Role) -> int:
    """Index of last message in message_history, with specified role.
    Return -1 if not found. Index = 0 is the first message in the history.
    """
    indices_with_role = [
        i for i, m in enumerate(self.message_history) if m.role == role
    ]
    if len(indices_with_role) == 0:
        return -1
    return indices_with_role[-1]

nth_message_idx_with_role(role, n)

Index of nth message in message_history, with specified role. (n is assumed to be 1-based, i.e. 1 is the first message with that role). Return -1 if not found. Index = 0 is the first message in the history.

Source code in langroid/agent/chat_agent.py

def nth_message_idx_with_role(self, role: Role, n: int) -> int:
    """Index of `n`th message in message_history, with specified role.
    (n is assumed to be 1-based, i.e. 1 is the first message with that role).
    Return -1 if not found. Index = 0 is the first message in the history.
    """
    indices_with_role = [
        i for i, m in enumerate(self.message_history) if m.role == role
    ]

    if len(indices_with_role) < n:
        return -1
    return indices_with_role[n - 1]

update_last_message(message, role=Role.USER)

Update the last message that has role role in the message history. Useful when we want to replace a long user prompt, that may contain context documents plus a question, with just the question. Args: message (str): new message to replace with role (str): role of message to replace

Source code in langroid/agent/chat_agent.py

def update_last_message(self, message: str, role: str = Role.USER) -> None:
    """
    Update the last message that has role `role` in the message history.
    Useful when we want to replace a long user prompt, that may contain context
    documents plus a question, with just the question.
    Args:
        message (str): new message to replace with
        role (str): role of message to replace
    """
    if len(self.message_history) == 0:
        return
    # find last message in self.message_history with role `role`
    for i in range(len(self.message_history) - 1, -1, -1):
        if self.message_history[i].role == role:
            self.message_history[i].content = message
            break

delete_last_message(role=Role.USER)

Delete the last message that has role role from the message history. Args: role (str): role of message to delete

Source code in langroid/agent/chat_agent.py

def delete_last_message(self, role: str = Role.USER) -> None:
    """
    Delete the last message that has role `role` from the message history.
    Args:
        role (str): role of message to delete
    """
    if len(self.message_history) == 0:
        return
    # find last message in self.message_history with role `role`
    for i in range(len(self.message_history) - 1, -1, -1):
        if self.message_history[i].role == role:
            self.message_history.pop(i)
            break

handle_message_fallback(msg)

Fallback method for the "no-tools" scenario, i.e., the current msg (presumably emitted by the LLM) does not have any tool that the agent can handle. NOTE: The msg may contain tools but either (a) the agent is not enabled to handle them, or (b) there's an explicit recipient field in the tool that doesn't match the agent's name.

Uses the self.config.non_tool_routing to determine the action to take.

This method can be overridden by subclasses, e.g., to create a "reminder" message when a tool is expected but the LLM "forgot" to generate one.

Parameters:

Name	Type	Description	Default
msg	str | ChatDocument	The input msg to handle	required

Returns: Any: The result of the handler method

Source code in langroid/agent/chat_agent.py

def handle_message_fallback(self, msg: str | ChatDocument) -> Any:
    """
    Fallback method for the "no-tools" scenario, i.e., the current `msg`
    (presumably emitted by the LLM) does not have any tool that the agent
    can handle.
    NOTE: The `msg` may contain tools but either (a) the agent is not
    enabled to handle them, or (b) there's an explicit `recipient` field
    in the tool that doesn't match the agent's name.

    Uses the self.config.non_tool_routing to determine the action to take.

    This method can be overridden by subclasses, e.g.,
    to create a "reminder" message when a tool is expected but the LLM "forgot"
    to generate one.

    Args:
        msg (str | ChatDocument): The input msg to handle
    Returns:
        Any: The result of the handler method
    """
    if (
        isinstance(msg, str)
        or msg.metadata.sender != Entity.LLM
        or self.config.handle_llm_no_tool is None
        or self.has_only_unhandled_tools(msg)
    ):
        return None
    # we ONLY use the `handle_llm_no_tool` config option when
    # the msg is from LLM and does not contain ANY tools at all.
    from langroid.agent.tools.orchestration import AgentDoneTool, ForwardTool

    no_tool_option = self.config.handle_llm_no_tool
    if no_tool_option in list(NonToolAction):
        # in case the `no_tool_option` is one of the special NonToolAction vals
        match self.config.handle_llm_no_tool:
            case NonToolAction.FORWARD_USER:
                return ForwardTool(agent="User")
            case NonToolAction.DONE:
                done_tool = AgentDoneTool(
                    content=msg.content, tools=msg.tool_messages
                )
                # Carry taint on this content+tools repackage, exactly as
                # DonePassTool does: msg may be a tainted doc re-emitted
                # with sender relabeled to LLM (#1035).
                done_tool._tainted = msg.metadata.tainted
                return done_tool
    elif is_callable(no_tool_option):
        return no_tool_option(msg)
    # Otherwise just return `no_tool_option` as is:
    # This can be any string, such as a specific nudge/reminder to the LLM,
    # or even something like ResultTool etc.
    return no_tool_option

unhandled_tools()

The set of tools that are known but not handled. Useful in task flow: an agent can refuse to accept an incoming msg when it only has unhandled tools.

Source code in langroid/agent/chat_agent.py

def unhandled_tools(self) -> set[str]:
    """The set of tools that are known but not handled.
    Useful in task flow: an agent can refuse to accept an incoming msg
    when it only has unhandled tools.
    """
    return self.llm_tools_known - self.llm_tools_handled

enable_message(message_class, use=True, handle=True, force=False, require_recipient=False, include_defaults=True)

Add the tool (message class) to the agent, and enable either - tool USE (i.e. the LLM can generate JSON to use this tool), - tool HANDLING (i.e. the agent can handle JSON from this tool),

Parameters:

Name	Type	Description	Default
message_class	Optional[Type[ToolMessage] | List[Type[ToolMessage]]]	The ToolMessage class OR List of such classes to enable, for USE, or HANDLING, or both. If this is a list of ToolMessage classes, then the remain args are applied to all classes. Optional; if None, then apply the enabling to all tools in the agent's toolset that have been enabled so far.	required
use	bool	IF True, allow the agent (LLM) to use this tool (or all tools), else disallow	True
handle	bool	if True, allow the agent (LLM) to handle (i.e. respond to) this tool (or all tools)	True
force	bool	whether to FORCE the agent (LLM) to USE the specific tool represented by message_class. force is ignored if message_class is None.	False
require_recipient	bool	whether to require that recipient be specified when using the tool message (only applies if use is True).	False
include_defaults	bool	whether to include fields that have default values, in the "properties" section of the JSON format instructions. (Normally the OpenAI completion API ignores these fields, but the Assistant fn-calling seems to pay attn to these, and if we don't want this, we should set this to False.)	True

Source code in langroid/agent/chat_agent.py

def enable_message(
    self,
    message_class: Optional[Type[ToolMessage] | List[Type[ToolMessage]]],
    use: bool = True,
    handle: bool = True,
    force: bool = False,
    require_recipient: bool = False,
    include_defaults: bool = True,
) -> None:
    """
    Add the tool (message class) to the agent, and enable either
    - tool USE (i.e. the LLM can generate JSON to use this tool),
    - tool HANDLING (i.e. the agent can handle JSON from this tool),

    Args:
        message_class: The ToolMessage class OR List of such classes to enable,
            for USE, or HANDLING, or both.
            If this is a list of ToolMessage classes, then the remain args are
            applied to all classes.
            Optional; if None, then apply the enabling to all tools in the
            agent's toolset that have been enabled so far.
        use: IF True, allow the agent (LLM) to use this tool (or all tools),
            else disallow
        handle: if True, allow the agent (LLM) to handle (i.e. respond to) this
            tool (or all tools)
        force: whether to FORCE the agent (LLM) to USE the specific
             tool represented by `message_class`.
             `force` is ignored if `message_class` is None.
        require_recipient: whether to require that recipient be specified
            when using the tool message (only applies if `use` is True).
        include_defaults: whether to include fields that have default values,
            in the "properties" section of the JSON format instructions.
            (Normally the OpenAI completion API ignores these fields,
            but the Assistant fn-calling seems to pay attn to these,
            and if we don't want this, we should set this to False.)
    """
    if message_class is not None and isinstance(message_class, list):
        for mc in message_class:
            self.enable_message(
                mc,
                use=use,
                handle=handle,
                force=force,
                require_recipient=require_recipient,
                include_defaults=include_defaults,
            )
        return None

    # Validate that use/handle are booleans, not accidentally passed tool classes
    if isclass(use) or isclass(handle):
        param = "use" if isclass(use) else "handle"
        raise TypeError(
            textwrap.dedent(
                f"""
                Invalid arguments to enable_message().
                It appears you passed multiple ToolMessage classes as separate
                arguments instead of as a list.

                Correct usage:
                    agent.enable_message([Tool1, Tool2, Tool3])

                Incorrect usage:
                    agent.enable_message(Tool1, Tool2, Tool3)

                The '{param}' parameter must be a boolean, not a class.
                """
            )
        )

    if require_recipient and message_class is not None:
        message_class = message_class.require_recipient()
    if message_class is not None:
        if not issubclass(message_class, ToolMessage):
            raise ValueError("message_class must be a subclass of ToolMessage")
        request = message_class.default_value("request")
        existing_class = self.llm_tools_map.get(request)
        existing_origin = (
            existing_class.__dict__.get("__tool_message_origin__", existing_class)
            if existing_class is not None
            else None
        )
        message_origin = message_class.__dict__.get(
            "__tool_message_origin__", message_class
        )
        if existing_class is not None and existing_origin is not message_origin:
            raise ValueError(
                f"Tool request name {request!r} is already registered to "
                f"{existing_class.__name__}; cannot register "
                f"{message_class.__name__}"
            )
    if isinstance(message_class, XMLToolMessage):
        # XMLToolMessage is not compatible with OpenAI's Tools/functions API,
        # so we disable use of functions API, enable langroid-native Tools,
        # which are prompt-based.
        self.config.use_functions_api = False
        self.config.use_tools = True
    super().enable_message_handling(message_class)  # enables handling only
    tools = self._get_tool_list(message_class)
    if message_class is not None:
        if request == "":
            raise ValueError(
                f"""
                ToolMessage class {message_class} must have a non-empty
                'request' field if it is to be enabled as a tool.
                """
            )
        llm_function = message_class.llm_function_schema(defaults=include_defaults)
        self.llm_functions_map[request] = llm_function
        if force:
            self.llm_function_force = dict(name=request)
        else:
            self.llm_function_force = None

    for t in tools:
        self.llm_tools_known.add(t)

        if handle:
            self.llm_tools_handled.add(t)
            self.llm_functions_handled.add(t)

            if (
                self.enabled_handling_output_format is not None
                and self.enabled_handling_output_format.name() == t
            ):
                # `t` was designated as "enabled for handling" ONLY for
                # output_format enforcement, but we are explicitly ]
                # enabling it for handling here, so we set the variable to None.
                self.enabled_handling_output_format = None
        else:
            self.llm_tools_handled.discard(t)
            self.llm_functions_handled.discard(t)

        if use:
            tool_class = self.llm_tools_map[t]
            allow_llm_use = tool_class._allow_llm_use
            if isinstance(allow_llm_use, ModelPrivateAttr):
                allow_llm_use = allow_llm_use.default
            if allow_llm_use:
                self.llm_tools_usable.add(t)
                self.llm_functions_usable.add(t)
            else:
                logger.warning(
                    f"""
                    ToolMessage class {tool_class} does not allow LLM use,
                    because `_allow_llm_use=False` either in the Tool or a
                    parent class of this tool;
                    so not enabling LLM use for this tool!
                    If you intended an LLM to use this tool,
                    set `_allow_llm_use=True` when you define the tool.
                    """
                )
            if (
                self.enabled_use_output_format is not None
                and self.enabled_use_output_format.default_value("request") == t
            ):
                # `t` was designated as "enabled for use" ONLY for output_format
                # enforcement, but we are explicitly enabling it for use here,
                # so we set the variable to None.
                self.enabled_use_output_format = None
        else:
            self.llm_tools_usable.discard(t)
            self.llm_functions_usable.discard(t)

    self._update_tool_instructions()

set_output_format(output_type, force_tools=None, use=None, handle=None, instructions=None, is_copy=False)

Sets output_format to output_type and, if force_tools is enabled, switches to the native Langroid tools mechanism to ensure that no tool calls not of output_type are generated. By default, force_tools follows the use_tools_on_output_format parameter in the config.

If output_type is None, restores to the state prior to setting output_format.

If use, we enable use of output_type when it is a subclass of ToolMesage. Note that this primarily controls instruction generation: the model will always generate output_type regardless of whether use is set. Defaults to the use_output_format parameter in the config. Similarly, handling of output_type is controlled by handle, which defaults to the handle_output_format parameter in the config.

instructions controls whether we generate instructions specifying the output format schema. Defaults to the instructions_output_format parameter in the config.

is_copy is set when called via __getitem__. In that case, we must copy certain fields to ensure that we do not overwrite the main agent's setings.

Source code in langroid/agent/chat_agent.py

def set_output_format(
    self,
    output_type: Optional[type],
    force_tools: Optional[bool] = None,
    use: Optional[bool] = None,
    handle: Optional[bool] = None,
    instructions: Optional[bool] = None,
    is_copy: bool = False,
) -> None:
    """
    Sets `output_format` to `output_type` and, if `force_tools` is enabled,
    switches to the native Langroid tools mechanism to ensure that no tool
    calls not of `output_type` are generated. By default, `force_tools`
    follows the `use_tools_on_output_format` parameter in the config.

    If `output_type` is None, restores to the state prior to setting
    `output_format`.

    If `use`, we enable use of `output_type` when it is a subclass
    of `ToolMesage`. Note that this primarily controls instruction
    generation: the model will always generate `output_type` regardless
    of whether `use` is set. Defaults to the `use_output_format`
    parameter in the config. Similarly, handling of `output_type` is
    controlled by `handle`, which defaults to the
    `handle_output_format` parameter in the config.

    `instructions` controls whether we generate instructions specifying
    the output format schema. Defaults to the `instructions_output_format`
    parameter in the config.

    `is_copy` is set when called via `__getitem__`. In that case, we must
    copy certain fields to ensure that we do not overwrite the main agent's
    setings.
    """
    # Disable usage of an output format which was not specifically enabled
    # by `enable_message`
    if self.enabled_use_output_format is not None:
        self.disable_message_use(self.enabled_use_output_format)
        self.enabled_use_output_format = None

    # Disable handling of an output format which did not specifically have
    # handling enabled via `enable_message`
    if self.enabled_handling_output_format is not None:
        self.disable_message_handling(self.enabled_handling_output_format)
        self.enabled_handling_output_format = None

    # Reset any previous instructions
    self.output_format_instructions = ""

    if output_type is None:
        self.output_format = None
        (
            requests_for_inference,
            use_functions_api,
            use_tools,
        ) = self.saved_requests_and_tool_setings
        self.config = self.config.model_copy()
        self.enabled_requests_for_inference = requests_for_inference
        self.config.use_functions_api = use_functions_api
        self.config.use_tools = use_tools
    else:
        if force_tools is None:
            force_tools = self.config.use_tools_on_output_format

        if not any(
            (isclass(output_type) and issubclass(output_type, t))
            for t in [ToolMessage, BaseModel]
        ):
            output_type = get_pydantic_wrapper(output_type)

        if self.output_format is None and force_tools:
            self.saved_requests_and_tool_setings = (
                self._requests_and_tool_settings()
            )

        self.output_format = output_type
        if issubclass(output_type, ToolMessage):
            name = output_type.default_value("request")
            if use is None:
                use = self.config.use_output_format

            if handle is None:
                handle = self.config.handle_output_format

            if use or handle:
                is_usable = name in self.llm_tools_usable.union(
                    self.llm_functions_usable
                )
                is_handled = name in self.llm_tools_handled.union(
                    self.llm_functions_handled
                )

                if is_copy:
                    if use:
                        # We must copy `llm_tools_usable` so the base agent
                        # is unmodified
                        self.llm_tools_usable = self.llm_tools_usable.copy()
                        self.llm_functions_usable = self.llm_functions_usable.copy()
                    if handle:
                        # If handling the tool, do the same for `llm_tools_handled`
                        self.llm_tools_handled = self.llm_tools_handled.copy()
                        self.llm_functions_handled = (
                            self.llm_functions_handled.copy()
                        )
                # Enable `output_type`
                self.enable_message(
                    output_type,
                    # Do not override existing settings
                    use=use or is_usable,
                    handle=handle or is_handled,
                )

                # If the `output_type` ToilMessage was not already enabled for
                # use, this means we are ONLY enabling it for use specifically
                # for enforcing this output format, so we set the
                # `enabled_use_output_forma  to this output_type, to
                # record that it should be disabled when `output_format` is changed
                if not is_usable:
                    self.enabled_use_output_format = output_type

                # (same reasoning as for use-enabling)
                if not is_handled:
                    self.enabled_handling_output_format = output_type

            generated_tool_instructions = name in self.llm_tools_usable.union(
                self.llm_functions_usable
            )
        else:
            generated_tool_instructions = False

        if instructions is None:
            instructions = self.config.instructions_output_format
        if issubclass(output_type, BaseModel) and instructions:
            if generated_tool_instructions:
                # Already generated tool instructions as part of "enabling for use",
                # so only need to generate a reminder to use this tool.
                name = cast(ToolMessage, output_type).default_value("request")
                self.output_format_instructions = textwrap.dedent(
                    f"""
                    === OUTPUT FORMAT INSTRUCTIONS ===

                    Please provide output using the `{name}` tool/function.
                    """
                )
            else:
                if issubclass(output_type, ToolMessage):
                    output_format_schema = output_type.llm_function_schema(
                        request=True,
                        defaults=self.config.output_format_include_defaults,
                    ).parameters
                else:
                    output_format_schema = output_type.model_json_schema()

                format_schema_for_strict(output_format_schema)

                self.output_format_instructions = textwrap.dedent(
                    f"""
                    === OUTPUT FORMAT INSTRUCTIONS ===
                    Please provide output as JSON with the following schema:

                    {output_format_schema}
                    """
                )

        if force_tools:
            if issubclass(output_type, ToolMessage):
                self.enabled_requests_for_inference = {
                    output_type.default_value("request")
                }
            if self.config.use_functions_api:
                self.config = self.config.model_copy()
                self.config.use_functions_api = False
                self.config.use_tools = True

disable_message_handling(message_class=None)

Disable this agent from RESPONDING to a message_class (Tool). If message_class is None, then disable this agent from responding to ALL. Args: message_class: The ToolMessage class to disable; Optional.

Source code in langroid/agent/chat_agent.py

def disable_message_handling(
    self,
    message_class: Optional[Type[ToolMessage]] = None,
) -> None:
    """
    Disable this agent from RESPONDING to a `message_class` (Tool). If
        `message_class` is None, then disable this agent from responding to ALL.
    Args:
        message_class: The ToolMessage class to disable; Optional.
    """
    super().disable_message_handling(message_class)
    for t in self._get_tool_list(message_class):
        self.llm_tools_handled.discard(t)
        self.llm_functions_handled.discard(t)

disable_message_use(message_class)

Disable this agent from USING a message class (Tool). If message_class is None, then disable this agent from USING ALL tools. Args: message_class: The ToolMessage class to disable. If None, disable all.

Source code in langroid/agent/chat_agent.py

def disable_message_use(
    self,
    message_class: Optional[Type[ToolMessage]],
) -> None:
    """
    Disable this agent from USING a message class (Tool).
    If `message_class` is None, then disable this agent from USING ALL tools.
    Args:
        message_class: The ToolMessage class to disable.
            If None, disable all.
    """
    for t in self._get_tool_list(message_class):
        self.llm_tools_usable.discard(t)
        self.llm_functions_usable.discard(t)

    self._update_tool_instructions()

disable_message_use_except(message_class)

Disable this agent from USING ALL messages EXCEPT a message class (Tool) Args: message_class: The only ToolMessage class to allow

Source code in langroid/agent/chat_agent.py

def disable_message_use_except(self, message_class: Type[ToolMessage]) -> None:
    """
    Disable this agent from USING ALL messages EXCEPT a message class (Tool)
    Args:
        message_class: The only ToolMessage class to allow
    """
    request = message_class.model_fields["request"].default
    to_remove = [r for r in self.llm_tools_usable if r != request]
    for r in to_remove:
        self.llm_tools_usable.discard(r)
        self.llm_functions_usable.discard(r)
    self._update_tool_instructions()

get_tool_messages(msg, all_tools=False)

Extracts messages and tracks whether any errors occurred. If strict mode was enabled, disables it for the tool, else triggers strict recovery.

Source code in langroid/agent/chat_agent.py

def get_tool_messages(
    self,
    msg: str | ChatDocument | None,
    all_tools: bool = False,
) -> List[ToolMessage]:
    """
    Extracts messages and tracks whether any errors occurred. If strict mode
    was enabled, disables it for the tool, else triggers strict recovery.
    """
    self.tool_error = False
    most_recent_sent_by_llm = (
        len(self.message_history) > 0
        and self.message_history[-1].role == Role.ASSISTANT
    )
    was_llm = most_recent_sent_by_llm or (
        isinstance(msg, ChatDocument) and msg.metadata.sender == Entity.LLM
    )
    try:
        tools = super().get_tool_messages(msg, all_tools)
    except ValidationError as ve:
        # Check if tool class was attached to the exception
        if hasattr(ve, "tool_class") and ve.tool_class:
            tool_class = ve.tool_class  # type: ignore
            if issubclass(tool_class, ToolMessage):
                was_strict = (
                    self.config.use_functions_api
                    and self.config.use_tools_api
                    and self._strict_mode_for_tool(tool_class)
                )
                # If the result of strict output for a tool using the
                # OpenAI tools API fails to parse, we infer that the
                # schema edits necessary for compatibility prevented
                # adherence to the underlying `ToolMessage` schema and
                # disable strict output for the tool
                if was_strict:
                    name = tool_class.default_value("request")
                    self.disable_strict_tools_set.add(name)
                    logging.warning(
                        f"""
                        Validation error occured with strict tool format.
                        Disabling strict mode for the {name} tool.
                        """
                    )
                else:
                    # We will trigger the strict recovery mechanism to force
                    # the LLM to correct its output, allowing us to parse
                    if isinstance(msg, ChatDocument):
                        self.tool_error = msg.metadata.sender == Entity.LLM
                    else:
                        self.tool_error = most_recent_sent_by_llm

        if was_llm:
            raise ve
        else:
            self.tool_error = False
            return []

    if not was_llm:
        self.tool_error = False

    return tools

truncate_message(idx, tokens=5, warning='...[Contents truncated!]', inplace=True)

Truncate message at idx in msg history to tokens tokens.

If inplace is True, the message is truncated in place, else it LEAVES the original message INTACT and returns a new message

Source code in langroid/agent/chat_agent.py

def truncate_message(
    self,
    idx: int,
    tokens: int = 5,
    warning: str = "...[Contents truncated!]",
    inplace: bool = True,
) -> LLMMessage:
    """
    Truncate message at idx in msg history to `tokens` tokens.

    If inplace is True, the message is truncated in place, else
    it LEAVES the original message INTACT and returns a new message
    """
    if inplace:
        llm_msg = self.message_history[idx]
    else:
        llm_msg = copy.deepcopy(self.message_history[idx])
    if llm_msg.content is None or (
        llm_msg.content == "" and _is_identified_result(llm_msg)
    ):
        return llm_msg
    orig_content = llm_msg.content
    new_content = (
        self.parser.truncate_tokens(orig_content, tokens)
        if self.parser is not None
        else orig_content[: tokens * 4]  # approx truncation
    )
    llm_msg.content = new_content + "\n" + warning
    return llm_msg

llm_response(message=None)

Respond to a single user message, appended to the message history, in "chat" mode Args: message (str|ChatDocument): message or ChatDocument object to respond to. If None, use the self.task_messages Returns: LLM response as a ChatDocument object

Source code in langroid/agent/chat_agent.py

def llm_response(
    self, message: Optional[str | ChatDocument] = None
) -> Optional[ChatDocument]:
    """
    Respond to a single user message, appended to the message history,
    in "chat" mode
    Args:
        message (str|ChatDocument): message or ChatDocument object to respond to.
            If None, use the self.task_messages
    Returns:
        LLM response as a ChatDocument object
    """
    if self.llm is None:
        return None

    # If enabled and a tool error occurred, we recover by generating the tool in
    # strict json mode
    if (
        self.tool_error
        and self.output_format is None
        and self._json_schema_available()
        and self.config.strict_recovery
    ):
        self.tool_error = False
        AnyTool = self._get_any_tool_message()
        if AnyTool is None:
            return None
        self.set_output_format(
            AnyTool,
            force_tools=True,
            use=True,
            handle=True,
            instructions=True,
        )
        recovery_message = self._strict_recovery_instructions(AnyTool)
        augmented_message = message
        if augmented_message is None:
            augmented_message = recovery_message
        elif isinstance(augmented_message, str):
            augmented_message = augmented_message + recovery_message
        else:
            augmented_message.content = augmented_message.content + recovery_message

        # only use the augmented message for this one response...
        result = self.llm_response(augmented_message)
        # ... restore the original user message so that the AnyTool recover
        # instructions don't persist in the message history
        # (this can cause the LLM to use the AnyTool directly as a tool)
        if message is None:
            self.delete_last_message(role=Role.USER)
        else:
            msg = message if isinstance(message, str) else message.content
            self.update_last_message(msg, role=Role.USER)
        return result

    hist, output_len = self._prep_llm_messages(message)
    if len(hist) == 0:
        return None
    tool_choice = (
        "auto"
        if isinstance(message, str)
        else (message.oai_tool_choice if message is not None else "auto")
    )
    with StreamingIfAllowed(self.llm, self.llm.get_stream()):
        try:
            response = self.llm_response_messages(hist, output_len, tool_choice)
        except openai.BadRequestError as e:
            if self.any_strict:
                self.disable_strict = True
                self.set_output_format(None)
                logging.warning(
                    f"""
                    OpenAI BadRequestError raised with strict mode enabled.
                    Message: {e.message}
                    Disabling strict mode and retrying.
                    """
                )
                return self.llm_response(message)
            else:
                raise e
    self.message_history.extend(ChatDocument.to_LLMMessage(response))
    response.metadata.msg_idx = len(self.message_history) - 1
    response.metadata.agent_id = self.id
    if isinstance(message, ChatDocument):
        self._reduce_raw_tool_results(message)
    # Preserve trail of tool_ids for OpenAI Assistant fn-calls
    response.metadata.tool_ids = (
        []
        if isinstance(message, str)
        else message.metadata.tool_ids if message is not None else []
    )

    return response

llm_response_async(message=None) async

Async version of llm_response. See there for details.

Source code in langroid/agent/chat_agent.py

async def llm_response_async(
    self, message: Optional[str | ChatDocument] = None
) -> Optional[ChatDocument]:
    """
    Async version of `llm_response`. See there for details.
    """
    if self.llm is None:
        return None

    # If enabled and a tool error occurred, we recover by generating the tool in
    # strict json mode
    if (
        self.tool_error
        and self.output_format is None
        and self._json_schema_available()
        and self.config.strict_recovery
    ):
        self.tool_error = False
        AnyTool = self._get_any_tool_message()
        self.set_output_format(
            AnyTool,
            force_tools=True,
            use=True,
            handle=True,
            instructions=True,
        )
        recovery_message = self._strict_recovery_instructions(AnyTool)
        augmented_message = message
        if augmented_message is None:
            augmented_message = recovery_message
        elif isinstance(augmented_message, str):
            augmented_message = augmented_message + recovery_message
        else:
            augmented_message.content = augmented_message.content + recovery_message

        # only use the augmented message for this one response...
        result = self.llm_response(augmented_message)
        # ... restore the original user message so that the AnyTool recover
        # instructions don't persist in the message history
        # (this can cause the LLM to use the AnyTool directly as a tool)
        if message is None:
            self.delete_last_message(role=Role.USER)
        else:
            msg = message if isinstance(message, str) else message.content
            self.update_last_message(msg, role=Role.USER)
        return result

    hist, output_len = self._prep_llm_messages(message)
    if len(hist) == 0:
        return None
    tool_choice = (
        "auto"
        if isinstance(message, str)
        else (message.oai_tool_choice if message is not None else "auto")
    )
    with StreamingIfAllowed(self.llm, self.llm.get_stream()):
        try:
            response = await self.llm_response_messages_async(
                hist, output_len, tool_choice
            )
        except openai.BadRequestError as e:
            if self.any_strict:
                self.disable_strict = True
                self.set_output_format(None)
                logging.warning(
                    f"""
                    OpenAI BadRequestError raised with strict mode enabled.
                    Message: {e.message}
                    Disabling strict mode and retrying.
                    """
                )
                return await self.llm_response_async(message)
            else:
                raise e
    self.message_history.extend(ChatDocument.to_LLMMessage(response))
    response.metadata.msg_idx = len(self.message_history) - 1
    response.metadata.agent_id = self.id
    if isinstance(message, ChatDocument):
        self._reduce_raw_tool_results(message)
    # Preserve trail of tool_ids for OpenAI Assistant fn-calls
    response.metadata.tool_ids = (
        []
        if isinstance(message, str)
        else message.metadata.tool_ids if message is not None else []
    )

    return response

init_message_history()

Initialize the message history with the system message and user message

Source code in langroid/agent/chat_agent.py

def init_message_history(self) -> None:
    """
    Initialize the message history with the system message and user message
    """
    self.message_history = [self._create_system_and_tools_message()]
    if self.user_message:
        self.message_history.append(
            LLMMessage(role=Role.USER, content=self.user_message)
        )

llm_response_messages(messages, output_len=None, tool_choice='auto')

Respond to a series of messages, e.g. with OpenAI ChatCompletion Args: messages: seq of messages (with role, content fields) sent to LLM output_len: max number of tokens expected in response. If None, use the LLM's default model_max_output_tokens. Returns: Document (i.e. with fields "content", "metadata")

Source code in langroid/agent/chat_agent.py

def llm_response_messages(
    self,
    messages: List[LLMMessage],
    output_len: Optional[int] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
) -> ChatDocument:
    """
    Respond to a series of messages, e.g. with OpenAI ChatCompletion
    Args:
        messages: seq of messages (with role, content fields) sent to LLM
        output_len: max number of tokens expected in response.
                If None, use the LLM's default model_max_output_tokens.
    Returns:
        Document (i.e. with fields "content", "metadata")
    """
    assert self.config.llm is not None and self.llm is not None
    output_len = output_len or self.config.llm.model_max_output_tokens
    streamer = noop_fn
    if self.llm.get_stream():
        streamer = self.callbacks.start_llm_stream()
    self.llm.config.streamer = streamer
    with ExitStack() as stack:  # for conditionally using rich spinner
        if not self.llm.get_stream() and not settings.quiet:
            # show rich spinner only if not streaming!
            # (Why? b/c the intent of showing a spinner is to "show progress",
            # and we don't need to do that when streaming, since
            # streaming output already shows progress.)
            cm = status(
                "LLM responding to messages...",
                log_if_quiet=False,
            )
            stack.enter_context(cm)
        if self.llm.get_stream() and not settings.quiet:
            console.print(f"[green]{self.indent}", end="")
        functions, fun_call, tools, force_tool, output_format = (
            self._function_args()
        )
        assert self.llm is not None
        response = self.llm.chat(
            messages,
            output_len,
            tools=tools,
            tool_choice=force_tool or tool_choice,
            functions=functions,
            function_call=fun_call,
            response_format=output_format,
        )
    if self.llm.get_stream():
        # Create temp ChatDocument for tool check, then clean up to avoid
        # polluting ObjectRegistry (see PR #939 discussion)
        temp_doc = ChatDocument.from_LLMResponse(
            response,
            displayed=True,
            recognize_recipient_in_content=self.config.recognize_recipient_in_content,
        )
        self._call_callback_with_reasoning(
            "finish_llm_stream",
            reasoning=response.reasoning,
            content=response.message or "",
            tools_content=response.tools_content(),
            is_tool=self.has_tool_message_attempt(temp_doc),
        )
        ObjectRegistry.remove(temp_doc.id())
    self.llm.config.streamer = noop_fn
    if response.cached:
        self.callbacks.cancel_llm_stream()
    self._render_llm_response(response)
    self.update_token_usage(
        response,  # .usage attrib is updated!
        messages,
        self.llm.get_stream(),
        chat=True,
        print_response_stats=self.config.show_stats and not settings.quiet,
    )
    chat_doc = ChatDocument.from_LLMResponse(
        response,
        displayed=True,
        recognize_recipient_in_content=self.config.recognize_recipient_in_content,
    )
    self.oai_tool_calls = response.oai_tool_calls or []
    self.oai_tool_id2call.update(
        {t.id: t for t in self.oai_tool_calls if t.id is not None}
    )

    # If using strict output format, parse the output JSON
    self._load_output_format(chat_doc)

    return chat_doc

llm_response_messages_async(messages, output_len=None, tool_choice='auto') async

Async version of llm_response_messages. See there for details.

Source code in langroid/agent/chat_agent.py

async def llm_response_messages_async(
    self,
    messages: List[LLMMessage],
    output_len: Optional[int] = None,
    tool_choice: ToolChoiceTypes | Dict[str, str | Dict[str, str]] = "auto",
) -> ChatDocument:
    """
    Async version of `llm_response_messages`. See there for details.
    """
    assert self.config.llm is not None and self.llm is not None
    output_len = output_len or self.config.llm.model_max_output_tokens
    functions, fun_call, tools, force_tool, output_format = self._function_args()
    assert self.llm is not None

    streamer_async = async_noop_fn
    if self.llm.get_stream():
        streamer_async = await self.callbacks.start_llm_stream_async()
    self.llm.config.streamer_async = streamer_async

    response = await self.llm.achat(
        messages,
        output_len,
        tools=tools,
        tool_choice=force_tool or tool_choice,
        functions=functions,
        function_call=fun_call,
        response_format=output_format,
    )
    if self.llm.get_stream():
        # Create temp ChatDocument for tool check, then clean up to avoid
        # polluting ObjectRegistry (see PR #939 discussion)
        temp_doc = ChatDocument.from_LLMResponse(
            response,
            displayed=True,
            recognize_recipient_in_content=self.config.recognize_recipient_in_content,
        )
        self._call_callback_with_reasoning(
            "finish_llm_stream",
            reasoning=response.reasoning,
            content=response.message or "",
            tools_content=response.tools_content(),
            is_tool=self.has_tool_message_attempt(temp_doc),
        )
        ObjectRegistry.remove(temp_doc.id())
    self.llm.config.streamer_async = async_noop_fn
    if response.cached:
        self.callbacks.cancel_llm_stream()
    self._render_llm_response(response)
    self.update_token_usage(
        response,  # .usage attrib is updated!
        messages,
        self.llm.get_stream(),
        chat=True,
        print_response_stats=self.config.show_stats and not settings.quiet,
    )
    chat_doc = ChatDocument.from_LLMResponse(
        response,
        displayed=True,
        recognize_recipient_in_content=self.config.recognize_recipient_in_content,
    )
    self.oai_tool_calls = response.oai_tool_calls or []
    self.oai_tool_id2call.update(
        {t.id: t for t in self.oai_tool_calls if t.id is not None}
    )

    # If using strict output format, parse the output JSON
    self._load_output_format(chat_doc)

    return chat_doc

llm_response_forget(message=None)

LLM Response to single message, and restore message_history. In effect a "one-off" message & response that leaves agent message history state intact.

Parameters:

Name	Type	Description	Default
message	str | ChatDocument	message to respond to.	None

Returns:

Type	Description
ChatDocument	A Document object with the response.

Source code in langroid/agent/chat_agent.py

def llm_response_forget(
    self, message: Optional[str | ChatDocument] = None
) -> ChatDocument:
    """
    LLM Response to single message, and restore message_history.
    In effect a "one-off" message & response that leaves agent
    message history state intact.

    Args:
        message (str|ChatDocument): message to respond to.

    Returns:
        A Document object with the response.

    """
    # explicitly call THIS class's respond method,
    # not a derived class's (or else there would be infinite recursion!)
    n_msgs = len(self.message_history)
    with StreamingIfAllowed(self.llm, self.llm.get_stream()):  # type: ignore
        response = cast(ChatDocument, ChatAgent.llm_response(self, message))
    # If there is a response, then we will have two additional
    # messages in the message history, i.e. the user message and the
    # assistant response. We want to (carefully) remove these two messages.
    if len(self.message_history) > n_msgs:
        msg = self.message_history.pop()
        self._drop_msg_update_tool_calls(msg)

    if len(self.message_history) > n_msgs:
        msg = self.message_history.pop()
        self._drop_msg_update_tool_calls(msg)

    # If using strict output format, parse the output JSON
    self._load_output_format(response)

    return response

llm_response_forget_async(message=None) async

Async version of llm_response_forget. See there for details.

Source code in langroid/agent/chat_agent.py

async def llm_response_forget_async(
    self, message: Optional[str | ChatDocument] = None
) -> ChatDocument:
    """
    Async version of `llm_response_forget`. See there for details.
    """
    # explicitly call THIS class's respond method,
    # not a derived class's (or else there would be infinite recursion!)
    n_msgs = len(self.message_history)
    with StreamingIfAllowed(self.llm, self.llm.get_stream()):  # type: ignore
        response = cast(
            ChatDocument, await ChatAgent.llm_response_async(self, message)
        )
    # If there is a response, then we will have two additional
    # messages in the message history, i.e. the user message and the
    # assistant response. We want to (carefully) remove these two messages.
    if len(self.message_history) > n_msgs:
        msg = self.message_history.pop()
        self._drop_msg_update_tool_calls(msg)

    if len(self.message_history) > n_msgs:
        msg = self.message_history.pop()
        self._drop_msg_update_tool_calls(msg)
    return response

chat_num_tokens(messages=None)

Total number of tokens in the message history so far.

Parameters:

Name	Type	Description	Default
messages	Optional[List[LLMMessage]]	if provided, compute the number of tokens in this list of messages, rather than the current message history.	None

Returns: int: number of tokens in message history

Source code in langroid/agent/chat_agent.py

def chat_num_tokens(self, messages: Optional[List[LLMMessage]] = None) -> int:
    """
    Total number of tokens in the message history so far.

    Args:
        messages: if provided, compute the number of tokens in this list of
            messages, rather than the current message history.
    Returns:
        int: number of tokens in message history
    """
    if self.parser is None:
        raise ValueError(
            "ChatAgent.parser is None. "
            "You must set ChatAgent.parser "
            "before calling chat_num_tokens()."
        )
    hist = messages if messages is not None else self.message_history
    return sum([self._message_num_tokens(m) for m in hist])

message_history_str(i=None)

Return a string representation of the message history Args: i: if provided, return only the i-th message when i is postive, or last k messages when i = -k. Returns:

Source code in langroid/agent/chat_agent.py

def message_history_str(self, i: Optional[int] = None) -> str:
    """
    Return a string representation of the message history
    Args:
        i: if provided, return only the i-th message when i is postive,
            or last k messages when i = -k.
    Returns:
    """
    if i is None:
        return "\n".join([str(m) for m in self.message_history])
    elif i > 0:
        return str(self.message_history[i])
    else:
        return "\n".join([str(m) for m in self.message_history[i:]])