LlamaIndex Workflow — steps not firing, wrong Event types, or missing StopEvent

from llama_index.core.workflow import (
    Workflow, StartEvent, StopEvent, Event, step, Context
)
from llama_index.llms.openai import OpenAI

# 1) Define your custom events
class RetrieveEvent(Event):
    query: str

class GenerateEvent(Event):
    query: str
    context: str


class RAGWorkflow(Workflow):
    def __init__(self, retriever, llm, **kwargs):
        super().__init__(**kwargs)
        self.retriever = retriever
        self.llm = llm

    # StartEvent -> RetrieveEvent
    @step
    async def start(self, ev: StartEvent) -> RetrieveEvent:
        return RetrieveEvent(query=ev.query)

    # RetrieveEvent -> GenerateEvent
    @step
    async def retrieve(self, ev: RetrieveEvent) -> GenerateEvent:
        nodes = await self.retriever.aretrieve(ev.query)
        context = "\n\n".join(n.text for n in nodes)
        return GenerateEvent(query=ev.query, context=context)

    # GenerateEvent -> StopEvent
    @step
    async def generate(self, ev: GenerateEvent) -> StopEvent:
        prompt = f"Context:\n{ev.context}\n\nQuestion: {ev.query}\nAnswer:"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))


# 2) Run it
async def main():
    workflow = RAGWorkflow(
        retriever=index.as_retriever(similarity_top_k=6),
        llm=OpenAI(model="gpt-4o-mini"),
        timeout=60,           # give it 60s max
        verbose=True,         # print steps and events
    )
    result = await workflow.run(query="What is the refund policy?")
    print(result)

import asyncio
asyncio.run(main())

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

class ProcessingEvent(Event):
    item_id: str

class ItemProcessedEvent(Event):
    item_id: str
    success: bool

class MultiItemWorkflow(Workflow):

    @step
    async def start(self, ctx: Context, ev: StartEvent) -> ProcessingEvent:
        # Store initial state in ctx
        await ctx.set("total_items", len(ev.items))
        await ctx.set("processed", 0)
        await ctx.set("errors", [])

        # Fan out — emit one event per item
        for item_id in ev.items:
            ctx.send_event(ProcessingEvent(item_id=item_id))

        return None   # no direct return; events sent above

    @step
    async def process_item(self, ctx: Context, ev: ProcessingEvent) -> ItemProcessedEvent:
        # Do the work
        try:
            await do_work(ev.item_id)
            return ItemProcessedEvent(item_id=ev.item_id, success=True)
        except Exception as e:
            errors = await ctx.get("errors", [])
            errors.append({"item": ev.item_id, "error": str(e)})
            await ctx.set("errors", errors)
            return ItemProcessedEvent(item_id=ev.item_id, success=False)

    @step
    async def gather(self, ctx: Context, ev: ItemProcessedEvent) -> StopEvent | None:
        # Collect results with ctx.collect_events until all items are processed
        total = await ctx.get("total_items")
        events = ctx.collect_events(ev, [ItemProcessedEvent] * total)
        if events is None:
            return None   # wait for more
        # All items processed
        errors = await ctx.get("errors", [])
        return StopEvent(result={
            "processed": len(events),
            "succeeded": sum(1 for e in events if e.success),
            "errors": errors,
        })

# Run
async def main():
    wf = MultiItemWorkflow(timeout=120)
    result = await wf.run(items=["item_1", "item_2", "item_3"])
    print(result)

from llama_index.core.workflow import Workflow, draw_all_possible_flows

# Static: visualize what CAN happen
draw_all_possible_flows(RAGWorkflow, filename="rag_workflow.html")
# Opens an HTML file with all possible event routes — inspect for orphaned events

# --- Runtime: verbose + custom logging ---
class DebuggableWorkflow(RAGWorkflow):
    @step
    async def start(self, ctx, ev):
        print(f"[start] ev={ev}")
        return await super().start(ctx, ev)

    @step
    async def retrieve(self, ctx, ev):
        print(f"[retrieve] query={ev.query}")
        result = await super().retrieve(ctx, ev)
        print(f"[retrieve] emitting {type(result).__name__}")
        return result

# --- Timeout diagnosis ---
async def diagnose():
    wf = RAGWorkflow(retriever=r, llm=llm, timeout=10, verbose=True)
    try:
        result = await wf.run(query="...")
    except Exception as e:
        print(f"Workflow failed: {e}")
        # Print any events collected mid-run
        for ev in wf._events:
            print(f"  event: {ev}")

# --- Streaming events ---
async def stream_events():
    wf = RAGWorkflow(retriever=r, llm=llm)
    handler = wf.run(query="...")

    async for event in handler.stream_events():
        print(f"streamed: {type(event).__name__}")

    result = await handler
    print(f"final: {result}")

asyncio.run(stream_events())