teff.yaml

teff.yaml

YAML serialisation and deserialisation for graphs.

Functions:

Name	Description
checkpointer_from_workflow	Build the checkpointer declared by a workflow's checkpoint: block.
from_yaml	Parse a YAML string or file path into a Graph.
graph_to_yaml	Serialize a Graph instance to a YAML string.
load_workflow	Load a complete workflow from a YAML file (graph + tools + state).
load_workflow_document	Read path and resolve env refs and include: blocks.
workflow_to_yaml	Serialize a Graph (plus optional tools/state) to a workflow YAML.

checkpointer_from_workflow

checkpointer_from_workflow(path)

Build the checkpointer declared by a workflow's checkpoint: block.

Returns None when the workflow has no checkpoint: block. Relative path values are resolved against the workflow file's directory; dsn values are passed through verbatim.

Source code in teff/yaml.py

def checkpointer_from_workflow(path: str):
    """Build the checkpointer declared by a workflow's ``checkpoint:`` block.

    Returns ``None`` when the workflow has no ``checkpoint:`` block.
    Relative ``path`` values are resolved against the workflow file's
    directory; ``dsn`` values are passed through verbatim.
    """
    from teff.checkpoint.from_config import (
        checkpointer_from_config,
        resolve_checkpoint_config,
    )

    with open(path) as f:
        data = _safe_load(f)
    if not isinstance(data, dict):
        return None
    if "checkpoint" not in data:
        return None
    base_dir = os.path.dirname(os.path.abspath(path))
    return checkpointer_from_config(
        resolve_checkpoint_config(data["checkpoint"], base_dir)
    )

from_yaml

from_yaml(source)

Parse a YAML string or file path into a Graph.

The YAML format::

name: my-graph
steps:
  - id: start
    type: transform
    config: {action: "uppercase"}
edges:
  - from: start
    to: next

If source is an existing file path it is read from disk; otherwise it is treated as a raw YAML string.

Parameters:

Name	Type	Description	Default
source	str	YAML string or path to a .yaml file.	required

Returns:

Type	Description
Graph	A compiled Graph ready for execution.

Source code in teff/yaml.py

def from_yaml(source: str) -> Graph:
    """Parse a YAML string or file path into a ``Graph``.

    The YAML format::

        name: my-graph
        steps:
          - id: start
            type: transform
            config: {action: "uppercase"}
        edges:
          - from: start
            to: next

    If *source* is an existing file path it is read from disk;
    otherwise it is treated as a raw YAML string.

    Args:
        source: YAML string or path to a ``.yaml`` file.

    Returns:
        A compiled ``Graph`` ready for execution.
    """
    if os.path.exists(source):
        data = _load_workflow_document(source)
    else:
        data = _safe_load(source)
        if data is None:
            data = {}
        if not isinstance(data, dict):
            raise ConfigError("workflow must be a mapping")

    base_dir = (
        os.path.dirname(os.path.abspath(source))
        if os.path.exists(source)
        else os.getcwd()
    )
    data = _interpolate_env(data)
    data = _resolve_includes(data, base_dir)
    data = _expand_interrupt_strategy(data)
    label = source if os.path.exists(source) else "workflow"
    raise_for_validation(validate_workflow(data), source=label)

    from teff.node.registry import default_registry

    nodes = {}
    edges = []
    entry_point = None

    for step in data.get("steps", []):
        sid = step["id"]
        stype = step["type"]
        config = step.get("config", {})
        node = default_registry.create(stype, config)
        if step.get("retry"):
            from teff.node.retry import wrap_with_retry

            node = wrap_with_retry(node, step["retry"])
        nodes[sid] = node
        if entry_point is None:
            entry_point = sid

    for edge_data in data.get("edges", []):
        edges.append(
            Edge(
                source_id=edge_data["from"],
                target_id=edge_data["to"],
                condition=edge_data.get("condition"),
            )
        )

    providers = _providers_from_data(data)
    _validate_provider_refs(data, providers)

    return Graph(
        nodes=nodes,
        edges=edges,
        entry_point=entry_point or "",
        providers=providers,
        default_provider=data.get("default_provider"),
        default_model=data.get("default_model"),
    )

graph_to_yaml

graph_to_yaml(graph)

Serialize a Graph instance to a YAML string.

This is shorthand for :func:workflow_to_yaml without tools or state.

Source code in teff/yaml.py

def graph_to_yaml(graph: Graph) -> str:
    """Serialize a ``Graph`` instance to a YAML string.

    This is shorthand for :func:`workflow_to_yaml` without tools or state.
    """
    return workflow_to_yaml(graph)

load_workflow

load_workflow(path)

Load a complete workflow from a YAML file (graph + tools + state).

YAML format::

name: my-workflow
tools:
  - type: calculator
  - type: shell
    config: {root_dir: /tmp}
state:
  schema:
    messages:
      reducer: append
      type: list
  initial:
    status: active
steps:
  - id: step1
    type: transform
    config: {action: uppercase, input_key: text, output_key: out}
edges:
  - from: step1
    to: step2

Returns:

Type	Description
tuple[Graph, list[Tool | McpToolGroup], dict, dict[str, Reducer]]	A (Graph, tools_list, initial_state, reducers) tuple ready for graph.run().

Source code in teff/yaml.py

def load_workflow(
    path: str,
) -> tuple[Graph, list[Tool | McpToolGroup], dict, dict[str, Reducer]]:
    """Load a complete workflow from a YAML file (graph + tools + state).

    YAML format::

        name: my-workflow
        tools:
          - type: calculator
          - type: shell
            config: {root_dir: /tmp}
        state:
          schema:
            messages:
              reducer: append
              type: list
          initial:
            status: active
        steps:
          - id: step1
            type: transform
            config: {action: uppercase, input_key: text, output_key: out}
        edges:
          - from: step1
            to: step2

    Returns:
        A ``(Graph, tools_list, initial_state, reducers)`` tuple ready for ``graph.run()``.
    """
    data = _load_workflow_document(path)

    # Resolve ${ENV} references across the whole document (tools, steps,
    # state), then load plugins so custom node/tool types validate below.
    base_dir = os.path.dirname(os.path.abspath(path))
    data = _interpolate_env(data)
    data = _resolve_includes(data, base_dir)

    # Authoring-layer documents (single-key idiom steps: ``team:``,
    # ``map:``, ``loop:``…) compile through the Flow builder; the classic
    # low-level surface (id/type steps + edges) goes through this loader.
    from teff.flow.compiler import looks_like_flow

    if looks_like_flow(data):
        from teff.flow.compiler import load_flow

        return load_flow(path, data=data)

    data = _expand_interrupt_strategy(data)
    from teff.plugins import load_plugins_from_document

    load_plugins_from_document(data, base_dir)

    import teff.rag  # noqa: F401 — registers the "rag" tool
    import teff.tool.builtin  # noqa: F401 — registers built-in tools

    raise_for_validation(validate_workflow(data), source=path)

    from teff.node.registry import default_registry

    nodes: dict[str, Node] = {}
    edges: list[Edge] = []
    entry_point: str | None = None

    for step in data.get("steps", []):
        sid = step["id"]
        stype = step["type"]
        config = step.get("config", {})
        node = default_registry.create(stype, config)
        if step.get("retry"):
            from teff.node.retry import wrap_with_retry

            node = wrap_with_retry(node, step["retry"])
        nodes[sid] = node
        if entry_point is None:
            entry_point = sid

    for edge_data in data.get("edges", []):
        edges.append(
            Edge(
                source_id=edge_data["from"],
                target_id=edge_data["to"],
                condition=edge_data.get("condition"),
            )
        )

    tools: list[Tool | McpToolGroup] = []
    base_dir = os.path.dirname(os.path.abspath(path))
    for td in data.get("tools", []):
        ttype = td["type"]
        tconfig = td.get("config", {})
        if ttype in ("rag", "rag_ingest"):
            tconfig = _resolve_rag_config(tconfig, base_dir)
        if ttype == "mcp":
            tools.append(_mcp_group_from_config(tconfig))
            continue
        tools.append(default_tool_registry.create(ttype, tconfig))

    providers = _providers_from_data(data)
    _validate_provider_refs(data, providers)

    graph = Graph(
        nodes=nodes,
        edges=edges,
        entry_point=entry_point or "",
        providers=providers,
        default_provider=data.get("default_provider"),
        default_model=data.get("default_model"),
    )

    state_block = data.get("state", {})
    if isinstance(state_block, dict):
        schema = state_block.get("schema", {})
        initial = state_block.get("initial", {})
    else:
        schema = {}
        initial = {}

    if not isinstance(initial, dict):
        raise ConfigError("state.initial must be a mapping")
    if schema:
        from teff.state.state import validate_state

        errors = validate_state(initial, schema)
        if errors:
            raise ConfigError(
                "state.initial does not match state.schema:\n"
                + "\n".join(f"  {e}" for e in errors)
            )

    reducers: dict[str, Reducer] = reducers_from_yaml_schema(schema)

    return graph, tools, initial, reducers

load_workflow_document

load_workflow_document(path)

Read path and resolve env refs and include: blocks.

Returns the fully expanded document (interpolated, includes merged) — the same surface :func:load_workflow validates — without building any nodes or tools.

Source code in teff/yaml.py

def load_workflow_document(path: str) -> dict:
    """Read *path* and resolve env refs and ``include:`` blocks.

    Returns the fully expanded document (interpolated, includes merged)
    — the same surface :func:`load_workflow` validates — without building
    any nodes or tools.
    """
    data = _load_workflow_document(path)
    base_dir = os.path.dirname(os.path.abspath(path))
    data = _interpolate_env(data)
    data = _resolve_includes(data, base_dir)
    return data

workflow_to_yaml

workflow_to_yaml(graph, *, tools=None, initial=None, reducers=None, name='graph')

Serialize a Graph (plus optional tools/state) to a workflow YAML.

steps and edges come from the graph; tools are written from their name and (when present) their config attribute; reducers become the state.schema block (string reducers only) and initial becomes state.initial.

The output validates with :func:validate_workflow and round-trips through :func:load_workflow.

Source code in teff/yaml.py

def workflow_to_yaml(
    graph: Graph,
    *,
    tools: list[Tool] | None = None,
    initial: dict | None = None,
    reducers: dict[str, Reducer] | None = None,
    name: str = "graph",
) -> str:
    """Serialize a ``Graph`` (plus optional tools/state) to a workflow YAML.

    ``steps`` and ``edges`` come from the graph; ``tools`` are written
    from their ``name`` and (when present) their ``config`` attribute;
    ``reducers`` become the ``state.schema`` block (string reducers only)
    and *initial* becomes ``state.initial``.

    The output validates with :func:`validate_workflow` and round-trips
    through :func:`load_workflow`.
    """
    steps = []
    for nid, node in graph.nodes.items():
        config = _serialize_node_config(node)
        steps.append(
            {
                "id": nid,
                "type": getattr(type(node), "type", None) or getattr(node, "type", nid),
                "config": config or {},
            }
        )

    edges = []
    for e in graph.edges:
        if callable(e.condition):
            raise ValueError(
                "cannot serialize a callable edge condition to YAML "
                "(callable conditions are programmatic-only; use a string "
                f"condition or a decider key for the edge {e.source_id!r} -> {e.target_id!r})"
            )
        entry = {"from": e.source_id, "to": e.target_id}
        if e.condition:
            entry["condition"] = e.condition
        edges.append(entry)

    data: dict = {"name": name, "steps": steps, "edges": edges}
    if getattr(graph, "default_provider", None):
        data["default_provider"] = graph.default_provider
    if getattr(graph, "default_model", None):
        data["default_model"] = graph.default_model
    if getattr(graph, "providers", None) and len(graph.providers):
        data["providers"] = _providers_to_block(graph.providers)
    if tools:
        data["tools"] = [{"type": t.name, "config": _tool_config(t)} for t in tools]
    if reducers or initial:
        from teff.state.state import reducers_to_yaml_schema

        state: dict = {}
        if reducers:
            state["schema"] = reducers_to_yaml_schema(reducers)
        if initial:
            state["initial"] = initial
        data["state"] = state
    return yaml.dump(data, default_flow_style=False, sort_keys=False)