symfonic.agent.transcript

transcript

Transcript-retrieval value types and read helpers (v7.27.0).

The verbatim conversation transcript lives in the LangGraph checkpointer's state['messages'] channel, persisted per thread_id. This module defines the public :class:TranscriptMessage value type returned by SymfonicAgent.get_transcript(...) (decoupling the public surface from LangChain's internal BaseMessage shape) plus the pure helpers that turn a checkpoint's message list into transcript rows.

Surface separation (v8.0 §14.a): this is surface 1 (verbatim replay), NOT episodic HMS (surface 3, lossy + similarity-only). See docs/concepts/conversation-persistence.md.

Non-goal (v7.27.0): semantic / embedding search over verbatim turns. The episodic embedding index is truncated (lossy) and re-using it would re-blur the transcript/HMS line. Adopters who need semantic search over verbatim turns use their own chat-store search. See docs/concepts/v8-roadmap.md.

TranscriptMessage dataclass

TranscriptMessage(
    index: int,
    role: TranscriptRole,
    content: str,
    message_id: str | None = None,
    timestamp: datetime | None = None,
)

A single verbatim transcript row.

Decouples the public get_transcript surface from LangChain's BaseMessage. index is the row's 0-based position in the speaker-filtered view that produced it (so an index returned for a speaker="user" query is the ordinal among user messages, which is exactly the T14 "first thing I asked" semantics).

timestamp is checkpoint-granularity (the wall-clock ts of the checkpoint that first introduced the message), NOT a per-message stamp -- LangChain BaseMessage carries no native timestamp (see docs/concepts/conversation-persistence.md). It is None when the introducing checkpoint cannot be resolved (e.g. a saver without alist support, where only ordinal/aget_tuple reads are available).

build_transcript_rows

build_transcript_rows(
    messages: list[Any],
    *,
    speaker: Speaker = "all",
    ts_by_message_id: dict[str, datetime] | None = None,
) -> list[TranscriptMessage]

Turn a checkpoint message list into speaker-filtered transcript rows.

index is assigned 0-based over the FILTERED view (so it counts only rows that pass speaker). ts_by_message_id maps a message id to the timestamp of the checkpoint that introduced it (checkpoint granularity); absent ids get timestamp=None.

Source code in src/symfonic/agent/transcript.py

def build_transcript_rows(
    messages: list[Any],
    *,
    speaker: Speaker = "all",
    ts_by_message_id: dict[str, datetime] | None = None,
) -> list[TranscriptMessage]:
    """Turn a checkpoint message list into speaker-filtered transcript rows.

    ``index`` is assigned 0-based over the FILTERED view (so it counts only
    rows that pass ``speaker``).  ``ts_by_message_id`` maps a message id to
    the timestamp of the checkpoint that introduced it (checkpoint
    granularity); absent ids get ``timestamp=None``.
    """
    ts_map = ts_by_message_id or {}
    rows: list[TranscriptMessage] = []
    filtered_index = 0
    for message in messages:
        role = _role_of(message)
        if not _matches_speaker(role, speaker):
            continue
        message_id = getattr(message, "id", None)
        timestamp = ts_map.get(message_id) if message_id else None
        rows.append(
            TranscriptMessage(
                index=filtered_index,
                role=role,
                content=_content_str(message),
                message_id=message_id,
                timestamp=timestamp,
            )
        )
        filtered_index += 1
    return rows

filter_by_time_range

filter_by_time_range(
    rows: list[TranscriptMessage],
    time_range: tuple[datetime, datetime],
) -> list[TranscriptMessage]

Keep rows whose checkpoint-granularity timestamp falls in [start, end].

Rows with timestamp is None (unresolved introducing checkpoint) are dropped from a time-range query -- a row we cannot place in time cannot honestly be claimed to fall inside a window. The bound is inclusive on both ends.

Source code in src/symfonic/agent/transcript.py

def filter_by_time_range(
    rows: list[TranscriptMessage],
    time_range: tuple[datetime, datetime],
) -> list[TranscriptMessage]:
    """Keep rows whose checkpoint-granularity timestamp falls in ``[start, end]``.

    Rows with ``timestamp is None`` (unresolved introducing checkpoint) are
    dropped from a time-range query -- a row we cannot place in time cannot
    honestly be claimed to fall inside a window.  The bound is inclusive on
    both ends.
    """
    start, end = time_range
    return [
        row
        for row in rows
        if row.timestamp is not None and start <= row.timestamp <= end
    ]

slice_by_index

slice_by_index(
    rows: list[TranscriptMessage], index: int
) -> list[TranscriptMessage]

Return a single-element list for the Nth row, or [] out-of-range.

Python-slice index semantics: negatives count from the end (index=-1 is the last row). Out-of-range NEVER raises -- it returns [] (the plan's locked contract). The returned row keeps its original index from the filtered view so callers can correlate.

Source code in src/symfonic/agent/transcript.py

def slice_by_index(
    rows: list[TranscriptMessage], index: int
) -> list[TranscriptMessage]:
    """Return a single-element list for the Nth row, or ``[]`` out-of-range.

    Python-slice index semantics: negatives count from the end
    (``index=-1`` is the last row).  Out-of-range NEVER raises -- it
    returns ``[]`` (the plan's locked contract).  The returned row keeps
    its original ``index`` from the filtered view so callers can correlate.
    """
    n = len(rows)
    if n == 0:
        return []
    resolved = index + n if index < 0 else index
    if resolved < 0 or resolved >= n:
        return []
    return [rows[resolved]]