Interactive Models Guide#

NeMo Agent toolkit provides interactive prompt and response Pydantic data models as a way to validate, serialize, and document data structures to support human input during the execution of an agent workflow. Note: All human in the loop interaction data models are supported by the nat serve command, while the nat run command only supports the nat.data_models.interactive.HumanPromptText data model. Ensure WebSocket mode is enabled by toggling the setting in the top-right corner of the webpage for proper interaction when using this feature with the front-end user interface.

How to Use Interactive Prompt and Response Data Models#

Start by acquiring an instance of the nat.builder.user_interaction_manager.UserInteractionManager class from the nat.builder.context.Context instance.

context = Context.get()
user_input_manager = context.user_interaction_manager

Once the nat.builder.user_interaction_manager.UserInteractionManager has been acquired, use the Interaction Prompt data models located here: nat.data_models.interactive to create a user defined prompt of your choosing i.e. nat.data_models.interactive.HumanPromptText to prompt user interaction during work flow execution.

human_prompt_text = HumanPromptText(text="Hello, how are you today?", required=True, placeholder="default")

Pass the interaction prompt instance to the prompt_user_input method from the nat.builder.user_interaction_manager.UserInteractionManager Once called the workflow will pause execution and wait for user input which can be handled by processing the returned interaction response instance.

response = await user_input_manager.prompt_user_input(human_prompt_text)

Finally, process the returned response from the user input. Note: The response will be an instance of the corresponding data model that matches the type of user-defined interactive prompt.

assert (isinstance(response.content, HumanResponseText))
return response.content.text

Complete example:

async def _inner(prompt: str) -> str:
    try:
        context = Context.get()
        user_input_manager = context.user_interaction_manager

        human_prompt_text = HumanPromptText(text="Hello, how are you today?", required=True, placeholder="default")

        response = await user_input_manager.prompt_user_input(human_prompt_text)

        assert (isinstance(response.content, HumanResponseText))

        return response.content.text

    except Exception as e:
        logger.error("An error occurred when getting interaction content: %s", e)

        raise