RuntimeEventData

事件元数据与运行态 API。

源码

from agently import TriggerFlow, TriggerFlowEventData


## TriggerFlow RuntimeEventData: data, emit, and layers
def runtime_event_data_demo():
    # Idea: inspect event metadata and emit a custom event in-flow.
    # Flow: to(inspect_event) -> emit(CustomEvent) -> when(CustomEvent)
    # Expect: prints a custom event payload and returns event details.
    flow = TriggerFlow()

    async def inspect_event(data: TriggerFlowEventData):
        # RuntimeEventData provides:
        # - trigger_event / trigger_type / value
        # - runtime_data accessors (execution-scoped)
        # - emit / async_emit to trigger custom events
        # - layer_marks for nested branches
        data.set_runtime_data("seen_event", data.trigger_event)
        await data.async_emit("CustomEvent", {"from": data.trigger_event, "value": data.value})
        return {
            "event": data.trigger_event,
            "type": data.trigger_type,
            "value": data.value,
            "runtime": data.get_runtime_data("seen_event"),
            "layer": data.layer_mark,
        }

    flow.to(inspect_event).end()
    flow.when("CustomEvent").to(lambda d: print("[custom]", d.value))

    result = flow.start("hello")
    print(result)


# runtime_event_data_demo()

讲解

• 可读取 trigger 信息并发出事件。

注释解读

• Idea 表示案例思路
• Flow 表示执行编排路径
• Expect 表示预期输出或行为

你学会了什么

• 掌握 RuntimeEventData 的关键字段
• 理解 trigger_event/trigger_type

练习任务

• 打印 layer_mark 并观察变化