Workflow Runtime Loop Step#

Source: examples/workflow/workflow_runtime_loop_step.py

Introduction#

Tree of Thoughts and ReAct each motivate iterative reasoning with explicit state updates, and AutoGen provides a practical framing for orchestrating repeated loop actions. This example demonstrates loop-step execution in the workflow runtime, including bounded iteration behavior and trace emission.

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_counter_loop<br/>LoopStep<br/>max_iterations=10"]
    subgraph loop_body_1["Loop Body: design_counter_loop"]
        direction TD
        loop_entry_1["design_counter_loop iteration entry"]
        step_2["design_counter_loop::increment<br/>LogicStep"]
        step_3["design_counter_loop::snapshot<br/>LogicStep"]
        loop_entry_1 --> step_2
        step_2 --> step_3
        step_3 -. "next iteration" .-> loop_entry_1
    end
    workflow_entry --> step_1
    step_1 -. "iterate" .-> loop_entry_1
    
 1from __future__ import annotations
 2
 3import json
 4from collections.abc import Mapping
 5from pathlib import Path
 6
 7import design_research_agents as drag
 8
 9WORKFLOW_DIAGRAM_DIRECTION = "LR"
10
11
12def _increment_handler(context: drag.WorkflowContext) -> Mapping[str, object]:
13    loop_state = context.get("loop_state")
14    state_mapping = loop_state if isinstance(loop_state, Mapping) else {}
15    return {"counter": int(state_mapping.get("counter", 0)) + 1}
16
17
18def _snapshot_handler(context: drag.WorkflowContext) -> Mapping[str, object]:
19    increment_output = context.dependency_output("increment")
20    counter = int(increment_output.get("counter", 0))
21    return {
22        "counter": counter,
23        "status": "threshold_met" if counter >= 3 else "looping",
24    }
25
26
27def _state_reducer(
28    state: Mapping[str, object],
29    iteration_result: object,
30    iteration: int,
31) -> Mapping[str, object]:
32    del state, iteration
33    step_results = getattr(iteration_result, "step_results", {})
34    if not isinstance(step_results, Mapping):
35        return {"counter": 0}
36    increment = step_results.get("increment")
37    increment_output = getattr(increment, "output", {})
38    if not isinstance(increment_output, Mapping):
39        return {"counter": 0}
40    return {
41        "counter": int(increment_output.get("counter", 0)),
42    }
43
44
45def build_example_workflow(*, tracer: drag.Tracer | None = None) -> drag.Workflow:
46    """Build the bounded loop workflow used for runtime illustration and docs diagrams."""
47    return drag.Workflow(
48        tool_runtime=None,
49        input_schema={"type": "object"},
50        tracer=tracer,
51        steps=[
52            drag.LoopStep(
53                step_id="design_counter_loop",
54                steps=(
55                    drag.LogicStep(step_id="increment", handler=_increment_handler),
56                    drag.LogicStep(
57                        step_id="snapshot",
58                        dependencies=("increment",),
59                        handler=_snapshot_handler,
60                    ),
61                ),
62                max_iterations=10,
63                initial_state={"counter": 0},
64                continue_predicate=lambda iteration, state: int(state.get("counter", 0)) < 3,
65                state_reducer=_state_reducer,
66                execution_mode="sequential",
67                failure_policy="skip_dependents",
68            )
69        ],
70    )
71
72
73def main() -> None:
74    """Run a small loop and print compact JSON summary."""
75    # Fixed request id keeps traces and docs output deterministic across runs.
76    request_id = "example-workflow-loop-design-001"
77    tracer = drag.Tracer(
78        enabled=True,
79        trace_dir=Path("artifacts/examples/traces"),
80        enable_jsonl=True,
81        enable_console=True,
82    )
83    # Build and run the loop workflow using public runtime APIs.
84    workflow = build_example_workflow(tracer=tracer)
85
86    result = workflow.run({}, execution_mode="sequential", request_id=request_id)
87    # Print the results
88    summary = result.summary()
89    print(json.dumps(summary, ensure_ascii=True, indent=2, sort_keys=True))
90
91
92if __name__ == "__main__":
93    main()

Expected Results#

Run Command

PYTHONPATH=src python3 examples/workflow/workflow_runtime_loop_step.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#