How to build a workflow in LlamaIndex

Application flows become harder to change when retrieval, generation, review, and packaging all live in one long Python function. LlamaIndex Workflows separates those stages into asynchronous steps whose typed events make each handoff visible.

The standalone llama-index-workflows package provides the workflows imports used here. A custom StartEvent defines the caller's input, an intermediate Event carries data between steps, and a custom StopEvent becomes the value returned by workflow.run().

The customer-brief checklist does not call an LLM or require an API key, so its event routing can be exercised locally. Validation must accept the step graph, and the caller must receive a BriefResult whose three items each use the request's topic or audience.

1. Install the standalone LlamaIndex Workflows package in the active Python environment.

   $ python3 -m pip install llama-index-workflows

2. Define the workflow event contracts at the top of customer_brief_workflow.py.

   customer_brief_workflow.py

   import asyncio
    
   from workflows import Workflow, step
   from workflows.events import Event, StartEvent, StopEvent
    
    
   class BriefRequest(StartEvent):
       topic: str
       audience: str = "team"
    
    
   class DraftEvent(Event):
       topic: str
       audience: str
       bullets: list[str]
    
    
   class BriefResult(StopEvent):
       topic: str
       bullets: list[str]
       summary: str

   BriefRequest is the input boundary, DraftEvent is the internal handoff, and BriefResult is returned to the caller when the workflow stops.

3. Append the workflow class after the BriefResult event.

   class CustomerBriefWorkflow(Workflow):
       @step
       async def draft(self, ev: BriefRequest) -> DraftEvent:
           bullets = [
               f"Define the {ev.topic} goal",
               f"Name the {ev.audience} handoff",
               f"Run one {ev.topic} smoke test before release",
           ]
           return DraftEvent(
               topic=ev.topic,
               audience=ev.audience,
               bullets=bullets,
           )
    
       @step
       async def package(self, ev: DraftEvent) -> BriefResult:
           summary = f"{len(ev.bullets)} checklist items ready for {ev.topic}"
           return BriefResult(
               topic=ev.topic,
               bullets=ev.bullets,
               summary=summary,
           )

   The @step annotations and event type hints connect draft() to package() without a separate graph declaration.

4. Append the asynchronous runner after the CustomerBriefWorkflow class.

   async def main() -> None:
       workflow = CustomerBriefWorkflow(timeout=30)
       workflow.validate()
       result = await workflow.run(
           start_event=BriefRequest(
               topic="support chatbot",
               audience="developer",
           )
       )
    
       print(f"result type: {type(result).__name__}")
       print(result.summary)
       for bullet in result.bullets:
           print(f"- {bullet}")
    
    
   if __name__ == "__main__":
       asyncio.run(main())

   validate() rejects missing event consumers or unreachable stop events before run() schedules the typed BriefRequest.

5. Run the completed workflow script.

   $ python3 customer_brief_workflow.py
   result type: BriefResult
   3 checklist items ready for support chatbot
   - Define the support chatbot goal
   - Name the developer handoff
   - Run one support chatbot smoke test before release

   The BriefResult type and computed checklist confirm that both workflow steps consumed and produced the intended events.