Build reliable workflows

Use a workflow when your application knows the shape of the process. Use an agent when the model should decide which tool, specialist, or next action to choose.

Typical workflow use cases include intake and review pipelines, parallel research followed by synthesis, bounded refinement, and resumable human approval.

Stability by language

Language	Current contract
TypeScript	createWorkflow(), runWorkflow(), replay, every built-in state service, workflow evaluation baselines and gates, and artifact helpers are stable. File workflow-state pruning remains beta.
Python	Sequential, parallel, loop, durable WorkflowGraph, checkpoint stores, resume, fork, leases, and external-runtime adapters are beta. Use top-level zhivex_ai imports and review minor-version changes before upgrading.

The two SDKs intentionally do not claim identical workflow APIs. Keep orchestration behind an application service if the same product supports both languages.

TypeScript: a deterministic workflow

This credential-free example runs two steps in order:

import {
  createAgent,
  createInMemorySessionService,
  createMockLanguageModel,
  createRunner,
  createTextMessage,
  createWorkflow,
  runWorkflow
} from "@zhivex-ai/sdk";

const model = createMockLanguageModel({
  responses: [
    {
      text: "intake-ok",
      finishReason: "stop",
      messages: [createTextMessage("assistant", "intake-ok")]
    },
    {
      text: "review-ok",
      finishReason: "stop",
      messages: [createTextMessage("assistant", "review-ok")]
    }
  ]
});

const sessionService = createInMemorySessionService();
const runner = createRunner({
  appName: "candidate-review",
  agent: createAgent({ id: "reviewer", model, maxSteps: 1 }),
  sessionService
});

const workflow = createWorkflow({
  id: "candidate-review",
  steps: [
    {
      id: "intake",
      runner,
      prompt: "Validate the candidate input.",
      outputKey: "intake"
    },
    {
      id: "review",
      runner,
      prompt: ({ outputs }) => `Review: ${String(outputs.intake)}`,
      outputKey: "review"
    }
  ]
});

const result = await runWorkflow(workflow, {
  userId: "user_123",
  sessionId: "candidate_456"
});

console.log(result.status, result.outputs);

Replace the mock model with a configured provider from Choose a provider when moving from an offline test to an integration environment.

Python: a sequential workflow

The Python workflow layer is beta, but it supports the same fixed-pipeline pattern:

import asyncio

from zhivex_ai import (
    Agent,
    GenerateResult,
    SequentialAgent,
    WorkflowStep,
    create_mock_language_model,
)


def mock_agent(name: str, text: str) -> Agent:
    model = create_mock_language_model(
        responses=[GenerateResult(text=text, finish_reason="stop")]
    )
    return Agent(name=name, model=model)


async def main() -> None:
    workflow = SequentialAgent(
        name="candidate_review",
        steps=[
            WorkflowStep(
                "intake",
                mock_agent("intake", "intake-ok"),
                prompt="Validate the input",
                output_key="intake",
            ),
            WorkflowStep(
                "review",
                mock_agent("review", "review-ok"),
                input_template="Review {intake}",
                output_key="review",
            ),
        ],
    )

    result = await workflow.run()
    print(result.status, result.state)


asyncio.run(main())

Use WorkflowBuilder and WorkflowGraph when Python needs validated DAG routing, append-only checkpoints, interruption, resume, or fork. A graph is acyclic; use LoopAgent for intentional bounded repetition.

Choose the smallest control flow

Need	TypeScript	Python
Fixed sequence	sequential steps	SequentialAgent
One fan-out wave	kind: "parallel"	ParallelAgent
Bounded refinement	kind: "loop" with maxIterations	LoopAgent with max_iterations
Durable branching graph	compose steps and persisted workflow state	beta WorkflowGraph
Model-directed delegation	agent tools or subagents	handoffs or subagent tools

Keep every loop bounded and choose an explicit failure policy. A workflow should never depend on an unlimited model loop to reach a terminal state.

Persist and resume deliberately

For long-running TypeScript processes, add a workflow state service to the definition. In-memory, file, SQLite, and Postgres implementations share a Stable contract. File-backed state is useful locally; a shared deployment should use a reviewed database backend and exercise the exact installed package against a production-shaped schema.

Python WorkflowGraph can append a checkpoint for every transition and resume only when the workflow name, definition version, and definition digest match. Reconstruct runtime clients and dependencies when resuming: they are not serialized into checkpoints.

In both languages:

• map the authenticated subject and tenant before loading workflow state;
• use stable step and workflow identifiers because they become replay evidence;
• store large or binary outputs as artifacts instead of embedding them in state;
• bind approval to the exact pending request and consume it once;
• propagate cancellation into networked tools and workers;
• expire, archive, and delete persisted state under application retention policy.

Gate TypeScript workflow regressions

TypeScript 1.5 adds Stable, versioned evaluation baselines and fail-closed gates. Keep the dataset, judge configuration, baseline, thresholds, and SDK version in review together; a model judge is still a provider call and must not silently replace deterministic assertions.

Use runWorkflowEvaluationFixture() to execute a fixture, createWorkflowEvaluationBaseline() to pin the reviewed report, and evaluateWorkflowEvaluationGate() to compare a candidate. The installed CLI exposes the same boundary through zhivex-ai workflow eval, workflow baseline, and workflow gate.

Workflow outputs and replay evidence can be saved through the Stable Artifact Service. Choose bounded payload limits and tenant-scoped identifiers; large production objects should stay in application-owned object storage with explicit external references.

Side effects are still app-owned

A durable checkpoint proves orchestration progress. It does not make an email, payment, deployment, or database write atomic with that checkpoint.

Use a stable idempotency key at the destination, and add an outbox, fencing, or reconciliation when the external outcome could be unknown. Resuming or retrying a workflow without this boundary can repeat an effect.

The application also owns authorization, tenant isolation, business policy, approval UI, artifact storage, workflow-engine operation, and audit retention. Python callback adapters for Temporal, Prefect, DBOS, and Restate are contracts—not embedded workers or certified deployments.

Test before using live providers

Start with mock models and deterministic expectations. Then add a credentialed smoke for the exact provider, model, workflow path, and release candidate that will run in production.

Use Observability and evaluations for replay, traces, regression fixtures, and CI gates. Review Production architecture before selecting durable stores or exposing resume and approval endpoints.

Full references

• TypeScript workflows guide
• TypeScript runnable workflow
• Python durable workflows
• Python workflow examples