merge all workflows docs into a single place (#234)

Author: logan-markewich

docs/src/content/docs/llamaagents/workflows/assets/arize.png

@@ -0,0 +1,107 @@
+---
+sidebar:
+  order: 2
+title: Branches and loops
+---
+
+A key feature of Workflows is their enablement of branching and looping logic, more simply and flexibly than graph-based approaches.
+
+## Loops in workflows
+
+To create a loop, we'll take a `LoopingWorkflow` that randomly loops. It will have a single event that we'll call `LoopEvent` (but it can have any arbitrary name).
+
+```python
+from workflows.events import Event
+
+class LoopEvent(Event):
+    num_loops: int
+```
+
+Now we'll `import random` and modify our `step_one` function to randomly decide either to loop or to continue:
+
+```python
+import random
+from workflows import Workflow, step
+from workflows.events import StartEvent, StopEvent
+
+class LoopingWorkflow(Workflow):
+    @step
+    async def prepare_input(self, ev: StartEvent) -> LoopEvent:
+        num_loops = random.randint(0, 10)
+        return LoopEvent(num_loops=num_loops)
+
+    @step
+    async def loop_step(self, ev: LoopEvent) -> LoopEvent | StopEvent:
+        if ev.num_loops <= 0:
+            return StopEvent(result="Done looping!")
+
+        return LoopEvent(num_loops=ev.num_loops-1)
+```
+
+Let's visualize this:
+
+![A simple loop](./assets/loop.png)
+
+You can create a loop from any step to any other step by defining the appropriate event types and return types.
+
+## Branches in workflows
+
+Closely related to looping is branching. As you've already seen, you can conditionally return different events. Let's see a workflow that branches into two different paths:
+
+```python
+import random
+from workflows import Workflow, step
+from workflows.events import Event, StartEvent, StopEvent
+
+class BranchA1Event(Event):
+    payload: str
+
+
+class BranchA2Event(Event):
+    payload: str
+
+
+class BranchB1Event(Event):
+    payload: str
+
+
+class BranchB2Event(Event):
+    payload: str
+
+
+class BranchWorkflow(Workflow):
+    @step
+    async def start(self, ev: StartEvent) -> BranchA1Event | BranchB1Event:
+        if random.randint(0, 1) == 0:
+            print("Go to branch A")
+            return BranchA1Event(payload="Branch A")
+        else:
+            print("Go to branch B")
+            return BranchB1Event(payload="Branch B")
+
+    @step
+    async def step_a1(self, ev: BranchA1Event) -> BranchA2Event:
+        print(ev.payload)
+        return BranchA2Event(payload=ev.payload)
+
+    @step
+    async def step_b1(self, ev: BranchB1Event) -> BranchB2Event:
+        print(ev.payload)
+        return BranchB2Event(payload=ev.payload)
+
+    @step
+    async def step_a2(self, ev: BranchA2Event) -> StopEvent:
+        print(ev.payload)
+        return StopEvent(result="Branch A complete.")
+
+    @step
+    async def step_b2(self, ev: BranchB2Event) -> StopEvent:
+        print(ev.payload)
+        return StopEvent(result="Branch B complete.")
+```
+
+Our imports are the same as before, but we've created 4 new event types. `start` randomly decides to take one branch or another, and then multiple steps in each branch complete the workflow. Let's visualize this:
+
+![A simple branch](./assets/branching.png)
+
+You can of course combine branches and loops in any order to fulfill the needs of your application. Later in this tutorial you'll learn how to run multiple branches in parallel using `send_event` and synchronize them using `collect_events`.

docs/src/content/docs/llamaagents/workflows/assets/branching.png

@@ -0,0 +1,170 @@
+---
+sidebar:
+  order: 5
+title: Concurrent execution of workflows
+---
+
+In addition to looping, branching, and streaming, workflows can run steps concurrently. This is useful when you have multiple steps that can be run independently of each other and they have time-consuming operations that they `await`, allowing other steps to run in parallel.
+
+## Emitting multiple events
+
+To emit multiple events to trigger multiple steps, you can use `ctx.send_event()`:
+
+```python
+import asyncio
+from workflows import Workflow, Context, step
+from workflows.events import Event, StartEvent, StopEvent
+
+class StepTwoEvent(Event):
+    query: str
+
+class ParallelFlow(Workflow):
+    @step
+    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
+        ctx.send_event(StepTwoEvent(query="Query 1"))
+        ctx.send_event(StepTwoEvent(query="Query 2"))
+        ctx.send_event(StepTwoEvent(query="Query 3"))
+
+    @step(num_workers=4)
+    async def step_two(self, ev: StepTwoEvent) -> StopEvent:
+        print("Running slow query ", ev.query)
+        await asyncio.sleep(random.randint(0, 5))
+
+        return StopEvent(result=ev.query)
+```
+
+In this example, our `start` step emits 3 `StepTwoEvent`s. The `step_two` step is decorated with `num_workers=4`, which tells the workflow to run up to 4 instances of this step concurrently (this is the default).
+
+## Collecting events
+
+If you execute the previous example, you'll note that the workflow stops after whichever query is first to complete. Sometimes that's useful, but other times you'll want to wait for all your slow operations to complete before moving on to another step. You can do this using `collect_events`:
+
+```python
+import asyncio
+from workflows import Workflow, Context, step
+from workflows.events import Event, StartEvent, StopEvent
+
+class StepTwoEvent(Event):
+    query: str
+
+class StepThreeEvent(Event):
+    result: str
+
+class ConcurrentFlow(Workflow):
+    @step
+    async def start(self, ctx: Context, ev: StartEvent) -> StepTwoEvent | None:
+        ctx.send_event(StepTwoEvent(query="Query 1"))
+        ctx.send_event(StepTwoEvent(query="Query 2"))
+        ctx.send_event(StepTwoEvent(query="Query 3"))
+
+    @step(num_workers=4)
+    async def step_two(self, ctx: Context, ev: StepTwoEvent) -> StepThreeEvent:
+        print("Running query ", ev.query)
+        await asyncio.sleep(random.randint(1, 5))
+        return StepThreeEvent(result=ev.query)
+
+    @step
+    async def step_three(
+        self, ctx: Context, ev: StepThreeEvent
+    ) -> StopEvent | None:
+        # wait until we receive 3 events
+        result = ctx.collect_events(ev, [StepThreeEvent] * 3)
+        if result is None:
+            return None
+
+        # do something with all 3 results together
+        print(result)
+        return StopEvent(result="Done")
+```
+
+The `collect_events` method lives on the `Context` and takes the event that triggered the step and an array of event types to wait for. In this case, we are awaiting 3 events of the same `StepThreeEvent` type.
+
+The `step_three` step is fired every time a `StepThreeEvent` is received, but `collect_events` will return `None` until all 3 events have been received. At that point, the step will continue and you can do something with all 3 results together.
+
+The `result` returned from `collect_events` is an array of the events that were collected, in the order that they were received.
+
+## Multiple event types
+
+Of course, you do not need to wait for the same type of event. You can wait for any combination of events you like, such as in this example:
+
+```python
+import asyncio
+from workflows import Workflow, Context, step
+from workflows.events import Event, StartEvent, StopEvent
+
+class StepAEvent(Event):
+    query: str
+
+class StepBEvent(Event):
+    query: str
+
+class StepCEvent(Event):
+    query: str
+
+class StepACompleteEvent(Event):
+    result: str
+
+class StepBCompleteEvent(Event):
+    result: str
+
+class StepCCompleteEvent(Event):
+    result: str
+
+
+class ConcurrentFlow(Workflow):
+    @step
+    async def start(
+        self, ctx: Context, ev: StartEvent
+    ) -> StepAEvent | StepBEvent | StepCEvent | None:
+        ctx.send_event(StepAEvent(query="Query 1"))
+        ctx.send_event(StepBEvent(query="Query 2"))
+        ctx.send_event(StepCEvent(query="Query 3"))
+
+    @step
+    async def step_a(self, ctx: Context, ev: StepAEvent) -> StepACompleteEvent:
+        print("Doing something A-ish")
+        return StepACompleteEvent(result=ev.query)
+
+    @step
+    async def step_b(self, ctx: Context, ev: StepBEvent) -> StepBCompleteEvent:
+        print("Doing something B-ish")
+        return StepBCompleteEvent(result=ev.query)
+
+    @step
+    async def step_c(self, ctx: Context, ev: StepCEvent) -> StepCCompleteEvent:
+        print("Doing something C-ish")
+        return StepCCompleteEvent(result=ev.query)
+
+    @step
+    async def step_three(
+        self,
+        ctx: Context,
+        ev: StepACompleteEvent | StepBCompleteEvent | StepCCompleteEvent,
+    ) -> StopEvent:
+        print("Received event ", ev.result)
+
+        # wait until we receive 3 events
+        if (
+            ctx.collect_events(
+                ev,
+                [StepCCompleteEvent, StepACompleteEvent, StepBCompleteEvent],
+            )
+            is None
+        ):
+            return None
+
+        # do something with all 3 results together
+        return StopEvent(result="Done")
+```
+
+There are several changes we've made to handle multiple event types:
+
+- `start` is now declared as emitting 3 different event types
+- `step_three` is now declared as accepting 3 different event types
+- `collect_events` now takes an array of the event types to wait for
+
+Note that the order of the event types in the array passed to `collect_events` is important. The events will be returned in the order they are passed to `collect_events`, regardless of when they were received.
+
+The visualization of this workflow is quite pleasing:
+
+![A concurrent workflow](./assets/different_events.png)

docs/src/content/docs/llamaagents/workflows/assets/different_events.png

@@ -1,8 +1,10 @@
 ---
+sidebar:
+  order: 7
 title: Customizing entry and exit points
 ---
 
-Most of the times, relying on the default entry and exit points we have seen in the [Getting Started](/python/workflows/) section is enough.
+Most of the times, relying on the default entry and exit points we have seen in the [Getting Started](/python/llamaagents/workflows/) section is enough.
 However, workflows support custom events where you normally would expect `StartEvent` and `StopEvent`, let's see how.
 
 ## Using a custom `StartEvent`
@@ -41,7 +43,7 @@ class JokeFlow(Workflow):
     ) -> JokeEvent:
         # Build a query engine using the index and the llm from the start event
         query_engine = ev.an_index.as_query_engine(llm=ev.an_llm)
-        topic = query_engine.query(
+        topic = await query_engine.aquery(
             f"What is the closest topic to {a_string_field}"
         )
         # Use the llm attached to the start event to instruct the model

docs/src/content/docs/llamaagents/workflows/assets/joke.png

@@ -1,4 +1,6 @@
 ---
+sidebar:
+  order: 12
 title: Run Your Workflow as a Server
 ---
 
@@ -33,8 +35,9 @@ class GreetingWorkflow(Workflow):
             ctx.write_event_to_stream(StreamEvent(sequence=i))
             await asyncio.sleep(0.3)
 
-        name = getattr(ev, "name", "World")
+        name = ev.get("name", "World")
         return StopEvent(result=f"Hello, {name}!")
+
 greet_wf = GreetingWorkflow()
 
 

docs/src/content/docs/llamaagents/workflows/assets/loop.png

@@ -1,8 +1,14 @@
 ---
+sidebar:
+  order: 8
 title: Drawing a Workflow
 ---
 
-Workflows can be visualized, using the power of type annotations in your step definitions. You can either draw all possible paths through the workflow, or the most recent execution, to help with debugging.
+Workflows can be visualized, using the power of type annotations in your step definitions.
+
+There are two main ways to visualize your workflows.
+
+## 1. Converting a Workflow to HTML
 
 First install:
 
@@ -19,35 +25,40 @@ from llama_index.utils.workflow import (
 )
 
 # Draw all
-draw_all_possible_flows(JokeFlow, filename="joke_flow_all.html")
+draw_all_possible_flows(MyWorkflow, filename="all_paths.html")
 
 # Draw an execution
-w = JokeFlow()
+w = MyWorkflow()
 handler = w.run(topic="Pirates")
 await handler
-draw_most_recent_execution(handler, filename="joke_flow_recent.html")
+draw_most_recent_execution(handler, filename="most_recent.html")
 ```
 
-<div id="working-with-global-context-state"></div>
-## Working with Global Context/State
+## 2. Using the `workflow-debugger`
 
-Optionally, you can choose to use global context between steps. For example, maybe multiple steps access the original `query` input from the user. You can store this in global context so that every step has access.
+Workflows ship with a [`WorkflowServer`](/python/llamaagents/workflows/deployment) that allows you to convert workflows to API's. As part of the `WorkflowServer`, a debugging UI is provided as the home `/` page.
 
-```python
-from workflows import Context
+Using this server app, you can visualize and run your workflows.
 
+![workflow debugger](./assets/ui_sample.png)
 
-@step
-async def query(self, ctx: Context, ev: MyEvent) -> StopEvent:
-    # retrieve from context
-    query = await ctx.store.get("query")
+Setting up the server is straightforward:
+
+```python
+import asyncio
+from workflows import Workflow, step
+from workflows.events import StartEvent, StopEvent
 
-    # do something with context and event
-    val = ...
-    result = ...
+class MyWorkflow(Workflow):
+    @step
+    async def my_step(self, ev: StartEvent) -> StopEvent:
+        return StopEvent(result="Done!")
 
-    # store in context
-    await ctx.store.set("key", val)
+async def main():
+    server = WorkflowServer()
+    server.add_workflow("my_workflow", MyWorkflow())
+    await server.serve("0.0.0.0", "8080")
 
-    return StopEvent(result=result)
+if __name__ == "__main__":
+    asyncio.run(main())
 ```