Nodes reference

Every node is a steps: entry in YAML (type: <name>) and a class in the Flow API. All nodes share the same contract: async def execute(ctx, state) -> dict. Plain functions work too — see Core concepts.

Built-in node types

type	Flow class	Purpose
transform	Transform	String transforms on state values
llm_chat	LLM	A single LLM call with prompts, structured output, tools
react_agent	ReActAgent / Harness	Tool-calling agent loop (LLM + tool_exec)
tool_exec	ToolExec	Execute tool calls signalled by an agent, in parallel
tool_call	ToolCall	Invoke a registered tool by name with fixed args
interrupt	Interrupt	Pause for human input; resume via checkpoint
parallel	Parallel	Run branch chains concurrently, merge with reducers
map	Map	Dynamic fan-out of a state list into parallel branches
context_builder	ContextBuilder	Compose a scratch prompt from state + conversation
append_assistant	AppendAssistant	Append the result as an assistant message
supervisor	Supervisor	Ask a model "which agent next" + deterministic guards
gate	Gate	Turn a verdict object into a loop decider + retry budget
validate	Validate	Decode an interrupt answer (raw or verdict) into a loop decider; capture a value
command	CommandNode	Declarative goto/STOP routing from state conditions
loop	Loop	Repeat a body chain until a state condition holds

Wrapper nodes

Retry

Wrap any node with retry logic — retries the inner node up to max_retries times with an optional delay (seconds) between attempts. On final failure the last exception propagates:

from teff import Retry, LLM

flow.step(Retry(LLM(model="gpt-4", output_key="answer"), max_retries=3, delay=1.0))

Key	Type	Default	Description
node	Node	—	The inner node to retry
max_retries	int	3	Attempts (including the first)
delay	float	0.0	Seconds to wait between attempts

Retries are recorded on the tracer (tracer.retry(...)) and surfaced in graph.stream() events.

transform

String/data transforms. Actions: uppercase, lowercase, trim, count_lines, value (set a literal), render (render a {key} template into a scalar), json_get (extract a field from a dict), append (render a template and accumulate into a list).

Pipeline-building actions for pure-YAML workflows:

Action	Reads	Writes to output_key
contains	input_key (string) vs value (needle)	"true" / "false"
compare	input_key vs value with op (eq/ne/gt/ge/lt/le)	"true" / "false"
split	input_key (string) by sep (default ,)	list
join	input_key (list) by sep (default ,)	string
replace	input_key, old → new (default "")	string
coalesce	input_key, falls back to value when empty	string
pick	field out of a dict in input_key (like json_get)	value
to_int / to_float	input_key (numeric string)	number as string
now	—	current UTC ISO timestamp

contains and compare emit "true"/"false" strings so they drive edges: conditions directly (e.g. condition: "has_refund=true").

Key	Type	Default	Description
action	str	—	One of the actions above
input_key	str	""	State key to read from
output_key	str	""	State key to write to
value	str	None	Literal value (action: value, needle for contains, RHS of compare, coalesce fallback)
field	str	None	Field to extract with action: json_get / action: pick
template	str	None	{key} template for action: render / action: append
raw	bool	False	Keep json_get/pick values without stringifying
sep	str	,	Separator for split / join
op	str	eq	Operator for action: compare
old / new	str	—	Replacement pair for action: replace

llm_chat

One model call. Provider, sampling, caching and retry keys are shared with react_agent.

Key	Type	Default	Description
model	str	—	Model name (required, or default_model on the graph)
provider	str	—	Provider key (must be declared in providers=; see Providers)
system	str	—	System prompt (supports {key} templates)
prompt	str	—	User prompt with {key} templates
input_key	str	—	Read a single state key as the user message
output_key	str	"output"	Where the reply lands
json_schema / output_type	dict/type	—	Structured output validation
parse	bool	False	Parse the reply as JSON into a dict (no validation)
use_tools	bool/list	—	Tool scope: True, False, or a list of tool names
skills / skill_dir	list/str	—	Mount skills
temperature / max_tokens	float/int	—	Sampling knobs
max_retries / fallbacks	int/list	—	Retry + model failover
cache	bool	False	Dedupe identical calls
max_tool_rounds	int	10	Max model calls per visit

react_agent / tool_exec

The ReAct loop is two nodes: the agent (react_agent) proposes tool calls and the executor (tool_exec) runs them in parallel, then routes back on _tool_call_name !=.

Key	Type	Default	Description
input_key / output_key	str	—	Entry question / final answer
messages_key	str	"messages"	Conversation state key
tool_call_key	str	"_tool_call_name"	Signal key for routing
use_tools	bool/list	True	Tool scope for the agent
tool_error_mode	str	"message"	"message" (model sees the error) or "raise"
tool_timeout / tool_retries	float/int	—	Bound and retry each tool call
tool_approval	str/callable	"auto"	"auto", "deny", "interactive", or callable
parse_text_tool_calls	bool	True	Decode tool calls from plain text (local models)
max_tool_rounds	int	10	Max model calls per graph visit
max_total_tokens	int	—	Token budget for the whole agent run
max_context_tokens / trim_messages	int/bool	—	Trim the conversation before each call

See Agents for the full harness surface.

tool_call

Invoke one registered tool with explicit arguments (no model involved).

Key	Type	Default	Description
tool	str	—	Registered tool name
args	dict	—	Tool args; values support {key} templates
output_key	str	"output"	State key for the result
on_error	str	"raise"	"raise" or "message"
max_chars	int	—	Truncate the result

interrupt

Key	Type	Default	Description
key	str	—	State key the resume value lands in
prompt	str	—	Question shown to the operator
messages_key	str	"messages"	Conversation state key
reset_keys	list	—	Scratch keys to clear before resume

Pauses raise GraphInterrupt; resume with the same checkpoint_id plus resume={key: answer}. Requires a checkpointer. See Durable execution.

To validate the answer (instead of comparing it verbatim) and capture a value, pair the interrupt with an Ask strategy — see validate and Ask below.

In YAML, a strategy: mapping on the interrupt step expands to the classifier + validate chain automatically ({id}-classifier, {id}-validate), the YAML counterpart of flow.interrupt(key, prompt, accept=...):

- id: gate
  type: interrupt
  config:
    key: approved
    prompt: "Approve the report? (yes / no)"
    strategy: {equals: да}   # or any_of: [да, ок] | regex: "^[A-Z0-9]{4}$" | llm: {system, user, schema, model, provider}

The llm strategy requires model and provider in the strategy block. Edges that would have sourced from the interrupt now source from {id}-validate, where the decision key (decision by default) is written.

command

Declarative goto/STOP routing from state — the YAML surface for Command. Returns a Command whose goto is the first matching route (when uses the same expression language as edges: conditions), falling back to goto; update merges state keys after routing.

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

Key	Type	Default	Description
routes	list	[]	{when, goto} pairs; first match wins
goto	str	—	Fallback target, or STOP to end the run
update	dict	—	State keys merged after routing

loop

Repeat a body chain until state[key] equals until — the self-contained sibling of flow.loop, expressible entirely in YAML. Each round runs body (a node or list of nodes, given as inline type: ... specs like map's processor), then checks key=until with the edges condition language (so until: "да" matches "Да" or "да."). max_rounds (default 10) bounds the repetition.

- id: refine
  type: loop
  config:
    key: approved
    until: "да"
    max_rounds: 3
    body:
      - {type: transform, config: {action: value, value: "нет", output_key: approved}}

Key	Type	Default	Description
body	node/list	—	Chain run each round (inline type: ... specs)
key	str	—	State key the condition reads
until	str	—	Value of key that stops the loop
max_rounds	int	10	Maximum body rounds before giving up

parallel / map

• parallel — config branches: list of branch chains (each a node or list of nodes, or an embedded sub-flow). Branches merge with per-key reducers.
• map — config input_keys (state list keys, zipped), output_key (list of per-item results), result_key (per-item result key), chunk_size (items per branch, default 1), max_concurrency (cap on branches).

See State.

context_builder / append_assistant

Compose the scratch prompt from named state sections and the conversation:

• context_builder — sections (state key → section label map), messages_key (default "messages"), output_key (default "input"), reset_keys.
• append_assistant — output_key (default "draft"), messages_key.

supervisor

The decider for a supervisor loop: ask the model for a one-word route, then apply deterministic guards. Wired with flow.supervisor() or used directly with flow.route().

Key	Default	Description
model / provider	—	LLM model and provider for the harness.
system	""	System prompt (list the reply values + finish).
output_key	"next_agent"	State key that receives the chosen route.
sections	{}	State key → label map rendered into the prompt as progress.
route_keys	{}	Map route value → output slot; a picked agent whose slot already has content is not re-routed.
done_keys / done_mode	{} / "all"	When these output slots are filled, return finish with no model call ("any" = just one).
fallback_agent	""	Route to this agent when finish is picked before anything is produced.
rounds_key / max_rounds	"supervisor_rounds" / 6	Force finish once the counter reaches max_rounds.
messages_key	"messages"	Source of the user message; "" means always consult the model.
agents	—	Explicit reply vocabulary (default: route_keys ∪ {"finish"} ∪ {fallback_agent}).

See Supervisors — a ready-made decider for the guards and the _needs_model / decide override hooks.

gate

Turn a verdict object (typically structured JSON from an LLM) into the discriminator value a flow.loop / flow.branch switches on — the "approve or fix" loop behind QA and review cycles. Each evaluation increments rounds_key; once it reaches max_rounds the gate is forced to pass_value so the loop terminates instead of raising a max_iterations error.

Key	Type	Default	Description
input_key	str	"verdict"	State key holding the verdict object (LLM(json_schema=...) output).
ok_field	str	"ok"	Field of the verdict treated as the pass flag.
output_key	str	"decision"	State key receiving pass_value / fail_value.
pass_value	str	"yes"	Written on pass (the value loop compares until against).
fail_value	str	"fix"	Written when the verdict fails.
rounds_key	str	"rounds"	Evaluation counter, incremented each run.
max_rounds	int	3	After this many evaluations the gate is forced to pass_value.
message_field	str	"message"	Field of the verdict with the remarks.
message_key	str	""	State key receiving the remarks (cleared on a pass); empty disables.
missing_is_ok	bool	True	A missing / non-dict verdict counts as a pass.

from teff.node import Gate, LLM

flow.step(qa_llm)  # LLM(json_schema=QaVerdict) -> state["qa_verdict"]
flow.step(Gate(input_key="qa_verdict", output_key="qa_ok", rounds_key="qa_rounds"))
flow.loop(
    key="qa_ok",
    until="yes",
    done=finalize,
    body=[planner, estimator, qa_llm],
)

validate

Like gate, but built for interrupt answers: it turns the raw answer (or a classifier verdict) into the discriminator value a flow.loop / flow.branch switches on, and can capture an arbitrary value (a discount code, a date, …) into value_key. Each evaluation increments rounds_key; once it reaches max_rounds the node is forced to pass_value so the loop terminates instead of raising a max_iterations error.

Key	Type	Default	Description
input_key	str	"answer"	State key holding the raw answer, or the verdict object for a model Ask.
strategy	str	""	"equals" / "any_of" / "regex" / "check" for raw answers; "model" when input_key holds a verdict.
equals / any_of / regex / check	—	—	Raw-answer matching: exact (normalized) value, a set of values, a regex, or a callable fn(value) -> bool / (bool, extracted).
verdict_key / ok_field	str	"verdict" / "ok"	Where a model classifier's verdict object lives and its pass flag.
output_key	str	"decision"	State key receiving pass_value / fail_value.
pass_value	str	"да"	Written on pass (the value loop compares until against).
fail_value	str	"нет"	Written when the answer fails.
value_key	str	""	State key receiving the captured value (cleared on a fail); empty disables.
value_field	str	""	Verdict field (for model) captured into value_key.
rounds_key	str	"rounds"	Evaluation counter, incremented each run.
max_rounds	int	100	After this many evaluations the node is forced to pass_value.
missing_is_ok	bool	False	A missing / empty answer counts as a pass.

Ask

Ask is not a node — it's the declarative strategy an interrupt uses to decide pass/fail and capture a value. flow.interrupt(key, prompt, accept=Ask(...)) validates a single answer; flow.interrupt_loop(key, accept=Ask(...), body=..., done=...) re-asks until it passes. Use the classmethod constructors:

from teff.flow import Flow
from teff.node import Ask, LLM, Transform

# exact (normalized) match
Ask.equals("да", decision_key="plan_ok")

# any of several values
Ask.any_of("да", "ок", "конечно", decision_key="plan_ok")

# regex + capture the value into state["discount_code"]
Ask.regex(r"^[A-Z]{2}-[0-9]{4}$", decision_key="code_ok", value_key="discount_code")

# callable: fn(value) -> bool, or (bool, extracted)
Ask.check(lambda v: len(v) >= 8, value_key="password")

# LLM classifier normalizes free-form answers into {ok: bool, ...}
Ask.llm(
    system="Ты классифицируешь ответ пользователя...",
    user="Ответ пользователя:\n{approved}\n\nОдобрил?",
    schema={"type": "object", "properties": {"ok": {"type": "boolean"}}},
    model="llama3.1:8b",
    provider="ollama",
    verdict_key="verdict",
    decision_key="approved_ok",
)

flow.interrupt_loop(
    key="code",
    prompt="Введите промокод (формат XX-1234):",
    accept=Ask.regex(
        r"^[A-Z]{2}-[0-9]{4}$", decision_key="code_ok", value_key="discount_code"
    ),
    body=Transform(action="value", value="неверный код", output_key="total"),
    done=Transform(action="value", value="скидка применена", output_key="total"),
)

Ask is auto-detected from the constructor kwargs, so Ask(equals="да") and Ask(regex=..., value_key="code") work too. See the runnable example ask_strategies for all three strategies in one checkout flow.

Extract

Ask's sibling: instead of deciding pass/fail on an interrupt answer it extracts a structured object from the conversation. Extract is not a node — it's a declarative recipe that builds [LLM extractor, *Fallback nodes] (Extract.nodes()) ready to drop into a done/finish chain. Use the classmethod constructors:

from teff.flow import Flow
from teff.node import Extract

extractor = Extract.model(
    system="Ты извлекаешь данные проекта из переписки...",
    schema=PROJECT_INFO_SCHEMA,
    model="llama3.1:8b", provider="ollama",
    messages_key="messages",          # conversation history to scan
    output_key="project_info",        # parsed object lands here
    fallbacks=[
        Extract.fallback("room_type", room_from_first_user),
    ],
)

flow.interrupt_loop(key="approved", ..., done=extractor.nodes())

Extract.model is equivalent to a plain LLM(json_schema=...) — core LLM prepends the system prompt to the messages_key history, so the extraction sees the whole conversation. Extract.fallback(field, fn) declares a deterministic fill: when the model leaves field empty (a common failure mode of small local models), fn(state) runs and its return value is merged into the extracted object; fn returns None to skip. Any extra kwargs are threaded through to the LLM (e.g. max_retries).

fallback

A standalone node that fills a field a model left empty. Reads a dict from input_key; when field in it is empty/None, calls fn(state) and merges the result under field. No-op when the field is already set or fn returns None.

Key	Type	Default	Description
input_key	str	"output"	State key holding the extracted dict.
field	str	""	Dict field to fill when empty.
fn	callable	—	fn(state) -> value \| None.

Registering custom types

Use decorators or subclasses — see Plugins. The current registry:

from teff.node.registry import default_registry

print(default_registry.list())