teff.flow.compiler

teff.flow.compiler

Declarative flow.yaml compiler — the authoring layer.

A flow.yaml is a small, high-level document that describes how the app should behave (teams, chains, gates) without spelling out every node and edge. This package compiles it into a :class:teff.flow.Flow, which in turn produces a regular Graph and can be exported as the compiled graph.yaml artifact::

flow.yaml ──compile()--> Flow ──compile()--> Graph
                           └── to_yaml() ──► graph.yaml

Invariant: the compiler only ever produces things a hand-written graph.yaml (or Flow) could express — every idiom below lowers onto the existing node types and the existing Flow methods. There is no second runtime.

The implementation is split by concern:

• :mod:teff.flow.compiler._common — shared node-level builders;
• :mod:teff.flow.compiler._nodes — single-node idioms (llm: etc.);
• :mod:teff.flow.compiler._flow — flow-control idioms (parallel:…);
• :mod:teff.flow.compiler._team — supervised-team idioms (team:…);
• :mod:teff.flow.compiler._state — tools/state extraction.

Document grammar (0.2 MVP — see docs/design/two-layer.md)::

name: my-app
description: ...
default_provider: ollama
default_model: llama3.1:8b
providers: [...]                    # pass-through (graph providers)
tools: [...]                        # pass-through (tool registry)
state: {schema: ..., initial: ...}  # pass-through (graph state)

steps:
  - llm: {id: replier, model: gpt-4, system: ..., output_key: answer}
  - transform: {id: shout, action: uppercase, input_key: answer,
                output_key: shout}
  - agent_step: {id: coder, system: "You are...", output_key: code,
                 model: ..., tools: [...]}
  - team:
      id: lead
      leader:
        system: "You are the team lead... route to coder or finish."
        model: llama3.1:8b
      roles:
        coder:   {system: "...", output_key: code}
        planner: {system: "...", output_key: plan}
      fallback: planner
      max_rounds: 6
  - supervisor:                          # native decider — no team wrapper
      id: lead
      system: "You are the team lead... route to coder or finish."
      route_keys: {coder: code, planner: plan}
      done_keys: [code, plan]
      fallback: planner
      max_rounds: 6
      agents:                            # optional — wires the whole loop
        coder:   [agent_step: {system: "...", output_key: code}]
        planner: [agent_step: {system: "...", output_key: plan}]
      finish:
        - transform: {action: now, output_key: delivered_at}
  - supervise:                           # route an existing decider (advanced)
      key: next_agent
      agents:
        coder:   [agent_step: {system: "...", output_key: code}]
        planner: [agent_step: {system: "...", output_key: plan}]
      finish:
        - transform: {action: now, output_key: delivered_at}
  - parallel:
      branches:
        - llm: {...}
        - agent: {...}
      converge: {transform: {...}}
  - map:
      input_keys: [items]
      output_key: results
      processor: {llm: {model: ..., system: "..."}}
  - branch:
      key: sentiment
      cases:
        - {value: positive, steps: [transform: {...}]}
        - {value: negative, steps: [transform: {...}]}
      default: {transform: {...}}
      converge: {transform: {...}}
  - loop:
      key: verdict
      until: pass
      body:
        - llm: {...}
      done:
        - llm: {...}
  - interrupt:
      id: approve
      key: decision
      prompt: "Send to work?"
      strategy: {any_of: [approve, ok], decision_key: decision,
                 pass_value: approve, fail_value: rework}
  - route:
      key: decision
      routes:
        - {when: "decision=approve", goto: final}
        - {when: "decision=rework", goto: refine}
      goto: STOP

Unknown steps raise :class:teff.errors.ConfigError.

Functions:

Name	Description
build_flow_to_yaml	Compile flow.yaml at path to a graph.yaml document string.
compile_flow_file	Compile a flow.yaml file into a plain Graph.
flow_from_yaml	Compile a parsed flow.yaml mapping into a :class:Flow.
load_flow	Load a flow.yaml as a (graph, tools, initial, reducers) tuple.
load_flow_yaml	Parse a flow.yaml file, interpolating ${ENV} references.
looks_like_flow	Return True when data is authored in the flow.yaml idiom surface.

build_flow_to_yaml

build_flow_to_yaml(path, output=None)

Compile flow.yaml at path to a graph.yaml document string.

Returns the compiled YAML text; when output is given the text is also written to that path.

Source code in teff/flow/compiler/__init__.py

def build_flow_to_yaml(path: str, output: str | None = None) -> str:
    """Compile ``flow.yaml`` at *path* to a ``graph.yaml`` document string.

    Returns the compiled YAML text; when *output* is given the text is
    also written to that path.
    """
    from ...yaml import workflow_to_yaml

    data = load_flow_yaml(path)
    base_dir = os.path.dirname(os.path.abspath(path))
    from teff.plugins import load_plugins_from_document

    load_plugins_from_document(data, base_dir)
    flow = flow_from_yaml(data)
    text = workflow_to_yaml(flow.compile(), name=data.get("name") or "graph")
    if output:
        with open(output, "w") as f:
            f.write(text)
    return text

compile_flow_file

compile_flow_file(path)

Compile a flow.yaml file into a plain Graph.

Source code in teff/flow/compiler/__init__.py

def compile_flow_file(path: str) -> "Graph":
    """Compile a ``flow.yaml`` file into a plain Graph."""
    doc = load_flow_yaml(path)
    return flow_from_yaml(doc).compile()

flow_from_yaml

flow_from_yaml(data)

Compile a parsed flow.yaml mapping into a :class:Flow.

Parameters:

Name	Type	Description	Default
data	dict	The document mapping returned by :func:load_flow_yaml.	required

Returns:

Name	Type	Description
A	Flow	class:teff.flow.Flow whose :meth:~Flow.compile renders a
	Flow	complete graph (and :meth:~Flow.to_yaml the graph.yaml).

Source code in teff/flow/compiler/__init__.py

def flow_from_yaml(data: dict) -> Flow:
    """Compile a parsed ``flow.yaml`` mapping into a :class:`Flow`.

    Args:
        data: The document mapping returned by :func:`load_flow_yaml`.

    Returns:
        A :class:`teff.flow.Flow` whose :meth:`~Flow.compile` renders a
        complete graph (and :meth:`~Flow.to_yaml` the ``graph.yaml``).
    """
    providers = _providers_from_data(data)
    flow = Flow(
        name=str(data.get("name") or ""),
        providers=providers,
        default_provider=data.get("default_provider"),
        default_model=data.get("default_model"),
    )
    for step in data.get("steps", []) or []:
        _compile_step(flow, step)
    if not flow._nodes:
        raise ConfigError("flow: `steps:` is empty — nothing to compile")
    return flow

load_flow

load_flow(path, data=None)

Load a flow.yaml as a (graph, tools, initial, reducers) tuple.

Mirrors :func:teff.yaml.load_workflow so callers (and the CLI) can treat flow.yaml and graph.yaml interchangeably::

graph, tools, state, reducers = load_flow("app/flow.yaml")

The compiled graph reflects all idioms in the authoring layer; the optional tools: / state: blocks are passed through unchanged. Pass data (a document already resolved by :func:teff.yaml.load_workflow — env interpolation and include: blocks applied) to reuse it instead of re-reading the file.

Source code in teff/flow/compiler/__init__.py

def load_flow(path: str, data: dict | None = None):
    """Load a ``flow.yaml`` as a ``(graph, tools, initial, reducers)`` tuple.

    Mirrors :func:`teff.yaml.load_workflow` so callers (and the CLI) can
    treat ``flow.yaml`` and ``graph.yaml`` interchangeably::

        graph, tools, state, reducers = load_flow("app/flow.yaml")

    The compiled ``graph`` reflects all idioms in the authoring layer; the
    optional ``tools:`` / ``state:`` blocks are passed through unchanged.
    Pass *data* (a document already resolved by :func:`teff.yaml.load_workflow`
    — env interpolation and ``include:`` blocks applied) to reuse it instead
    of re-reading the file.
    """
    if data is None:
        data = load_flow_yaml(path)
    base_dir = os.path.dirname(os.path.abspath(path))
    from teff.plugins import load_plugins_from_document

    load_plugins_from_document(data, base_dir)
    graph = flow_from_yaml(data).compile()
    tools, initial, reducers = _build_state(data, base_dir)
    return graph, tools, initial, reducers

load_flow_yaml

load_flow_yaml(path)

Parse a flow.yaml file, interpolating ${ENV} references.

Returns:

Type	Description
dict	The document mapping. Raises :class:ConfigError on a parse
dict	error or a non-mapping document.

Source code in teff/flow/compiler/__init__.py

def load_flow_yaml(path: str) -> dict:
    """Parse a ``flow.yaml`` file, interpolating ``${ENV}`` references.

    Returns:
        The document mapping.  Raises :class:`ConfigError` on a parse
        error or a non-mapping document.
    """
    if not os.path.exists(path):
        raise ConfigError(f"flow file not found: {path}")
    with open(path) as f:
        data = _safe_load(f)
    if data is None:
        data = {}
    if not isinstance(data, dict):
        raise ConfigError(f"{path}: flow must be a mapping")
    return _interpolate_env(data)

looks_like_flow

looks_like_flow(data)

Return True when data is authored in the flow.yaml idiom surface.

A low-level graph documents steps with explicit id/type keys; the authoring layer uses the shorthand idiom keys instead (llm:, team:, map:, …).

Source code in teff/flow/compiler/__init__.py

def looks_like_flow(data: dict) -> bool:
    """Return ``True`` when *data* is authored in the flow.yaml idiom surface.

    A low-level graph documents ``steps`` with explicit ``id``/``type``
    keys; the authoring layer uses the shorthand idiom keys instead
    (``llm:``, ``team:``, ``map:``, …).
    """
    steps = data.get("steps")
    if not isinstance(steps, list) or not steps:
        return False
    for step in steps:
        if not isinstance(step, dict):
            continue
        for key, value in step.items():
            if key == "type":
                # ``type:`` is only an idiom when its value is a mapping
                # (``- type: {type: csv, config: {...}}``); a low-level
                # ``type: transform`` string is a plain step key.
                if isinstance(value, dict) and value.get("type"):
                    return True
                continue
            if key in _HANDLERS:
                return True
    return False