chat_agent

langroid/agent/chat_agent.py

ChatAgentConfig

Bases: AgentConfig

Configuration for ChatAgent Attributes: system_message: 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: user message to include in message sequence. Used only if task is not specified in the constructor. use_tools: whether to use our own ToolMessages mechanism use_functions_api: whether to use functions/tools native to the LLM API (e.g. OpenAI's function_call or tool_call mechanism) use_tools_api: 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: whether to enable strict schema recovery when there is a tool-generation error. enable_orchestration_tool_handling: whether to enable handling of orchestration tools, e.g. ForwardTool, DoneTool, PassTool, etc. output_format: 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: When output_format is a ToolMessage T, controls whether T is "enabled for handling". use_output_format: 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: Controls whether we generate instructions for output_format in the system message. use_tools_on_output_format: 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: Whether to include fields with default arguments in the output schema

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._fn_call_available())
    self.message_history: List[LLMMessage] = []
    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
        # 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()

    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: List[LLMMessage] 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: set[str] 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)
    config_copy = copy.deepcopy(self.config)
    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_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
    # Caution - we are copying the vector-db, maybe we don't always want this?
    new_agent.vecdb = self.vecdb
    new_agent.id = ObjectRegistry.new_id()
    if self.config.add_to_registry:
        ObjectRegistry.register_object(new_agent)
    return new_agent

clear_history(start=-2)

Clear the message history, starting at the index start

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

Source code in langroid/agent/chat_agent.py

def clear_history(self, start: int = -2) -> None:
    """
    Clear the message history, starting at the index `start`

    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)
    """
    if start < 0:
        n = len(self.message_history)
        start = max(0, n + start)
    dropped = self.message_history[start:]
    # 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)
    self.message_history = self.message_history[:start]

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),
        ]
    )

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 "You can ask questions in natural language."
    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 (
            hasattr(msg_cls, "instructions")
            and inspect.ismethod(msg_cls.instructions)
            and msg_cls.default_value("request") in self.llm_tools_usable
        ):
            # 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
            class_instructions = msg_cls.instructions()
            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]

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

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
    if require_recipient and message_class is not None:
        message_class = message_class.require_recipient()
    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:
        request = message_class.default_value("request")
        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]
            if tool_class._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)

    # Set tool instructions and JSON format instructions
    if self.config.use_tools:
        self.system_tool_format_instructions = self.tool_format_rules()
    self.system_tool_instructions = self.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.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 = copy.copy(self.llm_tools_usable)
                        self.llm_functions_usable = copy.copy(
                            self.llm_functions_usable
                        )
                    if handle:
                        # If handling the tool, do the same for `llm_tools_handled`
                        self.llm_tools_handled = copy.copy(self.llm_tools_handled)
                        self.llm_functions_handled = copy.copy(
                            self.llm_functions_handled
                        )
                # 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.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.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)

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

get_tool_messages(msg, all_tools=False)

Extracts messages and tracks whether any errors occured. 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 occured. If strict mode
    was enabled, disables it for the tool, else triggers strict recovery.
    """
    self.tool_error = False
    try:
        tools = super().get_tool_messages(msg, all_tools)
    except ValidationError as ve:
        tool_class = ve.model
        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
                self.tool_error = True

        raise ve

    return tools

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

Truncate message at idx in msg history to tokens tokens

Source code in langroid/agent/chat_agent.py

def truncate_message(
    self,
    idx: int,
    tokens: int = 5,
    warning: str = "...[Contents truncated!]",
) -> LLMMessage:
    """Truncate message at idx in msg history to `tokens` tokens"""
    llm_msg = self.message_history[idx]
    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
    ):
        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)

        if message is None:
            message = recovery_message
        elif isinstance(message, str):
            message = message + recovery_message
        else:
            message.content = message.content + recovery_message

        return self.llm_response(message)

    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
    ):
        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)

        if message is None:
            message = recovery_message
        elif isinstance(message, str):
            message = message + recovery_message
        else:
            message.content = message.content + recovery_message

        return self.llm_response(message)

    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 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 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.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():
        self.callbacks.finish_llm_stream(
            content=str(response),
            is_tool=self.has_tool_message_attempt(
                ChatDocument.from_LLMResponse(response, displayed=True),
            ),
        )
    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)
    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.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():
        self.callbacks.finish_llm_stream(
            content=str(response),
            is_tool=self.has_tool_message_attempt(
                ChatDocument.from_LLMResponse(response, displayed=True),
            ),
        )
    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)
    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)

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	user message	required

Returns:

Type	Description
ChatDocument	A Document object with the response.

Source code in langroid/agent/chat_agent.py

def llm_response_forget(self, message: str) -> 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): user message

    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) 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: str) -> 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.parser.num_tokens(m.content) 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:]])