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_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)