datarobot_genai.dragent.inline

inline

In-process dragent execution for use as a DRUM alternative in run_agent.py.

Mirrors the entry shape of datarobot_drum's execute_drum_inline so the host script can route between DRUM and dragent with a single env-var-gated branch. The function always returns a single aggregated OpenAI ChatCompletion; the request's stream flag is ignored because the agentic playground only renders the final assistant message.

execute_dragent_inline_async async

execute_dragent_inline_async(chat_completion: CompletionCreateParamsBase, custom_model_dir: Path, *, config_file: Path | None = None) -> ChatCompletion

Execute a dragent workflow in-process and return the final OpenAI ChatCompletion.

The function delegates aggregation to NAT: runner.result(to_type=ChatResponse) is the same call the dragent FastAPI route uses for non-streaming /v1/chat/completions requests. NAT workflows already declare how to produce a single response — per_user_tool_calling_agent registers single_output_type=ChatResponse natively, and dragent-native agents (base, langgraph, crewai, llamaindex) declare Streaming(convert=aggregate_dragent_event_responses) so NAT collapses their stream and then chains the registered global converter DRAgentEventResponse → ChatResponseChunk through to ChatResponse. The OpenAI ChatCompletion is then a structural re-validation of NAT's OpenAI-compatible ChatResponse.

Parameters

chat_completion OpenAI Chat Completions create-params dict (stream is ignored). custom_model_dir Directory containing the agent code. workflow.yaml is loaded from here when config_file is not supplied. config_file Optional explicit override of the workflow YAML path.

Source code in datarobot_genai/dragent/inline.py

async def execute_dragent_inline_async(
    chat_completion: CompletionCreateParamsBase,
    custom_model_dir: Path,
    *,
    config_file: Path | None = None,
) -> ChatCompletion:
    """Execute a dragent workflow in-process and return the final OpenAI ``ChatCompletion``.

    The function delegates aggregation to NAT: ``runner.result(to_type=ChatResponse)``
    is the same call the dragent FastAPI route uses for non-streaming
    ``/v1/chat/completions`` requests. NAT workflows already declare how to
    produce a single response — ``per_user_tool_calling_agent`` registers
    ``single_output_type=ChatResponse`` natively, and dragent-native agents
    (``base``, ``langgraph``, ``crewai``, ``llamaindex``) declare
    ``Streaming(convert=aggregate_dragent_event_responses)`` so NAT collapses
    their stream and then chains the registered global converter
    ``DRAgentEventResponse → ChatResponseChunk`` through to ``ChatResponse``.
    The OpenAI ``ChatCompletion`` is then a structural re-validation of NAT's
    OpenAI-compatible ``ChatResponse``.

    Parameters
    ----------
    chat_completion
        OpenAI Chat Completions create-params dict (``stream`` is ignored).
    custom_model_dir
        Directory containing the agent code. ``workflow.yaml`` is loaded from
        here when ``config_file`` is not supplied.
    config_file
        Optional explicit override of the workflow YAML path.
    """
    # Local imports keep the optional NAT dependency out of any path that just
    # imports the symbol but never calls it (e.g. when DRUM is selected).
    # The dragent front-end's global type converters are registered by NAT
    # plugin discovery via the ``nat.front_ends`` entry point ``dragent`` when
    # ``load_workflow`` runs.
    from nat.data_models.api_server import ChatResponse
    from nat.runtime.loader import load_workflow

    from datarobot_genai.core.chat.completions import backfill_model
    from datarobot_genai.core.chat.completions import (
        convert_chat_completion_params_to_run_agent_input,
    )
    from datarobot_genai.core.config import default_response_model
    from datarobot_genai.dragent.frontends.request import DRAgentRunAgentInput
    from datarobot_genai.dragent.workflow_paths import publish_dragent_config_file_env

    workflow_path = _resolve_config_path(Path(custom_model_dir), config_file)
    logger.info("Running dragent workflow from %s", workflow_path)
    publish_dragent_config_file_env(workflow_path)

    # Build the workflow input as ``DRAgentRunAgentInput`` (a ``RunAgentInput``
    # subclass) so NAT's registered converters (e.g.
    # ``DRAgentRunAgentInput -> ChatRequest`` /
    # ``DRAgentRunAgentInput -> ChatRequestOrMessage``) match by exact type
    # for workflows whose input type isn't ``RunAgentInput``.
    base_input = convert_chat_completion_params_to_run_agent_input(chat_completion)
    run_agent_input = DRAgentRunAgentInput.model_validate(base_input.model_dump())

    async with (
        load_workflow(workflow_path) as session_manager,
        session_manager.session(user_id=INLINE_USER_ID) as session,
        _seed_nat_workflow_trace_id(),
        session.run(run_agent_input) as runner,
    ):
        response: ChatResponse = await runner.result(to_type=ChatResponse)

    # NAT's ``ChatResponse`` is documented as OpenAI Chat Completions API
    # compatible (same field layout); ``mode="json"`` serialises ``created`` as
    # an int via the field serializer so the resulting dict satisfies OpenAI's
    # stricter typing. Report the configured LLM the agent actually ran (NAT defaults
    # to ``"unknown-model"`` and the request's ``model`` is ignored for execution).
    payload = response.model_dump(mode="json")
    payload["model"] = backfill_model(payload.get("model"), default_response_model())
    return ChatCompletion.model_validate(payload)

execute_dragent_inline

execute_dragent_inline(chat_completion: CompletionCreateParamsBase, custom_model_dir: Path, *, config_file: Path | None = None) -> ChatCompletion

Run :func:execute_dragent_inline_async synchronously for run_agent.py.

Sync entry point used by datarobot-user-models's run_agent.py when the dragent inline path is enabled. The host process may already have an asyncio loop running on the current thread (for example when run_agent_procedure is invoked from an ipykernel driven agentic notebook) without exposing that loop to this call site; in that case asyncio.run cannot be used here and the coroutine is executed in a worker thread with its own event loop.

Source code in datarobot_genai/dragent/inline.py

def execute_dragent_inline(
    chat_completion: CompletionCreateParamsBase,
    custom_model_dir: Path,
    *,
    config_file: Path | None = None,
) -> ChatCompletion:
    """Run :func:`execute_dragent_inline_async` synchronously for ``run_agent.py``.

    Sync entry point used by ``datarobot-user-models``'s ``run_agent.py`` when the
    dragent inline path is enabled. The host process may already have an asyncio
    loop running on the current thread (for example when ``run_agent_procedure``
    is invoked from an ``ipykernel`` driven agentic notebook) without exposing
    that loop to this call site; in that case ``asyncio.run`` cannot be used here
    and the coroutine is executed in a worker thread with its own event loop.
    """
    coro = execute_dragent_inline_async(
        chat_completion=chat_completion,
        custom_model_dir=custom_model_dir,
        config_file=config_file,
    )

    try:
        asyncio.get_running_loop()
    except RuntimeError:
        # No loop on this thread: run inline. The current otel context is
        # already active here, so the coroutine inherits it automatically.
        return asyncio.run(coro)

    # A loop is already running on this thread, so we hand off to a worker
    # thread with its own loop. contextvars are NOT copied into worker
    # threads, so capture the current otel context here and re-attach it
    # inside the worker before the coroutine runs.
    ctx = get_current()

    def _run_in_worker() -> ChatCompletion:
        token = attach(ctx)
        try:
            return asyncio.run(coro)
        finally:
            detach(token)

    with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
        return executor.submit(_run_in_worker).result()