TriggerFlow Orchestration Playbook

Scenario

Multi-step workflows need conditional branching, routing, parallelism, async behavior, and state signals.

1. Orchestration capability map

How to read this diagram

• These are not isolated APIs. They are composable orchestration primitives.
• The engineering value comes from combining the main chain, concurrency, signals, convergence, and side-channel observability into one observable flow.

2. Capability (key traits)

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

3. Operations

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

4. Full code (parallel + runtime_stream)

import asyncio
from agently import TriggerFlow, TriggerFlowRuntimeData

flow = TriggerFlow()


@flow.chunk("normalize")
async def normalize(data: TriggerFlowRuntimeData):
    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: TriggerFlowRuntimeData):
    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: TriggerFlowRuntimeData):
    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: TriggerFlowRuntimeData):
    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)

5. 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)'}

6. Validation

• parallel tasks complete
• runtime_stream emits stage events
• aggregated result is correct