datarobot_genai.core.agents.history

history

Chat history extraction and normalization utilities.

This module provides helpers for extracting and summarizing prior chat messages from RunAgentInput so that agent templates can inject conversation history into their prompts.

NormalizedHistoryMessage

Bases: _NormalizedHistoryMessageRequired

Normalized representation of a single prior chat message.

This structure is intentionally minimal but preserves enough optional metadata for tool-heavy agents to reconstruct richer history when needed.

Required fields: - role: str - content: str

Optional fields (best-effort, may be absent): - tool_call_id: str | None - name: str | None - tool_calls: Any | None # e.g. OpenAI-style tool_calls payload

Source code in datarobot_genai/core/agents/history.py

class NormalizedHistoryMessage(_NormalizedHistoryMessageRequired, total=False):
    """Normalized representation of a single prior chat message.

    This structure is intentionally minimal but preserves enough optional
    metadata for tool-heavy agents to reconstruct richer history when needed.

    Required fields:
    - role: str
    - content: str

    Optional fields (best-effort, may be absent):
    - tool_call_id: str | None
    - name: str | None
    - tool_calls: Any | None  # e.g. OpenAI-style tool_calls payload
    """

    tool_call_id: str | None
    name: str | None
    tool_calls: Any | None

extract_history_messages

extract_history_messages(run_agent_input: RunAgentInput, max_history: int) -> list[NormalizedHistoryMessage]

Return normalized prior messages to use as chat history.

Behaviour: - Considers run_agent_input.messages in order. - If the final message is a "user" message, treats everything before it as history (so the latest user turn can be handled separately). - Otherwise treats all provided messages as history. - Converts messages into {role, content} dicts with string content. - Truncates to the most recent max_history entries.

Special cases: - When there are no messages, returns an empty list. - When max_history <= 0, history is disabled and an empty list is returned. This matches the documented semantics where 0 means "no history".

Source code in datarobot_genai/core/agents/history.py

def extract_history_messages(
    run_agent_input: RunAgentInput,
    max_history: int,
) -> list[NormalizedHistoryMessage]:
    r"""Return normalized prior messages to use as chat history.

    Behaviour:
    - Considers ``run_agent_input.messages`` in order.
    - If the *final* message is a ``"user"`` message, treats everything *before*
      it as history (so the latest user turn can be handled separately).
    - Otherwise treats all provided messages as history.
    - Converts messages into ``{role, content}`` dicts with string content.
    - Truncates to the most recent ``max_history`` entries.

    Special cases:
    - When there are no messages, returns an empty list.
    - When ``max_history <= 0``, history is disabled and an empty list is returned.
      This matches the documented semantics where 0 means "no history".
    """
    # Fold standalone reasoning messages into their following assistant turn before
    # any truncation, so reasoning travels with its answer and never consumes a slot.
    raw_messages = _fold_reasoning_messages(list(run_agent_input.messages))
    if not raw_messages or max_history <= 0:
        return []

    # Only exclude the final user message when it is actually the last message
    # in the provided list. This avoids dropping trailing assistant/tool messages
    # that some runtimes include after the last user message.
    last_role = _get_message_field(raw_messages[-1], "role")
    history_slice = raw_messages[:-1] if last_role == "user" else raw_messages

    # Keep only the most recent N messages in history to avoid unbounded growth
    # when callers provide long transcripts.
    if len(history_slice) > max_history:
        history_slice = history_slice[-max_history:]

    history: list[NormalizedHistoryMessage] = []
    for message in history_slice:
        role = _get_message_field(message, "role")
        content = _get_message_field(message, "content")
        tool_call_id = _get_message_field(message, "tool_call_id")
        name = _get_message_field(message, "name")
        tool_calls = _get_message_field(message, "tool_calls")

        text = str(content) if content is not None else ""
        if not text and str(role or "") == "tool":
            # Tool outputs should generally have content, but if they don't,
            # keep a minimal placeholder so downstream adapters don't silently
            # drop tool steps.
            label = str(name) if name is not None else ""
            if not label and tool_call_id is not None:
                label = str(tool_call_id)
            text = f"[tool] {label}".strip() if label else "[tool]"
        # Keep tool-call-only assistant turns even with empty content: the
        # structured ``tool_calls`` carry them. ``content`` stays as-is (possibly
        # "") so structured adapters reconstruct it faithfully; the text renderer
        # surfaces the calls separately (see build_history_summary_from_messages).
        if not text and tool_calls is None:
            continue

        entry: NormalizedHistoryMessage = {
            "role": str(role or "user"),
            "content": text,
        }

        # Preserve optional tool metadata when present so downstream agents can
        # reconstruct richer tool histories if desired.
        if tool_call_id is not None:
            entry["tool_call_id"] = str(tool_call_id)
        if name is not None:
            entry["name"] = str(name)
        if tool_calls is not None:
            entry["tool_calls"] = tool_calls

        history.append(entry)

    return history

build_history_summary_from_messages

build_history_summary_from_messages(run_agent_input: RunAgentInput, max_history: int) -> str

Build a plain-text summary of prior turns for prompts.

This is a convenience helper around extract_history_messages that: - Uses the same history selection semantics as extract_history_messages - Truncates to the most recent max_history entries - Normalizes them into {role, content} dicts and then renders a newline-separated transcript of the form role: content.

Returns an empty string when there is no history to include. Callers that embed the result in a prompt section (e.g. "Prior conversation:\n{summary}") should check for an empty return value and omit the section entirely to avoid dangling headers.

Source code in datarobot_genai/core/agents/history.py

def build_history_summary_from_messages(
    run_agent_input: RunAgentInput,
    max_history: int,
) -> str:
    r"""Build a plain-text summary of prior turns for prompts.

    This is a convenience helper around ``extract_history_messages`` that:
    - Uses the same history selection semantics as ``extract_history_messages``
    - Truncates to the most recent ``max_history`` entries
    - Normalizes them into ``{role, content}`` dicts
    and then renders a newline-separated transcript of the form
    ``role: content``.

    Returns an empty string when there is no history to include. Callers that
    embed the result in a prompt section (e.g. "Prior conversation:\\n{summary}")
    should check for an empty return value and omit the section entirely to
    avoid dangling headers.
    """
    history = extract_history_messages(run_agent_input, max_history)
    if not history:
        return ""

    lines = []
    for msg in history:
        # Render content and tool calls in both cases: a turn may have text, a
        # tool call, or both. Appending the tool-call summary whenever tool_calls
        # are present (not only when content is empty) keeps tool steps visible
        # for assistant turns that also say something.
        parts = []
        if msg["content"]:
            parts.append(msg["content"])
        if msg.get("tool_calls"):
            parts.append(_summarize_tool_calls(msg["tool_calls"]))
        lines.append(f"{msg['role']}: {' '.join(parts)}".rstrip())
    return "\n".join(lines)

drop_unpaired_boundary_tool_turns

drop_unpaired_boundary_tool_turns(history: list[NormalizedHistoryMessage]) -> list[NormalizedHistoryMessage]

Drop tool-call/result turns left unpaired at the history boundaries.

Two boundary cases, both of which most chat APIs reject; the structured (native-message) path drops them (the plain-text summary flattens and is unaffected, so it does not call this):

• Leading tool result with no preceding tool call — produced by max_history truncation, which keeps the most recent N turns (removing from the front) and so can slice a tool-call/result pair, leaving the result first.
• Trailing assistant message with tool_calls but no following tool result. This is not from truncation (truncation only trims the front): it is the supplied history simply ending on a tool-call turn (the current user turn is handled separately, so its result may not be in these messages).

Source code in datarobot_genai/core/agents/history.py

def drop_unpaired_boundary_tool_turns(
    history: list[NormalizedHistoryMessage],
) -> list[NormalizedHistoryMessage]:
    """Drop tool-call/result turns left unpaired at the history boundaries.

    Two boundary cases, both of which most chat APIs reject; the structured
    (native-message) path drops them (the plain-text summary flattens and is unaffected,
    so it does not call this):

    - **Leading** ``tool`` result with no preceding tool call — produced by
      ``max_history`` truncation, which keeps the most recent N turns (removing from the
      *front*) and so can slice a tool-call/result pair, leaving the result first.
    - **Trailing** ``assistant`` message with ``tool_calls`` but no following ``tool``
      result. This is *not* from truncation (truncation only trims the front): it is the
      supplied history simply *ending* on a tool-call turn (the current user turn is
      handled separately, so its result may not be in these messages).
    """
    result = list(history)
    # Leading tool result whose originating tool call was truncated off the front.
    while result and result[0].get("role") == "tool":
        result.pop(0)
    # Trailing assistant tool-call the history ends on, with no following tool result.
    if result and result[-1].get("role") == "assistant" and result[-1].get("tool_calls"):
        result.pop()
    return result