Graph Flow (LamaIndex)

How to use LlamaIndex worklows as a generic workflow engine or state machine.

Note: you can also find the LangGraph flavor of this setup.

LlamaIndex has a very different approach to graph flows than LangGraph, based on events rather than explicit (static or dynamic) connections. You can achieve the same results in both frameworks and they have both plus and minus.

Let’s assemble the same number-trials logic as in our LangGraph example.

• the API has two special events StartEvent and StopEvent which define the starting node and the end node respectively.
• you define custom events in function of the logic you wish to implement
• the type of event accepted by a step function creates the graph flow
• there is a global context where you can store whatever is necessary outside the local handler.

from llama_index.core.workflow import (
    Event,
    StartEvent,
    StopEvent,
    Workflow,
    step,
    Context
)
import random

class GenerateEvent(Event):
    pass
class EvaluateEvent(Event):
    num:int   

class ResponseEvent(Event):
    final: int


class MyFlow(Workflow):

    @step
    async def generate(self, ctx:Context, ev: StartEvent|GenerateEvent) -> EvaluateEvent:
       
        if random.random() > 0.9:
            return EvaluateEvent(num= random.randint(1, 100))
        else:
            return EvaluateEvent(num=0)

    @step
    async def evaluate(self, ctx:Context, ev: EvaluateEvent) -> ResponseEvent|GenerateEvent:
       
        if "tried" not in ctx.data:
            ctx.data["tried"] =  1
        else:
            ctx.data["tried"] +=  1
        if ev.num==0:
            return GenerateEvent()
        return ResponseEvent(final=ev.num)

    @step
    def response(self, ctx:Context, ev:ResponseEvent)->StopEvent:
        return StopEvent(f"{ev.final} [tried {ctx.data['tried']} time(s)]")


w = MyFlow(timeout=60, verbose=False)
result = await w.run()
print(str(result))

64 [tried 18 time(s)]

You can also render the flow, but I find it less satisfactory than the LangGraph rendering:

import nest_asyncio
nest_asyncio.apply()
from llama_index.utils.workflow import draw_all_possible_flows
draw_all_possible_flows(MyFlow)

workflow_all_flows.html

from IPython.display import IFrame

IFrame(src='./workflow_all_flows.html', width=700, height=600)

Just as before with LangGraph, the workflow infrastructure was developed to embrace complex LLM logic beyond chains but you can use it for arbitrary workflow needs.

For oldtimers like me who witnessed Windows Workflow Foundations, API’s like this are a joy and so much easier than how one used to assemble flows. At the same time, it’s interesting to notice that finite state machines, flows (including things like n8n and PowerApps)) appear again and again in software development. Not everyone likes it but it’s clear that certain paradigns like workflows, visual programming, literate programming, declarative programming…have a place in any software stack.