Blocks

This section describes the predefined building blocks available in core.

Overview

The blocks are divided into five categories: Prompt Builders, Prompt Modifiers, Reasoners, Data Modifiers, and Special Blocks.

Prompt builders

Prompt builder generates a prompt using the data in the context, state, and store.

Prompt builder takes a list of messages with placeholders and attempts to fill them with the actual data during the pipeline run.

from agent_dingo.core.blocks import PromptBuilder
from agent_dingo.core.message import UserMessage
from agent_dingo.core.state import KVData, Store, Context


builder = PromptBuilder([UserMessage("What is the capital of {country}?")])
print(builder.forward(None, Context(country="France"), Store()))

By default, the prompt builder will look for the values in the context, but it can be configured to look in the state or store as well.

state = KVData(_out_0="France")
builder = PromptBuilder(
    [UserMessage("What is the capital of {country}?")], from_state=["country"]
)
print(builder.forward(state, Context(), Store()))

Usually, whenever the KVData state is produced by a standard block, its output is stored under the "_out_0" key (or "_out_1" for the second output, etc.). This is a convention that allows the blocks to be easily connected together without the need to manually specify the keys. However, this is not a strict requirement and the key names can be customized as needed.

state = KVData(non_standard_key="France")
builder = PromptBuilder(
    [UserMessage("What is the capital of {country}?")],
    from_state={"country": "non_standard_key"},
)
print(builder.forward(state, Context(), Store()))

Similarly, the prompt builder can be configured to look for the values in the store:

store = Store()
store.update("key", KVData(country="France"))
builder = PromptBuilder(
    [UserMessage("What is the capital of {country}?")],
    from_store={"country": "key.country"},
)
print(builder.forward(None, Context(), store))

Note that when using the "from_store" option, the location of the value in the store is specified using a dot notation. The first part of the part is the key under which the KVData object is stored, and the second part is the key under which the value is stored in the KVData object.

Prompt modifiers

Prompt modifier takes a prompt and modifies it in some way such that the result is still a valid prompt. Dingo only provides a BasePromptModifier class in the core submodule, which can be used as a base class for custom prompt modifiers.

An example of a prompt modifier is a agent_dingo.rag.prompt_modifiers.RAGPromptModifier that modifies the prompt by adding relevant information retrieved from the external source.

Reasoners

Reasoners process the prompt and generate a KVData response. Currently, Dingo has two sub-types of reasoners: LLMs, which generate a single response for a given prompt, and Agents, that can autonomously use the external tools to generate a response in several steps.

Data processors

Data processors work similarly to prompt modifiers, but they operate on the KVData response generated by the reasoners. For example, a Squash block can be used to generate a single string from responses of multiple LLMs.

from agent_dingo.core.blocks import Squash
from agent_dingo.core.state import KVData, Context, Store

state = KVData(gpt4="This is response #1", mistral="This is response #2")
block = Squash("gpt4: {0} mistral: {1}")
print(block.forward(state, Context(), Store()))

Special blocks

Special blocks do not fit into any of the above categories and perform unique tasks.

Parallel

The first special block is Parallel, which allows to run multiple blocks in parallel and merge their outputs into a single KVData object.

from agent_dingo.core.blocks import Parallel
from agent_dingo.llm.openai import OpenAI
from agent_dingo.core.message import UserMessage
from agent_dingo.core.state import ChatPrompt, Context, Store

gpt3 = OpenAI(model="gpt-3.5-turbo")
gpt4 = OpenAI(model="gpt-4-0125-preview")
parallel = Parallel()
parallel.add_block(gpt3)
parallel.add_block(gpt4)

prompt = ChatPrompt([UserMessage("What is your knowledge cut-off date?")])

print(parallel.forward(prompt, Context(), Store()))

When composing a pipeline, it is possible to define a parallel block using the & operator:

from agent_dingo.core.blocks import Parallel, Squash
from agent_dingo.llm.openai import OpenAI
from agent_dingo.core.message import UserMessage
from agent_dingo.core.state import ChatPrompt, Context, Store

gpt3 = OpenAI(model="gpt-3.5-turbo")
gpt4 = OpenAI(model="gpt-4-0125-preview")
prompt = ChatPrompt([UserMessage("What is your knowledge cut-off date?")])

pipeline = (gpt3 & gpt4) >> Squash("gpt3: {0} gpt4: {1}")
print(pipeline.run(prompt))

Identity

The second special block is Identity, which simply returns the input state without any modifications. This block can be useful for defining the "skip-connection" in the parallel blocks.

from agent_dingo.core.blocks import Identity, Squash
from agent_dingo.llm.openai import OpenAI
from custom_block import SanitizeLLMOutput # this block does not exist, but you can create it

gpt3 = OpenAI(model="gpt-3.5-turbo")

pipeline = gpt3 >> (SanitizeLLMOutput() & Identity()) >> Squash("sanitized: {0} original: {1}")

SaveState

The third special block is SaveState, which allows to save the input of the block in the store under a specified key and pass it to the next block.

from agent_dingo.core.blocks import SaveState
from agent_dingo.core.state import KVData, Context, Store

store = Store()
state = KVData(_out_0="This is response #1")

save = SaveState("gpt4")

out = save.forward(state, Context(), store)

assert out is state
assert store.get_data("gpt4")["_out_0"] == "This is response #1"

LoadState

Similarly, the LoadState block allows to load the data from the store and pass it as the state to the next block.

from agent_dingo.core.blocks import LoadState
from agent_dingo.core.state import KVData, Context, Store

store = Store()
state = KVData(_out_0="This is response #1")
store.update("gpt4", state)

load = LoadState("data", "gpt4")  # loads KVDAta state, to load ChatPrompt use "prompts"

out = load.forward(None, Context(), store)

assert out is state

Pipeline

The Pipeline class that we have already seen can also act as a block. This allows to nest pipelines within pipelines.

from agent_dingo.core.blocks import Parallel, Squash, Identity, Pipeline
from agent_dingo.llm.openai import OpenAI
from agent_dingo.core.message import UserMessage
from agent_dingo.core.state import ChatPrompt, Context, Store

gpt3 = OpenAI(model="gpt-3.5-turbo")
gpt4 = OpenAI(model="gpt-4-0125-preview")
prompt = ChatPrompt([UserMessage("What is your knowledge cut-off date?")])

# Add Identity to demonstrate the nested pipeline
gpt3_pipeline = gpt3 >> Identity()
gpt4_pipeline = gpt4 >> Identity()
assert isinstance(gpt3_pipeline, Pipeline)
assert isinstance(gpt4_pipeline, Pipeline)
pipeline = (gpt3_pipeline & gpt4_pipeline) >> Squash("gpt3: {0} gpt4: {1}")
print(pipeline.run(prompt))

InlineBlock

The final special block we will discuss in this section is InlineBlock, which allows to define a custom block as a function without the need to create a separate class. Inline blocks are further discussed in the next section about the custom blocks.

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

@InlineBlock()
def custom_block(state, context, store):
    return KVData(_out_0="This is a custom block.")

print(custom_block.forward(None, Context(), Store()))