LlamaIndex Workflows (pași și evenimente) în LlamaIndex & Data RAG

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

class CautareEvent(Event):
    context: str

class RagWorkflow(Workflow):
    @step
    async def cauta(self, ev: StartEvent) -> CautareEvent:
        return CautareEvent(context=f"date pentru: {ev.intrebare}")

    @step
    async def raspunde(self, ev: CautareEvent) -> StopEvent:
        return StopEvent(result=f"Răspuns bazat pe {ev.context}")

w = RagWorkflow(timeout=60, verbose=False)
rezultat = await w.run(intrebare="Ce spune contractul?")

Un workflow este împărțit în pași. Un pas primește un eveniment, face ceva (apel LLM, regăsire, cerere de input uman, actualizare de stare) și întoarce un alt eveniment, care declanșează pasul ce îl acceptă prin adnotarea de tip. Evenimentele sunt obiecte Pydantic definite de tine; StartEvent și StopEvent sunt speciale. Argumentele date la run() devin câmpuri ale StartEvent (ev.intrebare sau ev.get("intrebare")), iar StopEvent(result=...) oprește fluxul și întoarce rezultatul.

Ramificările sunt if-uri care întorc tipuri diferite de evenimente (adnotare A | B), buclele sunt pași care întorc un eveniment tratat de un pas anterior, iar munca în paralel se face cu un pas care întoarce list[Event] sau cu ctx.send_event() și ctx.collect_events(). Înainte de rulare, workflow-ul își validează graful: evenimente produse fără consumator, consumate fără producător sau lipsa unui StopEvent dau eroare.

Pașii pot fi async def sau def simplu; pașii sincroni sunt mutați automat într-un thread pool. Starea comună unei rulări stă în Context: un pas care declară ctx: Context citește și scrie cu await ctx.store.get(...) / set(...) sau atomic cu async with ctx.store.edit_state(). run() întoarce un handler: îl aștepți cu await sau citești evenimentele intermediare cu stream_events(). Biblioteca se instalează și separat (llama-index-workflows, import workflows); cu llama-index-core e inclusă sub llama_index.core.workflow. Agenții LlamaIndex sunt construiți pe workflow-uri.

Concepte cheie și caracteristici

Concept / Parametru	Descriere practică
@step	Marchează o metodă ca pas; tipul argumentului eveniment și tipul întors definesc legăturile.
Event / StartEvent / StopEvent	Clasa de bază pentru evenimente proprii; evenimentul de start (câmpuri din run()) și cel de oprire (result=).
Workflow(timeout=..., verbose=...)	Timeout în secunde și afișare de diagnostic.
Context / ctx.store	Starea unei rulări; get, set, edit_state(). Starea tipizată: Context[ModelPydantic].
ctx.send_event / ctx.collect_events / ctx.write_event_to_stream	Trimitere dinamică de evenimente, așteptarea unui set de evenimente, publicare pentru stream_events().
InputRequiredEvent / HumanResponseEvent	Evenimentele pentru intervenție umană în flux.

Exemple practice și scenarii reale

Exemplul 1: Buclă cu ramificare

Ca în ghidul „Branches and loops”: un pas care întoarce propriul tip de eveniment formează o buclă.

class Ciorna(Event):
    text: str
    incercare: int

class Flux(Workflow):
    @step
    async def scrie(self, ev: StartEvent) -> Ciorna:
        return Ciorna(text=ev.subiect, incercare=1)

    @step
    async def verifica(self, ev: Ciorna) -> Ciorna | StopEvent:
        if len(ev.text) < 20 and ev.incercare < 3:
            return Ciorna(text=ev.text + " (completat)", incercare=ev.incercare + 1)
        return StopEvent(result=ev.text)

Exemplul 2: Stare comună și rulare dintr-un script

import asyncio
from llama_index.core.workflow import Context

class Contor(Workflow):
    @step
    async def numara(self, ctx: Context, ev: StartEvent) -> StopEvent:
        n = await ctx.store.get("n", default=0)
        await ctx.store.set("n", n + 1)
        return StopEvent(result=n + 1)

async def main():
    print(await Contor(timeout=30).run())

asyncio.run(main())

Greșeli frecvente și capcane

• Un pas care întoarce un eveniment pe care niciun alt pas nu îl acceptă. Validarea oprește workflow-ul înainte de rulare.
• Clienți, indecși sau conexiuni puse în ctx.store. Starea trebuie să fie serializabilă; documentația recomandă pentru dependențe Resource.
• Apeluri sincrone blocante (de exemplu requests.get) într-un pas async def. Blochează bucla de evenimente; scrie pasul ca def sau folosește asyncio.to_thread.

Noțiuni și module înrudite

• function-agent
• router-query-engine
• sub-question-query-engine
• structured-predict-pydantic