Building custom blocks

It is intended that developers heavily rely on creating custom blocks tailored to their specific use cases. To facilitate this process, Dingo blocks are designed to be lightweight and with minial abstractions over the underlying logic, which allows for maximum flexibility and control over the pipeline.

Depending on the complexity of the task, a custom block can be as simple as a single function or as complex as a multi-step pipeline. The only requirement is that the block should be able to accept State, Context and Store objects as input and return a State object as output.

Inline blocks

The simplest way to create a custom block is by defining a function with the required signature and wrapping it with the @InlineBlock decorator.

from agent_dingo.core.blocks import InlineBlock
from agent_dingo.core.state import State, Context, Store, KVData

@InlineBlock(required_context_keys=["key1"])
def my_custom_block(state: State, context: Context, store: Store) -> State:
    return KVData(_out_0=context["key1"])

pipeline = my_custom_block.as_pipeline()
print(pipeline.run(key1="some_value"))

In this example, the my_custom_block function extracts the value of key1 from the Context and returns it inside a State (KVData) object. The @InlineBlock decorator takes the function and transforms it into an instance of a Block, which can be used in a pipeline like any other block.

Notice that the @InlineBlock decorator requires the required_context_keys argument, which specifies the keys that the block expects to find in the Context. This information gets passed to the pipeline, such that it can validate the input before running.

Class-based blocks

This @InlineBlocks approach demonstrated in the previous section is mostly suitable for small, self-contained blocks that do not require any external dependencies. For more complex scenarios, it might be beneficial to define a block as a class. This allows for more flexibility in terms of initialization, configuration, and reuse.

Even when using the class-based approach, the block interface remains simple: the class should implement the forward method, which accepts State, Context, and Store objects as input and returns a State object as output. In addition, a get_required_context_keys method should be implemented to specify the required context keys.

from agent_dingo.core.blocks import Block
from agent_dingo.core.state import State, Context, Store, KVData

class MyCustomBlock(Block):

    def get_required_context_keys(self):
        return ["key1"]

    def forward(self, state: State, context: Context, store: Store) -> State:
        return KVData(_out_0=context["key1"])

my_custom_block = MyCustomBlock()

pipeline = my_custom_block.as_pipeline()
pipeline.run(key1="some_value")