google/lightflow

Creating Lightflow Workflows (create_lightflows)

Lightflow coordinates multi-step operations between AI agents and human operators using an append-only state ledger (passport.json), safe AST-based CEL-like expressions, and selective subgraph re-arming.

Note: If you only need to execute, inspect, or resume an existing Lightflow workflow, read the run_lightflows skill instead.

1. Scaffolding a New Workflow via create_lightflow (3-Phase Meta-Workflow)

Instead of authoring a workflow from scratch in one shot, run the built-in 3-phase meta-workflow, which gates progress on deterministic AST signature checks, sample_payload return-contract smoke tests, unit test execution, and payload.outputs.<stage> DAG cross-checks. The meta-workflow ships in the Lightflow repository under examples/create_lightflow/: run it from a clone of github.com/google/lightflow, or pass the absolute path to examples/create_lightflow in your clone:

lightflow start \
  --lightflow=/path/to/lightflow/examples/create_lightflow \
  --log_id=scaffold_<workflow_name>

(python3 -m lightflow is equivalent to lightflow whenever the console script is not on PATH.)

1. align_on_design (operator_action): Align with the user on stage boundaries (separate read/compute stages from external write/mutate stages), operator_action gates, and the output keys each stage produces under payload.outputs.<stage>.
2. implement_and_test_actions $\rightarrow$ verify_actions: Write actions.py + unit tests (or pass sample_payload); verify_actions parses the AST, smoke-tests that every action returns a JSON-serializable (dict, str) tuple, and runs unittest.
3. draft_and_explain_manifest $\rightarrow$ prove_manifest_dry_run: Write lightflow.yaml and explain the DAG to the user; prove_manifest_dry_run compiles the DAG, verifies that every payload.outputs.<stage> reference points to an upstream dependency in run_after, runs dry_run, and emits visualizer.html.

2. Workflow Manifest Reference (lightflow.yaml)

Define actions and stages in lightflow.yaml (or .json / .textproto). python_import resolves relative to the workflow's directory (e.g. actions.fetch_metrics for actions.py in the same folder) as well as package paths:

name: release_verification_pipeline
description: Fetch metrics, pause for operator review, and publish report.

actions:
  - id: fetch_metrics
    python_import: actions.fetch_metrics
  - id: publish_report
    python_import: actions.publish_report
  - id: cleanup_temp
    python_import: actions.cleanup_temp

stages:
  - name: fetch
    description: Fetch latest metrics
    timeout_seconds: 60
    python_action:
      action_id: fetch_metrics
      static_kwargs:
        limit: 500
    retry_policy:
      max_attempts: 3
      initial_backoff_seconds: 5
      backoff_multiplier: 2.0
    rollback_action:
      action_id: cleanup_temp

  - name: approve
    description: Operator approval gate
    run_after: [fetch]
    operator_action:
      # CEL-like expression: string literals MUST be quoted inside the string.
      instructions: "'Ready to publish: ' + payload.outputs.fetch.summary"
      json_schema:
        type: object
        required: [approved]
        properties:
          approved:
            type: boolean

  - name: publish
    description: Publish verified report
    run_after: [approve]
    run_if: "payload.outputs.approve.approved == true"
    python_action:
      action_id: publish_report

  - name: always_cleanup
    run_after: [publish]
    trigger_rule: ALL_DONE
    python_action:
      action_id: cleanup_temp

3. Three Rules for Crash-Free Python Actions

from typing import Any

def fetch_metrics(
    payload: dict[str, Any], **static_kwargs: Any
) -> tuple[dict[str, Any], str]:
  limit = static_kwargs.get("limit", 100)
  return {"row_count": limit, "summary": f"{limit} rows"}, "Fetched metrics"

1. One Side Effect per Stage: Separate read/compute stages from external write/mutate stages. Lightflow's Passport never re-runs COMPLETED stages on resume; make write stages safe to retry if they fail mid-execution (overwrite files instead of appending, or attach a rollback_action).
2. Consistent Output Keys (payload.outputs.<stage_name>): Every return path of an action (including any offline test mode for external APIs) must return a JSON-serializable (dict, str) with the exact same dictionary keys so downstream run_if and instructions expressions never crash. If an action cannot do its job (including its own plumbing failing), raise an exception: the stage is stamped FAILED and resume re-runs only it (or use lightflow resume --lightflow=. --log_id=<id> --rerun=<stage> to re-run an already completed stage and its downstream dependents). Never return a failure verdict as a normal result.
3. Cross-Check payload.outputs.<stage> Against run_after: Any stage whose run_if or instructions reads payload.outputs.<upstream>.<key> must list <upstream> (directly or transitively) in run_after.

4. Validating Your New Workflow

Before handing off or running in production, always verify compilation, CEL-like expression evaluation, and gate schemas:

lightflow dry_run --lightflow=.
lightflow render --lightflow=.
lightflow visualize --lightflow=. --out=/tmp/visualizer.html