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
    if self.function_call is not None:
        tool_type = "FUNC"
        tool = self.function_call.name
    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,
        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,
        ),
    )

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_name = None
    sender_role = Role.USER
    fun_call = None
    oai_tool_calls = None
    tool_id = ""  # for OpenAI Assistant
    chat_document_id: str = ""
    if isinstance(message, ChatDocument):
        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
        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,
                        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,
                        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,
                        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
    else:
        # LLM can only respond to text content, so extract it
        content = message

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