YAML Workflows

Every Dynamiq workflow can be expressed as a YAML document — connections, prompts, nodes, flows, and workflows — and loaded back into live Python objects. This is how you keep workflow definitions in version control, run the same definition locally and in CI, and exchange definitions with the platform's serialized workflow format.

The YAML schema

A workflow file has up to five top-level sections. Every entity is keyed by its ID, and type is the dotted Python path of the class to instantiate:

connections:
  openai-conn:
    type: dynamiq.connections.OpenAI
    api_key: ${oc.env:OPENAI_API_KEY}

prompts:
  answer-prompt:
    messages:
      - role: user
        content: |
          Please answer the following question.
          **User Question:** {{query}}
          Answer:

nodes:
  openai-1:
    type: dynamiq.nodes.llms.OpenAI
    name: OpenAI-1
    model: gpt-4o
    connection: openai-conn
    prompt: answer-prompt
    error_handling:
      timeout_seconds: 60
      retry_interval_seconds: 1
      max_retries: 0
      backoff_rate: 1
    input_transformer:
      path: null
      selector:
        "query": "$.query"
    output_transformer:
      path: null
      selector:
        "answer": "$.content"

flows:
  answering-flow:
    name: LLM answering flow
    nodes:
      - openai-1

workflows:
  answering-workflow:
    flow: answering-flow

Key rules, all enforced by the loader:

• connections are declared once and referenced by ID from nodes (connection: openai-conn). Nodes can also declare a connection inline as a dict with a type field.
• prompts are referenced by ID, or defined inline under the node's prompt key.
• nodes can express ordering with depends: [{node: <node-id>}]; every dependency must be part of the same flow.
• flows list node IDs; workflows point at one flow each. A single file can define multiple workflows.
• Files are parsed with OmegaConf, so ${oc.env:VAR_NAME} interpolates environment variables — keep secrets out of the file itself.

Nested components serialize the same way: an Agent node's llm is itself a node dict with its own type, model, and connection reference.

Loading

Workflow.from_yaml_file is the high-level entry point. Pass init_components=True to initialize node components (clients, vector stores) during parsing, and a connection manager to share clients across nodes:

from dynamiq import Workflow, runnables
from dynamiq.connections.managers import get_connection_manager

with get_connection_manager() as cm:
    wf = Workflow.from_yaml_file(
        file_path="workflow.yaml",
        connection_manager=cm,
        init_components=True,
    )
    result = wf.run(
        input_data={"query": "What is Dynamiq?"},
        config=runnables.RunnableConfig(callbacks=[]),
    )
print(result.output)

If the file defines multiple workflows, select one with wf_id="answering-workflow" — omitting it raises a ValueError when more than one workflow is present.

For lower-level control, dynamiq.serializers.loaders.yaml.WorkflowYAMLLoader exposes the pipeline directly: loads(file_path) reads raw data, parse(data) builds a WorkflowYamlData object holding all connections, nodes, flows, and workflows. Malformed definitions raise WorkflowYAMLLoaderException with the offending entity ID in the message.

Dumping

Any workflow built in Python (or loaded from YAML) serializes back with to_yaml_file:

from dynamiq import Workflow
from dynamiq.connections import OpenAI as OpenAIConnection
from dynamiq.flows import Flow
from dynamiq.nodes.llms import OpenAI
from dynamiq.prompts import Message, Prompt

llm = OpenAI(
    connection=OpenAIConnection(),
    model="gpt-4o-mini",
    prompt=Prompt(messages=[Message(role="user", content="{{query}}")]),
)
wf = Workflow(id="answering-workflow", flow=Flow(nodes=[llm]))

wf.to_yaml_file("workflow_dump.yaml")

The dumper (dynamiq.serializers.dumpers.yaml.WorkflowYAMLDumper) hoists node connections into the shared connections section and replaces them with ID references, and rewrites depends entries to node IDs. Connection secrets are written into the dump (include_secure_params=True), so treat dumped files like credentials — or replace secret values with ${oc.env:...} interpolations before committing.

Round-tripping is symmetric: load → dump → load produces an equivalent workflow, which is a cheap way to validate hand-written YAML:

from dynamiq import Workflow
from dynamiq.connections.managers import get_connection_manager

with get_connection_manager() as cm:
    original = Workflow.from_yaml_file(
        file_path="workflow.yaml", connection_manager=cm, init_components=True
    )
    original.to_yaml_file("workflow_dump.yaml")
    reloaded = Workflow.from_yaml_file(
        file_path="workflow_dump.yaml", connection_manager=cm, init_components=True
    )
    assert reloaded.id == original.id

Requirements: $type / $id placeholders

A YAML definition can reference resources that are resolved externally before parsing — this is how definitions exchanged with the platform refer to platform-managed entities (such as Connections) without embedding their secrets. Any dict carrying both $type and $id is a requirement:

nodes:
  my-llm:
    type: dynamiq.nodes.llms.OpenAI
    model: gpt-4o
    connection:
      $type: connection
      $id: 7e2f7c1e-9a1b-4f6e-8d2c-3b1a2c4d5e6f

The resolution flow has four explicit steps:

from dynamiq.serializers.loaders.yaml import WorkflowYAMLLoader

data = WorkflowYAMLLoader.loads("workflow.yaml")

# 1. Collect requirements without initializing anything
requirements = WorkflowYAMLLoader.get_requirements(data)

# 2. Resolve each $id against your own source (API, vault, config)
resolved = {req.id: fetch_resource(req.id) for req in requirements}

# 3. Substitute resolved values into the raw data (mutates in place)
WorkflowYAMLLoader.apply_resolved_requirements(data, resolved)

# 4. Parse into live objects
result = WorkflowYAMLLoader.parse(data)
workflow = list(result.workflows.values())[0]

A requirement may also carry a value_path — a JSONPath expression applied to the resolved value to extract a single field (for example $.account_id). The loader raises WorkflowYAMLLoaderException if an $id is unresolved, or if a value_path matches zero or multiple values.

Next steps