Interactive Models Guide

AgentIQ 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 aiq serve command, while the aiq run command only supports the aiq.data_models.interactive.HumanPromptText data model.

How to Use Interactive Prompt and Response Data Models

Start by acquiring an instance of the aiq.builder.user_interaction_manager.AIQUserInteractionManager class from the aiq.builder.context.AIQContext instance.

aiq_context = AIQContext.get()
user_input_manager = aiq_context.user_interaction_manager

Once the aiq.builder.user_interaction_manager.AIQUserInteractionManager has been acquired, use the Interaction Prompt data models located here: aiq.data_models.interactive to create a user defined prompt of your choosing i.e. aiq.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 aiq.builder.user_interaction_manager.AIQUserInteractionManager 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:
        aiq_context = AIQContext.get()
        user_input_manager = aiq_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