TriggerFlow Orchestration Playbook

Scenario

Multi‑step workflows need routing, parallelism, async, and signals.

Capability (key traits)

• to: main chain
• if_condition / match: routing
• for_each: list processing
• batch + concurrency: parallel control
• runtime_data: state signals

Operations

1. Chain with to.
2. Route with if_condition / match.
3. Parallelize with for_each or batch.

Full code (parallel + runtime_stream)

import asyncio
from agently import TriggerFlow, TriggerFlowEventData

flow = TriggerFlow()


@flow.chunk("normalize")
async def normalize(data: TriggerFlowEventData):
    topic = str(data.value).strip()
    data.set_runtime_data("topic", topic)
    data.put_into_stream({"stage": "normalized", "topic": topic})
    return topic


@flow.chunk("fetch_facts")
async def fetch_facts(data: TriggerFlowEventData):
    await asyncio.sleep(0.05)
    data.put_into_stream({"stage": "facts_ready", "topic": data.value})
    return f"facts({data.value})"


@flow.chunk("fetch_risks")
async def fetch_risks(data: TriggerFlowEventData):
    await asyncio.sleep(0.03)
    data.put_into_stream({"stage": "risks_ready", "topic": data.value})
    return f"risks({data.value})"


@flow.chunk("compile_report")
async def compile_report(data: TriggerFlowEventData):
    topic = data.get_runtime_data("topic")
    report = {
        "topic": topic,
        "facts": data.value.get("fetch_facts"),
        "risks": data.value.get("fetch_risks"),
    }
    data.put_into_stream({"stage": "compiled", "report": report})
    data.stop_stream()
    return report


flow.to(normalize)
flow.when({"runtime_data": "topic"}).batch(fetch_facts, fetch_risks, concurrency=2).to(compile_report).end()

execution = flow.create_execution(concurrency=2)

for item in execution.get_runtime_stream("Agently TriggerFlow", timeout=5):
    print("STREAM:", item)

result = execution.get_result(timeout=5)
print("RESULT:", result)

Real output

STREAM: {'stage': 'facts_ready', 'topic': 'Agently TriggerFlow'}
STREAM: {'stage': 'risks_ready', 'topic': 'Agently TriggerFlow'}
STREAM: {'stage': 'compiled', 'report': {'topic': 'Agently TriggerFlow', 'facts': 'facts(Agently TriggerFlow)', 'risks': 'risks(Agently TriggerFlow)'}}
RESULT: {'topic': 'Agently TriggerFlow', 'facts': 'facts(Agently TriggerFlow)', 'risks': 'risks(Agently TriggerFlow)'}

Validation

• Parallel tasks finish.
• runtime_stream emits stages.
• Aggregated result is correct.