Agentic Workflows

Stabilize ships a small set of opt-in building blocks for LLM-agent systems. They are additive — existing workflows are unaffected — and layer on top of the engine’s durable, event-sourced execution so agent runs inherit crash recovery, replay, and the full control-flow pattern set.

Four capabilities are covered here:

Streaming — consume a workflow’s progress (and LLM tokens) live.
Human-in-the-loop (HITL) — pause for an approval and resume with a value.
Fan-in reducers — combine parallel branches instead of clobbering them.
LLM toolkit — a chat client, tool-calling, and a ReAct agent loop.

A complete, runnable program that uses all four (a multi-agent “software team” that designs, writes, tests, reviews, and packages a small library, with a chaos-recovery mode) lives in examples/agent_team/.

Streaming Progress

Tasks can emit custom progress events, and a caller can consume them live via WorkflowStream. This is built on the event bus, so it works whether or not durable event sourcing is configured.

from stabilize import WorkflowStream, emit_progress

# Inside a task's execute():
def execute(self, stage):
    emit_progress(stage, "generating candidate", percent=10)
    ...
    emit_progress(stage, "done", percent=100)
    return TaskResult.success(outputs={...})

# In the caller — callback form (non-blocking):
stream = WorkflowStream(execution_id)
stream.on_event(lambda item: print(item.data.get("message", "")))
# ... run the workflow ...
stream.close()

# Or iterate live until the workflow ends:
stream = WorkflowStream(execution_id, event_store=event_store)
for item in stream.follow(include_history=True, timeout=30):
    print(item.source, item.event_type, item.data)

Each item is a StreamItem with event_type, entity_id, workflow_id, data, sequence, and source ("replay" or "live"). Progress events carry event_type == "custom.progress"; lifecycle events (stage.completed, workflow.completed, …) flow through the same stream.

Human-in-the-Loop Approvals

The ApprovalTask suspends a stage until an approval signal arrives, then routes on it. It wraps the engine’s durable TaskResult.suspend() plus persistent signal buffering, so an approval sent before the stage suspends is not lost, and the whole gate survives a crash.

from stabilize import ApprovalTask, approve, reject

registry.register("approval", ApprovalTask)

gate = StageExecution(
    ref_id="gate",
    type="approval",
    requisite_stage_ref_ids={"prepare"},
    tasks=[TaskExecution.create(
        name="wait", implementing_class="approval",
        stage_start=True, stage_end=True,
    )],
)

# ... the stage runs, suspends, and waits ...

# A human (or another system) approves — the payload lands in the stage's
# outputs under "approval":
approve(queue, execution_id, gate.id, {"user": "alice"})
# or: reject(queue, execution_id, gate.id, {"reason": "needs work"})

By default a rejection is terminal; set context["approval_reject_continues"] = True to continue the pipeline (FAILED_CONTINUE) instead. For custom suspend/resume tasks, use send_signal() to deliver an arbitrary signal and get_signal() inside execute to read which signal woke the stage.

Declarative Fan-in Reducers

When several parallel branches write the same output key, the default ancestor-output merge is last-write-wins for scalar values (list values already concatenate). Set output_reducers on a join stage to combine the per-branch values declaratively instead of losing all but one.

from stabilize import StageExecution, JoinType

gather = StageExecution(
    ref_id="gather",
    join_type=JoinType.AND,
    requisite_stage_ref_ids={"gen_1", "gen_2", "gen_3"},
    output_reducers={"candidate": "collect", "score": "sum"},
    tasks=[...],
)

Each branch that produced candidate contributes to a list under that key; score values are summed. Built-in reducers: collect / append, extend, sum, max, min, merge, first, last. Register your own with register_reducer(). Keys without a reducer keep the existing last-write-wins behavior, so this is fully backward compatible.

LLM Toolkit

stabilize.llm is a dependency-light (standard-library only) toolkit for LLM agents. It is not imported by the core engine, so it adds no cost to workflows that do not use it. The model client is injectable, targeting OpenAI-compatible endpoints or Ollama (local or cloud).

from stabilize.llm import LLMClient, LLMTask, AgentLoopTask, tool, ToolRegistry

# A chat client (credentials from OLLAMA_API_KEY / OPENAI_API_KEY by default)
client = LLMClient(model="glm-5.2", base_url="https://ollama.com", api="ollama")

# A one-shot completion as a workflow task:
registry.register("llm", LLMTask(client=client))

# A tool-calling ReAct agent as a single durable task:
@tool
def get_weather(city: str) -> str:
    "Return the weather for a city."
    return lookup(city)

tools = ToolRegistry().add(get_weather)
registry.register("agent", AgentLoopTask(client=client, tools=tools))

AgentLoopTask runs a bounded model → tool-calls → results → model loop inside one task execution (capped by max_iterations), so the whole agent turn is one durable, retryable unit of work. The @tool decorator generates an OpenAI-style JSON schema from the function signature and docstring; ToolRegistry dispatches the model’s tool calls back to your functions.

Note

Provide API keys via environment variables (OLLAMA_API_KEY / OPENAI_API_KEY) — never hard-code them. The example in examples/agent_team/ reads OLLAMA_API_KEY and falls back to offline stubs when it is unset, so it always runs end to end.

Agentic Workflows

Streaming Progress

Human-in-the-Loop Approvals

Declarative Fan-in Reducers

LLM Toolkit

See Also