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
        metadata_text = dump_metadata(self.metadata)
        if metadata_text is not None:
            if indent > 0:
                metadata_text = textwrap.indent(metadata_text, " " * indent)
            inner = f"{children_text}\n<|{self.name} metadata|>\n{metadata_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
    metadata_text = dump_metadata(self.metadata)
    if metadata_text is not None:
        if indent > 0:
            metadata_text = textwrap.indent(metadata_text, " " * indent)
        inner = f"{children_text}\n<|{self.name} metadata|>\n{metadata_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 _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(
        self,
        token_limit: int = sys.maxsize,
        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"<|transcript metadata|>\n{yaml.dump(metadata_obj, width=yaml_width)}\n</|transcript 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

    ##############################
    # 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
        metadata_text = dump_metadata(self.metadata)
        if metadata_text is not None:
            if indent > 0:
                metadata_text = textwrap.indent(metadata_text, " " * indent)
            content_str += f"\n<|T{transcript_idx} metadata|>\n{metadata_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

to_str(token_limit: int = maxsize, 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.

Parameters:

Name	Type	Description	Default
token_limit	int	Maximum tokens per returned string	maxsize
transcript_idx	int	Index of the transcript	0
agent_run_idx	int | None	Optional agent run index	None
use_action_units	bool	If True, group messages into action units. If False, use individual blocks.	True
highlight_action_unit	int | None	Optional action unit to highlight (only used with action units)	None

Returns:

Type	Description
list[str]	list[str]: List of strings, each within token limit

Source code in docent/data_models/transcript.py

def to_str(
    self,
    token_limit: int = sys.maxsize,
    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"<|transcript metadata|>\n{yaml.dump(metadata_obj, width=yaml_width)}\n</|transcript 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