runtime_data & flow_data

In TriggerFlow, state changes are signals. set_runtime_data() and set_flow_data() trigger events that when() can capture.

runtime_data (execution-scoped)

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
async def set_runtime(data: TriggerFlowEventData):
  data.set_runtime_data("user_id", "u-001")
  return "runtime ok"

@flow.chunk
def print_value(data: TriggerFlowEventData):
  print(data.value)

flow.when({"runtime_data": "user_id"}).to(print_value)

flow_data (flow-scoped)

from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()

@flow.chunk
def print_value(data: TriggerFlowEventData):
  print(data.value)

flow.set_flow_data("env", "prod")
flow.when({"flow_data": "env"}).to(print_value)

Note: flow_data affects all running and future executions under the same Flow. Use it only when you are certain. If you use a Flow as a shared service object to create per-user executions, avoid flow_data to prevent data leakage or cross-request contamination.

collect

collect() aggregates branches and emits Collect-xxx when all branches fill.