Guide to Advanced Agent Orchestration: Graph Orchestrator

Often, the logic of Linear or Adaptive orchestrators falls short when deeper customization of agent interactions is required. Dynamiq introduces a customizable Graph Orchestrator, empowering you to design and implement architectures tailored precisely to your unique use case.

Graph Orchestrator basics

The Graph orchestrator allows you to create a sequence of states, each of the states can have multiple of the tasks (other nodes).

The Graph Orchestrator allows to create a sequence of states, where each state can contain multiple tasks (represented as nodes), allowing to create highly flexible workflows.

These states are connected by edges, which define the flow between them. There are several types of edges:

• Ordinary Edge – Specifies the next state to execute in a straightforward, linear flow.

• Conditional Edge – Routes the flow based on defined conditions, enabling dynamic branching between multiple states.

Simple edges

To configure a simple edge that directly connects one node to the next, you can create multiple nodes and link them under the Edges section by specifying the names of the source and destination states.

Conditional edges

Conditional edges enable the development of more complex logic by adding conditions that determine the routing between states. A conditional edge is automatically created when multiple destination states are defined. Once created, an additional menu will appear under the "Conditional Edges" tab for further configuration.

To configure a conditional edge, a conditional function needs to be defined. This function can contain custom logic but must return the name of the next state to route to.

For example, a routing function might check if the context value under the key "correct" is true. If so, it will route to the END state; otherwise, the Graph Orchestrator will proceed with executing the refinement state.

from typing import Literal
def run(input_data) -> Literal['END', 'refinement']:
    if input_data.get('correct', False):
        return 'END'
    return 'refinement'

You can access the context in the same way as you would in a Python node, more details in Context management section below.

Context management

The Graph Orchestrator maintains an internal context to facilitate the sharing of critical information between multiple nodes. This context also records the execution history of workflow messages. It is passed as an object in both the Python node and the conditional edge router function.

Example of context structure:

{
"history": [
        {
        "content": "<message_content>",
        "type": "<'user' or 'system' or 'admin'>"
        }, ... 
    ],
    "key": <value>, ...
}

Context management in Agent Nodes

Upon the execution of an Agent Node, its result is automatically added to the context's history, allowing the result to be easily referenced throughout the workflow.

Context management in Python Nodes

Every Python Node must return a dictionary containing a reserved key, "result". The value associated with this key will automatically be added to the context's history. If the "result" key is not returned, an error will be triggered. To add additional values to the context of the Orchestrator, simply include them in the returned dictionary.

For example, to add "finished" to the history and save "test_value" under the key "test_key," the returned dictionary should look like this:

def run(input_data) -> dict:
    ...

    return {"result": "finished", "test_key": "test_value"}

Input handling for Agent Nodes

The input to the agent can be determined in multiple ways:

• Using an Input Transformer: If an Input Transformer is provided, it will be used to transform or generate the required input for the agent.

• Automatic Input Generation: If an Input Transformer is not provided or is incorrect, the Graph Agent Manager will automatically generate the input for the agent based on the available history, ensuring the process continues smoothly.

Example of setting up an Input Transformer for an Agent Node, which retrieves input from the context.agent_input key and passes it to the agent: