Workflow Runtime#

Source: examples/workflow/workflow_runtime.py

Introduction#

Human-AI collaboration by design motivates transparent orchestration boundaries, AutoGen motivates composable multi-component execution, and HELM motivates repeatable runtime instrumentation for comparisons. This example is the minimal workflow-runtime build for observing step execution semantics directly.

Technical Implementation#

  1. Configure Tracer with JSONL + console output so each run emits machine-readable traces and lifecycle logs.

  2. Build the runtime surface (public APIs only) and execute Workflow.run(...) with a fixed request_id.

  3. Capture structured outputs from runtime execution and preserve termination metadata for analysis.

  4. Print a compact JSON payload including trace_info for deterministic tests and docs examples.

The diagram below is generated from the example’s configured Workflow.

        flowchart LR
    workflow_entry["Workflow Entrypoint"]
    step_1["design_runtime_ready<br/>LogicStep"]
    workflow_entry --> step_1
    
 1from __future__ import annotations
 2
 3import json
 4from pathlib import Path
 5
 6import design_research_agents as drag
 7
 8WORKFLOW_DIAGRAM_DIRECTION = "LR"
 9
10
11def build_example_workflow(*, tracer: drag.Tracer | None = None) -> drag.Workflow:
12    """Build the minimal workflow used for runtime illustration and docs diagrams."""
13    return drag.Workflow(
14        tool_runtime=None,
15        input_schema={"type": "object"},
16        tracer=tracer,
17        steps=[
18            drag.LogicStep(
19                step_id="design_runtime_ready",
20                handler=lambda _context: {
21                    "message": "Design runtime orchestration validated.",
22                    "check": "workflow-runtime-ready",
23                },
24            )
25        ],
26    )
27
28
29def main() -> None:
30    """Run a minimal logic workflow and print literal payload."""
31    # Fixed request id keeps traces and docs output deterministic across runs.
32    request_id = "example-workflow-runtime-design-001"
33    tracer = drag.Tracer(
34        enabled=True,
35        trace_dir=Path("artifacts/examples/traces"),
36        enable_jsonl=True,
37        enable_console=True,
38    )
39    # Build and run the minimal workflow using public runtime APIs.
40    workflow = build_example_workflow(tracer=tracer)
41    result = workflow.run({}, execution_mode="sequential", request_id=request_id)
42    # Print the results
43    summary = result.summary()
44    print(json.dumps(summary, ensure_ascii=True, indent=2, sort_keys=True))
45
46
47if __name__ == "__main__":
48    main()

Expected Results#

Run Command

PYTHONPATH=src python3 examples/workflow/workflow_runtime.py

Example output shape (values vary by run):

{
  "success": true,
  "final_output": "<example-specific payload>",
  "terminated_reason": "<string-or-null>",
  "error": null,
  "trace": {
    "request_id": "<request-id>",
    "trace_dir": "artifacts/examples/traces",
    "trace_path": "artifacts/examples/traces/run_<timestamp>_<request_id>.jsonl"
  }
}

References#