Transcript

A Transcript object represents a sequence of chat messages (user, assistant, system, tool) from the perspective of a single agent. See here for more details on the chat message schemas.

docent.data_models.transcript

TranscriptGroup

Bases: BaseModel

Represents a group of transcripts that are logically related.

A transcript group can contain multiple transcripts and can have a hierarchical structure with parent groups. This is useful for organizing transcripts into logical units like experiments, tasks, or sessions.

Attributes:

Name	Type	Description
id	str	Unique identifier for the transcript group, auto-generated by default.
name	str | None	Optional human-readable name for the transcript group.
description	str | None	Optional description of the transcript group.
collection_id	str | None	ID of the collection this transcript group belongs to.
agent_run_id	str	ID of the agent run this transcript group belongs to.
parent_transcript_group_id	str | None	Optional ID of the parent transcript group.
metadata	dict[str, Any]	Additional structured metadata about the transcript group.

Source code in docent/data_models/transcript.py

class TranscriptGroup(BaseModel):
    """Represents a group of transcripts that are logically related.

    A transcript group can contain multiple transcripts and can have a hierarchical
    structure with parent groups. This is useful for organizing transcripts into
    logical units like experiments, tasks, or sessions.

    Attributes:
        id: Unique identifier for the transcript group, auto-generated by default.
        name: Optional human-readable name for the transcript group.
        description: Optional description of the transcript group.
        collection_id: ID of the collection this transcript group belongs to.
        agent_run_id: ID of the agent run this transcript group belongs to.
        parent_transcript_group_id: Optional ID of the parent transcript group.
        metadata: Additional structured metadata about the transcript group.
    """

    id: str = Field(default_factory=lambda: str(uuid4()))
    name: str | None = None
    description: str | None = None
    agent_run_id: str
    parent_transcript_group_id: str | None = None
    created_at: datetime | None = None
    metadata: dict[str, Any] = Field(default_factory=dict)

    @field_validator("metadata", mode="before")
    @classmethod
    def _validate_metadata_type(cls, v: Any) -> Any:
        if v is not None and not isinstance(v, dict):
            raise ValueError(f"metadata must be a dictionary, got {type(v).__name__}")
        return v  # type: ignore

    def to_text_new(self, children_text: str, indent: int = 0) -> str:
        """Render this transcript group with its children and metadata.

        Metadata appears below the rendered children content.

        Args:
            children_text: Pre-rendered text of this group's children (groups/transcripts).
            indent: Number of spaces to indent the rendered output.

        Returns:
            str: XML-like wrapped text including the group's metadata.
        """
        # Prepare YAML metadata
        yaml_text = yaml_dump_metadata(self.metadata)
        if yaml_text is not None:
            if indent > 0:
                yaml_text = textwrap.indent(yaml_text, " " * indent)
            inner = (
                f"{children_text}\n<|{self.name} metadata|>\n{yaml_text}\n</|{self.name} metadata|>"
            )
        else:
            inner = children_text

        # Compose final text: content first, then metadata, all inside the group wrapper
        if indent > 0:
            inner = textwrap.indent(inner, " " * indent)
        return f"<|{self.name}|>\n{inner}\n</|{self.name}|>"

to_text_new

to_text_new(children_text: str, indent: int = 0) -> str

Render this transcript group with its children and metadata.

Metadata appears below the rendered children content.

Parameters:

Name	Type	Description	Default
children_text	str	Pre-rendered text of this group's children (groups/transcripts).	required
indent	int	Number of spaces to indent the rendered output.	0

Returns:

Name	Type	Description
str	str	XML-like wrapped text including the group's metadata.

Source code in docent/data_models/transcript.py

def to_text_new(self, children_text: str, indent: int = 0) -> str:
    """Render this transcript group with its children and metadata.

    Metadata appears below the rendered children content.

    Args:
        children_text: Pre-rendered text of this group's children (groups/transcripts).
        indent: Number of spaces to indent the rendered output.

    Returns:
        str: XML-like wrapped text including the group's metadata.
    """
    # Prepare YAML metadata
    yaml_text = yaml_dump_metadata(self.metadata)
    if yaml_text is not None:
        if indent > 0:
            yaml_text = textwrap.indent(yaml_text, " " * indent)
        inner = (
            f"{children_text}\n<|{self.name} metadata|>\n{yaml_text}\n</|{self.name} metadata|>"
        )
    else:
        inner = children_text

    # Compose final text: content first, then metadata, all inside the group wrapper
    if indent > 0:
        inner = textwrap.indent(inner, " " * indent)
    return f"<|{self.name}|>\n{inner}\n</|{self.name}|>"

Transcript

Bases: BaseModel

Represents a transcript of messages in a conversation with an AI agent.

A transcript contains a sequence of messages exchanged between different roles (system, user, assistant, tool) and provides methods to organize these messages into logical units of action.

Attributes:

Name	Type	Description
id	str	Unique identifier for the transcript, auto-generated by default.
name	str | None	Optional human-readable name for the transcript.
description	str | None	Optional description of the transcript.
transcript_group_id	str | None	Optional ID of the transcript group this transcript belongs to.
messages	list[ChatMessage]	List of chat messages in the transcript.
metadata	dict[str, Any]	Additional structured metadata about the transcript.

Source code in docent/data_models/transcript.py

class Transcript(BaseModel):
    """Represents a transcript of messages in a conversation with an AI agent.

    A transcript contains a sequence of messages exchanged between different roles
    (system, user, assistant, tool) and provides methods to organize these messages
    into logical units of action.

    Attributes:
        id: Unique identifier for the transcript, auto-generated by default.
        name: Optional human-readable name for the transcript.
        description: Optional description of the transcript.
        transcript_group_id: Optional ID of the transcript group this transcript belongs to.
        messages: List of chat messages in the transcript.
        metadata: Additional structured metadata about the transcript.
    """

    id: str = Field(default_factory=lambda: str(uuid4()))
    name: str | None = None
    description: str | None = None
    transcript_group_id: str | None = None
    created_at: datetime | None = None

    messages: list[ChatMessage]
    metadata: dict[str, Any] = Field(default_factory=dict)
    _units_of_action: list[list[int]] | None = PrivateAttr(default=None)

    @field_validator("metadata", mode="before")
    @classmethod
    def _validate_metadata_type(cls, v: Any) -> Any:
        if v is not None and not isinstance(v, dict):
            raise ValueError(f"metadata must be a dict, got {type(v).__name__}")
        return v  # type: ignore

    @property
    def units_of_action(self) -> list[list[int]]:
        """Get the units of action in the transcript.

        A unit of action represents a logical group of messages, such as a system message
        on its own or a user message followed by assistant responses and tool outputs.

        For precise details on how action units are determined, refer to the _compute_units_of_action method implementation.

        Returns:
            list[list[int]]: List of units of action, where each unit is a list of message indices.
        """
        if self._units_of_action is None:
            self._units_of_action = self._compute_units_of_action()
        return self._units_of_action

    def __init__(self, *args: Any, **kwargs: Any):
        super().__init__(*args, **kwargs)
        self._units_of_action = self._compute_units_of_action()

    def _compute_units_of_action(self) -> list[list[int]]:
        """Compute the units of action in the transcript.

        A unit of action is defined as:
        - A system prompt by itself
        - A group consisting of a user message, assistant response, and any associated tool outputs

        Returns:
            list[list[int]]: A list of units of action, where each unit is a list of message indices.
        """
        if not self.messages:
            return []

        units: list[list[int]] = []
        current_unit: list[int] = []

        def _start_new_unit():
            nonlocal current_unit
            if current_unit:
                units.append(current_unit.copy())
            current_unit = []

        for i, message in enumerate(self.messages):
            role = message.role
            prev_message = self.messages[i - 1] if i > 0 else None

            # System messages are their own unit
            if role == "system":
                # Start a new unit if there's a current unit in progress
                if current_unit:
                    _start_new_unit()
                units.append([i])

            # User message always starts a new unit UNLESS the previous message was a user message
            elif role == "user":
                if current_unit and prev_message and prev_message.role != "user":
                    _start_new_unit()
                current_unit.append(i)

            # Start a new unit if the previous message was not a user or assistant message
            elif role == "assistant":
                if (
                    current_unit
                    and prev_message
                    and prev_message.role != "user"
                    and prev_message.role != "assistant"
                ):
                    _start_new_unit()
                current_unit.append(i)

            # Tool messages are part of the current unit
            elif role == "tool":
                current_unit.append(i)

            else:
                raise ValueError(f"Unknown message role: {role}")

        # Add the last unit if it exists
        _start_new_unit()

        return units

    def get_first_block_in_action_unit(self, action_unit_idx: int) -> int | None:
        """Get the index of the first message in a given action unit.

        Args:
            action_unit_idx: The index of the action unit.

        Returns:
            int | None: The index of the first message in the action unit,
                        or None if the action unit doesn't exist.

        Raises:
            IndexError: If the action unit index is out of range.
        """
        if not self._units_of_action:
            self._units_of_action = self._compute_units_of_action()

        if 0 <= action_unit_idx < len(self._units_of_action):
            unit = self._units_of_action[action_unit_idx]
            return unit[0] if unit else None
        return None

    def get_action_unit_for_block(self, block_idx: int) -> int | None:
        """Find the action unit that contains the specified message block.

        Args:
            block_idx: The index of the message block to find.

        Returns:
            int | None: The index of the action unit containing the block,
                        or None if no action unit contains the block.
        """
        if not self._units_of_action:
            self._units_of_action = self._compute_units_of_action()

        for unit_idx, unit in enumerate(self._units_of_action):
            if block_idx in unit:
                return unit_idx
        return None

    def set_messages(self, messages: list[ChatMessage]):
        """Set the messages in the transcript and recompute units of action.

        Args:
            messages: The new list of chat messages to set.
        """
        self.messages = messages
        self._units_of_action = self._compute_units_of_action()

    def to_str(
        self,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
        highlight_action_unit: int | None = None,
    ) -> str:
        return self._to_str_with_token_limit_impl(
            token_limit=sys.maxsize,
            transcript_idx=transcript_idx,
            agent_run_idx=agent_run_idx,
            use_action_units=True,
            highlight_action_unit=highlight_action_unit,
        )[0]

    def _generate_formatted_blocks(
        self,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
        use_action_units: bool = True,
        highlight_action_unit: int | None = None,
    ) -> list[str]:
        """Generate formatted blocks for transcript representation.

        Args:
            transcript_idx: Index of the transcript
            agent_run_idx: Optional agent run index
            use_action_units: If True, group messages into action units. If False, use individual blocks.
            highlight_action_unit: Optional action unit to highlight (only used with action units)

        Returns:
            list[str]: List of formatted blocks
        """
        if use_action_units:
            if highlight_action_unit is not None and not (
                0 <= highlight_action_unit < len(self._units_of_action or [])
            ):
                raise ValueError(f"Invalid action unit index: {highlight_action_unit}")

            blocks: list[str] = []
            for unit_idx, unit in enumerate(self._units_of_action or []):
                unit_blocks: list[str] = []
                for msg_idx in unit:
                    unit_blocks.append(
                        format_chat_message(
                            self.messages[msg_idx],
                            msg_idx,
                            transcript_idx,
                            agent_run_idx,
                        )
                    )

                unit_content = "\n".join(unit_blocks)

                # Add highlighting if requested
                if highlight_action_unit and unit_idx == highlight_action_unit:
                    blocks_str_template = "<HIGHLIGHTED>\n{}\n</HIGHLIGHTED>"
                else:
                    blocks_str_template = "{}"
                blocks.append(
                    blocks_str_template.format(
                        f"<action unit {unit_idx}>\n{unit_content}\n</action unit {unit_idx}>"
                    )
                )
        else:
            # Individual message blocks
            blocks = []
            for msg_idx, message in enumerate(self.messages):
                blocks.append(
                    format_chat_message(
                        message,
                        msg_idx,
                        transcript_idx,
                        agent_run_idx,
                    )
                )

        return blocks

    def _to_str_with_token_limit_impl(
        self,
        token_limit: int,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
        use_action_units: bool = True,
        highlight_action_unit: int | None = None,
    ) -> list[str]:
        """Core implementation for string representation with token limits.

        Args:
            token_limit: Maximum tokens per returned string
            transcript_idx: Index of the transcript
            agent_run_idx: Optional agent run index
            use_action_units: If True, group messages into action units. If False, use individual blocks.
            highlight_action_unit: Optional action unit to highlight (only used with action units)

        Returns:
            list[str]: List of strings, each within token limit
        """
        blocks = self._generate_formatted_blocks(
            transcript_idx, agent_run_idx, use_action_units, highlight_action_unit
        )
        blocks_str = "\n".join(blocks)

        # Gather metadata
        metadata_obj = to_jsonable_python(self.metadata)
        yaml_width = float("inf")
        block_str = f"<blocks>\n{blocks_str}\n</blocks>\n"
        metadata_str = f"<metadata>\n{yaml.dump(metadata_obj, width=yaml_width)}\n</metadata>"

        if token_limit == sys.maxsize:
            return [f"{block_str}" f"{metadata_str}"]

        metadata_token_count = get_token_count(metadata_str)
        block_token_count = get_token_count(block_str)

        if metadata_token_count + block_token_count <= token_limit:
            return [f"{block_str}" f"{metadata_str}"]
        else:
            results: list[str] = []
            block_token_counts = [get_token_count(block) for block in blocks]
            ranges = group_messages_into_ranges(
                block_token_counts, metadata_token_count, token_limit
            )
            for msg_range in ranges:
                if msg_range.include_metadata:
                    cur_blocks = "\n".join(blocks[msg_range.start : msg_range.end])
                    results.append(f"<blocks>\n{cur_blocks}\n</blocks>\n" f"{metadata_str}")
                else:
                    assert (
                        msg_range.end == msg_range.start + 1
                    ), "Ranges without metadata should be a single message"
                    result = str(blocks[msg_range.start])
                    if msg_range.num_tokens > token_limit - 10:
                        result = truncate_to_token_limit(result, token_limit - 10)
                    results.append(f"<blocks>\n{result}\n</blocks>\n")

            return results

    def to_str_blocks(
        self,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
    ) -> str:
        """Represents the transcript as a string using individual message blocks.

        Unlike to_str() which groups messages into action units, this method
        formats each message as an individual block.

        Returns:
            str: A string representation with individual message blocks.
        """
        return self._to_str_with_token_limit_impl(
            token_limit=sys.maxsize,
            transcript_idx=transcript_idx,
            agent_run_idx=agent_run_idx,
            use_action_units=False,
        )[0]

    def to_str_with_token_limit(
        self,
        token_limit: int,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
        highlight_action_unit: int | None = None,
    ) -> list[str]:
        """Represents the transcript as a list of strings using action units with token limit handling."""
        return self._to_str_with_token_limit_impl(
            token_limit=token_limit,
            transcript_idx=transcript_idx,
            agent_run_idx=agent_run_idx,
            use_action_units=True,
            highlight_action_unit=highlight_action_unit,
        )

    def to_str_blocks_with_token_limit(
        self,
        token_limit: int,
        transcript_idx: int = 0,
        agent_run_idx: int | None = None,
    ) -> list[str]:
        """Represents the transcript as individual blocks with token limit handling."""
        return self._to_str_with_token_limit_impl(
            token_limit=token_limit,
            transcript_idx=transcript_idx,
            agent_run_idx=agent_run_idx,
            use_action_units=False,
        )

    ##############################
    # New text rendering methods #
    ##############################

    def to_text_new(self, transcript_idx: int = 0, indent: int = 0) -> str:
        # Format individual message blocks
        blocks: list[str] = []
        for msg_idx, message in enumerate(self.messages):
            block_text = format_chat_message(message, msg_idx, transcript_idx)
            blocks.append(block_text)
        blocks_str = "\n".join(blocks)
        if indent > 0:
            blocks_str = textwrap.indent(blocks_str, " " * indent)

        content_str = f"<|T{transcript_idx} blocks|>\n{blocks_str}\n</|T{transcript_idx} blocks|>"

        # Gather metadata and add to content
        yaml_text = yaml_dump_metadata(self.metadata)
        if yaml_text is not None:
            if indent > 0:
                yaml_text = textwrap.indent(yaml_text, " " * indent)
            content_str += (
                f"\n<|T{transcript_idx} metadata|>\n{yaml_text}\n</|T{transcript_idx} metadata|>"
            )

        # Format content and return
        if indent > 0:
            content_str = textwrap.indent(content_str, " " * indent)
        return f"<|T{transcript_idx}|>\n{content_str}\n</|T{transcript_idx}|>\n"

units_of_action property

units_of_action: list[list[int]]

Get the units of action in the transcript.

A unit of action represents a logical group of messages, such as a system message on its own or a user message followed by assistant responses and tool outputs.

For precise details on how action units are determined, refer to the _compute_units_of_action method implementation.

Returns:

Type	Description
list[list[int]]	list[list[int]]: List of units of action, where each unit is a list of message indices.

get_first_block_in_action_unit

get_first_block_in_action_unit(action_unit_idx: int) -> int | None

Get the index of the first message in a given action unit.

Parameters:

Name	Type	Description	Default
action_unit_idx	int	The index of the action unit.	required

Returns:

Type	Description
int | None	int | None: The index of the first message in the action unit, or None if the action unit doesn't exist.

Raises:

Type	Description
IndexError	If the action unit index is out of range.

Source code in docent/data_models/transcript.py

def get_first_block_in_action_unit(self, action_unit_idx: int) -> int | None:
    """Get the index of the first message in a given action unit.

    Args:
        action_unit_idx: The index of the action unit.

    Returns:
        int | None: The index of the first message in the action unit,
                    or None if the action unit doesn't exist.

    Raises:
        IndexError: If the action unit index is out of range.
    """
    if not self._units_of_action:
        self._units_of_action = self._compute_units_of_action()

    if 0 <= action_unit_idx < len(self._units_of_action):
        unit = self._units_of_action[action_unit_idx]
        return unit[0] if unit else None
    return None

get_action_unit_for_block

get_action_unit_for_block(block_idx: int) -> int | None

Find the action unit that contains the specified message block.

Parameters:

Name	Type	Description	Default
block_idx	int	The index of the message block to find.	required

Returns:

Type	Description
int | None	int | None: The index of the action unit containing the block, or None if no action unit contains the block.

Source code in docent/data_models/transcript.py

def get_action_unit_for_block(self, block_idx: int) -> int | None:
    """Find the action unit that contains the specified message block.

    Args:
        block_idx: The index of the message block to find.

    Returns:
        int | None: The index of the action unit containing the block,
                    or None if no action unit contains the block.
    """
    if not self._units_of_action:
        self._units_of_action = self._compute_units_of_action()

    for unit_idx, unit in enumerate(self._units_of_action):
        if block_idx in unit:
            return unit_idx
    return None

set_messages

set_messages(messages: list[ChatMessage])

Set the messages in the transcript and recompute units of action.

Parameters:

Name	Type	Description	Default
messages	list[ChatMessage]	The new list of chat messages to set.	required

Source code in docent/data_models/transcript.py

def set_messages(self, messages: list[ChatMessage]):
    """Set the messages in the transcript and recompute units of action.

    Args:
        messages: The new list of chat messages to set.
    """
    self.messages = messages
    self._units_of_action = self._compute_units_of_action()

to_str_blocks

to_str_blocks(transcript_idx: int = 0, agent_run_idx: int | None = None) -> str

Represents the transcript as a string using individual message blocks.

Unlike to_str() which groups messages into action units, this method formats each message as an individual block.

Returns:

Name	Type	Description
str	str	A string representation with individual message blocks.

Source code in docent/data_models/transcript.py

def to_str_blocks(
    self,
    transcript_idx: int = 0,
    agent_run_idx: int | None = None,
) -> str:
    """Represents the transcript as a string using individual message blocks.

    Unlike to_str() which groups messages into action units, this method
    formats each message as an individual block.

    Returns:
        str: A string representation with individual message blocks.
    """
    return self._to_str_with_token_limit_impl(
        token_limit=sys.maxsize,
        transcript_idx=transcript_idx,
        agent_run_idx=agent_run_idx,
        use_action_units=False,
    )[0]

to_str_with_token_limit

to_str_with_token_limit(token_limit: int, transcript_idx: int = 0, agent_run_idx: int | None = None, highlight_action_unit: int | None = None) -> list[str]

Represents the transcript as a list of strings using action units with token limit handling.

Source code in docent/data_models/transcript.py

def to_str_with_token_limit(
    self,
    token_limit: int,
    transcript_idx: int = 0,
    agent_run_idx: int | None = None,
    highlight_action_unit: int | None = None,
) -> list[str]:
    """Represents the transcript as a list of strings using action units with token limit handling."""
    return self._to_str_with_token_limit_impl(
        token_limit=token_limit,
        transcript_idx=transcript_idx,
        agent_run_idx=agent_run_idx,
        use_action_units=True,
        highlight_action_unit=highlight_action_unit,
    )

to_str_blocks_with_token_limit

to_str_blocks_with_token_limit(token_limit: int, transcript_idx: int = 0, agent_run_idx: int | None = None) -> list[str]

Represents the transcript as individual blocks with token limit handling.

Source code in docent/data_models/transcript.py

def to_str_blocks_with_token_limit(
    self,
    token_limit: int,
    transcript_idx: int = 0,
    agent_run_idx: int | None = None,
) -> list[str]:
    """Represents the transcript as individual blocks with token limit handling."""
    return self._to_str_with_token_limit_impl(
        token_limit=token_limit,
        transcript_idx=transcript_idx,
        agent_run_idx=agent_run_idx,
        use_action_units=False,
    )