Agentic

Opt-in building blocks for LLM-agent workflows. See the Agentic Workflows guide for usage.

Streaming

Streaming surface for live workflow progress (agentic ergonomics).

LangGraph exposes stream/astream to feed live values, state updates, and LLM tokens to a caller. Stabilize already has the plumbing — an event-sourced log plus a thread-safe EventBus — so this module only adds the missing consumer surface:

  • emit_progress() lets a task push a custom progress event (a step narration, an LLM token chunk, a percentage) from inside execute.

  • WorkflowStream lets a caller consume a workflow’s events, either live (subscribe to the bus) or by replaying the durable log, or both (replay history, then follow live) — all additive and opt-in.

Nothing here changes existing behavior: workflows that never call emit_progress and callers that never open a WorkflowStream are unaffected.

class stabilize.streaming.StreamItem(event_type, entity_type, entity_id, workflow_id, data, sequence, source)[source]

Bases: object

One item yielded by a WorkflowStream.

Parameters:

  • event_type (str)

  • entity_type (str)

  • entity_id (str)

  • workflow_id (str)

  • data (dict[str, Any])

  • sequence (int)

  • source (str)

event_type: str
entity_type: str
entity_id: str
workflow_id: str
data: dict[str, Any]
sequence: int
source: str
classmethod from_event(event, source)[source]
Parameters:

  • event (Event)

  • source (str)

Return type:

StreamItem

stabilize.streaming.emit_progress(stage, message='', *, recorder=None, **data)[source]

Emit a custom progress event for a running stage.

Publishes a CUSTOM event carrying message and any extra keyword data to the event bus (and, if a recorder is given, the durable log). Callers watching a WorkflowStream see it live. Safe to call from a task’s execute — if event sourcing is not configured, this is a cheap no-op against the always-available global bus.

Parameters:

  • stage (StageExecution) – The stage emitting progress (must have an execution).

  • message (str) – Human-readable progress line.

  • recorder (EventRecorder | None) – Optional recorder to also persist the event durably.

  • **data (Any) – Extra structured fields (e.g. token=”…”, percent=42).

Return type:

None

class stabilize.streaming.WorkflowStream(execution_id, event_store=None, subscription_id=None)[source]

Bases: object

Consume a workflow’s events live and/or from the durable log.

Example (live follow until the workflow ends):

stream = WorkflowStream(execution_id, event_store=store)
for item in stream.follow():
    print(item.event_type, item.data.get("message", ""))

Example (callback form, non-blocking):

stream = WorkflowStream(execution_id)
stream.on_event(lambda item: print(item))
...  # run the workflow
stream.close()
Parameters:

  • execution_id (str)

  • event_store (EventStore | None)

  • subscription_id (str | None)

replay()[source]

Yield all durably-recorded events for this workflow, in order.

Return type:

Iterator[StreamItem]

on_event(callback, *, mode=SubscriptionMode.SYNC)[source]

Register a callback invoked with each live StreamItem.

Non-blocking. Call close() to unsubscribe.

Parameters:
Return type:

None

follow(include_history=False, timeout=None)[source]

Yield events for this workflow as they occur, until it ends.

Parameters:

  • include_history (bool) – First replay the durable log (requires an event_store), then continue live. Events already replayed are de-duplicated against the live feed by sequence.

  • timeout (float | None) – Stop after this many seconds of no new events (None waits indefinitely, but always stops on a workflow-terminal event).

Return type:

Iterator[StreamItem]

close()[source]

Unsubscribe from the live event bus.

Return type:

None

Human-in-the-Loop

Human-in-the-loop (HITL) helpers (agentic ergonomics).

LangGraph’s interrupt pauses a graph and returns the human’s value inline on resume. Stabilize already has the durable machinery — TaskResult.suspend sets a stage SUSPENDED and SignalStage(persistent=True) buffers a signal sent before the stage suspends, all surviving a crash via the atomic queue. This module wraps that into a friendly, one-call API plus a builtin task:

  • ApprovalTask — a ready-made task that suspends until an approve/reject signal arrives, then routes accordingly.

  • approve() / reject() / send_signal() — send the resume signal (with a payload) to a suspended stage via an orchestrator/queue.

  • get_signal() — read the delivered signal name/data inside a task.

All additive: existing suspend/signal users are unaffected.

class stabilize.hitl.ApprovalTask[source]

Bases: Task

A task that waits for a human approval signal, then routes on it.

Behavior: - First execution (no signal yet): returns TaskResult.suspend(); the

stage becomes SUSPENDED and waits.

  • Resumed by an approve signal: succeeds, exposing the approver’s payload in outputs under approval.

  • Resumed by a reject signal: fails. By default this is terminal; set context['approval_reject_continues'] = True to continue the pipeline (FAILED_CONTINUE) instead.

Register it and use it like any task:

registry.register("approval", ApprovalTask)
StageExecution(ref_id="gate", type="approval", tasks=[...])

Send the decision with approve() / reject().

aliases = ['human_approval', 'hitl_approval']
execute(stage)[source]

Execute the task.

Parameters:

stage (StageExecution) – The stage execution context

Returns:

TaskResult indicating status and any outputs

Raises:

Exception – Any exception will be caught and handled by the runner

Return type:

TaskResult

stabilize.hitl.send_signal(queue, execution_id, stage_id, signal_name, signal_data=None, *, execution_type='PIPELINE', persistent=True)[source]

Send a resume signal to a (possibly not-yet-)suspended stage.

Parameters:

  • queue (Queue) – The workflow queue.

  • execution_id (str) – Workflow id.

  • stage_id (str) – The suspended stage’s id.

  • signal_name (str) – Signal name the task checks for (e.g. “approve”).

  • signal_data (dict[str, Any] | None) – Payload delivered to the task as _signal_data.

  • execution_type (str) – Workflow type value (default “PIPELINE”).

  • persistent (bool) – Buffer the signal if the stage has not suspended yet (WCP-24). True by default so an early approval is not lost.

Return type:

None

stabilize.hitl.approve(queue, execution_id, stage_id, data=None, *, execution_type='PIPELINE')[source]

Approve a stage waiting on an ApprovalTask.

Parameters:

  • queue (Queue)

  • execution_id (str)

  • stage_id (str)

  • data (dict[str, Any] | None)

  • execution_type (str)

Return type:

None

stabilize.hitl.reject(queue, execution_id, stage_id, data=None, *, execution_type='PIPELINE')[source]

Reject a stage waiting on an ApprovalTask.

Parameters:

  • queue (Queue)

  • execution_id (str)

  • stage_id (str)

  • data (dict[str, Any] | None)

  • execution_type (str)

Return type:

None

stabilize.hitl.get_signal(stage)[source]

Return (signal_name, signal_data) delivered to a resumed stage.

Use inside a task’s execute to branch on which signal woke it. Returns (None, {}) when the stage was not resumed by a signal.

Parameters:

stage (StageExecution)

Return type:

tuple[str | None, dict[str, Any]]

Fan-in Reducers

Declarative fan-in reducers for parallel branches (agentic ergonomics).

LangGraph channels take a reducer (e.g. operator.add) so parallel nodes APPEND to a shared list instead of overwriting it. Stabilize’s default ancestor-output merge is last-write-wins for scalar keys (list-valued keys already concatenate), so N parallel branches each writing {"result": ...} silently lose all but one.

This module adds opt-in, declarative reducers applied when a join stage collects its upstream branches’ outputs. A stage sets output_reducers:

StageExecution(

ref_id=”gather”, join_type=JoinType.AND, requisite_stage_ref_ids={“gen_1”, “gen_2”, “gen_3”}, output_reducers={“candidate”: “collect”, “score”: “sum”},

)

Each upstream branch that produced candidate contributes to a list under that key in the join stage’s context; score values are summed. Keys without a reducer keep the existing last-write-wins behavior, so this is fully backward compatible.

Built-in reducers (name -> behavior): - collect / append: gather each branch’s value into a list (a

branch’s list value is extended, a scalar is appended).

  • extend: flatten list values from all branches into one list.

  • sum: numeric sum across branches.

  • max / min: extremum across branches.

  • merge: shallow-merge dict values across branches.

  • first / last: first/last branch’s value (last == current default).

Custom reducers can be registered with register_reducer().

stabilize.reducers.register_reducer(name, reducer)[source]

Register a custom named reducer usable in output_reducers.

Parameters:

  • name (str)

  • reducer (Callable[[list[Any]], Any])

Return type:

None

stabilize.reducers.get_reducer(name)[source]

Look up a reducer by name (custom overrides built-in), or None.

Parameters:

name (str)

Return type:

Callable[[list[Any]], Any] | None

stabilize.reducers.apply_output_reducers(reducers, branch_outputs)[source]

Combine per-branch outputs for the reducer-controlled keys.

Parameters:

  • reducers (dict[str, str]) – Mapping of output key -> reducer name.

  • branch_outputs (list[dict[str, Any]]) – Each upstream branch’s outputs dict.

Returns:

A dict of reduced values for exactly the keys named in reducers that at least one branch produced. Keys not named here are left to the caller’s normal merge.

Return type:

dict[str, Any]

LLM Toolkit

Dependency-light chat client for OpenAI-compatible / Ollama endpoints.

Uses only the standard library (urllib) so the LLM toolkit adds no runtime dependency and runs on a laptop. The endpoint and credentials are injectable so the same code targets Ollama (local or ollama.com cloud), OpenAI, or any OpenAI-compatible gateway.

class stabilize.llm.client.ChatMessage(role, content=None, tool_calls=None, tool_call_id=None, name=None)[source]

Bases: object

One chat message. tool_calls / tool_call_id support tool loops.

Parameters:

  • role (str)

  • content (str | None)

  • tool_calls (list[dict[str, Any]] | None)

  • tool_call_id (str | None)

  • name (str | None)

role: str
content: str | None = None
tool_calls: list[dict[str, Any]] | None = None
tool_call_id: str | None = None
name: str | None = None
to_dict()[source]
Return type:

dict[str, Any]

class stabilize.llm.client.ChatResponse(content, tool_calls=<factory>, raw=<factory>)[source]

Bases: object

A model response: assistant text and any requested tool calls.

Parameters:

  • content (str)

  • tool_calls (list[dict[str, Any]])

  • raw (dict[str, Any])

content: str
tool_calls: list[dict[str, Any]]
raw: dict[str, Any]
property has_tool_calls: bool
class stabilize.llm.client.LLMClient(model, base_url=None, api_key=None, api='openai', timeout=120.0, default_options=None)[source]

Bases: object

Minimal OpenAI-compatible / Ollama chat client (stdlib only).

Two wire formats are supported: - api="openai" (default): POST {base_url}/chat/completions with

the OpenAI schema. Works for OpenAI and OpenAI-compatible gateways, including ollama.com cloud’s OpenAI endpoint.

  • api="ollama": POST {base_url}/api/chat with the native Ollama schema (for a local ollama serve).

Credentials default to the OLLAMA_API_KEY / OPENAI_API_KEY env vars; never hard-code keys.

Parameters:

  • model (str)

  • base_url (str | None)

  • api_key (str | None)

  • api (str)

  • timeout (float)

  • default_options (dict[str, Any] | None)

chat(messages, tools=None, temperature=None, **options)[source]

Send a chat request and return the assistant response.

Parameters:

  • messages (list[ChatMessage]) – Conversation so far.

  • tools (list[dict[str, Any]] | None) – Optional OpenAI-style tool/function schemas to expose.

  • temperature (float | None) – Sampling temperature (falls back to default_options).

  • **options (Any) – Extra provider options merged into the request.

Return type:

ChatResponse

exception stabilize.llm.client.LLMError[source]

Bases: Exception

Raised when an LLM chat request fails.

Tool definitions for LLM tool-calling loops.

The @tool decorator turns a plain Python function into a callable tool with an auto-generated OpenAI-style JSON schema (derived from its signature and type hints). A ToolRegistry collects tools, exposes their schemas to the model, and dispatches the model’s tool calls back to the functions.

class stabilize.llm.tools.ToolSpec(func, name, description)[source]

Bases: object

A registered tool: the callable plus its generated JSON schema.

Parameters:

  • func (Callable[..., Any])

  • name (str)

  • description (str)

call(arguments)[source]
Parameters:

arguments (dict[str, Any])

Return type:

Any

stabilize.llm.tools.tool(_func=None, *, name=None, description=None)[source]

Decorate a function as an LLM tool.

The schema is derived from the signature and docstring:

@tool
def get_weather(city: str) -> str:
    "Return the weather for a city."
    ...
Parameters:

  • _func (Callable[[...], Any] | None)

  • name (str | None)

  • description (str | None)

Return type:

Any

class stabilize.llm.tools.ToolRegistry[source]

Bases: object

Collects tools and dispatches model tool calls to them.

add(func)[source]

Register a @tool-decorated function (or a plain callable).

Parameters:

func (Callable[[...], Any])

Return type:

ToolRegistry

schemas()[source]

OpenAI-style tool schemas for every registered tool.

Return type:

list[dict[str, Any]]

dispatch(tool_call)[source]

Execute one model tool call and return its result.

Accepts both OpenAI ({"function": {"name","arguments"}}) and Ollama ({"function": {"name","arguments": {...}}}) shapes.

Parameters:

tool_call (dict[str, Any])

Return type:

Any

LLM tasks: a plain chat task and a ReAct tool-calling agent loop.

Built on the engine’s existing primitives so they inherit its durability: - LLMTask calls the model once and returns the completion. - AgentLoopTask runs a bounded ReAct loop (model -> tool calls ->

results -> model …) inside a single task execution, so the whole agent turn is one durable, retryable unit of work.

The model client is injectable (constructor or the LLMTask.build_client hook) so these stay dependency-light and target Ollama or OpenAI-compatible endpoints without hard-coding anything.

class stabilize.llm.tasks.LLMTask(client=None)[source]

Bases: Task

Call an LLM once with a prompt/messages and return the completion.

Context keys (all optional unless noted): - prompt (str): user prompt. Required unless messages is given. - system (str): system prompt. - messages (list): full message list (overrides prompt/system). - model (str): model id (else the injected client’s model). - temperature (float): sampling temperature. - output_key (str): outputs key for the completion (default

"completion").

Outputs: {output_key: <text>, "llm_raw": <provider response>}.

Parameters:

client (LLMClient | None)

aliases = ['llm', 'chat']
build_client(stage)[source]

Return the LLM client. Override to construct one from context, or pass a client to the constructor.

Parameters:

stage (StageExecution)

Return type:

LLMClient

execute(stage)[source]

Execute the task.

Parameters:

stage (StageExecution) – The stage execution context

Returns:

TaskResult indicating status and any outputs

Raises:

Exception – Any exception will be caught and handled by the runner

Return type:

TaskResult

class stabilize.llm.tasks.AgentLoopTask(client=None, tools=None)[source]

Bases: LLMTask

A bounded ReAct tool-calling loop inside one task execution.

Each iteration: send the conversation (plus tool schemas) to the model; if it returns tool calls, dispatch them via the ToolRegistry, append the results, and loop; otherwise return the final answer. Bounded by max_iterations so a misbehaving model cannot loop forever.

Provide the tools by passing a ToolRegistry to the constructor or by overriding build_tools().

Context keys add to LLMTask: - max_iterations (int): loop cap (default 8). - output_key (str): outputs key for the final answer (default

"answer").

Outputs: {output_key: <final text>, "tool_invocations": [...], "iterations": <n>}.

Parameters:
aliases = ['agent', 'react_agent', 'agent_loop']
build_tools(stage)[source]

Return the tool registry for this agent. Override or pass to ctor.

Parameters:

stage (StageExecution)

Return type:

ToolRegistry

execute(stage)[source]

Execute the task.

Parameters:

stage (StageExecution) – The stage execution context

Returns:

TaskResult indicating status and any outputs

Raises:

Exception – Any exception will be caught and handled by the runner

Return type:

TaskResult