LlamaIndex Workflows, agenci na zdarzeniach

LlamaIndex Workflows, czyli agent zbudowany ze zdarzeń

Większość bibliotek do budowania agentów opisuje przepływ jako graf: węzły, krawędzie i warunki przejścia. Workflows wybiera inne pojęcie. Tutaj są kroki, które nasłuchują zdarzeń określonego typu i same emitują zdarzenia, a wykonanie polega na tym, że silnik dostarcza każde zdarzenie do kroku, który się na nie zapisał.

Różnica wygląda na kosmetyczną i taka nie jest. W grafie musisz z góry narysować wszystkie krawędzie. W układzie zdarzeniowym dodanie nowego kroku polega na napisaniu funkcji nasłuchującej istniejącego zdarzenia, bez ruszania czegokolwiek innego.

Warto od razu wyjaśnić nazwę, bo bywa myląca. Workflows wyrosło z biblioteki LlamaIndex, znanej z wyszukiwania po dokumentach, ale jest dziś osobnym pakietem z własnym repozytorium i własnym rozwojem. Można go używać bez reszty tamtego ekosystemu.

Pierwszy przepływ

pip install llama-index-workflows

Krok to metoda z dekoratorem, przyjmująca zdarzenie jednego typu i zwracająca zdarzenie innego typu. Cała logika sterowania siedzi w tych typach.

from workflows import Workflow, step, Context
from workflows.events import StartEvent, StopEvent, Event

class PytanieRozwiniete(Event):
    pytanie: str

class FragmentyZnalezione(Event):
    fragmenty: list[str]

class Asystent(Workflow):
    @step
    async def rozwin(self, ev: StartEvent) -> PytanieRozwiniete:
        pelne = await przeformuluj(ev.pytanie)
        return PytanieRozwiniete(pytanie=pelne)

    @step
    async def szukaj(self, ev: PytanieRozwiniete) -> FragmentyZnalezione:
        return FragmentyZnalezione(fragmenty=await baza.szukaj(ev.pytanie))

    @step
    async def odpowiedz(self, ev: FragmentyZnalezione) -> StopEvent:
        return StopEvent(result=await model.zapytaj(ev.fragmenty))

wynik = await Asystent(timeout=60).run(pytanie="Jaki jest okres wypowiedzenia?")

Nigdzie tu nie napisałeś, że po rozwinięciu pytania ma iść wyszukiwanie. Wynika to z typów: krok rozwijający zwraca zdarzenie, na które zapisał się krok wyszukujący. Kolejność jest konsekwencją deklaracji, nie osobnym opisem, który mógłby się z nią rozjechać.

To ma praktyczną zaletę, którą docenia się przy trzecim tygodniu pracy. Nie da się zbudować przepływu z krawędzią prowadzącą donikąd, bo zdarzenie bez odbiorcy jest widoczne od razu.

Kontekst i stan

Kroki nie widzą swoich zmiennych, więc do przekazywania stanu poza treścią zdarzeń służy kontekst przypisany do przebiegu.

@step
async def zbierz(self, ctx: Context, ev: FragmentyZnalezione) -> StopEvent:
    licznik = await ctx.store.get("liczba_prob", default=0)
    await ctx.store.set("liczba_prob", licznik + 1)
    return StopEvent(result=...)

Rozdział między zdarzeniami a kontekstem warto przemyśleć na początku, bo później zmienia się go niechętnie. Zasada, która sprawdza się w praktyce, brzmi tak: zdarzenie niesie to, czego potrzebuje następny krok, a kontekst to, co dotyczy całego przebiegu.

Numer sprawy, identyfikator użytkownika i licznik prób należą do kontekstu. Wynik wyszukiwania przekazywany dalej należy do zdarzenia. Wrzucanie wszystkiego do kontekstu daje przepływ, w którym nie widać, co od czego zależy, czyli dokładnie to, przed czym ten model miał chronić.

Kontekst da się zapisać do postaci słownika i odtworzyć z powrotem. To otwiera dwie możliwości: wznowienie przerwanego przebiegu i przechowanie stanu między żądaniami HTTP, bez trzymania procesu w pamięci.

Serializacja radzi sobie z obiektami opisanymi modelem danych, zapisując razem z wartościami pełną nazwę klasy, dzięki czemu odtworzenie zwraca właściwy typ, a nie sam słownik. Warto natomiast trzymać w kontekście dane, a nie obiekty niosące połączenia albo uchwyty do zasobów, bo te ostatnie po odtworzeniu nie będą już wskazywać na nic sensownego.

Punkty kontrolne i wznawianie

Wbudowanego mechanizmu punktów kontrolnych tu nie ma i lepiej wiedzieć to od razu. Biblioteka daje surowiec: po każdym zakończonym kroku silnik wystawia w strumieniu zdarzenie wewnętrzne ze zrzutem stanu, nazwą zdarzenia wejściowego i nazwą zdarzenia wyjściowego, a zapisanie tego zrzutu leży po Twojej stronie.

W praktyce to kilkanaście linii kodu. Nasłuchujesz strumienia przebiegu z odsłoniętymi zdarzeniami wewnętrznymi, przy każdym zakończeniu kroku zrzucasz kontekst do słownika i odkładasz go tam, gdzie chcesz. Wznowienie polega na odtworzeniu kontekstu z tego słownika i podaniu go do kolejnego uruchomienia przepływu, także w innym procesie.

Znaczenie tego widać przy rachunkach za model. Przepływ o ośmiu krokach, w którym siódmy wywala się na błędzie zewnętrznego API, przy braku punktów kontrolnych kosztuje sześć wywołań modelu przy każdej próbie naprawy. Z punktami kontrolnymi kosztuje zero, bo wznawiasz od miejsca awarii.

Jest przy tym zastrzeżenie, które trzeba znać przed produkcją. Wznowienie ma semantykę co najmniej raz. Kroki zakończone nie ruszają ponownie, bo ich wynik siedzi już w odtworzonym stanie, natomiast krok przerwany w połowie wykonuje się od początku, więc jego skutki uboczne muszą być bezpieczne przy powtórzeniu.

Druga zaleta dotyczy przepływów z udziałem człowieka. Krok może wystawić zdarzenie oczekujące na decyzję, a stan poczekać zapisany dowolnie długo. Agent przygotowuje propozycję odpowiedzi dla klienta, człowiek ją zatwierdza w panelu, przepływ rusza dalej, a między jednym a drugim mija sześć godzin i restart serwera.

Warto zaplanować, gdzie ten stan trafia, zanim napiszesz pierwszy przepływ produkcyjny. Pamięć procesu wystarcza do prototypu i zawodzi przy pierwszym wdrożeniu, bo przebiegi rozpoczęte przed wdrożeniem przepadają.

Równoległość i rozgałęzienia

To obszar, w którym model zdarzeniowy pokazuje przewagę nad grafem, bo równoległość wychodzi tu naturalnie.

Krok może wyemitować kilka zdarzeń naraz, a wtedy kroki nasłuchujące ruszają jednocześnie. Zebranie wyników polega na poczekaniu na określoną liczbę zdarzeń danego typu.

@step
async def rozeslij(self, ctx: Context, ev: PytanieRozwiniete) -> None:
    for zrodlo in ["dokumenty", "baza", "sieć"]:
        ctx.send_event(SzukajW(zrodlo=zrodlo, pytanie=ev.pytanie))

@step
async def zbierz(self, ctx: Context, ev: WynikCzastkowy) -> StopEvent | None:
    wyniki = ctx.collect_events(ev, [WynikCzastkowy] * 3)
    if wyniki is None:
        return None
    return StopEvent(result=polacz(wyniki))

Zwrócenie wartości pustej oznacza, że krok nie ma jeszcze kompletu i czeka na kolejne zdarzenia. Silnik wywołuje go przy każdym przychodzącym wyniku, aż komplet się uzbiera.

Ten wzorzec pokrywa większość realnych potrzeb: przeszukanie kilku źródeł naraz, ocenę odpowiedzi przez kilku recenzentów, wykonanie niezależnych wywołań API. Zysk czasowy jest równy różnicy między sumą czasów a czasem najwolniejszego kroku, co przy trzech wywołaniach modelu oznacza zwykle dwie trzecie oszczędności.

Pułapka jest jedna i warto ją znać dobrze. Jeśli któreś ze zdarzeń nigdy nie przyjdzie, bo krok je emitujący zawiódł albo zwrócił inny typ, zbieranie czeka do wyczerpania limitu czasu. Limit ustawiony na przepływie jest tu zabezpieczeniem, a nie formalnością.

Pętle, warunki i powroty

Model zdarzeniowy obsługuje rozgałęzienia inaczej niż graf i warto zobaczyć, jak wyglądają w nim trzy najczęstsze konstrukcje.

Warunek to krok zwracający jedno z kilku możliwych zdarzeń. Typ zwracany opisujesz jako sumę typów, a silnik dostarcza wyemitowane zdarzenie do właściwego odbiorcy.

@step
async def oceny(self, ev: OdpowiedzGotowa) -> StopEvent | PoprawZnowu:
    if await ocen(ev.tresc) >= 0.8:
        return StopEvent(result=ev.tresc)
    return PoprawZnowu(tresc=ev.tresc, uwagi=await zbierz_uwagi(ev.tresc))

Pętla powstaje z tego samego mechanizmu: krok poprawiający emituje zdarzenie, które wraca do kroku oceniającego. Nie ma tu osobnego pojęcia cyklu, bo cykl jest po prostu zdarzeniem wracającym w górę przepływu.

Ta prostota ma cenę, którą trzeba zapłacić świadomie. Pętla bez licznika kręci się aż do wyczerpania limitu czasu, a każdy obrót kosztuje wywołania modelu. Licznik prób trzymany w kontekście i sprawdzany w warunku to trzy linie kodu, których pominięcie potrafi kosztować kilkadziesiąt dolarów przy jednym nieudanym wdrożeniu.

Trzecia konstrukcja to wcześniejsze zakończenie. Krok zwracający zdarzenie kończące przerywa cały przebieg niezależnie od tego, co dzieje się w gałęziach równoległych. Przydaje się przy walidacji wejścia i przy wykryciu, że dalsza praca nie ma sensu.

Testowanie przepływów

Struktura oparta na krokach ma zaletę, którą łatwo przeoczyć: każdy krok jest zwykłą funkcją przyjmującą obiekt i zwracającą obiekt, więc testuje się go bez uruchamiania czegokolwiek innego.

To znaczy, że logikę rozgałęzień da się sprawdzić bez wywoływania modelu. Krok decydujący, czy odpowiedź jest wystarczająco dobra, testujesz podając mu gotowe zdarzenie i sprawdzając, jaki typ zwrócił. Sekundę zamiast trzydziestu i zero kosztu zamiast wywołania modelu.

Drugi poziom to test całego przepływu z podmienionymi wywołaniami zewnętrznymi. Podstawiając atrapy w miejsce modelu i wyszukiwania, sprawdzasz, czy kroki układają się w oczekiwaną kolejność i czy warunki prowadzą tam, gdzie powinny.

Trzeci poziom, najbardziej wartościowy i najczęściej pomijany, to zestaw realnych przypadków z oczekiwanym wynikiem, uruchamiany na prawdziwym modelu. Tylko on odpowiada na pytanie, czy zmiana w instrukcji systemowej cokolwiek poprawiła. Bez niego ocena sprowadza się do wrażenia po kilku ręcznych próbach, a wrażenie myli regularnie.

Agenci na wierzchu

Biblioteka daje też gotowe klasy agentów zbudowane na tym samym mechanizmie, więc nie musisz pisać pętli narzędziowej od zera.

from llama_index.core.agent.workflow import FunctionAgent

agent = FunctionAgent(
    tools=[sprawdz_stan, zloz_zamowienie],
    llm=model,
    system_prompt="Obslugujesz zamowienia. Zawsze sprawdzasz stan przed zlozeniem.",
)

odpowiedz = await agent.run("Zamow dwie pary butow w rozmiarze 42.")

Ponieważ agent jest przepływem, można go osadzić jako krok w większym przepływie. To jest właściwy sposób łączenia obu poziomów: proces o znanych krokach opisujesz przepływem, a pojedynczy krok wymagający oceny sytuacji oddajesz agentowi.

Odwrotna kolejność, czyli oddanie całego procesu jednemu agentowi z piętnastoma narzędziami, daje rozwiązanie działające na pokazie i nienadające się do diagnozy, gdy zawiedzie.

Workflows kontra alternatywy

Pierwsze kryterium jest prozaiczne: jeśli warstwa wyszukiwania po dokumentach stoi już na LlamaIndeksie, ten wybór oszczędza tłumaczenia między pojęciami dwóch bibliotek.

Drugie dotyczy sposobu myślenia o przepływie. Graf jest wygodniejszy, gdy proces ma jawną maszynę stanów z powrotami i pętlami. Zdarzenia są wygodniejsze, gdy proces jest zbiorem reakcji na to, co się wydarzyło, i gdy dużo rzeczy dzieje się równolegle.

Trzecie, jak zawsze, to pytanie o konieczność. Trzy wywołania modelu po kolei nie potrzebują żadnej z tych bibliotek, a dołożenie orkiestracji do czegoś, co mieści się w jednej funkcji, dodaje pojęć do nauczenia się bez żadnego zysku.

Uruchamianie na produkcji

Przepływ da się wystawić jako usługę, i to jest część, którą warto przemyśleć przed napisaniem drugiego przepływu.

Wykonanie jest asynchroniczne, więc naturalnym układem jest przyjęcie zadania, zwrócenie identyfikatora i udostępnienie wyniku po zakończeniu. Strumieniowanie zdarzeń pośrednich pozwala pokazać użytkownikowi postęp, co przy przepływie trwającym czterdzieści sekund bywa różnicą między działającym produktem a porzuconą stroną.

Obserwowalność opiera się na tym, że każde zdarzenie da się zapisać. Dziennik zawierający wszystkie zdarzenia jednego przebiegu, z czasami i użyciem tokenów, odpowiada na pytanie, co poszło nie tak, bez powtarzania przebiegu. To jest największa praktyczna zaleta tego modelu i warto ją wykorzystać od pierwszego dnia, a nie po pierwszej awarii.

Limit czasu i limit kosztu ustawiaj na poziomie przepływu. Pojedynczy krok wywołujący model w pętli potrafi wykonać kilkadziesiąt zapytań, zanim ktokolwiek zauważy, a górny próg jest tu jedynym mechanizmem, który zadziała bez Twojej uwagi.

Warto też przemyśleć, co się dzieje z przebiegiem przerwanym w połowie przy wdrożeniu nowej wersji kodu. Jeśli kroki zmieniły nazwy albo zdarzenia zmieniły pola, zapisany stan przestaje pasować do nowego przepływu i wznowienie zawodzi. Najprostsze wyjście to poczekanie z wdrożeniem na dokończenie trwających przebiegów, przy krótkich przepływach zwykle wystarczające. Przy przepływach czekających na decyzję człowieka przez wiele godzin trzeba wersjonować kształt zdarzeń, tak jak wersjonuje się schemat bazy danych.

Typowe błędy

Pierwszy to trzymanie całego stanu w kontekście zamiast w zdarzeniach. Przepływ przestaje wtedy pokazywać zależności między krokami, a to była główna korzyść z tego modelu.

Drugi to brak limitu czasu przy zbieraniu równoległych wyników. Jedno zdarzenie, które nigdy nie przyjdzie, zawiesza cały przebieg do wyczerpania limitu.

Trzeci to punkty kontrolne trzymane w pamięci procesu. Przy wdrożeniu przepadają razem z restartem, więc do produkcji potrzebny jest trwały magazyn.

Czwarty to oddanie całego procesu jednemu agentowi z kilkunastoma narzędziami. Znane kroki opisuj przepływem, bo zyskujesz powtarzalność i tańszą diagnozę.

Piąty to pominięcie zapisu zdarzeń. Model zdarzeniowy daje gotową ścieżkę audytu za darmo, a niewykorzystanie jej to zmarnowanie największej przewagi tej biblioteki.

Szósty to zakładanie, że biblioteka wymaga reszty ekosystemu. Pakiet jest samodzielny, instaluje się osobno i działa bez warstwy wyszukiwania po dokumentach.

FAQ

Czy muszę używać całego LlamaIndeksu?

Nie. Workflows jest samodzielnym pakietem z własnym repozytorium i własnym rozwojem, instalowanym osobno. Gotowe klasy agentów pochodzą z szerszego ekosystemu, natomiast sam mechanizm kroków i zdarzeń działa bez niego.

Czym różni się od LangGraph?

Pojęciem opisującym przepływ. LangGraph buduje graf z węzłami i krawędziami, tutaj masz kroki nasłuchujące zdarzeń. Graf jest wygodniejszy przy jawnej maszynie stanów, zdarzenia przy dużej równoległości i przy procesie rosnącym o kolejne reakcje.

Czy da się wznowić przerwany przepływ?

Tak, choć zapis jest po Twojej stronie. Wbudowanego mechanizmu punktów kontrolnych nie ma: silnik po każdym kroku wystawia zdarzenie wewnętrzne ze zrzutem stanu, a Ty zapisujesz kontekst w postaci słownika i odtwarzasz go przy wznowieniu. Zaplanuj na te zapisy trwały magazyn i pamiętaj, że krok przerwany w połowie wykona się ponownie.

Czy działa poza Pythonem?

Podstawowa biblioteka jest dostępna również w wariancie dla TypeScriptu, przy czym ekosystem wokół Pythona jest wyraźnie bogatszy. Przy projekcie w całości opartym na środowisku przeglądarki warto sprawdzić zakres wariantu przed decyzją.

Kiedy nie warto tego używać?

Przy procesie będącym sekwencją dwóch albo trzech wywołań modelu, bez rozgałęzień i bez potrzeby wznawiania. Zwykły kod jest wtedy krótszy, tańszy i czytelniejszy niż jakakolwiek biblioteka orkiestrująca.

Dokumentacja stoi na stronie projektu, a pakiet znajdziesz w rejestrze PyPI.