ContextBuilderElement

The ContextBuilder helps you build an ordered list of messages from incoming payloads and preset messages. It provides flexible ways to combine messages from different sources and emit them in a specific order. For example, a resulting message sequence might look like:

1. main_system_prompt_constant,
2. user_query,
3. secondary_system_prompt_template,
4. additional_query

Instantiation

When initializing a ContextBuilderElement, you must provide an input_map, which defines the necessary components such as inputs, constants, and templates with optional keys such as ports, payload_type, persist, callback, and depends_on to be converted to messages and emitted. In conjunction, you may provide optional arguments like emit_order, trigger_map, build_fn, and outgoing_input_ports as described below.

Inputs Setup

Absent emit_order, trigger_map, and build_fn, the ContextBuilder will wait until all regular ports have received payloads and then emit messages in the order defined by the input_map.

1. input_map

This is a dictionary used to create input ports, constants, and templates. Each key maps to a configuration dictionary with specific properties based on the entry type:

Keys:

• Regular ports: Keys without special suffixes create input ports
• Constants: Keys ending with _constant create preset messages
• Templates: Keys ending with _template create dynamic messages using Jinja2 templates

Port Configuration Options:

• role: The role to assign to the message (e.g., ‘user’, ‘assistant’, ‘system’)
• payload_type: The expected payload type (e.g., MessagePayload, list[MessagePayload])
• ports: Optional list of output ports to connect this input port to. (Automatic inference of payload_type so it’s not necessary to specify when ports are provided - the type is inferred from the first port connection)
• persist: Boolean flag indicating whether the payload should persist after being emitted (defaults to False)
• depends_on: Optional port name or list of port names that must have payloads before this entry is included.
• callback: Optional function to transform the payload when received
  (e.g., lambda payload: MessagePayload(content=payload.model.content.strip()))

Constant Configuration Options:

• role: The role to assign to the message (e.g., ‘system’, ‘user’, ‘assistant’)
• message: The content of the constant message
• depends_on: Optional port name or list of port names that must have payloads before this constant is included.

Template Configuration Options:

• role: The role to assign to the message
• template: Jinja2 template string that can reference other port contents
• depends_on: Optional port name or list of port names that must have payloads before this template is included.

Example:

input_map = {
    'port_a': {
        'role': 'user', 
        'payload_type': MessagePayload, 
        'persist': True,
        'callback': lambda payload: payload.model.content.strip()
    },
    'port_b': {'role': 'assistant', 'payload_type': list[MessagePayload]}, 
    'user_constant': {'role': 'user', 'message': "This text will be a user message"}, 
    'system_template': {'role': 'system', 'template': "{{ port_a }}  --  {{ port_b }}"}
}

This creates two input ports (with a callback for port_a), a constant user message, and a template that combines content from both ports.

Emission Condition:

For the case where emit_order, trigger_map, and build_fn parameters are absent – When all ports receive payloads, all of the Payloads from the ports will be coerced into MessagePayloads and emitted in their key order. If persist is False, the payloads on the ports will be cleared, otherwise they will be retained for future emissions.

---

2. emit_order

The emit_order parameter provides a simple way to specify the order in which messages should be emitted. The ContextBuilder will wait until all required payloads are available before emitting.

Example:

emit_order = ['port_a', 'port_b', 'system_constant', 'system_template']

This specifies that messages should be emitted in the order listed, once all required payloads are available.

Optional Ports:

To mark a port as optional, use square brackets around the port name:

emit_order = ['port_a', '[port_b]', 'system_constant', 'system_template']

In this case, if port_b is not provided, the ContextBuilder will still emit the other messages without it, so make sure that you are aware of the order in which payloads arrive at the ContextBuilder.

---

3. trigger_map

The trigger_map establishes rules for when the ContextBuilder should emit messages. Each key corresponds to a triggering port and points to an ordered list of ports, constants, or templates that should be included in the emission after all of the ports in the list have received payloads.(as long as they are not marked optional with square brackets e.g. ‘[port_b]’)

Example:

trigger_map = {
    'port_a': ['port_a', 'port_b', 'system_constant'],
    'port_b': ['port_b', 'system_constant']
}

In this example, if port_a receives a payload, the builder will stay in that trigger’s mode until all of the ports in the list have received payloads(port_a and port_b), after which the list of messages will be emitted and the trigger mechanism reset, awaiting a new trigger to go off to determine the next list of messages to emit. Keep in mind that the incoming payloads are stored regardless of what trigger mode we’re in, however, the payloads within the trigger’s list will be cleared after emission.

---

4. build_fn

For more complex control over the message building process, you can define a custom build function. This function is called every time any port receives a payload and receives:

• All defined ports and their payloads
• The active_input_port parameter indicating which port just received a payload
• A persistent dictionary c for storing state between calls

Your function should return a list of port/message names that should be processed and emitted. Remember that persist flags and the provided callback still apply.

Example:

def build_fn(port_a, port_b, system_constant, active_input_port, c):
    if active_input_port == port_a:
        return [port_a, port_b, system_constant]
    else:
        return [port_b, system_constant]

Comment: This sample build function adjusts the emission order based on which port was activated.

---

Output setup

outgoing_input_ports

This parameter identifies the port(s) to which the ContextBuilderElement’s output will be sent. You can assign it either during instantiation or manually connect it afterwards.

Example (during instantiation):

cb = ContextBuilderElement(..., outgoing_input_ports=[some_element.ports.some_input_port])

Alternatively, manually connecting after initialization:

cb = ContextBuilderElement(...)
cb.ports.messages_output > some_element.ports.some_input_port

This ensures that the accumulated messages are correctly routed to the intended downstream component.