chat_document

langroid/agent/chat_document.py

StatusCode

Bases: str, Enum

Codes meant to be returned by task.run(). Some are not used yet.

ChatDocument(**data)

Bases: Document

Represents a message in a conversation among agents. All responders of an agent have signature ChatDocument -> ChatDocument (modulo None, str, etc), and so does the Task.run() method.

Attributes:

Name	Type	Description
oai_tool_calls	Optional[List[OpenAIToolCall]]	Tool-calls from an OpenAI-compatible API
oai_tool_id2results	Optional[OrderedDict[str, str]]	Results of tool-calls from OpenAI (dict is a map of tool_id -> result)
oai_tool_choice	ToolChoiceTypes | Dict[str, Dict[str, str] | str]	ToolChoiceTypes | Dict[str, str]: Param controlling how the LLM should choose tool-use in its response (auto, none, required, or a specific tool)
function_call	Optional[LLMFunctionCall]	Function-call from an OpenAI-compatible API (deprecated by OpenAI, in favor of tool-calls)
tool_messages	List[ToolMessage]	Langroid ToolMessages extracted from - content field (via JSON parsing), - oai_tool_calls, or - function_call
metadata	ChatDocMetaData	Metadata for the message, e.g. sender, recipient.
attachment	None | ChatDocAttachment	Any additional data attached.

Source code in langroid/agent/chat_document.py

def __init__(self, **data: Any):
    super().__init__(**data)
    ObjectRegistry.register_object(self)

delete_id(id) staticmethod

Remove ChatDocument with given id from ObjectRegistry, and all its descendants.

Source code in langroid/agent/chat_document.py

@staticmethod
def delete_id(id: str) -> None:
    """Remove ChatDocument with given id from ObjectRegistry,
    and all its descendants.
    """
    chat_doc = ChatDocument.from_id(id)
    # first delete all descendants
    while chat_doc is not None:
        next_chat_doc = chat_doc.child
        ObjectRegistry.remove(chat_doc.id())
        chat_doc = next_chat_doc

get_tool_names()

Get names of attempted tool usages (JSON or non-JSON) in the content of the message. Returns: List[str]: list of attempted tool names (We say "attempted" since we ONLY look at the request component of the tool-call representation, and we're not fully parsing it into the corresponding tool message class)

Source code in langroid/agent/chat_document.py

def get_tool_names(self) -> List[str]:
    """
    Get names of attempted tool usages (JSON or non-JSON) in the content
        of the message.
    Returns:
        List[str]: list of *attempted* tool names
        (We say "attempted" since we ONLY look at the `request` component of the
        tool-call representation, and we're not fully parsing it into the
        corresponding tool message class)

    """
    tool_candidates = XMLToolMessage.find_candidates(self.content)
    if len(tool_candidates) == 0:
        tool_candidates = extract_top_level_json(self.content)
        if len(tool_candidates) == 0:
            return []
        tools = [json.loads(tc).get("request") for tc in tool_candidates]
    else:
        tool_dicts = [
            XMLToolMessage.extract_field_values(tc) for tc in tool_candidates
        ]
        tools = [td.get("request") for td in tool_dicts if td is not None]
    return [str(tool) for tool in tools if tool is not None]

log_fields()

Fields for logging in csv/tsv logger Returns: List[str]: list of fields

Source code in langroid/agent/chat_document.py

def log_fields(self) -> ChatDocLoggerFields:
    """
    Fields for logging in csv/tsv logger
    Returns:
        List[str]: list of fields
    """
    tool_type = ""  # FUNC or TOOL
    tool = ""  # tool name or function name

    # Skip tool detection for system messages - they contain tool instructions,
    # not actual tool calls
    if self.metadata.sender != Entity.SYSTEM:
        oai_tools = (
            []
            if self.oai_tool_calls is None
            else [t for t in self.oai_tool_calls if t.function is not None]
        )
        if self.function_call is not None:
            tool_type = "FUNC"
            tool = self.function_call.name
        elif len(oai_tools) > 0:
            tool_type = "OAI_TOOL"
            tool = ",".join(t.function.name for t in oai_tools)  # type: ignore
        else:
            try:
                json_tools = self.get_tool_names()
            except Exception:
                json_tools = []
            if json_tools != []:
                tool_type = "TOOL"
                tool = json_tools[0]
    recipient = self.metadata.recipient
    content = self.content
    sender_entity = self.metadata.sender
    sender_name = self.metadata.sender_name
    if tool_type == "FUNC":
        content += str(self.function_call)
    return ChatDocLoggerFields(
        sender_entity=sender_entity,
        sender_name=sender_name,
        recipient=recipient,
        block=self.metadata.block,
        tool_type=tool_type,
        tool=tool,
        content=content,
    )

pop_tool_ids()

Pop the last tool_id from the stack of tool_ids.

Source code in langroid/agent/chat_document.py

def pop_tool_ids(self) -> None:
    """
    Pop the last tool_id from the stack of tool_ids.
    """
    if len(self.metadata.tool_ids) > 0:
        self.metadata.tool_ids.pop()

from_LLMResponse(response, displayed=False) staticmethod

Convert LLMResponse to ChatDocument. Args: response (LLMResponse): LLMResponse to convert. displayed (bool): Whether this response was displayed to the user. Returns: ChatDocument: ChatDocument representation of this LLMResponse.

Source code in langroid/agent/chat_document.py

@staticmethod
def from_LLMResponse(
    response: LLMResponse,
    displayed: bool = False,
) -> "ChatDocument":
    """
    Convert LLMResponse to ChatDocument.
    Args:
        response (LLMResponse): LLMResponse to convert.
        displayed (bool): Whether this response was displayed to the user.
    Returns:
        ChatDocument: ChatDocument representation of this LLMResponse.
    """
    recipient, message = response.get_recipient_and_message()
    message = message.strip()
    if message in ["''", '""']:
        message = ""
    if response.function_call is not None:
        ChatDocument._clean_fn_call(response.function_call)
    if response.oai_tool_calls is not None:
        # there must be at least one if it's not None
        for oai_tc in response.oai_tool_calls:
            ChatDocument._clean_fn_call(oai_tc.function)
    return ChatDocument(
        content=message,
        reasoning=response.reasoning,
        content_any=message,
        oai_tool_calls=response.oai_tool_calls,
        function_call=response.function_call,
        metadata=ChatDocMetaData(
            source=Entity.LLM,
            sender=Entity.LLM,
            usage=response.usage,
            displayed=displayed,
            cached=response.cached,
            recipient=recipient,
        ),
    )

from_LLMMessage(message, sender_name='', recipient='') staticmethod

Convert LLMMessage to ChatDocument.

Parameters:

Name	Type	Description	Default
message	LLMMessage	LLMMessage to convert.	required
sender_name	str	Name of the sender. Defaults to "".	''
recipient	str	Name of the recipient. Defaults to "".	''

Returns:

Name	Type	Description
ChatDocument	'ChatDocument'	ChatDocument representation of this LLMMessage.

Source code in langroid/agent/chat_document.py

@staticmethod
def from_LLMMessage(
    message: LLMMessage,
    sender_name: str = "",
    recipient: str = "",
) -> "ChatDocument":
    """
    Convert LLMMessage to ChatDocument.

    Args:
        message (LLMMessage): LLMMessage to convert.
        sender_name (str): Name of the sender. Defaults to "".
        recipient (str): Name of the recipient. Defaults to "".

    Returns:
        ChatDocument: ChatDocument representation of this LLMMessage.
    """
    # Map LLMMessage Role to ChatDocument Entity
    role_to_entity = {
        Role.USER: Entity.USER,
        Role.SYSTEM: Entity.SYSTEM,
        Role.ASSISTANT: Entity.LLM,
        Role.FUNCTION: Entity.LLM,
        Role.TOOL: Entity.LLM,
    }

    sender_entity = role_to_entity.get(message.role, Entity.USER)

    return ChatDocument(
        content=message.content or "",
        content_any=message.content,
        files=message.files,
        function_call=message.function_call,
        oai_tool_calls=message.tool_calls,
        metadata=ChatDocMetaData(
            source=sender_entity,
            sender=sender_entity,
            sender_name=sender_name,
            recipient=recipient,
            oai_tool_id=message.tool_call_id,
            tool_ids=[message.tool_id] if message.tool_id else [],
        ),
    )

to_LLMMessage(message, oai_tools=None) staticmethod

Convert to list of LLMMessage, to incorporate into msg-history sent to LLM API. Usually there will be just a single LLMMessage, but when the ChatDocument contains results from multiple OpenAI tool-calls, we would have a sequence LLMMessages, one per tool-call result.

Parameters:

Name	Type	Description	Default
message	str | ChatDocument	Message to convert.	required
oai_tools	Optional[List[OpenAIToolCall]]	Tool-calls currently awaiting response, from the ChatAgent's latest message.	None

Returns: List[LLMMessage]: list of LLMMessages corresponding to this ChatDocument.

Source code in langroid/agent/chat_document.py

@staticmethod
def to_LLMMessage(
    message: Union[str, "ChatDocument"],
    oai_tools: Optional[List[OpenAIToolCall]] = None,
) -> List[LLMMessage]:
    """
    Convert to list of LLMMessage, to incorporate into msg-history sent to LLM API.
    Usually there will be just a single LLMMessage, but when the ChatDocument
    contains results from multiple OpenAI tool-calls, we would have a sequence
    LLMMessages, one per tool-call result.

    Args:
        message (str|ChatDocument): Message to convert.
        oai_tools (Optional[List[OpenAIToolCall]]): Tool-calls currently awaiting
            response, from the ChatAgent's latest message.
    Returns:
        List[LLMMessage]: list of LLMMessages corresponding to this ChatDocument.
    """

    sender_role = Role.USER
    if isinstance(message, str):
        message = ChatDocument.from_str(message)
    content = message.content or to_string(message.content_any) or ""
    fun_call = message.function_call
    oai_tool_calls = message.oai_tool_calls
    if message.metadata.sender == Entity.USER and fun_call is not None:
        # This may happen when a (parent agent's) LLM generates a
        # a Function-call, and it ends up being sent to the current task's
        # LLM (possibly because the function-call is mis-named or has other
        # issues and couldn't be handled by handler methods).
        # But a function-call can only be generated by an entity with
        # Role.ASSISTANT, so we instead put the content of the function-call
        # in the content of the message.
        content += " " + str(fun_call)
        fun_call = None
    if message.metadata.sender == Entity.USER and oai_tool_calls is not None:
        # same reasoning as for function-call above
        content += " " + "\n\n".join(str(tc) for tc in oai_tool_calls)
        oai_tool_calls = None
    # some LLM APIs (e.g. gemini) don't like empty msg
    content = content or " "
    sender_name = message.metadata.sender_name
    tool_ids = message.metadata.tool_ids
    tool_id = tool_ids[-1] if len(tool_ids) > 0 else ""
    chat_document_id = message.id()
    if message.metadata.sender == Entity.SYSTEM:
        sender_role = Role.SYSTEM
    if (
        message.metadata.parent is not None
        and message.metadata.parent.function_call is not None
    ):
        # This is a response to a function call, so set the role to FUNCTION.
        sender_role = Role.FUNCTION
        sender_name = message.metadata.parent.function_call.name
    elif oai_tools is not None and len(oai_tools) > 0:
        pending_tool_ids = [tc.id for tc in oai_tools]
        # The ChatAgent has pending OpenAI tool-call(s),
        # so the current ChatDocument contains
        # results for some/all/none of them.

        if len(oai_tools) == 1:
            # Case 1:
            # There was exactly 1 pending tool-call, and in this case
            # the result would be a plain string in `content`
            return [
                LLMMessage(
                    role=Role.TOOL,
                    tool_call_id=oai_tools[0].id,
                    content=content,
                    files=message.files,
                    chat_document_id=chat_document_id,
                )
            ]

        elif (
            message.metadata.oai_tool_id is not None
            and message.metadata.oai_tool_id in pending_tool_ids
        ):
            # Case 2:
            # ChatDocument.content has result of a single tool-call
            return [
                LLMMessage(
                    role=Role.TOOL,
                    tool_call_id=message.metadata.oai_tool_id,
                    content=content,
                    files=message.files,
                    chat_document_id=chat_document_id,
                )
            ]
        elif message.oai_tool_id2result is not None:
            # Case 2:
            # There were > 1 tool-calls awaiting response,
            assert (
                len(message.oai_tool_id2result) > 1
            ), "oai_tool_id2result must have more than 1 item."
            return [
                LLMMessage(
                    role=Role.TOOL,
                    tool_call_id=tool_id,
                    content=result or " ",
                    files=message.files,
                    chat_document_id=chat_document_id,
                )
                for tool_id, result in message.oai_tool_id2result.items()
            ]
    elif message.metadata.sender == Entity.LLM:
        sender_role = Role.ASSISTANT

    return [
        LLMMessage(
            role=sender_role,
            tool_id=tool_id,  # for OpenAI Assistant
            content=content,
            files=message.files,
            function_call=fun_call,
            tool_calls=oai_tool_calls,
            name=sender_name,
            chat_document_id=chat_document_id,
        )
    ]