teff.node

teff.node

Modules:

Name	Description
agent	ReAct agent: graph-visible tool-calling loop.
ask	Ask — declarative validation strategy for interrupt answers.
command	Command — a node return value that combines state updates with control flow.
command_node	Declarative command node — route the graph from YAML state.
context	Execution context and context-building nodes for agent flows.
extract	Extract — declarative structured-extraction recipe.
gate	Gate — deterministic loop decider with a retry budget.
interrupt	Interrupt node — pause a workflow for external (human) input.
llm	LLM chat node — multi-provider, tool calling, structured output.
loop	Loop node — repeat a body chain until a state condition holds.
map	Map node — dynamically fan a state list out across concurrent branches.
node	Abstract base for all graph nodes.
parallel	Parallel node — runs independent branches concurrently.
registry	Node registry and decorator for registering node types.
retry	Retry wrapper node with configurable attempts, backoff, and timeout.
supervisor	Supervisor node — decide which routed agent runs next.
tool_call	Tool-call node — invoke a registered tool deterministically.
transform	Transform node — simple string/data transformations.

Classes:

Name	Description
AppendAssistant	Append an agent's response to the shared conversation as assistant.
Ask	Declarative validation strategy for an interrupt answer.
Command	Combine a state update with an explicit next-node route.
CommandNode	Declarative command node: route the graph from YAML state.
ContextBuilder	Compose a plain-text input for an agent from shared state.
ExecContext	Context available to nodes during graph execution.
Extract	Declarative structured-extraction recipe (Ask's sibling).
Fallback	Deterministic fallback that fills a field the model left empty.
GraphInterrupt	Raised by graph.run() when a workflow pauses for human input.
Interrupt	Pause the workflow and wait for external (human) input.
LLM	Call an LLM chat API with tool calling and structured output.
Loop	Repeat a body chain until state[key] equals until.
Map	Run a processor over each item of a state list, in parallel.
Node	Abstract base class for all graph nodes.
NodeRegistry	Registry mapping node type names to factory functions.
Parallel	Execute several branch chains concurrently and merge their results.
ReActAgent	Single-step LLM node for a graph-level ReAct loop.
Retry	Wrap a node with retry logic.
StructuredOutputError	Raised when an LLM response fails structured-output parsing/validation.
Supervisor	Decide which agent handles the latest user message.
ToolCall	Call a registered tool by name with config-driven arguments.
ToolExec	Executes tools signalled by :class:ReActAgent in parallel and feeds
Transform	Apply a transform to state values.
Validate	Decode an interrupt answer into a flow.loop decider value.

Functions:

Name	Description
last_user_message	Return the most recent user message from a conversation list.

AppendAssistant

Bases: Node

Append an agent's response to the shared conversation as assistant.

Source code in teff/node/context.py

class AppendAssistant(Node):
    """Append an agent's response to the shared conversation as assistant."""

    type = "append_assistant"

    def __init__(
        self,
        config: dict | None = None,
        *,
        output_key: str = "draft",
        messages_key: str = "messages",
        **kwargs,
    ):
        merged = {
            "output_key": output_key,
            "messages_key": messages_key,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        content = state.get(self.config["output_key"], "")
        if not content:
            return {}
        return {
            self.config["messages_key"]: [{"role": "assistant", "content": content}]
        }

Ask

Declarative validation strategy for an interrupt answer.

Use the classmethod constructors to pick a strategy::

Ask.equals("yes")
Ask.any_of("yes", "ok", "sure")
Ask.regex(r"^[A-Z0-9]{4,12}$", value_key="discount_code")
Ask.check(lambda v: v.lower() in {"yes", "ok"})
Ask.llm(system=..., user=..., schema=..., model=..., provider=...)

The strategy is auto-detected from the constructor kwargs, so plain Ask(equals="yes", value_key="code") also works.

Methods:

Name	Description
classifier	Build the verdict classifier LLM for the "llm" strategy.
from_mapping	Build an :class:Ask from a declarative strategy mapping.
validate_node	Build the :class:Validate node wired to input_key.

Source code in teff/node/ask.py

class Ask:
    """Declarative validation strategy for an interrupt answer.

    Use the classmethod constructors to pick a strategy::

        Ask.equals("yes")
        Ask.any_of("yes", "ok", "sure")
        Ask.regex(r"^[A-Z0-9]{4,12}$", value_key="discount_code")
        Ask.check(lambda v: v.lower() in {"yes", "ok"})
        Ask.llm(system=..., user=..., schema=..., model=..., provider=...)

    The strategy is auto-detected from the constructor kwargs, so plain
    ``Ask(equals="yes", value_key="code")`` also works.
    """

    def __init__(
        self,
        *,
        equals: Optional[str] = None,
        any_of: Optional[list] = None,
        regex: Optional[str] = None,
        check: Optional[Callable] = None,
        system: str = "",
        user: str = "",
        schema: Optional[dict] = None,
        model: str = "",
        provider: str = "",
        verdict_key: str = "verdict",
        ok_field: str = "ok",
        value_key: str = "",
        value_field: str = "",
        decision_key: str = "decision",
        pass_value: str = "да",
        fail_value: str = "нет",
        clear_field: str = "",
        clarify_value: str = "",
        rounds_key: str = "rounds",
        max_rounds: int = 100,
    ):
        # Internal names avoid colliding with the classmethod constructors.
        self._expected = equals
        self._allowed = list(any_of) if any_of else None
        self._pattern = regex
        self._predicate = check
        self.system = system
        self.user = user
        self.schema = schema
        self.model_name = model
        self.provider = provider
        self.verdict_key = verdict_key
        self.ok_field = ok_field
        self.value_key = value_key
        self.value_field = value_field
        self.decision_key = decision_key
        self.pass_value = pass_value
        self.fail_value = fail_value
        self.clear_field = clear_field
        self.clarify_value = clarify_value
        self.rounds_key = rounds_key
        self.max_rounds = max_rounds

    @property
    def strategy(self) -> str:
        if self._predicate is not None:
            return "check"
        if self._pattern:
            return "regex"
        if self._allowed:
            return "any_of"
        if self._expected is not None:
            return "equals"
        if self.system or self.schema:
            return "llm"
        return ""

    def needs_classifier(self) -> bool:
        return self.strategy == "llm"

    @classmethod
    def equals(cls, value, **kwargs) -> "Ask":
        return cls(equals=value, **kwargs)

    @classmethod
    def any_of(cls, *values, **kwargs) -> "Ask":
        return cls(any_of=list(values), **kwargs)

    @classmethod
    def regex(cls, pattern: str, **kwargs) -> "Ask":
        return cls(regex=pattern, **kwargs)

    @classmethod
    def check(cls, fn: Callable, **kwargs) -> "Ask":
        return cls(check=fn, **kwargs)

    @classmethod
    def llm(
        cls,
        *,
        system: str,
        user: str,
        schema: dict,
        model: str,
        provider: str,
        **kwargs,
    ) -> "Ask":
        return cls(
            system=system,
            user=user,
            schema=schema,
            model=model,
            provider=provider,
            **kwargs,
        )

    @classmethod
    def from_mapping(cls, mapping: dict) -> "Ask":
        """Build an :class:`Ask` from a declarative strategy mapping.

        Mirrors the YAML shorthand on an ``interrupt`` step::

            strategy:
              equals: yes
            # or: any_of: [yes, ok]  |  regex: "^[A-Z0-9]{4}$"
            # or: llm: {system, user, schema, model, provider}

        The mapping's other keys (``value_key``, ``decision_key``,
        ``pass_value``, ``fail_value``, ``verdict_key``, ``ok_field``,
        ``clear_field``, ``clarify_value``, ``rounds_key``, ``max_rounds``)
        are passed through to the chosen strategy constructor.

        Raises:
            ValueError: When no known strategy key is present.
        """
        if "equals" in mapping:
            spec = {k: v for k, v in mapping.items() if k != "equals"}
            return cls(equals=mapping["equals"], **spec)
        if "any_of" in mapping:
            spec = {k: v for k, v in mapping.items() if k != "any_of"}
            return cls(any_of=list(mapping["any_of"]), **spec)
        if "regex" in mapping:
            spec = {k: v for k, v in mapping.items() if k != "regex"}
            return cls(regex=mapping["regex"], **spec)
        if isinstance(mapping.get("llm"), dict):
            llm_cfg = mapping["llm"]
            spec = {k: v for k, v in mapping.items() if k != "llm"}
            return cls(
                system=llm_cfg.get("system", ""),
                user=llm_cfg.get("user", ""),
                schema=llm_cfg.get("schema"),
                model=llm_cfg.get("model", ""),
                provider=llm_cfg.get("provider", ""),
                **spec,
            )
        raise ValueError(
            "strategy requires one of equals / any_of / regex / llm, "
            f"got {sorted(mapping)}"
        )

    def classifier(self) -> LLM:
        """Build the verdict classifier ``LLM`` for the ``"llm"`` strategy."""
        return LLM(
            system=self.system,
            prompt=self.user,
            output_key=self.verdict_key,
            json_schema=self.schema or {},
            model=self.model_name,
            provider=self.provider,
        )

    def validate_node(self, input_key: str) -> "Validate":
        """Build the :class:`Validate` node wired to *input_key*."""
        return Validate(
            input_key=input_key,
            verdict_key=self.verdict_key,
            ok_field=self.ok_field,
            output_key=self.decision_key,
            pass_value=self.pass_value,
            fail_value=self.fail_value,
            clear_field=self.clear_field,
            clarify_value=self.clarify_value,
            value_key=self.value_key,
            value_field=self.value_field,
            rounds_key=self.rounds_key,
            max_rounds=self.max_rounds,
            strategy=self.strategy,
            equals=self._expected,
            any_of=self._allowed,
            regex=self._pattern,
            check=self._predicate,
        )

classifier

classifier()

Build the verdict classifier LLM for the "llm" strategy.

Source code in teff/node/ask.py

def classifier(self) -> LLM:
    """Build the verdict classifier ``LLM`` for the ``"llm"`` strategy."""
    return LLM(
        system=self.system,
        prompt=self.user,
        output_key=self.verdict_key,
        json_schema=self.schema or {},
        model=self.model_name,
        provider=self.provider,
    )

from_mapping classmethod

from_mapping(mapping)

Build an :class:Ask from a declarative strategy mapping.

Mirrors the YAML shorthand on an interrupt step::

strategy:
  equals: yes
# or: any_of: [yes, ok]  |  regex: "^[A-Z0-9]{4}$"
# or: llm: {system, user, schema, model, provider}

The mapping's other keys (value_key, decision_key, pass_value, fail_value, verdict_key, ok_field, clear_field, clarify_value, rounds_key, max_rounds) are passed through to the chosen strategy constructor.

Raises:

Type	Description
ValueError	When no known strategy key is present.

Source code in teff/node/ask.py

@classmethod
def from_mapping(cls, mapping: dict) -> "Ask":
    """Build an :class:`Ask` from a declarative strategy mapping.

    Mirrors the YAML shorthand on an ``interrupt`` step::

        strategy:
          equals: yes
        # or: any_of: [yes, ok]  |  regex: "^[A-Z0-9]{4}$"
        # or: llm: {system, user, schema, model, provider}

    The mapping's other keys (``value_key``, ``decision_key``,
    ``pass_value``, ``fail_value``, ``verdict_key``, ``ok_field``,
    ``clear_field``, ``clarify_value``, ``rounds_key``, ``max_rounds``)
    are passed through to the chosen strategy constructor.

    Raises:
        ValueError: When no known strategy key is present.
    """
    if "equals" in mapping:
        spec = {k: v for k, v in mapping.items() if k != "equals"}
        return cls(equals=mapping["equals"], **spec)
    if "any_of" in mapping:
        spec = {k: v for k, v in mapping.items() if k != "any_of"}
        return cls(any_of=list(mapping["any_of"]), **spec)
    if "regex" in mapping:
        spec = {k: v for k, v in mapping.items() if k != "regex"}
        return cls(regex=mapping["regex"], **spec)
    if isinstance(mapping.get("llm"), dict):
        llm_cfg = mapping["llm"]
        spec = {k: v for k, v in mapping.items() if k != "llm"}
        return cls(
            system=llm_cfg.get("system", ""),
            user=llm_cfg.get("user", ""),
            schema=llm_cfg.get("schema"),
            model=llm_cfg.get("model", ""),
            provider=llm_cfg.get("provider", ""),
            **spec,
        )
    raise ValueError(
        "strategy requires one of equals / any_of / regex / llm, "
        f"got {sorted(mapping)}"
    )

validate_node

validate_node(input_key)

Build the :class:Validate node wired to input_key.

Source code in teff/node/ask.py

def validate_node(self, input_key: str) -> "Validate":
    """Build the :class:`Validate` node wired to *input_key*."""
    return Validate(
        input_key=input_key,
        verdict_key=self.verdict_key,
        ok_field=self.ok_field,
        output_key=self.decision_key,
        pass_value=self.pass_value,
        fail_value=self.fail_value,
        clear_field=self.clear_field,
        clarify_value=self.clarify_value,
        value_key=self.value_key,
        value_field=self.value_field,
        rounds_key=self.rounds_key,
        max_rounds=self.max_rounds,
        strategy=self.strategy,
        equals=self._expected,
        any_of=self._allowed,
        regex=self._pattern,
        check=self._predicate,
    )

Command

Combine a state update with an explicit next-node route.

Attributes:

Name	Type	Description
update		State keys merged after the node (same as returning a plain dict; per-key reducers apply).
goto		Node id to route to next — any node in the graph (a dynamic edge), or :data:STOP to end the run. None keeps normal edge (condition) routing.

Source code in teff/node/command.py

class Command:
    """Combine a state update with an explicit next-node route.

    Attributes:
        update: State keys merged after the node (same as returning a
            plain dict; per-key reducers apply).
        goto: Node id to route to next — any node in the graph (a dynamic
            edge), or :data:`STOP` to end the run.  ``None`` keeps normal
            edge (condition) routing.
    """

    #: Sentinel for :attr:`goto`: terminate the run from a node.
    STOP: typing.Any = object()

    def __init__(
        self,
        update: dict | None = None,
        goto: str | object | None = None,
    ):
        self.update = dict(update or {})
        self.goto = goto

    def __repr__(self) -> str:
        return f"Command(update={self.update!r}, goto={self.goto!r})"

CommandNode

Bases: Node

Declarative command node: route the graph from YAML state.

Returns a :class:~teff.node.command.Command whose goto is chosen from routes (the first route whose when condition matches state, using the same expressions as edges: conditions) and falls back to goto. update merges state keys after routing (reducers apply).

Use STOP as a target to terminate the run::

- id: route
  type: command
  config:
    routes:
      - when: score >= 0.8
        goto: approve
      - when: score < 0.3
        goto: reject
    goto: review
    update: {routed: true}

Source code in teff/node/command_node.py

class CommandNode(Node):
    """Declarative ``command`` node: route the graph from YAML state.

    Returns a :class:`~teff.node.command.Command` whose ``goto`` is chosen
    from ``routes`` (the first route whose ``when`` condition matches
    *state*, using the same expressions as ``edges:`` conditions) and falls
    back to ``goto``.  ``update`` merges state keys after routing (reducers
    apply).

    Use ``STOP`` as a target to terminate the run::

        - id: route
          type: command
          config:
            routes:
              - when: score >= 0.8
                goto: approve
              - when: score < 0.3
                goto: reject
            goto: review
            update: {routed: true}
    """

    type = "command"

    async def execute(self, ctx, state: dict) -> Command:
        from teff.graph.conditions import evaluate

        goto: str | object | None = None
        for route in self.config.get("routes", []) or []:
            when = route.get("when")
            if when and evaluate(when, state):
                goto = _resolve_target(route.get("goto"))
                break
        if goto is None:
            goto = _resolve_target(self.config.get("goto"))
        return Command(update=dict(self.config.get("update") or {}), goto=goto)

ContextBuilder

Bases: Node

Compose a plain-text input for an agent from shared state.

Renders each configured section as <label>:\n<value> plus the latest user message, and clears scratch keys so a routed agent starts clean.

Parameters:

Name	Type	Description	Default
sections	dict[str, str] | None	State key → section label mapping.	None
messages_key	str	State key holding the conversation.	'messages'
output_key	str	State key receiving the composed text.	'input'
reset_keys	tuple[str, ...]	Scratch state keys to clear before the agent runs.	()

Source code in teff/node/context.py

class ContextBuilder(Node):
    """Compose a plain-text ``input`` for an agent from shared state.

    Renders each configured section as ``<label>:\\n<value>`` plus the latest
    user message, and clears scratch keys so a routed agent starts clean.

    Args:
        sections: State key → section label mapping.
        messages_key: State key holding the conversation.
        output_key: State key receiving the composed text.
        reset_keys: Scratch state keys to clear before the agent runs.
    """

    type = "context_builder"

    def __init__(
        self,
        config: dict | None = None,
        *,
        sections: dict[str, str] | None = None,
        messages_key: str = "messages",
        output_key: str = "input",
        reset_keys: tuple[str, ...] = (),
        **kwargs,
    ):
        merged = {
            "sections": sections or {},
            "messages_key": messages_key,
            "output_key": output_key,
            "reset_keys": list(reset_keys),
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        parts: list[str] = []
        for key, label in self.config["sections"].items():
            value = state.get(key)
            if not value:
                continue
            if isinstance(value, list):
                value = "\n".join(str(item) for item in value)
            parts.append(f"{label}:\n{value}")
        last_user = last_user_message(state.get(self.config["messages_key"], []))
        if last_user:
            parts.append(f"User: {last_user}")
        output_key = self.config["output_key"]
        out: dict = {output_key: "\n\n".join(parts)}
        for key in self.config["reset_keys"]:
            if key != output_key:
                out[key] = []
        return out

ExecContext

Context available to nodes during graph execution.

Provides access to registered tools and a placeholder for LLM calls (overridden by the built-in LLM node).

Attributes:

Name	Type	Description
state		Current workflow state dict.
tools		Dict of tool name to Tool instance.
node_id		Graph node id of the running node.
node_type		Node type string of the running node.
tracer		Optional :class:~teff.trace.RunTracer collecting observability events for the current run.
reducers		Per-key merge strategies for state updates.
emit		Optional async sink receiving :class:~teff.stream.StreamEvent objects as the run progresses (used by graph.stream()). None for plain graph.run().
providers		Optional {name: Provider} map or :class:~teff.provider.ProviderRegistry for LLM nodes (custom providers declared in a workflow / passed to graph.run(providers=...)). None uses the built-in presets.
default_provider		Optional default provider name for the graph. LLM nodes use it when they don't set provider themselves (the graph-level Graph(default_provider=...) / workflow default_provider:).
default_model		Optional default model name for the graph. LLM nodes use it when they don't set model themselves (the graph-level Graph(default_model=...)).
hooks		Observability hooks dict (forwarded to nested runs, e.g. :class:~teff.flow.sub_flow.SubFlow).
node_timeout		Per-node timeout for nested runs (seconds).
checkpointer		Optional persistence backend, forwarded to nested runs so interrupts inside a subflow stay resumable.
checkpoint_id		Run key of the enclosing run, used to namespace nested run checkpoints.
owner		Owner scope of the enclosing run.
resume		Resume dict of the enclosing run, forwarded to nested runs so a sub-flow interrupted by human input resumes in place.
on_llm_payload		Optional async callback receiving the raw request / response of every LLM call: (provider, model, messages, completion, usage, latency_ms, cached). Observability layers (tracing, exporters) set this so node harnesses can forward the full payload, not just token counts.

Methods:

Name	Description
llm	Placeholder for LLM calls (not used by built-in LLM node).
tool	Look up a tool by name.

Source code in teff/node/context.py

class ExecContext:
    """Context available to nodes during graph execution.

    Provides access to registered tools and a placeholder for
    LLM calls (overridden by the built-in LLM node).

    Attributes:
        state: Current workflow state dict.
        tools: Dict of tool name to Tool instance.
        node_id: Graph node id of the running node.
        node_type: Node type string of the running node.
        tracer: Optional :class:`~teff.trace.RunTracer` collecting
            observability events for the current run.
        reducers: Per-key merge strategies for state updates.
        emit: Optional async sink receiving :class:`~teff.stream.StreamEvent`
            objects as the run progresses (used by ``graph.stream()``).
            ``None`` for plain ``graph.run()``.
        providers: Optional ``{name: Provider}`` map or
            :class:`~teff.provider.ProviderRegistry` for LLM nodes
            (custom providers declared in a workflow / passed to
            ``graph.run(providers=...)``).  ``None`` uses the built-in
            presets.
        default_provider: Optional default provider name for the graph.  LLM
            nodes use it when they don't set ``provider`` themselves
            (the graph-level ``Graph(default_provider=...)`` / workflow
            ``default_provider:``).
        default_model: Optional default model name for the graph.  LLM
            nodes use it when they don't set ``model`` themselves
            (the graph-level ``Graph(default_model=...)``).
        hooks: Observability hooks dict (forwarded to nested runs, e.g.
            :class:`~teff.flow.sub_flow.SubFlow`).
        node_timeout: Per-node timeout for nested runs (seconds).
        checkpointer: Optional persistence backend, forwarded to nested
            runs so interrupts inside a subflow stay resumable.
        checkpoint_id: Run key of the enclosing run, used to namespace
            nested run checkpoints.
        owner: Owner scope of the enclosing run.
        resume: Resume dict of the enclosing run, forwarded to nested runs
            so a sub-flow interrupted by human input resumes in place.
        on_llm_payload: Optional async callback receiving the raw request /
            response of every LLM call: ``(provider, model, messages,
            completion, usage, latency_ms, cached)``.  Observability layers
            (tracing, exporters) set this so node harnesses can forward the
            full payload, not just token counts.
    """

    def __init__(
        self,
        state: dict,
        tools: dict[str, Tool],
        *,
        node_id: str | None = None,
        node_type: str | None = None,
        tracer: "RunTracer | None" = None,
        reducers: dict[str, Any] | None = None,
        emit: "Callable[[StreamEvent], Awaitable[None]] | None" = None,
        providers: "dict | ProviderRegistry | None" = None,
        default_provider: str | None = None,
        default_model: str | None = None,
        hooks: "dict | None" = None,
        node_timeout: float | None = None,
        checkpointer: Any = None,
        checkpoint_id: str | None = None,
        owner: str | None = None,
        resume: dict | None = None,
        on_llm_payload: "Callable[..., Awaitable[None]] | None" = None,
    ):
        self.state = state
        self.tools = tools
        self.node_id = node_id
        self.node_type = node_type
        self.tracer = tracer
        self.reducers = reducers
        self.emit = emit
        self.providers = providers
        self.default_provider = default_provider
        self.default_model = default_model
        self.hooks = hooks
        self.node_timeout = node_timeout
        self.checkpointer = checkpointer
        self.checkpoint_id = checkpoint_id
        self.owner = owner
        self.resume = resume
        self.on_llm_payload = on_llm_payload

    def tool(self, name: str) -> Tool:
        """Look up a tool by name.

        Args:
            name: Tool name registered in the tool registry.

        Returns:
            Tool instance.

        Raises:
            KeyError: If the tool is not registered.
        """
        if name not in self.tools:
            msg = f"unknown tool: {name}"
            raise KeyError(msg)
        return self.tools[name]

    async def llm(self, model: str, messages: list) -> str:
        """Placeholder for LLM calls (not used by built-in LLM node)."""
        raise NotImplementedError("LLM provider not configured")

llm async

llm(model, messages)

Placeholder for LLM calls (not used by built-in LLM node).

Source code in teff/node/context.py

async def llm(self, model: str, messages: list) -> str:
    """Placeholder for LLM calls (not used by built-in LLM node)."""
    raise NotImplementedError("LLM provider not configured")

tool

tool(name)

Look up a tool by name.

Parameters:

Name	Type	Description	Default
name	str	Tool name registered in the tool registry.	required

Returns:

Type	Description
Tool	Tool instance.

Raises:

Type	Description
KeyError	If the tool is not registered.

Source code in teff/node/context.py

def tool(self, name: str) -> Tool:
    """Look up a tool by name.

    Args:
        name: Tool name registered in the tool registry.

    Returns:
        Tool instance.

    Raises:
        KeyError: If the tool is not registered.
    """
    if name not in self.tools:
        msg = f"unknown tool: {name}"
        raise KeyError(msg)
    return self.tools[name]

Extract

Declarative structured-extraction recipe (Ask's sibling).

Builds [LLM extractor, *Fallback nodes] from a single spec — the extraction half of a done chain::

extractor = Extract.model(
    system="You extract project data...",
    schema=PROJECT_INFO_SCHEMA,
    model="llama3.1:8b",
    provider="ollama",
    messages_key="messages",
    output_key="project_info",
    fallbacks=[
        Extract.fallback("room_type", room_from_first_user),
    ],
)
flow.interrupt_loop(key="approved", ..., done=extractor.nodes())

Use :meth:model to configure the LLM pass (equivalent to a plain LLM with json_schema) and :meth:fallback to declare a deterministic fill for a field the model may drop. Everything else is threaded through to :class:~teff.node.LLM.

Methods:

Name	Description
fallback	Declare a deterministic fill for field via fn(state).
llm	Build the extraction LLM node.
model	Build an extraction recipe around a structured LLM pass.
nodes	Build [LLM extractor, *Fallback nodes] for flow wiring.

Source code in teff/node/extract.py

class Extract:
    """Declarative structured-extraction recipe (``Ask``'s sibling).

    Builds ``[LLM extractor, *Fallback nodes]`` from a single spec — the
    extraction half of a ``done`` chain::

        extractor = Extract.model(
            system="You extract project data...",
            schema=PROJECT_INFO_SCHEMA,
            model="llama3.1:8b",
            provider="ollama",
            messages_key="messages",
            output_key="project_info",
            fallbacks=[
                Extract.fallback("room_type", room_from_first_user),
            ],
        )
        flow.interrupt_loop(key="approved", ..., done=extractor.nodes())

    Use :meth:`model` to configure the LLM pass (equivalent to a plain
    ``LLM`` with ``json_schema``) and :meth:`fallback` to declare a
    deterministic fill for a field the model may drop.  Everything else is
    threaded through to :class:`~teff.node.LLM`.
    """

    def __init__(
        self,
        *,
        system: str = "",
        schema: dict | None = None,
        output_type: Any | None = None,
        model: str = "",
        provider: str = "",
        messages_key: str | None = None,
        output_key: str = "output",
        parse: bool = False,
        fallbacks: list | None = None,
        id: str = "",
        **llm_kwargs,
    ):
        self.system = system
        self.schema = schema
        self.output_type = output_type
        self.model_name = model
        self.provider = provider
        self.messages_key = messages_key
        self.output_key = output_key
        self.parse = parse
        self._id = id
        self._fallbacks = list(fallbacks or [])
        self._llm_kwargs = dict(llm_kwargs)

    @classmethod
    def model(
        cls,
        *,
        system: str,
        schema: dict,
        model: str,
        provider: str,
        **kwargs,
    ) -> "Extract":
        """Build an extraction recipe around a structured ``LLM`` pass.

        ``id`` (optional) names the built nodes in the compiled graph: the
        extractor ``LLM`` becomes ``<id>`` and each fallback
        ``<id>-fallback-<n>``, so the topology shows ``extractor`` instead of
        an auto-generated ``llm_chat_7``.
        """
        return cls(
            system=system,
            schema=schema,
            model=model,
            provider=provider,
            **kwargs,
        )

    @classmethod
    def fallback(cls, field: str, fn: Callable) -> "_FallbackSpec":
        """Declare a deterministic fill for *field* via ``fn(state)``.

        *fn* receives the whole workflow state and returns the field value
        (or ``None`` to skip).  Runs after the LLM pass, only when the
        model left *field* empty.
        """
        return _FallbackSpec(field=field, fn=fn)

    def llm(self) -> LLM:
        """Build the extraction ``LLM`` node."""
        node = LLM(
            system=self.system,
            json_schema=self.schema,
            output_type=self.output_type,
            model=self.model_name,
            provider=self.provider,
            messages_key=self.messages_key,
            output_key=self.output_key,
            parse=self.parse,
            **self._llm_kwargs,
        )
        if self._id:
            node.config["id"] = self._id
        return node

    def nodes(self) -> list[Node]:
        """Build ``[LLM extractor, *Fallback nodes]`` for flow wiring."""
        nodes: list[Node] = [self.llm()]
        for i, spec in enumerate(self._fallbacks, start=1):
            fb = Fallback(
                input_key=self.output_key,
                field=spec.field,
                fn=spec.fn,
            )
            if self._id:
                fb.config["id"] = f"{self._id}-fallback-{i}"
            nodes.append(fb)
        return nodes

fallback classmethod

fallback(field, fn)

Declare a deterministic fill for field via fn(state).

fn receives the whole workflow state and returns the field value (or None to skip). Runs after the LLM pass, only when the model left field empty.

Source code in teff/node/extract.py

@classmethod
def fallback(cls, field: str, fn: Callable) -> "_FallbackSpec":
    """Declare a deterministic fill for *field* via ``fn(state)``.

    *fn* receives the whole workflow state and returns the field value
    (or ``None`` to skip).  Runs after the LLM pass, only when the
    model left *field* empty.
    """
    return _FallbackSpec(field=field, fn=fn)

llm

llm()

Build the extraction LLM node.

Source code in teff/node/extract.py

def llm(self) -> LLM:
    """Build the extraction ``LLM`` node."""
    node = LLM(
        system=self.system,
        json_schema=self.schema,
        output_type=self.output_type,
        model=self.model_name,
        provider=self.provider,
        messages_key=self.messages_key,
        output_key=self.output_key,
        parse=self.parse,
        **self._llm_kwargs,
    )
    if self._id:
        node.config["id"] = self._id
    return node

model classmethod

model(*, system, schema, model, provider, **kwargs)

Build an extraction recipe around a structured LLM pass.

id (optional) names the built nodes in the compiled graph: the extractor LLM becomes <id> and each fallback <id>-fallback-<n>, so the topology shows extractor instead of an auto-generated llm_chat_7.

Source code in teff/node/extract.py

@classmethod
def model(
    cls,
    *,
    system: str,
    schema: dict,
    model: str,
    provider: str,
    **kwargs,
) -> "Extract":
    """Build an extraction recipe around a structured ``LLM`` pass.

    ``id`` (optional) names the built nodes in the compiled graph: the
    extractor ``LLM`` becomes ``<id>`` and each fallback
    ``<id>-fallback-<n>``, so the topology shows ``extractor`` instead of
    an auto-generated ``llm_chat_7``.
    """
    return cls(
        system=system,
        schema=schema,
        model=model,
        provider=provider,
        **kwargs,
    )

nodes

nodes()

Build [LLM extractor, *Fallback nodes] for flow wiring.

Source code in teff/node/extract.py

def nodes(self) -> list[Node]:
    """Build ``[LLM extractor, *Fallback nodes]`` for flow wiring."""
    nodes: list[Node] = [self.llm()]
    for i, spec in enumerate(self._fallbacks, start=1):
        fb = Fallback(
            input_key=self.output_key,
            field=spec.field,
            fn=spec.fn,
        )
        if self._id:
            fb.config["id"] = f"{self._id}-fallback-{i}"
        nodes.append(fb)
    return nodes

Fallback

Bases: Node

Deterministic fallback that fills a field the model left empty.

Reads a dict from input_key; when field in it is empty / None, calls fn(state) and merges the returned value under field. No-op when the dict already has the field or fn returns None.

Config

input_key: State key holding the extracted dict. field: Dict field to fill when empty. fn: Callable fn(state) -> value | None.

Source code in teff/node/extract.py

class Fallback(Node):
    """Deterministic fallback that fills a field the model left empty.

    Reads a dict from ``input_key``; when *field* in it is empty / ``None``,
    calls ``fn(state)`` and merges the returned value under *field*.  No-op
    when the dict already has the field or *fn* returns ``None``.

    Config:
        input_key: State key holding the extracted dict.
        field: Dict field to fill when empty.
        fn: Callable ``fn(state) -> value | None``.
    """

    type = "fallback"

    def __init__(
        self,
        config: dict | None = None,
        *,
        input_key: str = "output",
        field: str = "",
        fn: Callable | None = None,
        **kwargs,
    ):
        merged = {
            "input_key": input_key,
            "field": field,
            "fn": fn,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        cfg = self.config
        input_key = cfg.get("input_key", "output")
        field = cfg.get("field")
        fn = cfg.get("fn")
        if not field or not callable(fn):
            return {}
        data = state.get(input_key)
        if not isinstance(data, dict) or data.get(field):
            return {}
        value = fn(state)
        if value is None:
            return {}
        return {input_key: {**data, field: value}}

GraphInterrupt

Bases: TeffError

Raised by graph.run() when a workflow pauses for human input.

Attributes:

Name	Type	Description
key		State key the resume value will be written to.
prompt		Human-readable question shown to the operator.
node_id		Id of the interrupt node that paused execution.
checkpoint_id		Pass this back to graph.run() with the same checkpointer together with resume to continue.
nested_checkpoint_id	str | None	When the interrupt fired inside a :class:~teff.flow.sub_flow.SubFlow, the checkpoint id the sub-flow paused under. Resuming routes back into the sub-flow instead of continuing past it.

Source code in teff/node/interrupt.py

class GraphInterrupt(TeffError):
    """Raised by ``graph.run()`` when a workflow pauses for human input.

    Attributes:
        key: State key the resume value will be written to.
        prompt: Human-readable question shown to the operator.
        node_id: Id of the interrupt node that paused execution.
        checkpoint_id: Pass this back to ``graph.run()`` with the same
            checkpointer together with ``resume`` to continue.
        nested_checkpoint_id: When the interrupt fired inside a
            :class:`~teff.flow.sub_flow.SubFlow`, the checkpoint id the
            sub-flow paused under.  Resuming routes back into the sub-flow
            instead of continuing past it.
    """

    def __init__(
        self,
        key: str,
        prompt: str = "",
        node_id: str | None = None,
        checkpoint_id: str | None = None,
    ):
        super().__init__(f"workflow paused for input: {prompt or key}")
        self.key = key
        self.prompt = prompt
        self.node_id = node_id
        self.checkpoint_id = checkpoint_id
        self.nested_checkpoint_id: str | None = None

Interrupt

Bases: Node

Pause the workflow and wait for external (human) input.

When execution reaches this node, graph.run() saves a checkpoint and raises :class:GraphInterrupt. The operator provides a value and the graph is resumed with the same checkpoint_id and resume::

try:
    await graph.run(state, checkpointer=cp, checkpoint_id="run-1")
except GraphInterrupt as interrupt:
    print(interrupt.prompt)
    value = input("> ")
    await graph.run(
        state, checkpointer=cp, checkpoint_id="run-1", resume=value
    )

The resumed value is written to the state under key before continuing with the node that follows this one.

Requires a checkpointer to be set on graph.run().

Config

key: State key that receives the resume value. prompt: Human-readable question for the operator.

Source code in teff/node/interrupt.py

class Interrupt(Node):
    """Pause the workflow and wait for external (human) input.

    When execution reaches this node, ``graph.run()`` saves a checkpoint
    and raises :class:`GraphInterrupt`.  The operator provides a value
    and the graph is resumed with the same ``checkpoint_id`` and
    ``resume``::

        try:
            await graph.run(state, checkpointer=cp, checkpoint_id="run-1")
        except GraphInterrupt as interrupt:
            print(interrupt.prompt)
            value = input("> ")
            await graph.run(
                state, checkpointer=cp, checkpoint_id="run-1", resume=value
            )

    The resumed value is written to the state under *key* before
    continuing with the node that follows this one.

    Requires a checkpointer to be set on ``graph.run()``.

    Config:
        key: State key that receives the resume value.
        prompt: Human-readable question for the operator.
    """

    type = "interrupt"

    async def execute(self, ctx, state: dict) -> dict:
        prompt = self.config.get("prompt", "")
        if "{" in prompt:
            prompt = render_template(prompt, state)
        raise GraphInterrupt(
            self.config.get("key", ""),
            prompt,
        )

LLM

Bases: Node

Call an LLM chat API with tool calling and structured output.

Parameters:

Name	Type	Description	Default
model	str | None	Model name (e.g. gpt-4, llama3.1:8b).	None
system	str	System prompt. Supports {key} placeholders rendered from state (see :func:teff.prompt.render_template).	''
prompt	str | None	User prompt template. Supports {key} placeholders rendered from state, e.g. "create a repair plan for {type} " "up to {summ}". Overrides input_key when set.	None
input_key	str | None	State key for user message (default: whole state).	None
output_key	str	State key for the response (default "output").	'output'
provider	str | None	Provider name ("openai", "ollama", etc.). Falls back to the graph-level default (Graph(default_provider=...) / workflow default_provider:) when unset.	None
use_tools	bool	Tool capability for the node: a list of names restricts it to exactly those tools; "all" uses every ctx.tools entry. None/[] (default) — no tools are surfaced.	False
temperature	float | None	Sampling temperature.	None
max_tokens	int | None	Max tokens in response.	None
response_format	dict | None	{"type": "json_object"} etc.	None
stream	bool	If True, use SSE streaming. Automatically disabled when tool calling is active.	False
on_token	Callable[[str], None] | None	Optional callback (token: str) -> None for streaming.	None
json_schema	dict | None	JSON Schema dict describing the expected response. When set, the response is parsed as JSON, validated against the schema, and re-asked (with the validation error fed back) up to max_retries times. The parsed object is stored under output_key. Adds response_format: {"type": "json_object"} for OpenAI-compatible providers (format: "json" for Ollama) unless response_format is already set.	None
output_type	Type[Any] | None	Python type spec — a TypedDict, dataclass, or dict[str, type] field map — converted to a JSON Schema. Alternative to json_schema.	None
parse	bool	If True without a schema, parse the response as a JSON object and store the dict under output_key (no validation).	False
max_retries	int	How many times to re-ask after a validation failure.	2
tool_timeout	float | None	Per-tool execution timeout in seconds.	None
tool_retries	int	Extra attempts per tool call after a failure.	0
tool_approval	Any	Gate on tool execution — "auto" (default), "deny", or a callable (name, args) -> bool | str (sync or async). "pause" is treated as "deny" in the internal multi-round loop (use a :class:~teff.node.agent.ToolExec node for pause/resume).	None
http_max_retries	int	HTTP request retries (429/5xx/timeouts).	2
fallbacks	list[str] | None	Fallback model names for provider failover.	None
base_url	str | None	Custom base URL (overrides provider default).	None
chat_path	str | None	Custom API path (overrides provider default).	None
auth_header	str | None	Custom auth header name.	None
auth_prefix	str | None	Custom auth header prefix.	None
api_key_env	str | None	Custom env var for API key.	None
tools	list[dict] | None	List of raw tool definition dicts.	None
messages_key	str | None	State key for message history. If set, the conversation history is read/written from/to state[messages_key] instead of being built fresh each call.	None
memory	MemoryConfig | dict | None	Optional long-term memory injection — a :class:~teff.memory.context.MemoryConfig or {store, namespace, k, header} (same as :class:~teff.node.agent.ReActAgent). Recalled memories for the most recent user message are prepended to the call as a system message.	None
response_path	str	Dot-separated path to extract content from response.	''
skills	list | None	Skills to mount on this call — a :class:~teff.skill.Skill, a path to a skill folder/SKILL.md, or a name resolved against skill_dir. Their instructions are merged into the system prompt and their allowed-tools/disallowed-tools narrow the visible tools.	None
skill_dir	str	Directory to resolve bare skill names from (default "skills").	'skills'

Source code in teff/node/llm.py

class LLM(Node):
    """Call an LLM chat API with tool calling and structured output.

    Parameters:
        model: Model name (e.g. ``gpt-4``, ``llama3.1:8b``).
        system: System prompt.  Supports ``{key}`` placeholders rendered
            from state (see :func:`teff.prompt.render_template`).
        prompt: User prompt template.  Supports ``{key}`` placeholders
            rendered from state, e.g. ``"create a repair plan for {type} "
            "up to {summ}"``.  Overrides *input_key* when set.
        input_key: State key for user message (default: whole state).
        output_key: State key for the response (default ``"output"``).
        provider: Provider name (``"openai"``, ``"ollama"``, etc.).
            Falls back to the graph-level default (``Graph(default_provider=...)``
            / workflow ``default_provider:``) when unset.
        use_tools: Tool capability for the node: a list of names restricts
            it to exactly those tools; ``"all"`` uses every ``ctx.tools``
            entry.  ``None``/``[]`` (default) — no tools are surfaced.
        temperature: Sampling temperature.
        max_tokens: Max tokens in response.
        response_format: ``{"type": "json_object"}`` etc.
        stream: If ``True``, use SSE streaming.
            Automatically disabled when tool calling is active.
        on_token: Optional callback ``(token: str) -> None`` for streaming.
        json_schema: JSON Schema dict describing the expected response.
            When set, the response is parsed as JSON, validated against
            the schema, and re-asked (with the validation error fed back)
            up to *max_retries* times.  The parsed object is stored under
            *output_key*.  Adds ``response_format: {"type": "json_object"}``
            for OpenAI-compatible providers (``format: "json"`` for Ollama)
            unless *response_format* is already set.
        output_type: Python type spec — a ``TypedDict``, dataclass, or
            ``dict[str, type]`` field map — converted to a JSON Schema.
            Alternative to *json_schema*.
        parse: If ``True`` without a schema, parse the response as a JSON
            object and store the dict under *output_key* (no validation).
        max_retries: How many times to re-ask after a validation failure.
        tool_timeout: Per-tool execution timeout in seconds.
        tool_retries: Extra attempts per tool call after a failure.
        tool_approval: Gate on tool execution — ``"auto"`` (default),
            ``"deny"``, or a callable ``(name, args) -> bool | str``
            (sync or async).  ``"pause"`` is treated as ``"deny"`` in the
            internal multi-round loop (use a :class:`~teff.node.agent.ToolExec`
            node for pause/resume).
        http_max_retries: HTTP request retries (429/5xx/timeouts).
        fallbacks: Fallback model names for provider failover.
        base_url: Custom base URL (overrides provider default).
        chat_path: Custom API path (overrides provider default).
        auth_header: Custom auth header name.
        auth_prefix: Custom auth header prefix.
        api_key_env: Custom env var for API key.
        tools: List of raw tool definition dicts.
        messages_key: State key for message history.
            If set, the conversation history is read/written from/to
            ``state[messages_key]`` instead of being built fresh each call.
        memory: Optional long-term memory injection — a
            :class:`~teff.memory.context.MemoryConfig` or ``{store,
            namespace, k, header}`` (same as
            :class:`~teff.node.agent.ReActAgent`). Recalled memories for
            the most recent user message are prepended to the call as a
            system message.
        response_path: Dot-separated path to extract content from response.
        skills: Skills to mount on this call — a :class:`~teff.skill.Skill`,
            a path to a skill folder/``SKILL.md``, or a name resolved
            against *skill_dir*.  Their instructions are merged into the
            system prompt and their ``allowed-tools``/``disallowed-tools``
            narrow the visible tools.
        skill_dir: Directory to resolve bare skill names from
            (default ``"skills"``).
    """

    type = "llm_chat"
    _MAX_TOOL_ROUNDS = 10

    def __init__(
        self,
        config: dict | None = None,
        *,
        model: str | None = None,
        system: str = "",
        prompt: str | None = None,
        input_key: str | None = None,
        output_key: str = "output",
        provider: str | None = None,
        use_tools: bool = False,
        temperature: float | None = None,
        max_tokens: int | None = None,
        response_format: dict | None = None,
        stream: bool = False,
        on_token: typing.Callable[[str], None] | None = None,
        json_schema: dict | None = None,
        output_type: typing.Type[typing.Any] | None = None,
        parse: bool = False,
        max_retries: int = 2,
        tool_timeout: float | None = None,
        tool_retries: int = 0,
        tool_approval: typing.Any = None,
        http_max_retries: int = 2,
        fallbacks: list[str] | None = None,
        base_url: str | None = None,
        chat_path: str | None = None,
        auth_header: str | None = None,
        auth_prefix: str | None = None,
        api_key_env: str | None = None,
        tools: list[dict] | None = None,
        messages_key: str | None = None,
        response_path: str = "",
        skills: list | None = None,
        skill_dir: str = "skills",
        memory: MemoryConfig | dict | None = None,
        **kwargs: typing.Any,
    ):
        merged = {
            "model": model,
            "system": system,
            "prompt": prompt,
            "input_key": input_key,
            "output_key": output_key,
            "provider": provider,
            "use_tools": use_tools,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "response_format": response_format,
            "stream": stream,
            "on_token": on_token,
            "json_schema": json_schema,
            "output_type": output_type,
            "parse": parse,
            "max_retries": max_retries,
            "tool_timeout": tool_timeout,
            "tool_retries": tool_retries,
            "tool_approval": tool_approval,
            "http_max_retries": http_max_retries,
            "fallbacks": fallbacks,
            "base_url": base_url,
            "chat_path": chat_path,
            "auth_header": auth_header,
            "auth_prefix": auth_prefix,
            "api_key_env": api_key_env,
            "messages_key": messages_key,
            "response_path": response_path,
            "skills": skills,
            "skill_dir": skill_dir,
            "memory": memory,
            **(config or {}),
            **kwargs,
        }
        # ensure tools is always a list
        merged.setdefault("tools", tools or [])
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        cfg = self.config

        skills = resolve_skills(cfg)
        skill_text = skills_instructions(skills)

        has_messages_key = cfg.get("messages_key") and state.get(cfg["messages_key"])
        if has_messages_key:
            messages = list(state[cfg["messages_key"]])
            system = render_template(cfg.get("system", ""), state)
            if skill_text:
                system = f"{system}\n\n{skill_text}" if system else skill_text
            # a conversation node injects its system prompt on every turn,
            # but must not duplicate one already persisted in history
            if system and not any(m.get("role") == "system" for m in messages):
                messages.insert(0, {"role": "system", "content": system})
        else:
            prompt = cfg.get("prompt")
            input_key = cfg.get("input_key")
            if prompt:
                user_message = render_template(prompt, state)
            elif input_key:
                user_message = str(state.get(input_key, ""))
            else:
                user_message = str(state)
            messages = []
            system = render_template(cfg.get("system", ""), state)
            if skill_text:
                system = f"{system}\n\n{skill_text}" if system else skill_text
            if system:
                messages.append({"role": "system", "content": system})
            if user_message:
                messages.append({"role": "user", "content": user_message})

        from teff.memory.context import memory_context_from_config

        memory_block = await memory_context_from_config(cfg, state=state, ctx=ctx)
        if memory_block:
            messages.insert(0, {"role": "system", "content": memory_block})

        tool_defs: list[dict] = list(cfg.get("tools", []))
        if cfg.get("use_tools", False):
            scoped_tools = scope_tools(ctx.tools, cfg, skills)
            for t in scoped_tools.values():
                tool_defs.append(tool_to_schema(t))
        else:
            scoped_tools = dict(ctx.tools)

        has_tools = bool(tool_defs)
        output_key = cfg.get("output_key", "output")

        schema = self._resolve_schema(cfg)
        structured = schema is not None
        parse_only = bool(cfg.get("parse", False)) and not structured

        harness = Harness.from_config(
            cfg,
            default_provider=getattr(ctx, "default_provider", None),
            default_model=getattr(ctx, "default_model", None),
            providers=getattr(ctx, "providers", None),
        )
        if cfg.get("http_max_retries") is not None:
            harness.max_retries = int(cfg.get("http_max_retries", 2))
        if cfg.get("fallbacks") is not None:
            harness.fallbacks = [str(f) for f in cfg["fallbacks"]]
        provider_key = harness.provider_key
        harness.on_llm = self._record_llm_cb(ctx, cfg, provider_key)
        payload_sink = getattr(ctx, "on_llm_payload", None)
        if payload_sink is not None:
            harness.on_llm_payload = payload_sink

        if structured and not cfg.get("response_format"):
            if harness.type == "ollama":
                harness._body_extra["format"] = "json"
            else:
                harness._body_extra["response_format"] = {"type": "json_object"}

        graph_stream = getattr(ctx, "emit", None) is not None
        content: str | dict = ""
        if (
            (cfg.get("stream", False) or graph_stream)
            and not has_tools
            and not structured
        ):
            harness.on_token = self._token_sink(ctx, cfg, provider_key)
            content = (await harness.call(messages, stream=True)).content
        else:
            max_retries = int(cfg.get("max_retries", 2))
            rounds = harness.max_rounds if has_tools else 1
            if structured:
                rounds = max(rounds, max_retries + 1)
            attempts = 0
            for _round in range(rounds):
                reply = await harness.call(
                    messages,
                    tools=tool_defs or None,
                    content_path=cfg.get("response_path", ""),
                )
                content = reply.content
                msg = reply.message
                tool_calls = msg.get("tool_calls")

                if has_tools and not tool_calls and harness.parse_text_tool_calls:
                    tool_calls, msg = normalize_text_tool_calls(
                        content, msg, seq=len(messages)
                    )

                if has_tools and tool_calls:
                    messages.append(msg)
                    tool_timeout = _opt_float_cfg(cfg.get("tool_timeout"))
                    tool_retries = int(cfg.get("tool_retries", 0))
                    results = await execute_tool_calls(
                        tool_calls,
                        scoped_tools,
                        harness.tool_error_mode,
                        tool_timeout,
                        tool_retries,
                        cfg.get("tool_approval"),
                    )
                    for tc, res in zip(tool_calls, results):
                        messages.append(
                            {
                                "role": "tool",
                                "tool_call_id": tc["id"],
                                "content": res,
                            }
                        )
                    continue

                if structured:
                    assert schema is not None
                    assert isinstance(content, str)
                    parsed_value, error = self._parse_structured(content, schema)
                    if error is None:
                        content = parsed_value
                        break
                    attempts += 1
                    await self._record_structured(ctx, cfg, schema, error, attempts)
                    if attempts > max_retries:
                        raise StructuredOutputError(
                            schema=schema,
                            content=content,
                            errors=error,
                            attempts=attempts,
                        )
                    messages.append({"role": "assistant", "content": content})
                    messages.append(
                        {
                            "role": "user",
                            "content": (
                                "Your previous response failed JSON schema "
                                f"validation: {error}\n"
                                "Respond with a single JSON object conforming "
                                f"to this schema:\n{json.dumps(schema)}"
                            ),
                        }
                    )
                    continue

                if parse_only:
                    raw = content
                    assert isinstance(raw, str)
                    try:
                        parsed = parse_json_object(raw)
                    except ValueError as exc:
                        raise StructuredOutputError(
                            content=raw,
                            errors=str(exc),
                            attempts=1,
                        ) from exc
                    content = parsed
                break

        if isinstance(content, dict):
            return {output_key: content}
        return {output_key: content or ""}

    def _record_llm_cb(self, ctx, cfg: dict, provider_key: str):
        """Build the ``on_llm`` callback recording usage + ``llm`` events."""

        async def record(
            provider: str, model: str, prompt: int, completion: int, duration: float
        ) -> None:
            tracer = getattr(ctx, "tracer", None)
            if tracer is not None:
                tracer.llm(provider, model, prompt, completion, duration)
            emit = getattr(ctx, "emit", None)
            if emit is not None:
                await emit(
                    StreamEvent(
                        "llm",
                        node_id=ctx.node_id,
                        node_type=ctx.node_type,
                        data={
                            "provider": provider,
                            "model": model,
                            "prompt_tokens": prompt,
                            "completion_tokens": completion,
                            "duration_ms": duration,
                        },
                    )
                )

        return record

    def _resolve_schema(self, cfg: dict) -> dict | None:
        """Return the JSON Schema for structured output, if configured."""
        if cfg.get("json_schema") is not None:
            return json_schema_from_type(cfg["json_schema"])
        if cfg.get("output_type") is not None:
            return json_schema_from_type(cfg["output_type"])
        return None

    def _parse_structured(
        self, content: str, schema: dict
    ) -> tuple[typing.Any, str | None]:
        """Parse *content* as JSON and validate it against *schema*.

        Returns:
            ``(value, None)`` on success, ``(None, error_message)`` otherwise.
        """
        try:
            value = parse_json_object(content)
        except ValueError as exc:
            return None, str(exc)
        errors = validate_json(value, schema)
        if errors:
            return None, "; ".join(errors)
        return value, None

    async def _record_structured(
        self, ctx, cfg: dict, schema: dict, errors: str, attempt: int
    ) -> None:
        """Record a structured-output validation failure."""
        tracer = getattr(ctx, "tracer", None)
        if tracer is not None:
            tracer.structured(ctx.node_id, ctx.node_type, errors, attempt)
        emit = getattr(ctx, "emit", None)
        if emit is not None:
            await emit(
                StreamEvent(
                    "structured",
                    node_id=ctx.node_id,
                    node_type=ctx.node_type,
                    data={"errors": errors, "attempt": attempt},
                )
            )

    def _token_sink(
        self, ctx, cfg: dict, provider_key: str
    ) -> typing.Callable[[str], typing.Any]:
        """Build the per-token callback for streaming.

        Forwards each token to the node's ``on_token`` config and, when
        running under ``graph.stream()``, emits a ``token`` :class:`StreamEvent`.
        """
        emit = getattr(ctx, "emit", None)
        on_token = cfg.get("on_token")

        async def sink(token: str) -> None:
            if on_token is not None:
                on_token(token)
            if emit is not None:
                await emit(
                    StreamEvent(
                        "token",
                        node_id=ctx.node_id,
                        node_type=ctx.node_type,
                        data={
                            "token": token,
                            "provider": provider_key,
                            "model": str(cfg.get("model", "")),
                        },
                    )
                )

        return sink

    @staticmethod
    def _tool_to_schema(tool: Tool) -> dict:
        """Alias of :func:`teff.harness.tool_to_schema` (backward compat)."""
        return tool_to_schema(tool)

Loop

Bases: Node

Repeat a body chain until state[key] equals until.

This is the self-contained sibling of :meth:Flow.loop <teff.flow.Flow.loop>: instead of wiring decider/done/body chains together with condition edges, the whole repeat lives inside one node, so a loop is expressible directly in YAML::

- id: refine
  type: loop
  config:
    key: approved
    until: "yes"
    max_rounds: 3
    body:
      - {type: transform, config: {action: value, value: "no", output_key: approved}}

Each round runs the body chain (a single node or a list), then evaluates the condition key=until against the merged state using the same expression language as edges: conditions (so until: "yes" matches "Yes" or "yes."). When the condition holds the loop stops; the body still runs at least once even if the condition already held on entry, which matches the Flow-loop contract where a decider writes key and then the body decides whether to re-run.

max_rounds (default 10) bounds the repetition so a body that never reaches until cannot hang the workflow.

Parameters:

Name	Type	Description	Default
body	Node | list[Node] | dict | list[dict]	Node or list of Nodes (or their declarative dict specs) run per round, sequentially.	required
key	str	State key the condition reads.	''
until	str	Value of key that stops the loop.	''
max_rounds	int	Maximum number of body rounds before giving up.	10

Source code in teff/node/loop.py

class Loop(Node):
    """Repeat a *body* chain until ``state[key]`` equals *until*.

    This is the self-contained sibling of :meth:`Flow.loop <teff.flow.Flow.loop>`:
    instead of wiring decider/done/body chains together with condition edges,
    the whole repeat lives inside one node, so a loop is expressible directly
    in YAML::

        - id: refine
          type: loop
          config:
            key: approved
            until: "yes"
            max_rounds: 3
            body:
              - {type: transform, config: {action: value, value: "no", output_key: approved}}

    Each round runs the *body* chain (a single node or a list), then evaluates
    the condition ``key=until`` against the merged state using the same
    expression language as ``edges:`` conditions (so ``until: "yes"`` matches
    ``"Yes"`` or ``"yes."``).  When the condition holds the loop stops; the body
    still runs at least once even if the condition already held on entry, which
    matches the Flow-loop contract where a decider writes *key* and then the
    body decides whether to re-run.

    ``max_rounds`` (default 10) bounds the repetition so a body that never
    reaches *until* cannot hang the workflow.

    Args:
        body: Node or list of Nodes (or their declarative dict specs) run per
            round, sequentially.
        key: State key the condition reads.
        until: Value of *key* that stops the loop.
        max_rounds: Maximum number of body rounds before giving up.
    """

    type = "loop"

    def __init__(
        self,
        body: Node | list[Node] | dict | list[dict],
        *,
        key: str = "",
        until: str = "",
        max_rounds: int = 10,
        config: dict | None = None,
        **kwargs,
    ):
        merged = {
            "key": key,
            "until": until,
            "max_rounds": max_rounds,
            "body": body,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)
        self._body = self._build_body(body)

    @staticmethod
    def _build_body(body: Node | list[Node] | dict | list[dict]) -> list[Node]:
        from teff.node.registry import default_registry

        def resolve(spec: Node | dict) -> Node:
            if isinstance(spec, Node):
                return spec
            if isinstance(spec, dict):
                spec = dict(spec)
                stype = spec.pop("type")
                cfg = spec.pop("config", None)
                if cfg is not None:
                    if not isinstance(cfg, dict):
                        msg = f"invalid loop body spec: {spec!r}"
                        raise TypeError(msg)
                    cfg = {**spec, **dict(cfg)}
                else:
                    cfg = spec
                return default_registry.create(stype, cfg)
            msg = f"invalid loop body spec: {spec!r}"
            raise TypeError(msg)

        if isinstance(body, list):
            return [resolve(spec) for spec in body]
        return [resolve(body)]

    async def execute(self, ctx: ExecContext | None, state: dict) -> dict:
        from teff.graph.conditions import evaluate

        key = self.config.get("key", "")
        until = self.config.get("until", "")
        max_rounds = int(self.config.get("max_rounds") or 10)
        if not key:
            raise ValueError("loop requires config.key")
        condition = f"{key}={until}"
        if ctx is None:
            ctx = ExecContext(state, {})
        reducers = getattr(ctx, "reducers", None) or {}

        for _ in range(max_rounds):
            for node_idx, node in enumerate(self._body):
                node_id = f"{ctx.node_id or self.type}.loop.{node_idx}"
                node_ctx = ExecContext(
                    state,
                    ctx.tools,
                    node_id=node_id,
                    node_type=node.type,
                    tracer=ctx.tracer,
                    reducers=reducers,
                    providers=getattr(ctx, "providers", None),
                    default_provider=getattr(ctx, "default_provider", None),
                    on_llm_payload=getattr(ctx, "on_llm_payload", None),
                )
                start = time.monotonic()
                if ctx.tracer is not None:
                    ctx.tracer.node_start(node_id, node.type)
                try:
                    result = await node.execute(node_ctx, state) or {}
                except Exception as exc:
                    if ctx.tracer is not None:
                        ctx.tracer.node_error(node_id, node.type, _ms(start), exc)
                    raise
                if ctx.tracer is not None:
                    ctx.tracer.node_end(node_id, node.type, _ms(start))
                apply_reducers(state, as_updates(result), reducers)
            if evaluate(condition, state):
                break

        return {}

Map

Bases: Node

Run a processor over each item of a state list, in parallel.

Reads one or more lists from input_keys, splits them into chunks, and runs the processor (a single node or a chain) concurrently on each chunk — with the chunk placed back under its key in an isolated state copy. The per-chunk result is gathered into a list stored at output_key, preserving the order of the source lists.

With several input_keys the lists are zipped: chunk i contains key[0][i], key[1][i], etc., so the processor can read multiple per-item values straight from state.

This is the dynamic sibling of :class:Parallel: branches are derived from data at runtime instead of being declared up front.

Parameters:

Name	Type	Description	Default
processor	Node | list[Node] | dict | list[dict]	Node or list of Nodes run per chunk (sequentially).	required
input_keys	str | list[str]	One or more state keys holding the lists to fan out. The processor reads these same keys from the branch state.	''
output_key	str	State key that receives the list of per-chunk results.	''
result_key	str | None	State key holding each chunk's result. Defaults to the processor's own output_key (single-node processors); pass it explicitly for multi-node chains.	None
chunk_size	int | None	Items per branch (default 1 = one item per branch).	None
max_concurrency	int | None	Limit on simultaneously running branches (default None = no limit).	None

Usage::

node = Map(
    processor=LLM(model="llama3.1:8b",
                  input_key="chunk", output_key="summary"),
    input_keys=["chunks"],
    output_key="summaries",
    chunk_size=4,
    max_concurrency=2,
)

Source code in teff/node/map.py

class Map(Node):
    """Run a processor over each item of a state list, in parallel.

    Reads one or more lists from *input_keys*, splits them into chunks,
    and runs the *processor* (a single node or a chain) concurrently on
    each chunk — with the chunk placed back under its key in an isolated
    state copy.  The per-chunk result is gathered into a list stored at
    *output_key*, preserving the order of the source lists.

    With several *input_keys* the lists are zipped: chunk ``i`` contains
    ``key[0][i]``, ``key[1][i]``, etc., so the processor can read
    multiple per-item values straight from state.

    This is the *dynamic* sibling of :class:`Parallel`: branches are
    derived from data at runtime instead of being declared up front.

    Args:
        processor: Node or list of Nodes run per chunk (sequentially).
        input_keys: One or more state keys holding the lists to fan out.
            The processor reads these same keys from the branch state.
        output_key: State key that receives the list of per-chunk results.
        result_key: State key holding each chunk's result.  Defaults to
            the processor's own ``output_key`` (single-node processors);
            pass it explicitly for multi-node chains.
        chunk_size: Items per branch (default 1 = one item per branch).
        max_concurrency: Limit on simultaneously running branches
            (default ``None`` = no limit).

    Usage::

        node = Map(
            processor=LLM(model="llama3.1:8b",
                          input_key="chunk", output_key="summary"),
            input_keys=["chunks"],
            output_key="summaries",
            chunk_size=4,
            max_concurrency=2,
        )
    """

    type = "map"

    def __init__(
        self,
        processor: Node | list[Node] | dict | list[dict],
        *,
        input_keys: str | list[str] = "",
        output_key: str = "",
        result_key: str | None = None,
        chunk_size: int | None = None,
        max_concurrency: int | None = None,
        config: dict | None = None,
        **kwargs,
    ):
        keys = [input_keys] if isinstance(input_keys, str) else list(input_keys)
        merged = {
            "input_keys": keys,
            "output_key": output_key,
            "result_key": result_key,
            "chunk_size": chunk_size,
            "max_concurrency": max_concurrency,
            "processor": processor,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)
        self._processor = self._build_processor(processor)

    @staticmethod
    def _build_processor(
        processor: Node | list[Node] | dict | list[dict],
    ) -> list[Node]:
        from teff.node.registry import default_registry

        def resolve(spec: Node | dict) -> Node:
            if isinstance(spec, Node):
                return spec
            if isinstance(spec, dict):
                spec = dict(spec)
                stype = spec.pop("type")
                cfg = spec.pop("config", None)
                if cfg is not None:
                    if not isinstance(cfg, dict):
                        msg = f"invalid processor spec: {spec!r}"
                        raise TypeError(msg)
                    cfg = {**spec, **dict(cfg)}
                else:
                    cfg = spec
                return default_registry.create(stype, cfg)
            msg = f"invalid processor spec: {spec!r}"
            raise TypeError(msg)

        if isinstance(processor, list):
            return [resolve(spec) for spec in processor]
        return [resolve(processor)]

    async def execute(self, ctx: ExecContext, state: dict) -> dict:
        input_keys = self.config.get("input_keys", [])
        output_key = self.config.get("output_key", "")
        result_key = self._resolve_result_key(output_key)
        chunk_size = self.config.get("chunk_size") or 1
        max_concurrency = self.config.get("max_concurrency")

        lists = [state.get(key, []) for key in input_keys]
        for key, items in zip(input_keys, lists):
            if not isinstance(items, Sequence) or isinstance(items, (str, bytes)):
                msg = f"map input_key '{key}' is not a list"
                raise TypeError(msg)
        if not lists or not lists[0]:
            return {output_key: []}

        length = len(lists[0])
        for key, items in zip(input_keys, lists):
            if len(items) != length:
                msg = (
                    f"map input_keys length mismatch: '{key}' has "
                    f"{len(items)} items, expected {length}"
                )
                raise ValueError(msg)

        chunks = [
            [items[i : i + chunk_size] for i in range(0, length, chunk_size)]
            for items in lists
        ]
        groups = list(zip(*chunks))

        semaphore = asyncio.Semaphore(max_concurrency) if max_concurrency else None
        reducers = getattr(ctx, "reducers", None) or {}

        async def run_group(group: tuple, idx: int) -> object:
            if semaphore is not None:
                async with semaphore:
                    return await self._run_chunk(
                        group,
                        idx,
                        ctx,
                        state,
                        reducers,
                        input_keys,
                        result_key,
                        chunk_size,
                    )
            return await self._run_chunk(
                group, idx, ctx, state, reducers, input_keys, result_key, chunk_size
            )

        results = await gather_or_cancel(
            *(run_group(group, idx) for idx, group in enumerate(groups))
        )
        return {output_key: list(results)}

    def _resolve_result_key(self, output_key: str) -> str:
        configured = self.config.get("result_key")
        if configured:
            return configured
        if len(self._processor) == 1:
            node_key = self._processor[0].config.get("output_key")
            if node_key:
                return node_key
        return output_key

    async def _run_chunk(
        self,
        group: tuple,
        chunk_idx: int,
        ctx: ExecContext,
        state: dict,
        reducers: dict,
        input_keys: list[str],
        result_key: str,
        chunk_size: int,
    ) -> object:
        branch_state = dict(state)
        for key, chunk in zip(input_keys, group):
            branch_state[key] = chunk[0] if chunk_size == 1 else chunk
        for node_idx, node in enumerate(self._processor):
            node_id = f"{ctx.node_id or self.type}.m{chunk_idx}.{node_idx}"
            node_ctx = ExecContext(
                branch_state,
                ctx.tools,
                node_id=node_id,
                node_type=node.type,
                tracer=ctx.tracer,
                reducers=reducers,
                providers=getattr(ctx, "providers", None),
                default_provider=getattr(ctx, "default_provider", None),
                default_model=getattr(ctx, "default_model", None),
                on_llm_payload=getattr(ctx, "on_llm_payload", None),
            )
            start = time.monotonic()
            if ctx.tracer is not None:
                ctx.tracer.node_start(node_id, node.type)
            try:
                result = await node.execute(node_ctx, branch_state) or {}
            except Exception as exc:
                if ctx.tracer is not None:
                    ctx.tracer.node_error(node_id, node.type, _ms(start), exc)
                raise
            if ctx.tracer is not None:
                ctx.tracer.node_end(node_id, node.type, _ms(start))
            apply_reducers(branch_state, as_updates(result), reducers)
        return branch_state.get(result_key)

Node

Bases: ABC

Abstract base class for all graph nodes.

Subclasses must set type and implement execute.

Attributes:

Name	Type	Description
type	str	Unique node type identifier used for registry lookups.
config		Configuration dict (merged from constructor kwargs).

Methods:

Name	Description
execute	Execute the node's logic.

Source code in teff/node/node.py

class Node(ABC):
    """Abstract base class for all graph nodes.

    Subclasses must set *type* and implement *execute*.

    Attributes:
        type: Unique node type identifier used for registry lookups.
        config: Configuration dict (merged from constructor kwargs).
    """

    type: str = ""

    def __init__(self, config: dict | None = None, **kwargs):
        self.config = {**(config or {}), **kwargs}

    @abstractmethod
    async def execute(self, ctx: typing.Any, state: dict) -> "dict | Command":
        """Execute the node's logic.

        Args:
            ctx: Execution context providing tool/LLM access.
            state: Current workflow state dict (shallow-merge in/out).

        Returns:
            State updates to shallow-merge into the workflow state, or a
            :class:`~teff.node.Command` that additionally routes the graph
            to a specific next node.
        """

execute abstractmethod async

execute(ctx, state)

Execute the node's logic.

Parameters:

Name	Type	Description	Default
ctx	Any	Execution context providing tool/LLM access.	required
state	dict	Current workflow state dict (shallow-merge in/out).	required

Returns:

Type	Description
dict | Command	State updates to shallow-merge into the workflow state, or a
dict | Command	class:~teff.node.Command that additionally routes the graph
dict | Command	to a specific next node.

Source code in teff/node/node.py

@abstractmethod
async def execute(self, ctx: typing.Any, state: dict) -> "dict | Command":
    """Execute the node's logic.

    Args:
        ctx: Execution context providing tool/LLM access.
        state: Current workflow state dict (shallow-merge in/out).

    Returns:
        State updates to shallow-merge into the workflow state, or a
        :class:`~teff.node.Command` that additionally routes the graph
        to a specific next node.
    """

NodeRegistry

Registry mapping node type names to factory functions.

Used by the YAML loader and pipeline compiler to instantiate nodes by their string type identifier.

Methods:

Name	Description
copy	Return a shallow copy with the same factory registrations.
create	Create a node instance by type name.
list	Return all registered node type names.
register	Register a node factory under a type name.

Source code in teff/node/registry.py

class NodeRegistry:
    """Registry mapping node type names to factory functions.

    Used by the YAML loader and pipeline compiler to instantiate
    nodes by their string type identifier.
    """

    def __init__(self) -> None:
        self._factories: dict[str, NodeFactory] = {}

    def register(self, name: str, factory: NodeFactory) -> None:
        """Register a node factory under a type name."""
        self._factories[name] = factory

    def create(self, name: str, config: dict | None = None, **kwargs: Any) -> Node:
        """Create a node instance by type name.

        Args:
            name: Registered node type name.
            config: Optional configuration dict (backward-compatible).
            **kwargs: Additional keyword arguments merged into config.

        Returns:
            A Node instance.

        Raises:
            ConfigError: If the type name is not registered
                (also a ``KeyError``).
        """
        if name not in self._factories:
            msg = f"unknown node type: {name}"
            raise ConfigError(msg)
        merged = {**(config or {}), **kwargs}
        return self._factories[name](merged)

    def list(self) -> list[str]:
        """Return all registered node type names."""
        return list(self._factories.keys())

    def copy(self) -> "NodeRegistry":
        """Return a shallow copy with the same factory registrations."""
        reg = NodeRegistry()
        reg._factories = dict(self._factories)
        return reg

copy

copy()

Return a shallow copy with the same factory registrations.

Source code in teff/node/registry.py

def copy(self) -> "NodeRegistry":
    """Return a shallow copy with the same factory registrations."""
    reg = NodeRegistry()
    reg._factories = dict(self._factories)
    return reg

create

create(name, config=None, **kwargs)

Create a node instance by type name.

Parameters:

Name	Type	Description	Default
name	str	Registered node type name.	required
config	dict | None	Optional configuration dict (backward-compatible).	None
**kwargs	Any	Additional keyword arguments merged into config.	{}

Returns:

Type	Description
Node	A Node instance.

Raises:

Type	Description
ConfigError	If the type name is not registered (also a KeyError).

Source code in teff/node/registry.py

def create(self, name: str, config: dict | None = None, **kwargs: Any) -> Node:
    """Create a node instance by type name.

    Args:
        name: Registered node type name.
        config: Optional configuration dict (backward-compatible).
        **kwargs: Additional keyword arguments merged into config.

    Returns:
        A Node instance.

    Raises:
        ConfigError: If the type name is not registered
            (also a ``KeyError``).
    """
    if name not in self._factories:
        msg = f"unknown node type: {name}"
        raise ConfigError(msg)
    merged = {**(config or {}), **kwargs}
    return self._factories[name](merged)

list

list()

Return all registered node type names.

Source code in teff/node/registry.py

def list(self) -> list[str]:
    """Return all registered node type names."""
    return list(self._factories.keys())

register

register(name, factory)

Register a node factory under a type name.

Source code in teff/node/registry.py

def register(self, name: str, factory: NodeFactory) -> None:
    """Register a node factory under a type name."""
    self._factories[name] = factory

Parallel

Bases: Node

Execute several branch chains concurrently and merge their results.

Each branch is a list of nodes run sequentially on an isolated copy of the state. Branches run concurrently via gather_or_cancel; only the updates each node returns are merged back (per-key reducers apply, so append branches accumulate instead of overwriting one another).

Because branches read from independent copies, direct in-place mutation of the passed state is not propagated. Nodes inside branches should return their updates — the constitution's contract: receive state → return state.

Parameters:

Name	Type	Description	Default
branches	list[Node | list[Node]]	Sequence of branches, each a single :class:Node or a list of nodes. Nodes inside a branch run sequentially.	required

Usage::

node = Parallel([[upper_node, count_node], [tag_node]])

Source code in teff/node/parallel.py

class Parallel(Node):
    """Execute several branch chains concurrently and merge their results.

    Each *branch* is a list of nodes run sequentially on an isolated
    copy of the state.  Branches run concurrently via ``gather_or_cancel``;
    only the updates each node *returns* are merged back (per-key
    reducers apply, so ``append`` branches accumulate instead of
    overwriting one another).

    Because branches read from independent copies, direct in-place
    mutation of the passed state is not propagated.  Nodes inside
    branches should return their updates — the constitution's contract:
    *receive state → return state*.

    Args:
        branches: Sequence of branches, each a single :class:`Node` or a
            list of nodes.  Nodes inside a branch run sequentially.

    Usage::

        node = Parallel([[upper_node, count_node], [tag_node]])
    """

    type = "parallel"

    def __init__(
        self,
        branches: list[Node | list[Node]],
        config: dict | None = None,
        **kwargs,
    ):
        super().__init__(config, **kwargs)
        self._branches: list[list[Node]] = [
            [b] if isinstance(b, Node) else list(b) for b in branches
        ]

    async def execute(self, ctx: ExecContext, state: dict) -> dict:
        reducers = getattr(ctx, "reducers", None) or {}
        deltas = await gather_or_cancel(
            *(
                self._run_branch(branch, idx, ctx, state, reducers)
                for idx, branch in enumerate(self._branches)
            )
        )

        merged: dict = {}
        for delta in deltas:
            apply_reducers(merged, delta, reducers)
        return merged

    async def _run_branch(
        self,
        branch: list[Node],
        branch_idx: int,
        ctx: ExecContext,
        state: dict,
        reducers: dict,
    ) -> dict:
        branch_state = dict(state)
        delta: dict = {}
        for node_idx, node in enumerate(branch):
            node_id = f"{ctx.node_id or self.type}.b{branch_idx}.{node_idx}"
            node_ctx = ExecContext(
                branch_state,
                ctx.tools,
                node_id=node_id,
                node_type=node.type,
                tracer=ctx.tracer,
                reducers=reducers,
                providers=getattr(ctx, "providers", None),
                default_provider=getattr(ctx, "default_provider", None),
                default_model=getattr(ctx, "default_model", None),
                on_llm_payload=getattr(ctx, "on_llm_payload", None),
            )
            start = time.monotonic()
            if ctx.tracer is not None:
                ctx.tracer.node_start(node_id, node.type)
            try:
                result = await node.execute(node_ctx, branch_state) or {}
            except Exception as exc:
                if ctx.tracer is not None:
                    ctx.tracer.node_error(node_id, node.type, _ms(start), exc)
                raise
            if ctx.tracer is not None:
                ctx.tracer.node_end(node_id, node.type, _ms(start))
            apply_reducers(branch_state, as_updates(result), reducers)
            apply_reducers(delta, as_updates(result), reducers)
        return delta

ReActAgent

Bases: Node

Single-step LLM node for a graph-level ReAct loop.

Executes one LLM call, then signals any requested tools by setting state["_tool_calls"] (a list of {id, name, args}) and a non-empty state["_tool_call_name"].

When the LLM responds without calling a tool, the output key is set and _tool_call_name is cleared — the parent graph stops because no outgoing edge matches.

Expected graph edges::

agent  ──(_tool_call_name!=)──→  tool_exec
tool_exec  ──(unconditional)──→  agent

Parameters:

Name	Type	Description	Default
model	str | None	Model name (e.g. gpt-4).	None
system	str	System prompt.	''
input_key	str	State key for user input (default "input").	'input'
output_key	str	State key for final response (default "output").	'output'
messages_key	str	State key for conversation (default "messages").	'messages'
tool_call_key	str	Signal key (default "_tool_call_name").	'_tool_call_name'
temperature	float | None	Sampling temperature.	None
max_tokens	int | None	Max tokens in response.	None
response_format	dict | None	{"type": "json_object"} etc.	None
provider	str | None	Force a provider (auto-detected from model).	None
base_url	str | None	Custom base URL.	None
api_key_env	str | None	Custom env var name for API key.	None
chat_path	str | None	Custom API path.	None
auth_header	str | None	Custom auth header name.	None
auth_prefix	str | None	Custom auth header prefix.	None
max_tool_rounds	int | None	Round limit used by the harness loop.	None
parse_text_tool_calls	bool | None	Decode text-embedded tool calls.	None
tool_error_mode	str | None	"message" (default) or "raise".	None
tool_timeout	float | None	Per-tool execution timeout in seconds.	None
tool_retries	int	Extra attempts per tool call after a failure.	0
max_retries	int	HTTP request retries (429/5xx/timeouts).	2
fallbacks	list[str] | None	Fallback model names for provider failover.	None
tool_approval	Any	Gate on tool execution — "auto" (default), "deny", "interactive" (ask on stdin), or a callable (name, args) -> bool | str. "pause" decisions pause the run as a :class:~teff.node.interrupt.GraphInterrupt.	None
memory	MemoryConfig | dict | None	Optional long-term memory injection — a :class:~teff.memory.context.MemoryConfig or {store, namespace, k, header}. store is a :class:~teff.memory.base.MemoryStore or a config dict; on every turn the top-k recalled memories for the last user message are prepended to the conversation as a system message.	None
stream	bool	Stream the final assistant text (tokens forwarded to on_token and stream events).	False
on_token	Callable[[str], None] | None	Callback (token: str) -> None for streaming.	None

Source code in teff/node/agent.py

class ReActAgent(Node):
    """Single-step LLM node for a graph-level ReAct loop.

    Executes one LLM call, then signals any requested tools by setting
    ``state["_tool_calls"]`` (a list of ``{id, name, args}``) and a
    non-empty ``state["_tool_call_name"]``.

    When the LLM responds without calling a tool, the output key is
    set and ``_tool_call_name`` is cleared — the parent graph stops
    because no outgoing edge matches.

    Expected graph edges::

        agent  ──(_tool_call_name!=)──→  tool_exec
        tool_exec  ──(unconditional)──→  agent

    Parameters:
        model: Model name (e.g. ``gpt-4``).
        system: System prompt.
        input_key: State key for user input (default ``"input"``).
        output_key: State key for final response (default ``"output"``).
        messages_key: State key for conversation (default ``"messages"``).
        tool_call_key: Signal key (default ``"_tool_call_name"``).
        temperature: Sampling temperature.
        max_tokens: Max tokens in response.
        response_format: ``{"type": "json_object"}`` etc.
        provider: Force a provider (auto-detected from model).
        base_url: Custom base URL.
        api_key_env: Custom env var name for API key.
        chat_path: Custom API path.
        auth_header: Custom auth header name.
        auth_prefix: Custom auth header prefix.
        max_tool_rounds: Round limit used by the harness loop.
        parse_text_tool_calls: Decode text-embedded tool calls.
        tool_error_mode: ``"message"`` (default) or ``"raise"``.
        tool_timeout: Per-tool execution timeout in seconds.
        tool_retries: Extra attempts per tool call after a failure.
        max_retries: HTTP request retries (429/5xx/timeouts).
        fallbacks: Fallback model names for provider failover.
        tool_approval: Gate on tool execution — ``"auto"`` (default),
            ``"deny"``, ``"interactive"`` (ask on stdin), or a callable
            ``(name, args) -> bool | str``.  ``"pause"`` decisions pause
            the run as a :class:`~teff.node.interrupt.GraphInterrupt`.
        memory: Optional long-term memory injection — a
            :class:`~teff.memory.context.MemoryConfig` or ``{store,
            namespace, k, header}``.  ``store`` is a
            :class:`~teff.memory.base.MemoryStore` or a config dict; on
            every turn the top-*k* recalled memories for the last user
            message are prepended to the conversation as a system message.
        stream: Stream the final assistant text (tokens forwarded to
            ``on_token`` and stream events).
        on_token: Callback ``(token: str) -> None`` for streaming.
    """

    type = "react_agent"

    def __init__(
        self,
        config: dict | None = None,
        *,
        model: str | None = None,
        system: str = "",
        input_key: str = "input",
        output_key: str = "output",
        messages_key: str = "messages",
        tool_call_key: str = "_tool_call_name",
        temperature: float | None = None,
        max_tokens: int | None = None,
        response_format: dict | None = None,
        provider: str | None = None,
        base_url: str | None = None,
        api_key_env: str | None = None,
        chat_path: str | None = None,
        auth_header: str | None = None,
        auth_prefix: str | None = None,
        use_tools: str | list[str] | None = None,
        skills: list | None = None,
        skill_dir: str = "skills",
        max_tool_rounds: int | None = None,
        parse_text_tool_calls: bool | None = None,
        tool_error_mode: str | None = None,
        tool_timeout: float | None = None,
        tool_retries: int = 0,
        max_retries: int = 2,
        fallbacks: list[str] | None = None,
        tool_approval: typing.Any = None,
        stream: bool = False,
        on_token: typing.Callable[[str], None] | None = None,
        memory: MemoryConfig | dict | None = None,
        **kwargs,
    ):
        merged = {
            "model": model,
            "system": system,
            "input_key": input_key,
            "output_key": output_key,
            "messages_key": messages_key,
            "tool_call_key": tool_call_key,
            "temperature": temperature,
            "max_tokens": max_tokens,
            "response_format": response_format,
            "provider": provider,
            "base_url": base_url,
            "api_key_env": api_key_env,
            "chat_path": chat_path,
            "auth_header": auth_header,
            "auth_prefix": auth_prefix,
            "use_tools": use_tools,
            "skills": skills,
            "skill_dir": skill_dir,
            "max_tool_rounds": max_tool_rounds,
            "parse_text_tool_calls": parse_text_tool_calls,
            "tool_error_mode": tool_error_mode,
            "tool_timeout": tool_timeout,
            "tool_retries": tool_retries,
            "max_retries": max_retries,
            "fallbacks": fallbacks,
            "tool_approval": tool_approval,
            "stream": stream,
            "on_token": on_token,
            "memory": memory,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        cfg = self.config
        system = cfg.get("system", "")
        input_key = cfg.get("input_key", "input")
        output_key = cfg.get("output_key", "output")
        messages_key = cfg.get("messages_key", "messages")
        tool_call_key = cfg.get("tool_call_key", "_tool_call_name")

        skills = resolve_skills(cfg)
        skill_text = skills_instructions(skills)
        if skill_text:
            system = f"{system}\n\n{skill_text}" if system else skill_text

        messages = list(state.get(messages_key, []))
        from teff.memory.context import memory_context_from_config

        block = await memory_context_from_config(cfg, state=state, ctx=ctx)
        if block:
            messages.insert(0, {"role": "system", "content": block})
        start = len(messages)
        if not messages:
            user_input = str(state.get(input_key, ""))
            if system:
                messages.append({"role": "system", "content": system})
            if user_input:
                messages.append({"role": "user", "content": user_input})

        tool_defs = [
            tool_to_schema(t) for t in scope_tools(ctx.tools, cfg, skills).values()
        ]

        harness = Harness.from_config(
            cfg,
            default_provider=getattr(ctx, "default_provider", None),
            default_model=getattr(ctx, "default_model", None),
            providers=getattr(ctx, "providers", None),
        )
        tracer = getattr(ctx, "tracer", None)
        if tracer is not None:

            async def on_llm(provider, model, prompt, completion, duration):
                tracer.llm(provider, model, prompt, completion, duration)

            harness.on_llm = on_llm

        payload_sink = getattr(ctx, "on_llm_payload", None)
        if payload_sink is not None:
            harness.on_llm_payload = payload_sink

        emit = getattr(ctx, "emit", None)
        on_token_cfg = cfg.get("on_token")

        async def token_sink(token: str) -> None:
            if on_token_cfg is not None:
                res = on_token_cfg(token)
                if asyncio.iscoroutine(res):
                    await res
            if emit is not None:
                await emit(
                    StreamEvent(
                        "token",
                        node_id=ctx.node_id,
                        node_type=ctx.node_type,
                        data={
                            "token": token,
                            "provider": harness.provider_key,
                            "model": str(cfg.get("model", "")),
                        },
                    )
                )

        want_stream = bool(cfg.get("stream", False))
        if want_stream and not tool_defs:
            harness.on_token = token_sink

        reply = await harness.call(
            messages,
            tools=tool_defs or None,
            stream=want_stream and not tool_defs,
        )

        result: dict = {}

        # The graph loop (agent → tool → agent) is what repeats, so the node
        # itself must track how many times it has been visited.  Once the
        # round budget is spent we stop signalling tools even if the model
        # keeps asking, letting the loop end on this node.
        round_key = f"_react_round_{ctx.node_id}"
        round_count = int(state.get(round_key, 0)) + 1
        result[round_key] = round_count
        max_rounds = cfg.get("max_tool_rounds")
        budget_spent = max_rounds is not None and round_count > max_rounds

        if budget_spent:
            content = reply.content or ""
            messages.append({"role": "assistant", "content": content})
            result[output_key] = content
            result[tool_call_key] = ""
            result["_tool_calls"] = []
        else:
            tool_calls = reply.message.get("tool_calls")

            if tool_calls:
                calls: list[dict] = []
                for tc in tool_calls:
                    fn = tc.get("function", {})
                    raw = fn.get("arguments", "{}")
                    if isinstance(raw, dict):
                        raw = json.dumps(raw)
                    calls.append(
                        {
                            "id": tc.get("id", ""),
                            "name": fn.get("name", ""),
                            "args": raw,
                        }
                    )
                result[tool_call_key] = "pending"
                result["_tool_calls"] = calls
                messages.append(reply.message)
            else:
                content = reply.content
                parse_cfg = cfg.get("parse_text_tool_calls", True)
                if parse_cfg is None:
                    parse_cfg = True
                parsed = (
                    parse_text_tool_call(content) if tool_defs and parse_cfg else None
                )
                if parsed:
                    name, args = parsed
                    result[tool_call_key] = "pending"
                    result["_tool_calls"] = [
                        {
                            "id": f"call_{len(messages)}",
                            "name": name,
                            "args": json.dumps(args),
                        }
                    ]
                    messages.append({"role": "assistant", "content": content})
                else:
                    result[output_key] = content
                    result[tool_call_key] = ""
                    result["_tool_calls"] = []
                    messages.append({"role": "assistant", "content": content})

        if reducer_appends((ctx.reducers or {}).get(messages_key)):
            result[messages_key] = messages[start:]
        else:
            result[messages_key] = messages
        return result

Retry

Bases: Node

Wrap a node with retry logic.

Retries the inner node up to max_retries attempts total. Between attempts it waits delay seconds (scaled by backoff per retry, e.g. backoff=2.0 gives delay, 2×, 4×, …). Each attempt is bounded by timeout seconds when set. retry_on restricts which failures are retried (exception type names or HTTP status codes); by default any exception is retried.

Config (all optional): max_retries (default 3), delay (default 0.0), backoff (default 1.0), timeout (default None), retry_on (default [] = all).

Source code in teff/node/retry.py

class Retry(Node):
    """Wrap a node with retry logic.

    Retries the inner node up to *max_retries* attempts total.  Between
    attempts it waits ``delay`` seconds (scaled by *backoff* per retry,
    e.g. ``backoff=2.0`` gives delay, 2×, 4×, …).  Each attempt is bounded
    by *timeout* seconds when set.  *retry_on* restricts which failures are
    retried (exception type names or HTTP status codes); by default any
    exception is retried.

    Config (all optional): ``max_retries`` (default 3), ``delay`` (default
    0.0), ``backoff`` (default 1.0), ``timeout`` (default None),
    ``retry_on`` (default [] = all).
    """

    type = "retry"

    def __init__(
        self,
        node: Node,
        max_retries: int = 3,
        delay: float = 0.0,
        backoff: float = 1.0,
        timeout: float | None = None,
        retry_on: list | None = None,
        config: dict | None = None,
        **kwargs,
    ):
        super().__init__(config, **kwargs)
        self._node = node
        self._max_retries = max(1, int(max_retries))
        self._delay = float(delay)
        self._backoff = float(backoff)
        self._timeout = float(timeout) if timeout else None
        self._retry_on = list(retry_on or [])

    async def execute(self, ctx: ExecContext, state: dict) -> dict | Command:
        last_exc: Exception | None = None
        for attempt in range(self._max_retries):
            try:
                coro = self._node.execute(ctx, state)
                if self._timeout:
                    coro = asyncio.wait_for(coro, timeout=self._timeout)
                return await coro
            except GraphInterrupt:
                raise
            except Exception as e:
                last_exc = e
                if attempt >= self._max_retries - 1:
                    break
                if not _match_exception(e, self._retry_on):
                    break
                tracer = getattr(ctx, "tracer", None)
                if tracer is not None:
                    tracer.retry(ctx.node_id, ctx.node_type, attempt + 1, e)
                wait = self._delay * (self._backoff**attempt)
                if wait:
                    await asyncio.sleep(wait)
        raise last_exc  # type: ignore[misc]

StructuredOutputError

Bases: TeffError, ValueError

Raised when an LLM response fails structured-output parsing/validation.

Attributes:

Name	Type	Description
schema		The JSON Schema the output was validated against (or None).
content		Raw text the LLM returned.
errors		Parse/validation error message from the last attempt.
attempts		Number of attempts made before giving up.

Source code in teff/node/llm.py

class StructuredOutputError(TeffError, ValueError):
    """Raised when an LLM response fails structured-output parsing/validation.

    Attributes:
        schema: The JSON Schema the output was validated against (or ``None``).
        content: Raw text the LLM returned.
        errors: Parse/validation error message from the last attempt.
        attempts: Number of attempts made before giving up.
    """

    def __init__(
        self,
        *,
        schema: dict | None = None,
        content: str = "",
        errors: str = "",
        attempts: int = 0,
    ):
        self.schema = schema
        self.content = content
        self.errors = errors
        self.attempts = attempts
        message = f"LLM output failed structured validation after {attempts} attempt(s): {errors}"
        super().__init__(message)

Supervisor

Bases: Node

Decide which agent handles the latest user message.

Reads the last user message (plus any work already produced), asks the model which agent fits it best (a single word), and writes the chosen route to output_key. When the round counter reached max_rounds or the done_keys are already filled, the conversation is finished without another model call.

fill_order turns the supervisor into a deterministic pipeline without a subclass: the model picks only the entry agent, then every mid-pipeline round runs the chain in order (planner → estimator → ... → finish) with no further model calls. A mid-chain agent picked directly (a targeted question) runs once and finishes. See examples/applications/repair-ai-chat for a chat that routes a direct branch through done_keys while chaining the repair agents through fill_order.

finish renames the terminator token the model answers with (default "finish"); the same value is written to output_key for the finish route branch. Set it to whatever your system prompt tells the model to reply, e.g. finish="<end>".

Methods:

Name	Description
decide	Resolve the route from the parsed proposal plus the guards.

Source code in teff/node/supervisor.py

class Supervisor(Node):
    """Decide which agent handles the latest user message.

    Reads the last user message (plus any work already produced), asks the
    model which agent fits it best (a single word), and writes the chosen
    route to ``output_key``.  When the round counter reached ``max_rounds``
    or the ``done_keys`` are already filled, the conversation is finished
    without another model call.

    ``fill_order`` turns the supervisor into a deterministic pipeline
    without a subclass: the model picks only the *entry* agent, then every
    mid-pipeline round runs the chain in order (``planner`` → ``estimator``
    → ... → ``finish``) with no further model calls.  A mid-chain agent
    picked directly (a targeted question) runs once and finishes.  See
    ``examples/applications/repair-ai-chat`` for a chat that routes a
    ``direct`` branch through ``done_keys`` while chaining the repair
    agents through ``fill_order``.

    ``finish`` renames the terminator token the model answers with (default
    ``"finish"``); the same value is written to ``output_key`` for the
    ``finish`` route branch.  Set it to whatever your system prompt tells
    the model to reply, e.g. ``finish="<end>"``.
    """

    type = "supervisor"

    def __init__(
        self,
        config: dict | None = None,
        *,
        system: str = "",
        model: str = "",
        provider: str = "",
        messages_key: str = "messages",
        output_key: str = "next_agent",
        rounds_key: str = "supervisor_rounds",
        max_rounds: int = 6,
        sections: dict[str, str] | None = None,
        agents: AbstractSet[str] | None = None,
        route_keys: dict[str, str] | None = None,
        done_keys: set[str] | None = None,
        done_mode: str = "all",
        fallback_agent: str = "",
        finish: str = "finish",
        fill_order: list[tuple[str, str]] | None = None,
        **kwargs,
    ):
        merged = {
            "system": system,
            "model": model,
            "provider": provider,
            "messages_key": messages_key,
            "output_key": output_key,
            "rounds_key": rounds_key,
            "max_rounds": max_rounds,
            "sections": sections or {},
            "agents": agents,
            "route_keys": route_keys or {},
            "done_keys": set(done_keys or ()),
            "done_mode": done_mode,
            "fallback_agent": fallback_agent,
            "finish": finish,
            "fill_order": [(agent, slot) for agent, slot in (fill_order or [])],
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    def _agents(self) -> set[str]:
        """The single-word vocabulary the model may answer with."""
        agents = self.config.get("agents")
        if agents:
            return set(agents)
        route_keys = self.config.get("route_keys") or {}
        chain = [agent for agent, _ in (self.config.get("fill_order") or [])]
        fallback = self.config.get("fallback_agent") or ""
        return (
            set(route_keys)
            | set(chain)
            | {self._finish()}
            | ({fallback} if fallback else set())
        )

    def _finish(self) -> str:
        """The token that ends the conversation (``finish`` by default)."""
        return self.config.get("finish") or "finish"

    def _parse_agent(self, text: str) -> str:
        """Return the last word of *text* that matches the agent vocabulary.

        Both sides are normalized by stripping enclosing punctuation and
        ``<>``, so a model that wraps its reply as ``<finish>`` matches the
        configured ``finish`` token and vice versa.  The canonical spelling
        from the vocabulary is returned.
        """
        canonical = {}
        for agent in self._agents():
            canonical[agent.strip(" .*\"'»«-<>").lower()] = agent
        for word in reversed(
            text.strip().lower().replace(",", " ").replace(":", " ").split()
        ):
            w = word.strip(" .*\"'»«-<>")
            if w in canonical:
                return canonical[w]
        return ""

    def _progress_text(self, state: dict) -> str:
        """Render the non-empty ``sections`` as ``Label:\\n<value>`` blocks."""
        parts = []
        for key, label in (self.config.get("sections") or {}).items():
            value = state.get(key)
            if value:
                parts.append(f"{label}:\n{value}")
        return "\n\n".join(parts)

    def _done(self, state: dict) -> bool:
        """Whether the ``done_keys`` guard is satisfied (no model needed).

        ``done_mode="all"`` requires every key non-empty, ``"any"`` requires
        at least one.
        """
        done_keys = set(self.config.get("done_keys") or ())
        if not done_keys:
            return False
        filled = [k for k in done_keys if state.get(k)]
        if self.config.get("done_mode", "all") == "any":
            return bool(filled)
        return len(filled) == len(done_keys)

    def _chain_route(self, state: dict) -> str:
        """The deterministic route a configured ``fill_order`` prescribes.

        Returns ``""`` when the model must still pick the entry agent.
        Once the entry slot is filled the pipeline runs the rest of the
        chain in order and finishes when every slot is full.  A mid-chain
        agent picked directly (a targeted question) runs once and finishes
        without dragging the whole pipeline in.
        """
        order = self.config.get("fill_order") or []
        if not order:
            return ""
        entry_slot = order[0][1]
        if state.get(entry_slot):
            for agent, slot in order:
                if not state.get(slot):
                    return agent
            return self._finish()
        for agent, slot in order[1:]:
            if state.get(slot):
                return self._finish()
        return ""

    def _needs_model(self, state: dict) -> bool:
        """Whether the model must be consulted this round.

        Default (chat routing): the model is needed only when there is a user
        message to route and no ``done_keys`` are already filled.  Set
        ``messages_key=""`` to always consult the model, or override this in
        a subclass whose :meth:`decide` resolves some states deterministically.
        With a ``fill_order`` the model is needed only for the entry decision;
        every mid-pipeline round is resolved deterministically.
        """
        cfg = self.config
        if self._done(state):
            return False
        if cfg.get("fill_order"):
            return not self._chain_route(state)
        messages_key = cfg.get("messages_key")
        if not messages_key:
            return True
        return bool(last_user_message(state.get(messages_key, [])))

    def decide(self, state: dict, proposal: str) -> str:
        """Resolve the route from the parsed *proposal* plus the guards.

        Default implements the chat guards on top of the model's single word:
        a filled ``done_keys`` set short-circuits to ``finish``, a premature
        ``finish`` falls back to *fallback_agent*, and a ``route_keys`` agent
        whose slot is already filled is not re-routed.  With a ``fill_order``
        the mid-pipeline route is deterministic (see :meth:`_chain_route`);
        only the entry decision comes from the model.  Subclasses override
        this for a deterministic policy; *proposal* is ``""`` when the model
        was not consulted.
        """
        proposal = proposal or self._finish()
        cfg = self.config
        finish = self._finish()
        if self._done(state):
            return finish
        fallback = cfg.get("fallback_agent") or ""
        if cfg.get("fill_order"):
            route = self._chain_route(state)
            if route:
                return route
            if proposal in self._agents() and proposal != finish:
                return proposal
            return fallback or cfg["fill_order"][0][0]
        if (
            proposal == finish
            and fallback
            and not any(state.get(k) for k in cfg.get("done_keys") or ())
        ):
            return fallback
        route_keys = cfg.get("route_keys") or {}
        if proposal in route_keys and state.get(route_keys[proposal]):
            return finish
        return proposal

    async def _ask_model(
        self, ctx, state: dict, *, rounds: int, max_rounds: int
    ) -> str:
        """Render the context, call the model, return the parsed proposal."""
        cfg = self.config
        harness = Harness.from_config(
            cfg,
            default_provider=getattr(ctx, "default_provider", None),
            default_model=getattr(ctx, "default_model", None),
            providers=getattr(ctx, "providers", None),
        )
        tracer = getattr(ctx, "tracer", None)
        if tracer is not None:

            async def on_llm(provider, model, prompt, completion, duration):
                tracer.llm(provider, model, prompt, completion, duration)

            harness.on_llm = on_llm

        payload_sink = getattr(ctx, "on_llm_payload", None)
        if payload_sink is not None:
            harness.on_llm_payload = payload_sink

        messages_key = cfg.get("messages_key", "messages")
        user = (
            f"User: {last_user_message(state.get(messages_key, []))}"
            if messages_key
            else ""
        )
        user_parts = [
            part
            for part in (
                self._progress_text(state),
                f"Round: {rounds}/{max_rounds}",
                user,
            )
            if part
        ]
        reply = await harness.call(
            [
                {"role": "system", "content": cfg.get("system", "")},
                {"role": "user", "content": "\n\n".join(user_parts)},
            ]
        )
        return self._parse_agent(reply.content)

    async def execute(self, ctx, state: dict) -> dict:
        cfg = self.config
        rounds_key = cfg.get("rounds_key", "supervisor_rounds")
        output_key = cfg.get("output_key", "next_agent")
        max_rounds = int(cfg.get("max_rounds", 6))
        rounds = int(state.get(rounds_key) or 0) + 1

        # Bounded loop: a model that never says "finish" cannot hang.
        if rounds >= max_rounds:
            return {output_key: self._finish(), rounds_key: rounds}

        if not self._needs_model(state):
            return {output_key: self.decide(state, ""), rounds_key: rounds}

        proposal = await self._ask_model(
            ctx, state, rounds=rounds, max_rounds=max_rounds
        )
        return {output_key: self.decide(state, proposal), rounds_key: rounds}

decide

decide(state, proposal)

Resolve the route from the parsed proposal plus the guards.

Default implements the chat guards on top of the model's single word: a filled done_keys set short-circuits to finish, a premature finish falls back to fallback_agent, and a route_keys agent whose slot is already filled is not re-routed. With a fill_order the mid-pipeline route is deterministic (see :meth:_chain_route); only the entry decision comes from the model. Subclasses override this for a deterministic policy; proposal is "" when the model was not consulted.

Source code in teff/node/supervisor.py

def decide(self, state: dict, proposal: str) -> str:
    """Resolve the route from the parsed *proposal* plus the guards.

    Default implements the chat guards on top of the model's single word:
    a filled ``done_keys`` set short-circuits to ``finish``, a premature
    ``finish`` falls back to *fallback_agent*, and a ``route_keys`` agent
    whose slot is already filled is not re-routed.  With a ``fill_order``
    the mid-pipeline route is deterministic (see :meth:`_chain_route`);
    only the entry decision comes from the model.  Subclasses override
    this for a deterministic policy; *proposal* is ``""`` when the model
    was not consulted.
    """
    proposal = proposal or self._finish()
    cfg = self.config
    finish = self._finish()
    if self._done(state):
        return finish
    fallback = cfg.get("fallback_agent") or ""
    if cfg.get("fill_order"):
        route = self._chain_route(state)
        if route:
            return route
        if proposal in self._agents() and proposal != finish:
            return proposal
        return fallback or cfg["fill_order"][0][0]
    if (
        proposal == finish
        and fallback
        and not any(state.get(k) for k in cfg.get("done_keys") or ())
    ):
        return fallback
    route_keys = cfg.get("route_keys") or {}
    if proposal in route_keys and state.get(route_keys[proposal]):
        return finish
    return proposal

ToolCall

Bases: Node

Call a registered tool by name with config-driven arguments.

Each argument value may contain {key} templates that are rendered from the current state before the call. The tool's string result is written to output_key (default: output). When on_error is "message" a failure is stored under output_key as "error: ..." instead of raising.

Config

tool: Registered tool name to invoke. args: Mapping of tool argument name to value or {key} template. output_key: State key for the result (default "output"). on_error: "raise" (default) or "message". max_chars: Truncate the result to this many characters.

Source code in teff/node/tool_call.py

class ToolCall(Node):
    """Call a registered tool by name with config-driven arguments.

    Each argument value may contain ``{key}`` templates that are rendered
    from the current state before the call.  The tool's string result is
    written to ``output_key`` (default: ``output``).  When ``on_error`` is
    ``"message"`` a failure is stored under ``output_key`` as ``"error: ..."``
    instead of raising.

    Config:
        tool: Registered tool name to invoke.
        args: Mapping of tool argument name to value or ``{key}`` template.
        output_key: State key for the result (default ``"output"``).
        on_error: ``"raise"`` (default) or ``"message"``.
        max_chars: Truncate the result to this many characters.
    """

    type = "tool_call"

    def __init__(
        self,
        config: dict | None = None,
        *,
        tool: str = "",
        args: dict | None = None,
        output_key: str = "output",
        on_error: str = "raise",
        max_chars: int | None = None,
        **kwargs,
    ):
        merged = {
            "tool": tool,
            "args": args or {},
            "output_key": output_key,
            "on_error": on_error,
            "max_chars": max_chars,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx: "ExecContext", state: dict) -> dict:
        tool_name = str(self.config.get("tool", ""))
        if not tool_name:
            raise ValueError("tool_call requires 'tool'")
        tool = ctx.tool(tool_name)

        raw_args = dict(self.config.get("args") or {})
        args = {}
        for key, value in raw_args.items():
            if isinstance(value, str):
                value = render_template(value, state)
            args[key] = value
        args = coerce_args(tool, args)

        output_key = str(self.config.get("output_key", "output"))
        on_error = self.config.get("on_error", "raise")
        max_chars = self.config.get("max_chars")

        try:
            result = await tool.arun(**args)
        except Exception as exc:  # noqa: BLE001 — surfaced per on_error mode
            if on_error == "message":
                return {output_key: f"error: {exc}"}
            raise
        text = str(result)
        if max_chars is not None:
            text = text[: int(max_chars)]
        return {output_key: text}

ToolExec

Bases: Node

Executes tools signalled by :class:ReActAgent in parallel and feeds the results back into the conversation history.

Handles multiple tool calls per round: the agent writes the whole _tool_calls list, which is executed concurrently and appended as tool messages in one go. Falls back to the legacy single-call signals (_tool_call_name / _tool_call_args / _tool_call_id).

Parameters:

Name	Type	Description	Default
messages_key	str	State key for messages (default "messages").	'messages'
tool_call_key	str	Signal key (default "_tool_call_name").	'_tool_call_name'
tool_error_mode	str	"message" (default) or "raise" — when "raise", a tool failure propagates to the graph error path (e.g. an __error__ edge) instead of becoming a tool message.	'message'
tool_timeout	float | None	Per-tool execution timeout in seconds.	None
tool_retries	int	Extra attempts per tool call after a failure.	0
tool_approval	Any	Gate on tool execution — "auto" (default), "deny", "interactive" (ask on stdin), or a callable (name, args) -> bool | str (sync or async). A "pause" decision pauses the run as a :class:GraphInterrupt; "deny" short-circuits the call with a "denied" tool message.	None
human_key		State key / tool name of the human-in-the-loop question (default "ask_human"). A call to this tool is intercepted before execution: without a pending reply it pauses the run as a :class:GraphInterrupt carrying the question; with a reply in the state (from resume) it consumes it and returns it as the tool result.	required

Source code in teff/node/agent.py

class ToolExec(Node):
    """Executes tools signalled by :class:`ReActAgent` in parallel and feeds
    the results back into the conversation history.

    Handles multiple tool calls per round: the agent writes the whole
    ``_tool_calls`` list, which is executed concurrently and appended as
    ``tool`` messages in one go.  Falls back to the legacy single-call
    signals (``_tool_call_name`` / ``_tool_call_args`` / ``_tool_call_id``).

    Parameters:
        messages_key: State key for messages (default ``"messages"``).
        tool_call_key: Signal key (default ``"_tool_call_name"``).
        tool_error_mode: ``"message"`` (default) or ``"raise"`` — when
            ``"raise"``, a tool failure propagates to the graph error path
            (e.g. an ``__error__`` edge) instead of becoming a tool message.
        tool_timeout: Per-tool execution timeout in seconds.
        tool_retries: Extra attempts per tool call after a failure.
        tool_approval: Gate on tool execution — ``"auto"`` (default),
            ``"deny"``, ``"interactive"`` (ask on stdin), or a callable
            ``(name, args) -> bool | str`` (sync or async).  A ``"pause"``
            decision pauses the run as a :class:`GraphInterrupt`; ``"deny"``
            short-circuits the call with a "denied" tool message.
        human_key: State key / tool name of the human-in-the-loop
            question (default ``"ask_human"``).  A call to this tool is
            intercepted *before* execution: without a pending reply it
            pauses the run as a :class:`GraphInterrupt` carrying the
            question; with a reply in the state (from ``resume``) it
            consumes it and returns it as the tool result.
    """

    type = "tool_exec"

    def __init__(
        self,
        config: dict | None = None,
        *,
        messages_key: str = "messages",
        tool_call_key: str = "_tool_call_name",
        tool_error_mode: str = "message",
        tool_timeout: float | None = None,
        tool_retries: int = 0,
        tool_approval: typing.Any = None,
        use_tools: str | list[str] | None = None,
        skills: list | None = None,
        skill_dir: str = "skills",
        **kwargs,
    ):
        merged = {
            "messages_key": messages_key,
            "tool_call_key": tool_call_key,
            "tool_error_mode": tool_error_mode,
            "tool_timeout": tool_timeout,
            "tool_retries": tool_retries,
            "tool_approval": tool_approval,
            "use_tools": use_tools,
            "skills": skills,
            "skill_dir": skill_dir,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        messages_key = self.config.get("messages_key", "messages")
        tool_call_key = self.config.get("tool_call_key", "_tool_call_name")
        tool_error_mode = self.config.get("tool_error_mode", "message")
        tool_timeout = self.config.get("tool_timeout")
        tool_retries = int(self.config.get("tool_retries", 0))
        approver = self.config.get("tool_approval")
        human_key = self.config.get("human_key", "ask_human")

        calls = list(state.get("_tool_calls") or [])
        if not calls and state.get(tool_call_key):
            calls = [
                {
                    "id": state.get("_tool_call_id", ""),
                    "name": state.get(tool_call_key, ""),
                    "args": state.get("_tool_call_args", "{}"),
                }
            ]

        skills = resolve_skills(self.config)
        scoped = scope_tools(ctx.tools, self.config, skills)

        # Human-in-the-loop: an ask_human call pauses the run for an
        # operator's answer.  On resume the answer arrives in the state under
        # *human_key* (the resume dict key), is consumed here and delivered
        # back as the tool result — the same re-invoke pattern as the
        # tool_approval "pause" decision.  The call never reaches
        # execute_tool_calls (AskHuman.arun raises NotImplementedError).
        ask_replies: dict[str, str] = {}
        normal_calls: list[dict] = []
        for call in calls:
            name, raw_args, _call_id = _tool_call_parts(call)
            if name == human_key:
                try:
                    args = json.loads(raw_args) if raw_args else {}
                except json.JSONDecodeError:
                    args = {}
                question = str(args.get("question", ""))
                pending = state.get(human_key)
                if pending is None:
                    raise GraphInterrupt(
                        key=human_key,
                        prompt=question or "The agent is asking for your input.",
                    )
                state.pop(human_key, None)  # consume the operator's answer
                ask_replies[call.get("id", "")] = str(pending)
            else:
                normal_calls.append(call)

        # After a pause/interrupt, the operator's decision comes back in the
        # resume payload under the interrupt key; use it instead of re-asking.
        resumed = state.get("tool_approval")
        resumed = resumed if resumed in ("approve", "deny") else None

        to_run = normal_calls
        denied: list[tuple[str, str, str]] = []
        if approver is not None and approver != "auto" and normal_calls:
            to_run = []
            for call in normal_calls:
                name = call.get("name", "")
                try:
                    args = (
                        json.loads(call.get("args", "{}")) if call.get("args") else {}
                    )
                except json.JSONDecodeError:
                    args = {}
                if resumed is not None:
                    decision = resumed
                else:
                    decision = await resolve_approval(approver, name, args)
                if decision == "pause":
                    raise GraphInterrupt(
                        key="tool_approval",
                        prompt=(
                            f"Approve tool call '{name}' with args {json.dumps(args)}?"
                        ),
                    )
                if decision != "approve":
                    denied.append((name, call.get("id", ""), decision))
                else:
                    to_run.append(call)

        results = await execute_tool_calls(
            to_run,
            scoped,
            tool_error_mode,
            tool_timeout,
            tool_retries,
            state=state,
            ctx=ctx,
        )

        result_by_id = {
            call.get("id", ""): str(res) if res is not None else ""
            for call, res in zip(to_run, results)
        }

        messages = list(state.get(messages_key, []))
        start = len(messages)
        for call in calls:  # keep the original call order in the conversation
            call_id = call.get("id", "")
            if call_id in result_by_id:
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call_id,
                        "content": result_by_id[call_id],
                    }
                )
            elif call_id in ask_replies:
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call_id,
                        "content": ask_replies[call_id],
                    }
                )
        for name, call_id, decision in denied:
            messages.append(
                {
                    "role": "tool",
                    "tool_call_id": call_id,
                    "content": f"Tool call '{name}' was not approved ({decision})",
                }
            )

        out: dict = {
            tool_call_key: "",
            "_tool_calls": [],
            "_tool_call_args": "",
            "_tool_call_id": "",
        }
        if reducer_appends((ctx.reducers or {}).get(messages_key)):
            out[messages_key] = messages[start:]
        else:
            out[messages_key] = messages
        return out

Transform

Bases: Node

Apply a transform to state values.

Supported actions: uppercase, lowercase, trim, count_lines, value, render, json_get, append, plus the pipeline-building actions contains, compare, split, join, replace, coalesce, pick, to_int, to_float, now.

render formats a template ({key} placeholders rendered from state) and stores the resulting string under output_key — the scalar counterpart of append (which accumulates into a list).

json_get extracts field from a dict in input_key. Non-string values are stringified by default; pass raw=True to keep the value as-is (e.g. to hand a parsed list to a Map).

append formats a template ({key} placeholders rendered from state) and appends the result to the list in output_key. When no template is given, input_key/value supplies the item instead. The list is created if absent — the common "accumulate formatted results" pattern (report sections, chapter text, step logs).

contains outputs "true"/"false" when input_key contains value; compare does the same for input_key against value with op in eq/ne/gt/ge/lt/le (numeric when both sides parse as numbers). split/join convert between strings and lists with sep (default ,). replace swaps old→new in input_key. coalesce returns input_key unless it is empty, then value. pick reads field out of a dict (like json_get). to_int/to_float coerce input_key to a number (as a string). now writes the current UTC ISO timestamp. Every action writes to output_key; boolean-like actions emit "true"/"false" so they can drive edges: conditions like has_refund=true.

Parameters:

Name	Type	Description	Default
action	str	Transform action name.	''
input_key	str	State key to read from.	''
output_key	str	State key to write to.	''
value	str | None	Literal value (used with action="value", the needle for contains, the right-hand side of compare, the fallback for coalesce).	None
template	str | None	Template string for action="append".	None
raw	bool	Return json_get/pick values without stringifying.	False

Source code in teff/node/transform.py

class Transform(Node):
    """Apply a transform to state values.

    Supported actions: ``uppercase``, ``lowercase``, ``trim``,
    ``count_lines``, ``value``, ``render``, ``json_get``, ``append``,
    plus the pipeline-building actions ``contains``, ``compare``, ``split``,
    ``join``, ``replace``, ``coalesce``, ``pick``, ``to_int``, ``to_float``,
    ``now``.

    ``render`` formats a ``template`` (``{key}`` placeholders rendered from
    state) and stores the resulting string under *output_key* — the scalar
    counterpart of ``append`` (which accumulates into a list).

    ``json_get`` extracts ``field`` from a dict in *input_key*.  Non-string
    values are stringified by default; pass ``raw=True`` to keep the value
    as-is (e.g. to hand a parsed list to a ``Map``).

    ``append`` formats a ``template`` (``{key}`` placeholders rendered from
    state) and appends the result to the list in *output_key*.  When no
    template is given, *input_key*/*value* supplies the item instead.  The
    list is created if absent — the common "accumulate formatted results"
    pattern (report sections, chapter text, step logs).

    ``contains`` outputs ``"true"``/``"false"`` when *input_key* contains
    ``value``; ``compare`` does the same for ``input_key`` against ``value``
    with ``op`` in ``eq/ne/gt/ge/lt/le`` (numeric when both sides parse as
    numbers).  ``split``/``join`` convert between strings and lists with
    ``sep`` (default ``,``).  ``replace`` swaps ``old``→``new`` in
    *input_key*.  ``coalesce`` returns *input_key* unless it is empty, then
    ``value``.  ``pick`` reads ``field`` out of a dict (like ``json_get``).
    ``to_int``/``to_float`` coerce *input_key* to a number (as a string).
    ``now`` writes the current UTC ISO timestamp.  Every action writes to
    *output_key*; boolean-like actions emit ``"true"``/``"false"`` so they
    can drive ``edges:`` conditions like ``has_refund=true``.

    Parameters:
        action: Transform action name.
        input_key: State key to read from.
        output_key: State key to write to.
        value: Literal value (used with ``action="value"``, the needle for
            ``contains``, the right-hand side of ``compare``, the fallback
            for ``coalesce``).
        template: Template string for ``action="append"``.
        raw: Return ``json_get``/``pick`` values without stringifying.
    """

    type = "transform"

    def __init__(
        self,
        config: dict | None = None,
        *,
        action: str = "",
        input_key: str = "",
        output_key: str = "",
        value: str | None = None,
        template: str | None = None,
        raw: bool = False,
        **kwargs,
    ):
        merged = {
            "action": action,
            "input_key": input_key,
            "output_key": output_key,
            "value": value,
            "template": template,
            "raw": raw,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    async def execute(self, ctx, state: dict) -> dict:
        action = self.config.get("action", "")
        input_key = self.config.get("input_key", "")
        output_key = self.config.get("output_key", "")
        value = self.config.get("value")
        field = self.config.get("field")

        if action == "json_get":
            data = state.get(input_key) if input_key else value
            result = self._json_get(
                data, field, raw=bool(self.config.get("raw", False))
            )
            state[output_key] = result
            return {output_key: result}

        if action == "pick":
            data = state.get(input_key) if input_key else value
            result = self._json_get(
                data, field, raw=bool(self.config.get("raw", False))
            )
            state[output_key] = result
            return {output_key: result}

        if action == "append":
            item = self._render_item(state, input_key, value)
            items = list(state.get(output_key, []))
            items.append(item)
            state[output_key] = items
            return {output_key: items}

        if action == "render":
            result = render_template(self.config.get("template", ""), state)
            state[output_key] = result
            return {output_key: result}

        if action == "contains":
            needle = value if value is not None else self.config.get("needle", "")
            result = "true" if str(needle) in str(state.get(input_key, "")) else "false"
            state[output_key] = result
            return {output_key: result}

        if action == "compare":
            op = self.config.get("op", "eq")
            result = (
                "true" if self._compare(state.get(input_key), value, op) else "false"
            )
            state[output_key] = result
            return {output_key: result}

        if action == "split":
            sep = self.config.get("sep", ",")
            result = str(state.get(input_key, "")).split(sep)
            state[output_key] = result
            return {output_key: result}

        if action == "join":
            sep = self.config.get("sep", ",")
            items = state.get(input_key) or []
            result = sep.join(str(i) for i in items)
            state[output_key] = result
            return {output_key: result}

        if action == "replace":
            old = self.config.get("old", "")
            new = self.config.get("new", "")
            result = str(state.get(input_key, "")).replace(old, new)
            state[output_key] = result
            return {output_key: result}

        if action == "coalesce":
            source = state.get(input_key)
            if source in (None, ""):
                source = value if value is not None else ""
            result = source if isinstance(source, str) else str(source)
            state[output_key] = result
            return {output_key: result}

        if action in ("to_int", "to_float"):
            source = str(state.get(input_key, "")).strip()
            result = (
                str(int(float(source))) if action == "to_int" else str(float(source))
            )
            state[output_key] = result
            return {output_key: result}

        if action == "now":
            result = _dt.datetime.now(_dt.timezone.utc).isoformat()
            state[output_key] = result
            return {output_key: result}

        source = value if value is not None else state.get(input_key, "")
        result = self._apply(action, source)
        state[output_key] = result
        return {output_key: result}

    def _render_item(self, state: dict, input_key: str, value: str | None) -> str:
        template = self.config.get("template")
        if template:
            return render_template(template, state)
        if value is not None:
            return value
        return str(state.get(input_key, ""))

    def _json_get(self, data, field: str | None, raw: bool = False) -> Any:
        if not isinstance(data, dict):
            raise ValueError(
                f"json_get requires a dict in state key, got {type(data).__name__}"
            )
        if not field:
            raise ValueError("json_get requires 'field'")
        if field not in data:
            raise KeyError(f"json_get: no field {field!r} in object")
        value = data[field]
        if raw:
            return value
        return value if isinstance(value, str) else str(value)

    def _compare(self, lhs, rhs: Any, op: str) -> bool:
        a, b = lhs, rhs
        try:
            a, b = float(str(a).strip()), float(str(b).strip())
        except (ValueError, TypeError):
            a, b = str(a), str(b)
        if op == "eq":
            return a == b
        if op == "ne":
            return a != b
        if op == "gt":
            return a > b
        if op == "ge":
            return a >= b
        if op == "lt":
            return a < b
        if op == "le":
            return a <= b
        raise ValueError(f"unknown compare op: {op}")

    def _apply(self, action: str, text: str) -> str:
        if action == "uppercase":
            return text.upper()
        if action == "lowercase":
            return text.lower()
        if action == "trim":
            return text.strip() if isinstance(text, str) else str(text).strip()
        if action == "count_lines":
            return str(len(text.splitlines()))
        if action == "value":
            return text if isinstance(text, str) else str(text)
        msg = f"unknown transform action: {action}"
        raise ValueError(msg)

Validate

Bases: Node

Decode an interrupt answer into a flow.loop decider value.

Works on two kinds of input:

• a raw answer (a string from the interrupt resume) matched by the equals / any_of / regex / check strategies;
• a verdict dict (from an LLM classifier) read via ok_field, with value_field captured into value_key.

Each evaluation increments rounds_key; once it reaches max_rounds the node is forced to pass_value so the enclosing loop terminates deterministically instead of spinning forever.

Config

input_key: State key holding the raw answer or verdict object. strategy: Matching strategy for raw answers. equals/any_of/regex/check: Strategy parameters (raw answers). verdict_key: State key holding the classifier's verdict object. ok_field: Pass-flag field of the verdict object. output_key: State key receiving pass_value / fail_value. pass_value/fail_value: Decision values written on pass / fail. clear_field: Optional verdict boolean naming "is this answer decipherable". When it is False the node writes clarify_value instead of pass/fail (re-ask, no body). clarify_value: Decision value written when clear_field is False (falls back to fail_value when empty). value_key: State key receiving the extracted value (cleared on a fail). Empty to skip. value_field: Verdict field captured into value_key. rounds_key: State key with the evaluation counter (incremented). max_rounds: After this many evaluations the node is forced to pass. missing_is_ok: Treat a missing / non-dict input as a pass.

Source code in teff/node/ask.py

class Validate(Node):
    """Decode an interrupt answer into a ``flow.loop`` decider value.

    Works on two kinds of input:

    * a **raw answer** (a string from the interrupt resume) matched by the
      ``equals`` / ``any_of`` / ``regex`` / ``check`` strategies;
    * a **verdict dict** (from an ``LLM`` classifier) read via *ok_field*,
      with *value_field* captured into *value_key*.

    Each evaluation increments ``rounds_key``; once it reaches
    ``max_rounds`` the node is forced to ``pass_value`` so the enclosing
    loop terminates deterministically instead of spinning forever.

    Config:
        input_key: State key holding the raw answer or verdict object.
        strategy: Matching strategy for raw answers.
        equals/any_of/regex/check: Strategy parameters (raw answers).
        verdict_key: State key holding the classifier's verdict object.
        ok_field: Pass-flag field of the verdict object.
        output_key: State key receiving ``pass_value`` / ``fail_value``.
        pass_value/fail_value: Decision values written on pass / fail.
        clear_field: Optional verdict boolean naming "is this answer
            decipherable".  When it is ``False`` the node writes
            ``clarify_value`` instead of pass/fail (re-ask, no body).
        clarify_value: Decision value written when *clear_field* is
            ``False`` (falls back to *fail_value* when empty).
        value_key: State key receiving the extracted value (cleared on a
            fail).  Empty to skip.
        value_field: Verdict field captured into *value_key*.
        rounds_key: State key with the evaluation counter (incremented).
        max_rounds: After this many evaluations the node is forced to pass.
        missing_is_ok: Treat a missing / non-dict input as a pass.
    """

    type = "validate"

    def __init__(
        self,
        config: dict | None = None,
        *,
        input_key: str = "answer",
        strategy: str = "",
        equals: Optional[str] = None,
        any_of: Optional[list] = None,
        regex: Optional[str] = None,
        check: Optional[Callable] = None,
        verdict_key: str = "verdict",
        ok_field: str = "ok",
        output_key: str = "decision",
        pass_value: str = "да",
        fail_value: str = "нет",
        clear_field: str = "",
        clarify_value: str = "",
        value_key: str = "",
        value_field: str = "",
        rounds_key: str = "rounds",
        max_rounds: int = 100,
        missing_is_ok: bool = False,
        **kwargs,
    ):
        merged = {
            "input_key": input_key,
            "strategy": strategy,
            "equals": equals,
            "any_of": any_of,
            "regex": regex,
            "check": check,
            "verdict_key": verdict_key,
            "ok_field": ok_field,
            "output_key": output_key,
            "pass_value": pass_value,
            "fail_value": fail_value,
            "clear_field": clear_field,
            "clarify_value": clarify_value,
            "value_key": value_key,
            "value_field": value_field,
            "rounds_key": rounds_key,
            "max_rounds": max_rounds,
            "missing_is_ok": missing_is_ok,
            **(config or {}),
            **kwargs,
        }
        super().__init__(**merged)

    def _match(self, raw):
        """Return ``(ok, extracted)`` for a raw answer."""
        cfg = self.config
        strategy = cfg["strategy"]
        if strategy == "equals":
            ok = _norm(raw) == _norm(cfg["equals"])
            return ok, (raw if ok else None)
        if strategy == "any_of":
            ok = _norm(raw) in {_norm(v) for v in cfg["any_of"]}
            return ok, (raw if ok else None)
        if strategy == "regex":
            m = re.search(cfg["regex"], str(raw or ""))
            ok = m is not None
            value = None
            if m:
                value = m.group(1) if m.groups() else m.group(0)
            return ok, value
        if strategy == "check":
            res = cfg["check"](raw)
            if isinstance(res, tuple):
                ok, value = res
                return bool(ok), value
            ok = bool(res)
            return ok, (raw if ok else None)
        if isinstance(raw, dict):
            ok = bool(raw.get(cfg["ok_field"], cfg["missing_is_ok"]))
            value = raw.get(cfg["value_field"]) if cfg["value_field"] else None
            return ok, value
        return bool(cfg["missing_is_ok"]), None

    async def execute(self, ctx, state: dict) -> dict:
        cfg = self.config
        rounds = int(state.get(cfg["rounds_key"], 0) or 0) + 1

        data = state.get(cfg["input_key"])
        if isinstance(data, dict):
            ok = bool(data.get(cfg["ok_field"], cfg["missing_is_ok"]))
            value = data.get(cfg["value_field"]) if cfg["value_field"] else None
            clear = cfg["clear_field"] == "" or bool(
                data.get(cfg["clear_field"], False)
            )
        else:
            ok, value = self._match(data)
            clear = True

        forced = rounds >= int(cfg["max_rounds"])
        if not forced and not clear:
            # the verdict is unclear — route to the "re-ask" branch (no body)
            decision = cfg["clarify_value"] or cfg["fail_value"]
            passed = False
        else:
            passed = bool(ok or forced)
            decision = cfg["pass_value"] if passed else cfg["fail_value"]

        out: dict = {
            cfg["rounds_key"]: rounds,
            cfg["output_key"]: decision,
        }
        if cfg["value_key"]:
            out[cfg["value_key"]] = value if passed else ""
        return out

last_user_message

last_user_message(messages)

Return the most recent user message from a conversation list.

Parameters:

Name	Type	Description	Default
messages	list	List of {"role": ..., "content": ...} dicts.	required

Returns:

Type	Description
str	The latest user content, or "" when there is none.

Source code in teff/node/context.py

def last_user_message(messages: list) -> str:
    """Return the most recent ``user`` message from a conversation list.

    Args:
        messages: List of ``{"role": ..., "content": ...}`` dicts.

    Returns:
        The latest user content, or ``""`` when there is none.
    """
    for message in reversed(messages):
        if message.get("role") == "user":
            return str(message.get("content", ""))
    return ""