run-llama/workflows-py
1
0
Fork 0
mirror of https://github.com/run-llama/workflows-py.git synced 2026-07-18 16:14:58 -04:00
Code Issues 18 Packages Projects Releases 10 Wiki Activity

[Feature Request] Improve the map-reduce (fan-out/gather) syntax #9

New Issue
Open
opened 2026-02-16 02:16:03 -05:00 by yindo · 0 comments
yindo commented 2026-02-16 02:16:03 -05:00
Owner
Copy Link
Copy Source

Originally created by @logan-markewich on GitHub (Jul 29, 2025).

Currently, to do a map reduce pattern, you need a decent amount of boilerplate

import random

from workflows import Context, Workflow, step
from workflows.events import Event, StartEvent, StopEvent


class TaskRequested(Event):
    pass


class TaskReady(Event):
    pass

class FanOutFanIn(Workflow):
    @step
    async def fan_out(self, ctx: Context, ev: StartEvent) -> TaskRequested | None:
        times = random.randint(1, 10)
        await ctx.store.set("times", times)
        for _ in range(times):
            ctx.send_event(TaskRequested())

    @step
    async def process_task(self, ev: TaskRequested) -> TaskReady:
        # do work
        return TaskReady()

    @step
    async def fan_in(self, ctx: Context, ev: TaskReady) -> StopEvent | None:
        num_to_collect = await ctx.store.get("times")
        events = ctx.collect_events(ev, [TaskReady] * num_to_collect)
        if events is not None
            return StopEvent(results=events)

Notice several things

  • extremely verbose, lots of places to make a mistake
  • technically incorrect step annotations to make the workflow validator happy (the map step doesn't actually "return" MapEvent)
  • needing to know ahead of time how many events to collect

I think this pattern really starts to break down when you get more complex patterns. What if I have several levels of mapping?

I think ideally, there is a better way to do this, or at the very least, nicer syntactic sugar around this

Originally created by @logan-markewich on GitHub (Jul 29, 2025). Currently, to do a map reduce pattern, you need a decent amount of boilerplate ```py import random from workflows import Context, Workflow, step from workflows.events import Event, StartEvent, StopEvent class TaskRequested(Event): pass class TaskReady(Event): pass class FanOutFanIn(Workflow): @step async def fan_out(self, ctx: Context, ev: StartEvent) -> TaskRequested | None: times = random.randint(1, 10) await ctx.store.set("times", times) for _ in range(times): ctx.send_event(TaskRequested()) @step async def process_task(self, ev: TaskRequested) -> TaskReady: # do work return TaskReady() @step async def fan_in(self, ctx: Context, ev: TaskReady) -> StopEvent | None: num_to_collect = await ctx.store.get("times") events = ctx.collect_events(ev, [TaskReady] * num_to_collect) if events is not None return StopEvent(results=events) ``` Notice several things - extremely verbose, lots of places to make a mistake - technically incorrect step annotations to make the workflow validator happy (the map step doesn't actually "return" MapEvent) - needing to know ahead of time how many events to collect I think this pattern really starts to break down when you get more complex patterns. What if I have several levels of mapping? I think ideally, there is a better way to do this, or at the very least, nicer syntactic sugar around this
yindo added the enhancement label 2026-02-16 02:16:03 -05:00
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
bug
enhancement
enhancement
good first issue
pull-request
Mirrored from GitHub Pull Request
 question
No Label enhancement
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: run-llama/workflows-py#9
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?