datarobot_genai.dragent.frontends.converters

converters

convert_run_agent_input_to_chat_request_or_message

convert_run_agent_input_to_chat_request_or_message(input: RunAgentInput) -> ChatRequestOrMessage

Bridge plain RunAgentInput to NAT chat completions for inner workflow agents.

DRAgent registers converters for DRAgentRunAgentInput only. The DRUM NatAgent.invoke path produces plain RunAgentInput at the streaming_memory_agent passthrough boundary, where inner per_user_tool_calling_agent expects ChatRequestOrMessage.

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_run_agent_input_to_chat_request_or_message(
    input: RunAgentInput,
) -> ChatRequestOrMessage:
    """Bridge plain RunAgentInput to NAT chat completions for inner workflow agents.

    DRAgent registers converters for ``DRAgentRunAgentInput`` only. The DRUM
    ``NatAgent.invoke`` path produces plain ``RunAgentInput`` at the
    ``streaming_memory_agent`` passthrough boundary, where inner
    ``per_user_tool_calling_agent`` expects ``ChatRequestOrMessage``.
    """
    return convert_dragent_run_agent_input_to_chat_request_or_message(
        DRAgentRunAgentInput.model_validate(input.model_dump(by_alias=True))
    )

convert_str_to_chat_response

convert_str_to_chat_response(data: str) -> ChatResponse

Convert a workflow's string output to a ChatResponse reporting the configured model.

Overrides NAT's built-in str -> ChatResponse converter, which calls ChatResponse.from_string(data, usage=usage) without a model and so falls back to its "unknown-model" default: https://github.com/NVIDIA/NeMo-Agent-Toolkit/blob/99e07260fe71872202cdcff1c899f15ef14f4852/packages/nvidia_nat_core/src/nat/data_models/api_server.py#L973 dragent ignores the request's model (the agent runs its configured LLM), so the response reports that configured model (:func:default_response_model), independent of what the caller sent.

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_str_to_chat_response(data: str) -> ChatResponse:
    """Convert a workflow's string output to a ChatResponse reporting the configured model.

    Overrides NAT's built-in ``str -> ChatResponse`` converter, which calls
    ``ChatResponse.from_string(data, usage=usage)`` without a ``model`` and so
    falls back to its ``"unknown-model"`` default:
    https://github.com/NVIDIA/NeMo-Agent-Toolkit/blob/99e07260fe71872202cdcff1c899f15ef14f4852/packages/nvidia_nat_core/src/nat/data_models/api_server.py#L973
    dragent ignores the request's ``model`` (the agent runs its configured LLM), so the
    response reports that configured model (:func:`default_response_model`), independent
    of what the caller sent.
    """
    word_count = len(data.split())
    usage = Usage(prompt_tokens=0, completion_tokens=word_count, total_tokens=word_count)
    return ChatResponse.from_string(data, usage=usage, model=default_response_model())

convert_dragent_event_response_to_chat_response

convert_dragent_event_response_to_chat_response(response: DRAgentEventResponse) -> ChatResponse

Convert a (usually aggregated) DRAgentEventResponse to a NAT ChatResponse.

Used by the non-streaming /chat/completions and inline paths so datarobot_moderations survive the DRAgentEventResponse -> ChatResponse boundary. NAT's default path would route through str (convert_dragent_event_response_to_str then convert_str_to_chat_response), dropping the moderation extra.

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_dragent_event_response_to_chat_response(
    response: DRAgentEventResponse,
) -> ChatResponse:
    """Convert a (usually aggregated) ``DRAgentEventResponse`` to a NAT ``ChatResponse``.

    Used by the non-streaming ``/chat/completions`` and inline paths so
    ``datarobot_moderations`` survive the ``DRAgentEventResponse -> ChatResponse``
    boundary. NAT's default path would route through ``str``
    (``convert_dragent_event_response_to_str`` then ``convert_str_to_chat_response``),
    dropping the moderation extra.
    """
    error_event = _run_error_event(response.events)
    if error_event is not None:
        # Raise so the failure surfaces instead of an empty-success response. RuntimeError, not
        # ValueError, to stay distinct from NAT's own "no conversion path" ValueError.
        raise RuntimeError(error_event.message)

    content = convert_dragent_event_response_to_str(response)

    model = response.model
    if model in (None, "unknown-model") and response.original_chunk is not None:
        model = response.original_chunk.model
    model = backfill_model(model, default_response_model())

    finish_reason: str = "stop"
    if response.original_chunk is not None and response.original_chunk.choices:
        chunk_finish = response.original_chunk.choices[0].finish_reason
        if chunk_finish is not None:
            finish_reason = chunk_finish

    chat_response = ChatResponse(
        id=uuid.uuid4().hex,
        model=model,
        created=datetime.datetime.now(datetime.UTC),
        choices=[
            ChatResponseChoice(
                index=0,
                message=ChoiceMessage(
                    content=content,
                    role=UserMessageContentRoleType.ASSISTANT,
                ),
                finish_reason=cast(_FINISH_REASON, finish_reason),
            )
        ],
        usage=_usage_from_usage_metrics(response.usage_metrics),
    )
    if response.datarobot_moderations is not None:
        chat_response = chat_response.model_copy(
            update={"datarobot_moderations": response.datarobot_moderations}
        )
    return chat_response

convert_nat_chat_response_chunk_to_openai_chat_completion_chunk

convert_nat_chat_response_chunk_to_openai_chat_completion_chunk(chunk: ChatResponseChunk) -> ChatCompletionChunk

Map NAT streaming chunk to OpenAI chat.completion.chunk.

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_nat_chat_response_chunk_to_openai_chat_completion_chunk(
    chunk: ChatResponseChunk,
) -> ChatCompletionChunk:
    """Map NAT streaming chunk to OpenAI ``chat.completion.chunk``."""
    if not chunk.choices:
        raise ValueError("ChatResponseChunk has no choices")
    c0 = chunk.choices[0]
    delta = c0.delta
    openai_delta = OpenAIChoiceDelta(
        content=delta.content,
        role=delta.role.value if delta.role is not None else None,
        tool_calls=_nat_choice_delta_tool_calls_to_openai(delta.tool_calls),
    )
    finish = cast(_FINISH_REASON | None, c0.finish_reason)
    choice = OpenAIChunkChoice(
        index=0,
        delta=openai_delta,
        finish_reason=finish,
    )
    created = chunk.created
    if created.tzinfo is None:
        created = created.replace(tzinfo=datetime.UTC)
    created_ts = int(created.timestamp())
    usage_openai: CompletionUsage | None = None
    if chunk.usage is not None:
        u = chunk.usage
        usage_openai = CompletionUsage(
            prompt_tokens=u.prompt_tokens,
            completion_tokens=u.completion_tokens,
            total_tokens=u.total_tokens,
        )
    openai_chunk = ChatCompletionChunk(
        id=chunk.id,
        choices=[choice],
        created=created_ts,
        model=chunk.model,
        object="chat.completion.chunk",
        usage=usage_openai,
    )
    return _openai_chat_completion_chunk_with_datarobot_moderations(
        openai_chunk, getattr(chunk, "datarobot_moderations", None)
    )

convert_dragent_event_response_to_openai_chat_completion_chunk

convert_dragent_event_response_to_openai_chat_completion_chunk(response: DRAgentEventResponse) -> ChatCompletionChunk

Convert one DRAgent stream chunk to an OpenAI chunk (dome / moderation streaming).

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_dragent_event_response_to_openai_chat_completion_chunk(
    response: DRAgentEventResponse,
) -> ChatCompletionChunk:
    """Convert one DRAgent stream chunk to an OpenAI chunk (dome / moderation streaming)."""
    if response.original_chunk is not None:
        chunk = convert_nat_chat_response_chunk_to_openai_chat_completion_chunk(
            response.original_chunk
        )
    else:
        content, tool_calls = _dragent_streaming_delta_from_events(response.events)
        created_ts = int(datetime.datetime.now(datetime.UTC).timestamp())
        chunk = ChatCompletionChunk(
            id=uuid.uuid4().hex,
            choices=[
                OpenAIChunkChoice(
                    index=0,
                    delta=OpenAIChoiceDelta(content=content, tool_calls=tool_calls or None),
                    finish_reason=None,
                )
            ],
            created=created_ts,
            model=response.model or "unknown-model",
            object="chat.completion.chunk",
            usage=None,
        )
    moderations = _resolve_datarobot_moderations_for_chunk(
        response.original_chunk, response.datarobot_moderations
    )
    return _openai_chat_completion_chunk_with_datarobot_moderations(chunk, moderations)

build_assistant_text_events

build_assistant_text_events(content: str | None) -> list[Event]

Build an assistant TextMessageStart/Content/End AG-UI event sequence.

Produces real assistant text deltas (not a CustomEvent) so downstream consumers that detect assistant text (e.g. moderation postscore) and :func:convert_dragent_event_response_to_str see the content.

Source code in datarobot_genai/dragent/frontends/converters.py

def build_assistant_text_events(content: str | None) -> list[Event]:
    """Build an assistant ``TextMessageStart/Content/End`` AG-UI event sequence.

    Produces real assistant text deltas (not a ``CustomEvent``) so downstream
    consumers that detect assistant text (e.g. moderation postscore) and
    :func:`convert_dragent_event_response_to_str` see the content.
    """
    message_id = str(uuid.uuid4())
    events: list[Event] = [TextMessageStartEvent(message_id=message_id, role="assistant")]
    text = content or ""
    if text:
        events.append(TextMessageContentEvent(message_id=message_id, delta=text))
    else:
        events.append(TextMessageChunkEvent(message_id=message_id, role="assistant", delta=""))
    events.append(TextMessageEndEvent(message_id=message_id))
    return events

convert_str_to_dragent_text_response

convert_str_to_dragent_text_response(response: str) -> DRAgentEventResponse

Convert a native-NAT str output to a text-bearing DRAgentEventResponse.

Emits assistant TextMessage* events so the result carries detectable assistant text — required for the non-streaming normalization + moderation path. Called directly by datarobot_dragent_normalization middleware (not via GlobalTypeConverter).

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_str_to_dragent_text_response(response: str) -> DRAgentEventResponse:
    """Convert a native-NAT ``str`` output to a text-bearing ``DRAgentEventResponse``.

    Emits assistant ``TextMessage*`` events so the result carries detectable assistant
    text — required for the non-streaming normalization + moderation path. Called
    directly by ``datarobot_dragent_normalization`` middleware (not via
    ``GlobalTypeConverter``).
    """
    return DRAgentEventResponse(
        usage_metrics=default_usage_metrics(),
        events=build_assistant_text_events(response),
    )

convert_chat_response_to_dragent_event_response

convert_chat_response_to_dragent_event_response(response: ChatResponse) -> DRAgentEventResponse

Convert a native-NAT ChatResponse to a text-bearing DRAgentEventResponse.

Preserves model and token usage; emits assistant TextMessage* events for the message content. NAT ChoiceMessage carries no tool calls, so only text is represented (matching the non-streaming NAT contract).

Source code in datarobot_genai/dragent/frontends/converters.py

def convert_chat_response_to_dragent_event_response(
    response: ChatResponse,
) -> DRAgentEventResponse:
    """Convert a native-NAT ``ChatResponse`` to a text-bearing ``DRAgentEventResponse``.

    Preserves ``model`` and token usage; emits assistant ``TextMessage*`` events for
    the message content. NAT ``ChoiceMessage`` carries no tool calls, so only text is
    represented (matching the non-streaming NAT contract).
    """
    content = ""
    if response.choices:
        content = response.choices[0].message.content or ""
    usage_metrics = {
        "prompt_tokens": response.usage.prompt_tokens or 0,
        "completion_tokens": response.usage.completion_tokens or 0,
        "total_tokens": response.usage.total_tokens or 0,
    }
    model = response.model if response.model not in (None, "unknown-model") else None
    return DRAgentEventResponse(
        events=build_assistant_text_events(content),
        model=model,
        usage_metrics=usage_metrics,
    )