Skip to content

Presenter

PresenterAgent

Bases: Agent

An agent that presents a form to the user when triggered by specific keywords in the input stream. The form schema and UI schema are defined in the agent's properties. The agent listens for specific triggers in the input stream and displays the form when triggered. The form data is collected and sent to a specified output stream when the user submits the form.

Properties (in addition to Agent properties):

Name Type Default Description
triggers list of str [] List of keywords that trigger the form display when found in the input stream.
schema dict {} The JSON schema defining the structure of the form to be presented.
form dict {} The UI schema defining the layout and appearance of the form.
output str None The output stream where the collected form data will be sent upon submission. If not specified, the data is returned as a message.

Inputs: - DEFAULT: The main input stream where the agent listens for trigger keywords.

Outputs: - DEFAULT: The output stream where the collected form data is sent in structured format (JSON) upon form submission, tagged as JSON. - FORM: Control output stream for form UI interactions.

Source code in blue/agents/presenter.py 
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
class PresenterAgent(Agent):
    """
    An agent that presents a form to the user when triggered by specific keywords in the input stream.
    The form schema and UI schema are defined in the agent's properties.
    The agent listens for specific triggers in the input stream and displays the form when triggered.
    The form data is collected and sent to a specified output stream when the user submits the form.

    Properties (in addition to Agent properties):
    ----------
    | Name           | Type                 | Default | Description |
    |----------------|--------------------|----------|---------|
    | `triggers`       | `list of str`        | `[]`       | List of keywords that trigger the form display when found in the input stream. |
    | `schema`         | `dict`                | `{}`       | The JSON schema defining the structure of the form to be presented. |
    | `form`           | `dict`                | `{}`       | The UI schema defining the layout and appearance of the form. |
    | `output`         | `str`                | `None`    | The output stream where the collected form data will be sent upon submission. If not specified, the data is returned as a message. |

    Inputs:
    - `DEFAULT`: The main input stream where the agent listens for trigger keywords.

    Outputs:
    - `DEFAULT`: The output stream where the collected form data is sent in structured format (JSON) upon form submission, tagged as JSON.
    - `FORM`: Control output stream for form UI interactions.

    """

    def __init__(self, **kwargs):
        if 'name' not in kwargs:
            kwargs['name'] = "PRESENTER"
        super().__init__(**kwargs)

    ####### inputs / outputs
    def _initialize_inputs(self):
        """Initialize input parameters for the presenter agent. By default, no specific inputs are defined."""
        return

    def _initialize_outputs(self):
        """Initialize outputs for the presenter agent, tagged as JSON."""
        self.add_output("DEFAULT", description="Form data in structured format (JSON)", tags=["JSON"])

    def triggered(self, text, properties):
        """Check if the input text contains any of the trigger keywords defined in properties.

        Parameters:
            text: The input text to check for triggers.
            properties: The properties dict containing trigger keywords.

        Returns:
            True if any trigger keyword is found in the text, False otherwise.
        """
        # if instructed, consider it triggered
        if 'instructable' in properties:
            if properties['instructable']:
                return True

        triggers = properties['triggers']
        for trigger in triggers:
            if trigger.lower() in text.lower():
                return True
        return False

    def default_processor(self, message, input="DEFAULT", properties=None, worker=None):
        """Process messages for the presenter agent, displaying a form when triggered and collecting form data upon submission.

        Parameters:
            message: The incoming message to process.
            input: The input stream name. Defaults to "DEFAULT".
            properties: Additional properties for processing.
            worker: The worker handling the processing.

        Returns:
            None or a response message.
        """
        stream = message.getStream()

        if input == "EVENT":
            if message.isData():
                if worker:
                    data = message.getData()
                    stream = message.getStream()
                    form_id = data["form_id"]
                    action = data["action"]

                    # get form stream
                    form_data_stream = stream.replace("EVENT", "OUTPUT:FORM")

                    # when the user clicked DONE
                    if action == "DONE":
                        # gather all data in the form from stream memory
                        schema = properties['schema']['properties'].keys()

                        form_data = {}
                        for element in schema:
                            form_data[element] = worker.get_stream_data(element + ".value", stream=form_data_stream)

                        # close form
                        args = {"form_id": form_id}
                        worker.write_control(ControlCode.CLOSE_FORM, args, output="FORM")

                        ### stream form data
                        # if output defined, write to output
                        if 'output' in self.properties:
                            output = self.properties['output']
                            worker.write_data(form_data, output=output)
                            worker.write_eos(output=output)
                        else:
                            return [form_data, Message.EOS]

                    else:
                        path = data["path"]
                        timestamp = worker.get_stream_data(path + ".timestamp", stream=form_data_stream)

                        # TODO: timestamp should be replaced by id to determine order
                        if timestamp is None or data["timestamp"] > timestamp:
                            # save data into stream memory
                            worker.set_stream_data(
                                path,
                                {
                                    "value": data["value"],
                                    "timestamp": data["timestamp"],
                                },
                                stream=form_data_stream,
                            )
        else:
            if message.isEOS():
                stream_message = ""
                if worker:
                    stream_message = pydash.to_lower(" ".join(worker.get_data(stream)))

                # check trigger condition, and output to stream form UI when triggered
                if self.triggered(stream_message, properties):
                    args = {
                        "schema": properties['schema'],
                        "uischema": {
                            "type": "VerticalLayout",
                            "elements": [
                                properties['form'],
                                {
                                    "type": "Button",
                                    "label": "Submit",
                                    "props": {
                                        "intent": "success",
                                        "action": "DONE",
                                        "large": True,
                                    },
                                },
                            ],
                        },
                    }
                    # write ui
                    worker.write_control(ControlCode.CREATE_FORM, args, output="FORM")

            elif message.isBOS():
                # init stream to empty array
                if worker:
                    worker.set_data(stream, [])
                pass
            elif message.isData():
                # store data value
                data = message.getData()

                if worker:
                    worker.append_data(stream, data)

default_processor(message, input='DEFAULT', properties=None, worker=None)

Process messages for the presenter agent, displaying a form when triggered and collecting form data upon submission.

Parameters:

Name Type Description Default
message

The incoming message to process.

 required
input

The input stream name. Defaults to "DEFAULT".

 'DEFAULT'
properties

Additional properties for processing.

 None
worker

The worker handling the processing.

 None

Returns:

Type Description

None or a response message.
Source code in blue/agents/presenter.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
def default_processor(self, message, input="DEFAULT", properties=None, worker=None):
    """Process messages for the presenter agent, displaying a form when triggered and collecting form data upon submission.

    Parameters:
        message: The incoming message to process.
        input: The input stream name. Defaults to "DEFAULT".
        properties: Additional properties for processing.
        worker: The worker handling the processing.

    Returns:
        None or a response message.
    """
    stream = message.getStream()

    if input == "EVENT":
        if message.isData():
            if worker:
                data = message.getData()
                stream = message.getStream()
                form_id = data["form_id"]
                action = data["action"]

                # get form stream
                form_data_stream = stream.replace("EVENT", "OUTPUT:FORM")

                # when the user clicked DONE
                if action == "DONE":
                    # gather all data in the form from stream memory
                    schema = properties['schema']['properties'].keys()

                    form_data = {}
                    for element in schema:
                        form_data[element] = worker.get_stream_data(element + ".value", stream=form_data_stream)

                    # close form
                    args = {"form_id": form_id}
                    worker.write_control(ControlCode.CLOSE_FORM, args, output="FORM")

                    ### stream form data
                    # if output defined, write to output
                    if 'output' in self.properties:
                        output = self.properties['output']
                        worker.write_data(form_data, output=output)
                        worker.write_eos(output=output)
                    else:
                        return [form_data, Message.EOS]

                else:
                    path = data["path"]
                    timestamp = worker.get_stream_data(path + ".timestamp", stream=form_data_stream)

                    # TODO: timestamp should be replaced by id to determine order
                    if timestamp is None or data["timestamp"] > timestamp:
                        # save data into stream memory
                        worker.set_stream_data(
                            path,
                            {
                                "value": data["value"],
                                "timestamp": data["timestamp"],
                            },
                            stream=form_data_stream,
                        )
    else:
        if message.isEOS():
            stream_message = ""
            if worker:
                stream_message = pydash.to_lower(" ".join(worker.get_data(stream)))

            # check trigger condition, and output to stream form UI when triggered
            if self.triggered(stream_message, properties):
                args = {
                    "schema": properties['schema'],
                    "uischema": {
                        "type": "VerticalLayout",
                        "elements": [
                            properties['form'],
                            {
                                "type": "Button",
                                "label": "Submit",
                                "props": {
                                    "intent": "success",
                                    "action": "DONE",
                                    "large": True,
                                },
                            },
                        ],
                    },
                }
                # write ui
                worker.write_control(ControlCode.CREATE_FORM, args, output="FORM")

        elif message.isBOS():
            # init stream to empty array
            if worker:
                worker.set_data(stream, [])
            pass
        elif message.isData():
            # store data value
            data = message.getData()

            if worker:
                worker.append_data(stream, data)

triggered(text, properties)

Check if the input text contains any of the trigger keywords defined in properties.

Parameters:

Name Type Description Default
text

The input text to check for triggers.

 required
properties

The properties dict containing trigger keywords.

 required

Returns:

Type Description

True if any trigger keyword is found in the text, False otherwise.
Source code in blue/agents/presenter.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def triggered(self, text, properties):
    """Check if the input text contains any of the trigger keywords defined in properties.

    Parameters:
        text: The input text to check for triggers.
        properties: The properties dict containing trigger keywords.

    Returns:
        True if any trigger keyword is found in the text, False otherwise.
    """
    # if instructed, consider it triggered
    if 'instructable' in properties:
        if properties['instructable']:
            return True

    triggers = properties['triggers']
    for trigger in triggers:
        if trigger.lower() in text.lower():
            return True
    return False
Last update: 2025-10-09