teff.flow¶

teff.flow ¶

Modules:

Name	Description
`agent`	High-level agent helpers built on :class:`~teff.flow.Flow`.
`base`	Linear chain builders for :class:`~teff.flow.Flow`.
`case`	Branch case for conditional routing.
`compile`	Compilation and serialization builders for :class:`~teff.flow.Flow`.
`compiler`	Declarative `flow.yaml` compiler — the authoring layer.
`control`	Branching, looping and routing builders for :class:`~teff.flow.Flow`.
`flow`	Fluid flow builder for constructing graphs.
`harness`	ReAct agent harness builders for :class:`~teff.flow.Flow`.
`sub_flow`	SubFlow — a node that executes a nested graph.
`team`	Supervised team builder for :class:`~teff.flow.Flow`.

Classes:

Name	Description
`AgentRole`	One routed agent role for :meth:`teff.flow.Flow.team`.
`Case`	A single branch case used with `Flow.branch()`.
`Flow`	Fluid builder for constructing graphs with branching.
`SubFlow`	A node that executes a sub-graph with optional key mapping.

Functions:

Name	Description
`agent_step`	One routed agent: context builder → ReAct harness → append to conversation.

AgentRole ¶

One routed agent role for :meth:teff.flow.Flow.team.

Describes how a role performs its slot: the system prompt, the state key that receives its final answer, and optional model/provider/tool knobs. :meth:build turns it into a :class:~teff.flow.SubFlow (the agent_step recipe) that plugs straight into the team's supervisor route loop::

flow.team(
    "Route to the coder, then finish.",
    roles={
        "coder": AgentRole("You write code.", output_key="code"),
        "planner": AgentRole(
            "You plan.", output_key="plan", use_tools=["web_search"]
        ),
    },
)

Parameters:

Name	Type	Description	Default
`system`	`str`	System prompt for the role.	`''`
`output_key`	`str`	State key that receives the role's final answer.	required
`model`	`str \| None`	Model override; when omitted the team/flow default is used.	`None`
`provider`	`str \| None`	Provider override; when omitted the team/flow default is used.	`None`
`sections`	`dict[str, str] \| None`	Shared state key → label mapping rendered into the agent's context (defaults to `{output_key: Capitalized}`).	`None`
`messages_key`	`str`	State key holding the shared conversation.	`'messages'`
`use_tools`	`str \| list[str] \| None`	`None`/`[]` (no tools), `"all"`, or an allowlist of tool names the role may call.	`None`
`stream`	`bool`	Emit tokens as stream events (live rendering).	`True`
`**config`		Extra kwargs forwarded to the ReAct harness.	`{}`

Methods:

Name	Description
`build`	Render the role as a routed `agent_step` SubFlow.
`from_mapping`	Build a role from a `{system, output_key, ...}` mapping.

Source code in teff/flow/agent.py

class AgentRole:
    """One routed agent role for :meth:`teff.flow.Flow.team`.

    Describes *how* a role performs its slot: the system prompt, the state
    key that receives its final answer, and optional model/provider/tool
    knobs.  :meth:`build` turns it into a :class:`~teff.flow.SubFlow`
    (the ``agent_step`` recipe) that plugs straight into the team's
    supervisor route loop::

        flow.team(
            "Route to the coder, then finish.",
            roles={
                "coder": AgentRole("You write code.", output_key="code"),
                "planner": AgentRole(
                    "You plan.", output_key="plan", use_tools=["web_search"]
                ),
            },
        )

    Args:
        system: System prompt for the role.
        output_key: State key that receives the role's final answer.
        model: Model override; when omitted the team/flow default is used.
        provider: Provider override; when omitted the team/flow default is
            used.
        sections: Shared state key → label mapping rendered into the agent's
            context (defaults to ``{output_key: Capitalized}``).
        messages_key: State key holding the shared conversation.
        use_tools: ``None``/``[]`` (no tools), ``"all"``, or an allowlist of
            tool names the role may call.
        stream: Emit tokens as stream events (live rendering).
        **config: Extra kwargs forwarded to the ReAct harness.
    """

    def __init__(
        self,
        system: str = "",
        *,
        output_key: str,
        model: str | None = None,
        provider: str | None = None,
        sections: dict[str, str] | None = None,
        messages_key: str = "messages",
        use_tools: str | list[str] | None = None,
        stream: bool = True,
        **config,
    ):
        self.system = system
        self.output_key = output_key
        self.model = model
        self.provider = provider
        self.sections = sections
        self.messages_key = messages_key
        self.use_tools = use_tools
        self.stream = stream
        self.config = dict(config)

    def build(
        self,
        *,
        model: str | None = None,
        provider: str | None = None,
        id: str = "",
    ) -> SubFlow:
        """Render the role as a routed ``agent_step`` SubFlow.

        *model*/*provider* fall back from the arguments (the team/flow
        defaults) to the role's own overrides.  Raises ``ValueError`` when
        neither provides them.
        """
        model = model or self.model
        provider = provider or self.provider
        if not model or not provider:
            raise ValueError(
                f"AgentRole {self.output_key!r} needs model and provider "
                "(set them on the role or on the team/flow)"
            )
        return agent_step(
            self.system,
            self.output_key,
            model=model,
            provider=provider,
            sections=self.sections,
            messages_key=self.messages_key,
            use_tools=self.use_tools,
            stream=self.stream,
            id=id,
            **self.config,
        )

    @classmethod
    def from_mapping(cls, data: dict, *, name: str = "") -> "AgentRole":
        """Build a role from a ``{system, output_key, ...}`` mapping.

        Normalises the YAML ``team.roles`` entries (and plain dicts passed
        to :meth:`Flow.team`) into an :class:`AgentRole`.  ``tools:`` is
        accepted as the legacy alias for ``use_tools:``; any other keys are
        kept and forwarded to the ReAct harness via :meth:`build`.
        """
        data = dict(data)
        out_key = data.get("output_key") or name
        if not out_key:
            raise ValueError("a team role needs an `output_key` (or a role name)")
        system = data.pop("system", "") or ""
        known = {
            "model",
            "provider",
            "sections",
            "messages_key",
            "use_tools",
            "stream",
        }
        kwargs = {k: data.pop(k) for k in list(data) if k in known}
        if "tools" in data and "use_tools" not in kwargs:
            kwargs["use_tools"] = data.pop("tools")
        if data:
            kwargs["config"] = data
        return cls(system, output_key=out_key, **kwargs)

build ¶

build(*, model=None, provider=None, id='')

Render the role as a routed agent_step SubFlow.

model/provider fall back from the arguments (the team/flow defaults) to the role's own overrides. Raises ValueError when neither provides them.

Source code in teff/flow/agent.py

def build(
    self,
    *,
    model: str | None = None,
    provider: str | None = None,
    id: str = "",
) -> SubFlow:
    """Render the role as a routed ``agent_step`` SubFlow.

    *model*/*provider* fall back from the arguments (the team/flow
    defaults) to the role's own overrides.  Raises ``ValueError`` when
    neither provides them.
    """
    model = model or self.model
    provider = provider or self.provider
    if not model or not provider:
        raise ValueError(
            f"AgentRole {self.output_key!r} needs model and provider "
            "(set them on the role or on the team/flow)"
        )
    return agent_step(
        self.system,
        self.output_key,
        model=model,
        provider=provider,
        sections=self.sections,
        messages_key=self.messages_key,
        use_tools=self.use_tools,
        stream=self.stream,
        id=id,
        **self.config,
    )

from_mapping `classmethod` ¶

from_mapping(data, *, name='')

Build a role from a {system, output_key, ...} mapping.

Normalises the YAML team.roles entries (and plain dicts passed to :meth:Flow.team) into an :class:AgentRole. tools: is accepted as the legacy alias for use_tools:; any other keys are kept and forwarded to the ReAct harness via :meth:build.

Source code in teff/flow/agent.py

@classmethod
def from_mapping(cls, data: dict, *, name: str = "") -> "AgentRole":
    """Build a role from a ``{system, output_key, ...}`` mapping.

    Normalises the YAML ``team.roles`` entries (and plain dicts passed
    to :meth:`Flow.team`) into an :class:`AgentRole`.  ``tools:`` is
    accepted as the legacy alias for ``use_tools:``; any other keys are
    kept and forwarded to the ReAct harness via :meth:`build`.
    """
    data = dict(data)
    out_key = data.get("output_key") or name
    if not out_key:
        raise ValueError("a team role needs an `output_key` (or a role name)")
    system = data.pop("system", "") or ""
    known = {
        "model",
        "provider",
        "sections",
        "messages_key",
        "use_tools",
        "stream",
    }
    kwargs = {k: data.pop(k) for k in list(data) if k in known}
    if "tools" in data and "use_tools" not in kwargs:
        kwargs["use_tools"] = data.pop("tools")
    if data:
        kwargs["config"] = data
    return cls(system, output_key=out_key, **kwargs)

Case ¶

A single branch case used with Flow.branch().

Methods:

Name	Description
`add`	Add a node to this case branch.

Source code in teff/flow/case.py

class Case:
    """A single branch case used with ``Flow.branch()``."""

    def __init__(self, value: str):
        self.value = value
        self._nodes: list[Node] = []
        self._ids: list[str | None] = []

    def add(self, node: Node, id: str | None = None) -> "Case":
        """Add a node to this case branch.

        *id* optionally names the node in the compiled graph instead of
        the auto-generated ``{type}_{n}``.
        """
        self._nodes.append(node)
        self._ids.append(id)
        return self

add ¶

add(node, id=None)

Add a node to this case branch.

id optionally names the node in the compiled graph instead of the auto-generated {type}_{n}.

Source code in teff/flow/case.py

def add(self, node: Node, id: str | None = None) -> "Case":
    """Add a node to this case branch.

    *id* optionally names the node in the compiled graph instead of
    the auto-generated ``{type}_{n}``.
    """
    self._nodes.append(node)
    self._ids.append(id)
    return self

Flow ¶

Fluid builder for constructing graphs with branching.

Parameters:

Name	Type	Description	Default
`name`	`str`	Optional flow name.	`''`
`default_provider`	`str \| None`	Optional default provider name used by LLM nodes that don't set `provider` themselves. Must be declared in `providers`.	`None`
`default_model`	`str \| None`	Optional default model name used by LLM nodes that don't set `model` themselves. `model=` on a node always wins.	`None`
`providers`	`dict \| ProviderRegistry \| None`	The `{name: Provider}` map, :class:`~teff.provider.ProviderRegistry`, or YAML-style list of preset names threaded into the compiled graph. Every provider the graph references must be declared here.	`None`

Usage::

flow = Flow(
    "my-flow",
    providers=ProviderRegistry.from_presets("ollama"),
    default_provider="ollama",
)
flow.step(LLM(model="llama3.1:8b"))
flow.branch("status", Case("ok").add(ok_node), default=err_node)
graph = flow.compile()

The graph-building methods are implemented by dedicated builders:

:mod:teff.flow.base — step/llm/transform & co. (linear chain).
:mod:teff.flow.team — team (supervised agent team).
:mod:teff.flow.control — parallel/map/branch/ interrupt/loop/route/command (control flow).
:mod:teff.flow.harness — harness/react (ReAct agent loop).
:mod:teff.flow.compile — compile/label/to_yaml (serialization).

Methods:

Name	Description
`add_flow`	Embed a sub-flow as a single node (SubFlow). See
`append_assistant`	Add a :class:`~teff.node.context.AppendAssistant` node. See
`branch`	Add conditional branching from the last added node. See
`command`	Add a declarative `command` node that routes by state. See
`compile`	Compile the flow into a `Graph` ready for execution. See
`context_builder`	Add a :class:`~teff.node.context.ContextBuilder` node. See
`converge`	Merge all branch ends into a single node. See
`default`	Add a fallback node for the most recent guarded `step()`. See
`harness`	Build a ReAct-style agent loop (LLM ↔ tools) inside this flow.
`interrupt`	Pause the flow for human input at this point. See
`interrupt_loop`	Ask the human through an interrupt and re-ask until the answer
`label`	Attach a route name to the most recently added node. See
`label_target`	Resolve a declarative `goto` against labels to a real node id.
`llm`	Add an :class:`~teff.node.llm.LLM` chat node. See
`loop`	Run a chain repeatedly until `state[key]` equals until. See
`map`	Dynamically fan a state list out across parallel branches. See
`parallel`	Run several branch chains concurrently from the last node. See
`react`	Alias for :meth:`harness` (ReAct agent loop). See
`route`	Route between agent chains under a supervisor decider. See
`step`	Append a node to the linear chain. See
`supervisor`	Add a :class:`~teff.node.supervisor.Supervisor` decider node. See
`team`	Compose a supervised agent team in one call. See
`to_yaml`	Export the compiled flow as a `workflow.yaml` document. See
`transform`	Add a :class:`~teff.node.transform.Transform` node. See