LangGraph

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 1 z 18. Większość skryptów dla dużych modeli językowych działa idealnie, gdy masz szybki prompt i szybką odpowiedź. Ale kiedy zadanie wykonuje się przez dwadzieścia minut, a w połowie zrywa ci połączenie, wszystko się sypie. Tracisz swój postęp, kontekst i budżet API. Odcinek Problem orkiestracji: Dlaczego LangGraph? jest dokładnie o tym, jak naprawić tę kruchość. LangGraph to framework do orkiestracji, stworzony do obsługi aplikacji typu stateful z wieloma aktorami. Możesz usłyszeć tę nazwę i założyć, że to tylko feature LangChaina. Wcale tak nie jest. LangGraph to niskopoziomowy silnik do orkiestracji i w ogóle nie musisz używać LangChaina, żeby z niego korzystać. Stworzono go specjalnie po to, by modelować workflow agentów jako grafy stateful, a nie proste, liniowe skrypty. Standardowe skrypty wykonują się w pamięci. Jeśli skrypt niespodziewanie się zatrzyma, wszystkie dane z runtime'u znikają. Wyobraź sobie scenariusz, w którym masz agenta działającego w tle, który analizuje stustronicowy dokument. Agent czyta, wyciąga fakty i krzyżowo weryfikuje informacje przez dwadzieścia minut bez przerwy. Jeśli w dziewiętnastej minucie wystąpi timeout serwera, standardowy skrypt traci cały ten stan. Musisz zaczynać całe zadanie od nowa. LangGraph rozwiązuje ten problem orkiestracji poprzez durable execution. Kiedy modelujesz swój workflow jako graf, każdy pojedynczy krok staje się nodem, a logiczne połączenia między nimi to krawędzie. Gdy aplikacja przechodzi z jednego noda do następnego, LangGraph automatycznie zapisuje jej postęp. Traktuje długo działające procesy jako serię bezpiecznych checkpointów. Jeśli system padnie, LangGraph wznawia działanie dokładnie w miejscu, w którym skończył. Ten mechanizm checkpointingu opiera się na kompleksowej pamięci. Pamięć w LangGraph to nie tylko bieżąca lista wiadomości na czacie. To cały stan grafu. Gdy node kończy przetwarzanie, aktualizuje współdzielony obiekt stanu. Kolejny node w sekwencji czyta swój input bezpośrednio z tego stanu. To oznacza, że pamięć utrzymuje się przez cały lifecycle aplikacji. Agent działający w tle, który analizuje twój dokument, nie zapomina o krytycznych danych znalezionych na piątej stronie, gdy w końcu dociera do strony dziewięćdziesiątej, ponieważ stan grafu bezpiecznie je przechowuje. Oto kluczowa kwestia. Ponieważ stan grafu jest pauzowany i czysto zapisywany pomiędzy krokami, zyskujesz możliwość dodania human in the loop. Czasami autonomiczny agent dociera do punktu decyzyjnego, w którym potrzebuje pozwolenia, zanim przejdzie dalej, na przykład przed wysłaniem ostatecznego maila czy wykonaniem transakcji finansowej. W standardowym skrypcie pauzowanie, by użytkownik mógł kliknąć przycisk, często prowadzi do timeoutów połączenia. W LangGraph po prostu konfigurujesz konkretny node, aby zatrzymał wykonywanie. System przechodzi w stan uśpienia i idealnie zachowuje obecny stan. Operator może wtedy przejrzeć zebrane dane, zatwierdzić kolejną akcję, a nawet ręcznie zmodyfikować stan agenta przed kliknięciem continue. Po zatwierdzeniu graf się wybudza i wznawia pracę ze zaktualizowanym kontekstem. Główny wniosek jest taki, że budowanie złożonych agentów w dużej mierze opiera się na zarządzaniu stanem i awariami. LangGraph przenosi twoją architekturę z kruchych, ograniczonych pamięcią skryptów na odporne grafy, które potrafią przetrwać przerwy w działaniu, pamiętają swoją przeszłość i cierpliwie czekają na wskazówki od człowieka. Jeśli chcesz wesprzeć nasz podcast, możesz wyszukać DevStoriesEU na Patreonie. To wszystko na dzisiaj. Dzięki za wysłuchanie i twórz dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 2 z 18. Tworzysz agenta sztucznej inteligencji, upychając instrukcje, edge cases i przykłady w jeden ogromny prompt, a potem wysyłasz to i masz nadzieję, że zrobi to, co trzeba. Działa, dopóki się nie wywali, a debugowanie wyniku jest frustrujące. Żeby to naprawić, musisz przestać pisać monolityczne prompty i zacząć projektować systemy. I to prowadzi nas do tematu: Myślenie w LangGraph, czyli model mentalny. LangGraph zmusza cię do odejścia od liniowych skryptów. Wymaga od ciebie myślenia o aplikacji jak o maszynie stanów. Pierwszym krokiem jest po prostu zdefiniowanie twojego procesu. Zanim napiszesz jakikolwiek kod, zastanów się, co chcesz, żeby ten system osiągnął, i podziel to na odrębne akcje. Weźmy na przykład system do triażu w obsłudze klienta. Użytkownik wysyła wiadomość. Człowiek przeczytałby tego maila, zdecydował, czy to problem z płatnościami, czy wsparcie techniczne, a następnie przygotował odpowiednią odpowiedź. Ta sekwencja to twój proces. Mapujesz ten proces na workflow w LangGraph, używając trzech głównych komponentów, którymi są State, Nodes i Edges. Zacznijmy od State. State to współdzielona pamięć twojego grafu. To struktura, która przechowuje kontekst całej operacji w dowolnym momencie. Każdy krok w twoim workflow będzie czytał z tego State i zapisywał w nim aktualizacje. Słuchacze często popełniają tutaj pewien błąd. Próbują przechowywać w pełni sformatowane stringi promptów wewnątrz State. Nie rób tego. State powinien przechowywać tylko surowe dane. Trzyma oryginalny tekst maila od klienta, wyciągniętą kategorię albo surową listę poprzednich wiadomości. Formatowanie odbywa się na żądanie, później, dokładnie w tym kroku, w którym jest potrzebne. Następnie mamy Nodes. Jeśli State to pamięć, Nodes to pracownicy wykonujący właściwe zadania. Node to po prostu funkcja w Pythonie, która wykonuje pojedynczy, logiczny krok twojego procesu. Odbiera obecny State, wykonuje akcję i zwraca aktualizację. W naszym przykładzie z triażem, utworzyłbyś trzy oddzielne Nodes. Pierwszy to Node typu Read. Pobiera przychodzącego maila i zapisuje surowy tekst do State. Drugi to Node typu Classify. Analizuje surowy tekst w State, prosi model językowy o sklasyfikowanie go jako problem z płatnościami lub techniczny, i zapisuje wynikową kategorię z powrotem do State. Trzeci to Node typu Draft. Odczytuje zarówno maila, jak i kategorię ze State, formatuje je lokalnie w prompt i generuje odpowiedź. Każdy Node wykonuje dokładnie jedno zadanie. Na koniec potrzebujesz sposobu, żeby połączyć tych pracowników. Taka jest rola Edges. Edges reprezentują logikę routingu. Dyktują, co ma się wydarzyć, gdy Node skończy swoją pracę. Standardowy Edge mówi po prostu: kiedy Node Read skończy, zawsze przejdź do Node Classify. Ale LangGraph używa też conditional edges. I tu zaczyna się robić ciekawie. Po Node Classify możesz użyć conditional edge, żeby sprawdzić State. Jeśli kategoria to płatności, ten edge kieruje flow do konkretnego Node Draft dla płatności. Jeśli jest techniczna, kieruje do technicznego Node Draft. Edges podejmują decyzje o ruchu na podstawie danych, które twoje Nodes właśnie wyprodukowały. Zaczynasz od procesu, izolujesz dane do współdzielonego State, definiujesz pracowników jako Nodes i dyktujesz flow za pomocą Edges. Rozbijając problem na części, izolujesz błędy. Traktuj swoją aplikację nie jak pojedynczy generator tekstu, ale jak skoordynowaną linię montażową, gdzie surowe dane systematycznie przepływają od jednego wyspecjalizowanego pracownika do następnego. Dzięki za wysłuchanie, miłego kodowania wszystkim!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 3 z 18. Dwie równoległe funkcje kończą działanie dokładnie w tej samej milisekundzie i próbują zapisać dane do tej samej współdzielonej listy. Zazwyczaj jedna nadpisuje drugą, a dane znikają. Aby temu zapobiec, LangGraph korzysta z Graph API: State and Reducers. Podstawą każdego workflow w LangGraph jest jego state. Definiujesz go za pomocą standardowego TypedDict w Pythonie. Ten słownik określa dokładne klucze, które twój graph będzie śledził, oraz typy danych dla każdego klucza. Pomyśl o tym jak o schemacie, który jest przekazywany z node'a do node'a podczas działania twojego graphu. Oto kluczowa sprawa. Node'y w LangGraph nie mutują state'u bezpośrednio. Node otrzymuje kopię obecnego state'u, robi swoje i zwraca słownik z update'ami. Częstym błędem jest zakładanie, że zwrócenie słownika zastępuje cały obiekt state'u. Wcale tak nie jest. Jeśli twój state ma pięć kluczy, a node zwraca słownik z tylko jednym kluczem, LangGraph zostawia pozostałe cztery nietknięte i aplikuje tylko ten konkretny update. Jak LangGraph aplikuje ten update? I tu do gry wchodzą reducery. Reducer to po prostu funkcja, która określa, jak zwrócona wartość łączy się z istniejącą wartością dla konkretnego klucza. Domyślnie LangGraph używa reducera typu overwrite. Jeśli twój node zwróci nowy string dla klucza status, stary string znika, całkowicie zastąpiony przez ten nowy. Czasami overwrite to dokładnie to, czego chcesz uniknąć. Weźmy na przykład równoległy workflow pobierania danych. Masz współdzielony klucz w state o nazwie results, który jest listą. Odpalasz dwa node'y działające w tym samym czasie, żeby pobrać różne paczki danych. Jeśli oba node'y zwrócą słownik aktualizujący klucz results, domyślne zachowanie overwrite sprawi, że ten node, który skończy jako ostatni, skasuje pracę tego pierwszego. Żeby to naprawić, dodajesz adnotację do klucza results w swoim TypedDict z konkretnym reducerem, na przykład wbudowanym w Pythona operatorem kropka add. Teraz, kiedy oba node'y zwracają swoje listy, reducer działa jak policjant kierujący ruchem. Bierze istniejącą listę i bezpiecznie robi append wyników z obu node'ów. Nic nie ginie. Jest tu jeden edge case. Co jeśli masz reducer typu append na swoim kluczu results, ale docierasz do punktu w swoim graphie, gdzie naprawdę musisz wyczyścić listę i zacząć od nowa? Jeśli twój node zwróci pustą listę, reducer po prostu doda pustą listę do tej istniejącej, zostawiając stare dane nienaruszone. Dla takiego scenariusza LangGraph dostarcza specjalny typ Overwrite. Kiedy twój node opakuje swój update w obiekt Overwrite, LangGraph to wykrywa i całkowicie pomija reducer. Wyrzuca starą listę i wymusza hard reset. State w złożonym graphie to nie jest krucha zmienna globalna, która jest ciągle mutowana, ale log typu append-only z kontrolowanymi update'ami, rządzony przez jasne zasady redukcji. To wszystko w tym odcinku. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 4 z 18. Czasami chcesz po prostu napisać standardowy skrypt w Pythonie ze zwykłymi if-statementami i for-loopami, ale nadal potrzebujesz enterprise-grade state persistence. Nie chcesz budować jawnego state machine tylko po to, by odpalić kilka wywołań modelu językowego po kolei. W tym miejscu Functional API, a konkretnie dekoratory entrypoint i task, rozwiązują ten problem. Budowanie aplikacji z jawnymi node'ami i edge'ami wymaga od ciebie ręcznego definiowania, jak dane są routowane z jednego kroku do następnego. Taka struktura daje ogromną kontrolę, ale może wydawać się ciężka, gdy twoja logika jest mocno sekwencyjna lub opiera się na standardowych pętlach. Functional API pozwala ci pisać normalny kod w Pythonie z góry na dół, zachowując wbudowane funkcje streamingu i recovery. Zamiast tworzyć instancję obiektu graph, nakładasz dekoratory na swoje istniejące funkcje w Pythonie. Zaczynasz od dekoratora task. Nakładasz go na pojedyncze jednostki pracy w twojej aplikacji. Pomyśl o tasku jako o odrębnym kroku, który robi coś konkretnego, jak odpytanie bazy danych, obliczenie metryki czy promptowanie modelu. Kiedy funkcja ma dekorator task, framework opakowuje ją w warstwę śledzącą, aby monitorować jej wykonanie. Następnie używasz dekoratora entrypoint. Umieszczasz go na głównej funkcji orkiestrującej, która zarządza całym flow. Wewnątrz tej funkcji entrypoint wywołujesz swoje udekorowane taski, używając standardowego control flow z Pythona. Przypisujesz output taska do zmiennej, a następnie przekazujesz tę zmienną do kolejnego taska. Możesz używać bloków try-except, list comprehensions czy while-loopów. Logika orkiestracji zachowuje się dokładnie tak, jak natywny Python. Ponieważ kod wygląda zupełnie standardowo, możesz założyć, że brakuje mu pamięci formalnej struktury stanu. To częste błędne przekonanie. Functional API wciąż automatycznie robi checkpointy twojego postępu w tle. Za każdym razem, gdy task się kończy, LangGraph przechwytuje return value i zapisuje je w persistent store. Framework bezpiecznie zapisuje inputy i outputy każdej udekorowanej funkcji na bieżąco. Weźmy na przykład zautomatyzowany skrypt do pisania esejów. Definiujesz trzy udekorowane taski: funkcję do generowania zarysu, funkcję do pisania akapitu i funkcję do recenzji draftu. Wewnątrz swojej głównej funkcji entrypoint najpierw wywołujesz generator zarysu. Następnie piszesz standardowy for-loop, który iteruje po sekcjach tego zarysu, wywołując task pisania akapitu dla każdej z nich. Wyniki dodajesz do lokalnej listy. Na koniec odpalasz task recenzji. Używasz prostego if-statementu, aby sprawdzić uzyskany wynik. Jeśli wynik jest słaby, twój kod po prostu odpala while-loopa, aby przepisać konkretne akapity, aż wynik się poprawi. I tu jest kluczowa sprawa. Dzięki ukrytemu checkpointingowi, jeśli twój skrypt napotka network timeout podczas pisania trzeciego akapitu, nie tracisz swojej pracy. Kiedy zrestartujesz proces z tym samym identyfikatorem threadu, LangGraph wie, że zarys i pierwsze dwa akapity są już gotowe. Całkowicie pomija wykonywanie tych tasków, pobiera ich zcache'owane outputy ze state store'a i wznawia działanie dokładnie od trzeciego akapitu. Functional API przenosi obciążenie poznawcze z wizualizowania abstrakcyjnych topologii routingu z powrotem na czytanie kodu z góry na dół, dając ci odporność state machine z prostotą zwykłego skryptu. To wszystko na teraz. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 5 z 18. Tworzysz aplikację do czatu, a użytkownik zauważa literówkę w swoim prompcie. Klika edytuj, poprawia i wysyła. Ale zamiast podmienić starą wiadomość, twój backend po prostu dokleja poprawioną wersję na końcu historii, zostawiając oryginalną literówkę dokładnie tam, gdzie była, jako taki widmowy duplikat. Zarządzanie historią konwersacji za pomocą MessagesState to sposób, żeby temu zapobiec. Kiedy programiści budują swój pierwszy graf, zazwyczaj definiują customowy słownik state, żeby trzymać w nim historię czatu. Częstym błędem jest używanie standardowego appendowania do listy, żeby zarządzać tą historią. Podpinają standardową funkcję operator dot add do swojej listy messages. To mówi grafowi, żeby po prostu brał nowe wiadomości i doklejał je na koniec istniejącego arraya. Takie podejście append-only działa świetnie w przypadku prostego bota typu ping-pong, gdzie użytkownik coś pisze, AI odpowiada, a historia rośnie sekwencyjnie. Ale sypie się całkowicie, gdy state musi być mutowalny. Jeśli człowiek zedytuje poprzedni prompt, albo agent zdecyduje się wygenerować ponownie swoją ostatnią odpowiedź, standardowe dodawanie sobie z tym nie poradzi. Kończysz z duplikatami. LangGraph dostarcza wbudowaną strukturę state, która to rozwiązuje, zwaną MessagesState. Zawiera ona pojedynczy klucz o nazwie messages. I tu jest kluczowa sprawa. Siłą MessagesState nie jest sam klucz, ale podpięta pod niego specyficzna funkcja reducer, o nazwie add messages. Reducer add messages nie tylko ślepo appenduje dane. Śledzi on ID wiadomości. Za każdym razem, gdy nowa wiadomość wchodzi do state, reducer sprawdza jej unikalne ID. Jeśli to ID już istnieje gdziekolwiek w historii konwersacji, reducer nadpisuje starą wiadomość nową. Jeśli ID jest nowe, albo jeśli wiadomość nie ma jeszcze ID, reducer appenduje ją na koniec listy. Wróćmy do naszego scenariusza z literówką. Użytkownik wysyła prompt. System nadaje tej wiadomości ID 123. Użytkownik orientuje się, że zrobił błąd, edytuje tekst i wysyła poprawkę. Frontend wysyła nowy tekst, jawnie tagując go jako ID 123. Kiedy te dane trafiają do grafu, reducer add messages skanuje historię, znajduje oryginalną wiadomość pod ID 123 i podmienia tekst w miejscu. Widmowy duplikat znika. Konwersacja płynie dokładnie tak, jak powinna. Poza zarządzaniem ID, reducer add messages ogarnia też deserializację danych. W aplikacji na produkcji, twoje wiadomości często przychodzą w różnych formatach. Twój frontend może wysyłać surowe słowniki JSON zawierające stringi role i content. Twoje wewnętrzne node'y grafu mogą generować natywne obiekty message z LangChain. Reducer działa jak uniwersalny tłumacz dla tych inputów. Jeśli przekażesz do state listę zwykłych pythonowych słowników, funkcja add messages automatycznie skonwertuje je na odpowiednie klasy message z LangChain. Nie musisz pisać kodu boilerplate, żeby sparsować słownik do HumanMessage albo AIMessage przed aktualizacją state. Normalizuje te dane za ciebie. Kiedy budujesz agenty czatowe, historia state to nie jest tylko log typu append-only, to żywy dokument, a wiązanie twoich aktualizacji z unikalnymi ID wiadomości to coś, co utrzymuje ten dokument w poprawności. Dzięki za odsłuch. Do usłyszenia następnym razem!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 6 z 18. Wybierz zły paradygmat projektowy na wczesnym etapie, a skończysz pisząc setki linijek boilerplate'u dla prostego skryptu, albo utkasz koszmar spaghetti z podstawowych funkcji Pythona. Dzisiaj skupimy się na wyborze abstrakcji: Graph kontra Functional. Łatwo założyć, że jedno z tych API jest z natury potężniejsze lub bardziej production-ready niż drugie. To nieprawda. Pod spodem, zarówno Graph API, jak i Functional API kompilują się do dokładnie tego samego silnika runtime. Oba obsługują persistence, streaming i kontrolę wykonania w dokładnie taki sam sposób. Wybór między nimi zależy wyłącznie od twojego modelu mentalnego i sposobu, w jaki chcesz wyrazić swoją logikę. Przyjrzyjmy się najpierw Functional API. Opiera się ono na standardowym, imperatywnym control flow w Pythonie. Piszesz normalne funkcje w Pythonie, oznaczasz je dekoratorem i sterujesz wykonaniem za pomocą standardowych ifów i pętli. Zarządzanie stanem jest tutaj w pełni function-scoped. Dane przepływają ściśle od wartości zwracanej przez jedną funkcję do argumentów kolejnej. W tle nie ma żadnego współdzielonego, globalnego obiektu pamięci. Jeśli twój workflow jest liniowy lub ma przewidywalną, ściśle określoną logikę, Functional API sprawia, że twój kod jest lekki i znajomy. Całkowicie unikasz overheadu związanego z definiowaniem struktur grafu. Graph API wymaga innego podejścia. Zamiast wywoływać funkcje bezpośrednio, definiujesz współdzielony, globalny schemat stanu. Następnie piszesz node'y, czyli małe funkcje, których jedynym zadaniem jest odczytywanie i mutowanie tego współdzielonego stanu. Na koniec jawnie łączysz te node'y ze sobą za pomocą edge'y. Routing nie jest obsługiwany przez instrukcję warunkową ukrytą głęboko w ciele funkcji. Zamiast tego, logika dyktująca, gdzie aplikacja ma przejść dalej, jest wyciągnięta do jawnych conditional edge'y zadeklarowanych na najwyższym poziomie grafu. I tu jest kluczowa sprawa. Wybierasz między nimi na podstawie tego, jak twój system obsługuje stan i routing w czasie. Wyobraź sobie developera budującego proste narzędzie do ekstrakcji danych. Uruchamia ono pojedynczy model językowy, parsuje output i go zapisuje. Functional API jest do tego idealne. Szybko się je pisze i łatwo czyta. Ale przewińmy o trzy miesiące do przodu. Ten prosty skrypt jest refaktorowany do złożonego systemu multi-agentowego. Teraz masz agenta researchera przekazującego dane do agenta writera, agenta krytyka, który odrzuca je z poprawkami, oraz pauzę w wykonaniu, czekającą aż ludzki manager zatwierdzi ostateczny draft. Jeśli spróbujesz zbudować ten asynchroniczny, multi-agentowy workflow za pomocą Functional API, imperatywne podejście przestaje się sprawdzać. W efekcie przekazujesz ogromne payloady danych w górę i w dół przez głębokie call stacki funkcji. Twoja logika routingu zostaje zakopana w głęboko zagnieżdżonych instrukcjach warunkowych. To jest dokładnie ten moment, w którym migrujesz na Graph API. Abstrakcja grafowa błyszczy tutaj, ponieważ rozdziela stan od wykonania. Ponieważ stan jest globalny i współdzielony, twoje poszczególne node'y agentów nie muszą przekazywać sobie nawzajem ciężkich struktur danych. Node po prostu odczytuje współdzielony stan, aktualizuje konkretny klucz, za który odpowiada, i kończy działanie. Jawne edge'e przejmują kontrolę, dzięki czemu routing jest bardzo widoczny. Możesz spojrzeć na definicję grafu i od razu zmapować cały workflow bez czytania ani jednej linijki logiki biznesowej. Używasz Functional API, gdy control flow jest na tyle prosty, że można go czytać z góry na dół, ale przełączasz się na Graph API, gdy routing staje się na tyle skomplikowany, że musisz go rozrysować na tablicy. Dzięki za wysłuchanie. Trzymajcie się wszyscy.

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 7 z 18. Hardcodowane instrukcje warunkowe wystarczają tylko do pewnego momentu, kiedy budujesz agenta. Jeśli użytkownik zada złożone pytanie, proste wyszukiwanie po słowach kluczowych nie jest w stanie niezawodnie zdecydować, co twoja aplikacja powinna zrobić dalej. A co, gdyby sztuczna inteligencja sama dyktowała ścieżkę twojego workflow? Właśnie tutaj wkraczają dynamiczny routing i conditional edges. W standardowej konfiguracji grafu łączysz node A z nodem B. Jest to statyczna, gwarantowana ścieżka. Ale kiedy budujesz inteligentny mechanizm routingu, ścieżka musi się zmieniać w zależności od przychodzących danych. Możesz zakładać, że edges w LangGraph przyjmują tylko hardcodowane połączenia w postaci stringów. Wcale tak nie jest. Edge może być funkcją w Pythonie, która odczytuje aktualny state twojego grafu i dynamicznie oblicza nazwę następnego node'a. Podpinasz tę logikę do swojego grafu używając metody add conditional edges. Ta metoda wymaga trzech komponentów. Po pierwsze, początkowego node'a. Po drugie, funkcji routingu. Po trzecie, słownika, który mapuje możliwe wyjściowe stringi twojej funkcji routingu na rzeczywiste docelowe node'y w twoim grafie. Oto kluczowa sprawa. Najbardziej niezawodnym sposobem na sterowanie conditional edge jest połączenie go z dużym modelem językowym generującym ustrukturyzowane dane. Nie chcesz, aby sama funkcja routingu wykonywała złożoną ewaluację czy przetwarzanie języka naturalnego. Zamiast tego masz upstream node, w którym model jest zmuszony zwrócić ścisłą strukturę, taką jak model Pydantic. Weźmy na przykład router obsługi klienta. Użytkownik wysyła wiadomość. Pierwszym nodem w twoim grafie jest klasyfikator intencji. Wewnątrz tego node'a przekazujesz wiadomość użytkownika do modelu językowego i wymagasz, aby zwrócił ustrukturyzowany output z jednym polem o nazwie intent. Model ocenia tekst i wypełnia to pole konkretną wartością, taką jak billing, tech support lub sales. Ta ustrukturyzowana odpowiedź jest następnie zapisywana w graph state. Teraz kontrolę przejmuje conditional edge. Ten edge jest podpięty do node'a klasyfikatora. Kiedy node klasyfikatora kończy działanie, conditional edge uruchamia krótką funkcję w Pythonie. Funkcja ta przyjmuje graph state jako swój input, zagląda do tego state'u i wyciąga wartość intent, którą model właśnie wygenerował. Jeśli intent to billing, funkcja zwraca string billing. Conditional edge patrzy na swój słownik mapowania, widzi, że string billing odpowiada twojemu node'owi billing, i przekazuje wykonanie do tego konkretnego node'a. Jeśli intent to tech support, zwraca inny string, kierując flow do node'a tech support. Używasz modelu językowego ze względu na jego zdolności wnioskowania, aby skategoryzować input, ale samą logikę routingu utrzymujesz jako deterministyczną. Funkcja w Pythonie wewnątrz conditional edge po prostu odczytuje zmienną i zwraca string. Jest wysoce przewidywalna i łatwa do przetestowania. Najbardziej przydatnym wnioskiem z tego jest to, że zawsze powinieneś oddzielać decyzję od kierunku. Pozwól modelowi językowemu zdecydować o intent i zapisać go w state, a następnie użyj czystego conditional edge w Pythonie, aby odczytać ten state i sterować grafem. Zanim skończymy, jeśli uważasz te odcinki za przydatne i chcesz wesprzeć nasz program, możesz wyszukać DevStoriesEU na Patreonie — to naprawdę nam pomaga. To wszystko w tym odcinku. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 8 z 18. Nie możesz zhardkodować swoich ścieżek egzekucji, kiedy nie masz pojęcia, ile podzadań twój agent zdecyduje się utworzyć, dopóki faktycznie się nie uruchomi. Jeśli twój system zdecyduje w locie, że musi przetworzyć trzy lub trzydzieści elementów, standardowy statyczny routing zawiedzie. Aby to naprawić, potrzebujesz workflowów Map-Reduce z Send API. Bardzo częstym błędem podczas budowania w LangGraph jest próba użycia standardowych conditional edges do dynamicznego fan-outu. Conditional edges są idealne, gdy chcesz wybierać między znanymi, z góry określonymi ścieżkami na podstawie sprawdzenia logiki. Jednak przestają działać, gdy musisz odpalić nieznaną liczbę identycznych, równoległych zadań w runtime'ie. Standardowa paralelizacja pozwala ci na routing do z góry ustalonej liczby node'ów. Nazywasz node'y, a graf je triggeruje. Ale co się dzieje, gdy musisz uruchomić dokładnie ten sam node wiele razy jednocześnie, za każdym razem z innymi danymi? Nie możesz tego zrobić za pomocą podstawowego routingu. To prowadzi nas do patternu Orchestrator-worker. W tej architekturze centralny node analizuje przychodzące dane, oblicza, ile oddzielnych zadań jest wymaganych, i wysyła dynamiczne workery, aby obsłużyły je współbieżnie. LangGraph umożliwia ten pattern konkretnie poprzez Send API. Wyobraź sobie agenta, którego zadaniem jest napisanie kompleksowego raportu badawczego. Pierwszy node działa jako orchestrator. Odczytuje prompt użytkownika i generuje konspekt. W zależności od złożoności tematu, ten konspekt może zawierać trzy sekcje, albo może ich zawierać dwanaście. Chcesz, aby osobny worker node przygotował draft każdej sekcji dokładnie w tym samym czasie. Aby to osiągnąć, definiujesz funkcję conditional edge bezpośrednio po swoim node'zie orchestratora. Zamiast zwracać prosty string, który wskazuje na kolejny statyczny node w grafie, ta funkcja edge zwraca listę obiektów Send. I tu jest kluczowa sprawa. Obiekt Send pakuje razem miejsce docelowe i jego dane. Przyjmuje dwa argumenty. Pierwszy argument to nazwa worker node'a, którego chcesz ztriggerować. Drugi argument to konkretny payload dla tego odizolowanego workera. W naszym scenariuszu z raportem, funkcja orchestratora iteruje po wygenerowanym konspekcie. Dla każdego tematu sekcji, który znajdzie, tworzy nowy obiekt Send wskazujący na pojedynczy node o nazwie draft_section, przekazując indywidualny string z tematem jako payload. Kiedy LangGraph ewaluuje tę funkcję edge, otrzymuje listę obiektów Send. Następnie dynamicznie odpala równoległą instancję node'a draft_section dla każdego pojedynczego elementu na tej liście. Jeśli orchestrator wygenerował konspekt z siedmioma sekcjami, LangGraph uruchamia siedem równoległych node'ów draftujących. Każdy node wykonuje identyczny kod, ale operuje na swoim własnym, unikalnym payloadzie. Generowanie tych dynamicznych workerów to faza map. Zbieranie ich niezależnych wyników to faza reduce. Ponieważ te worker node'y działają współbieżnie, nie mogą bezpiecznie nadpisać pojedynczego stringa w twoim state grafu. Twój ogólny state musi być skonfigurowany tak, aby jednocześnie zbierać wiele przychodzących aktualizacji. Robisz to, podpinając funkcję reducer do konkretnego pola w state, które będzie trzymać twoje drafty sekcji, instruując ją, aby dodawała nowe elementy do listy, zamiast nadpisywać poprzednią wartość. Gdy każdy równoległy draft worker kończy pisanie, zwraca swój blok tekstu. LangGraph przechwytuje te odpowiedzi i używa twojego reducera do bezpiecznego dorzucenia każdego bloku tekstu we współdzielonej tablicy. Kiedy każdy dynamiczny worker zakończy swoją egzekucję, cały równoległy krok się rozwiązuje. Wtedy workflow idzie dalej, niosąc ze sobą kompletną, wypełnioną listę wszystkich zedraftowanych sekcji. Send API wyciąga równoległą egzekucję z twojej statycznej definicji grafu i oddaje ją bezpośrednio w ręce twoich danych w runtime'ie. Dzięki, że wpadłeś. Mam nadzieję, że dowiedziałeś się czegoś nowego.

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 9 z 18. Twój serwer pada w połowie procesu, podczas gdy agent przetwarza ogromny dataset. Nie powinien zaczynać od zera. Powinien wznowić pracę dokładnie tam, gdzie skończył. Właśnie tę odporność omówimy dzisiaj w kontekście persistence, a konkretnie threadów i checkpointów. Aby dodać persistence do aplikacji w LangGraph, musisz zrozumieć koncepcję threada. Thread reprezentuje pojedynczą, izolowaną sekwencję wykonania lub konkretną konwersację z użytkownikiem. Przechowuje roboczy state grafu, gdy ten przechodzi z node'a do node'a. Pozwól, że od razu coś wyjaśnię. Ludzie często mylą pamięć threada z długoterminową pamięcią pomiędzy sesjami. Thread to nie globalna baza danych, w której twój agent zapamiętuje fakty z różnych tasków na zawsze. To krótkoterminowa pamięć robocza ściśle powiązana z jedną trwającą sekwencją. Włączasz tę pamięć, przekazując checkpointer podczas kompilacji grafu. Checkpointer to obiekt, który obsługuje zapisywanie i ładowanie state'u grafu do storage backendu. Gdy twój graf jest już skompilowany z checkpointerem, uruchamiasz persistence, przekazując obiekt konfiguracji z thread ID za każdym razem, gdy wywołujesz graf. Ten ID to unikalny klucz, którego checkpointer używa do śledzenia historii tego konkretnego runa. Kiedy odpalasz graf z tym thread ID, checkpointer automatycznie zapisuje state. Ale nie robi tego w sposób ciągły. Zapisuje go w określonych granicach, zwanych super-steps. Super-step to odrębny cykl wykonania w grafie. Jeśli twój graf uruchamia node A, a następnie node B, to są to dwa super-steps. Jeśli twój graf się rozgałęzia i uruchamia node C oraz node D w tym samym czasie, to to równoległe wykonanie jest grupowane w jeden pojedynczy super-step. Checkpointer nie przerywa działania node'a w trakcie pracy. Czeka na granicę super-stepu. Gdy wszystkie node'y zaplanowane dla tego super-stepu zakończą działanie i zwrócą swoje update'y, LangGraph tworzy checkpoint. Ten checkpoint zawiera state snapshot, dokładnie rejestrując, jak wyglądają zmienne state'u grafu w tym konkretnym momencie. Spójrzmy, jak to działa w praktyce. Załóżmy, że masz agenta analizującego ogromny dataset. Graf ma cztery kroki. Krok pierwszy pobiera dane. Krok drugi je czyści. Krok trzeci odpala kosztowną analizę. Krok czwarty formatuje podsumowanie. Startujesz run, przekazując konfigurację z thread ID jeden dwa trzy. Graf pomyślnie kończy kroki pobierania, czyszczenia i analizy. Na koniec kroku trzeciego checkpointer zapisuje state snapshot. Następnie, zanim krok czwarty zdąży się zakończyć, twój serwer pada. Ponieważ użyłeś checkpointera i thread ID, state jest bezpieczny. Kiedy twój serwer wstanie, po prostu wywołujesz graf ponownie, przekazując dokładnie ten sam thread ID jeden dwa trzy. Checkpointer wyszukuje najnowszy checkpoint. Znajduje state snapshot zapisany zaraz po kroku analizy. Graf ładuje ten state i wznawia działanie od razu od kroku czwartego. Node'y pobierania, czyszczenia i analizy są całkowicie pomijane, ponieważ ich outputy są już bezpiecznie zapisane w checkpoincie threada. I to jest kluczowa sprawa. Kompilując swój graf z checkpointerem i wiążąc wykonanie z thread ID, zamieniasz kruche operacje in-memory w trwałe workflowy, które automatycznie przetrwają przerwy. To wszystko w tym odcinku. Dzięki za wysłuchanie i budujcie dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 10 z 18. Twój workflow przetwarza płatność, w kolejnym kroku trafia na rate limit i crashuje. Kiedy system się podnosi i wznawia workflow, twój klient zostaje obciążony po raz drugi. Twój kod się nie zmienił, ale twoje założenie o tym, jak wznawia się graf, było błędne. Naprawa tego wymaga zrozumienia durable execution i idempotentności. Wielu programistów zakłada, że gdy długo działający proces pauzuje lub kończy się błędem, jego wznowienie rusza dokładnie od tej samej linijki kodu w Pythonie, na której się zatrzymał. Oczekują, że runtime magicznie zapamięta zmienne lokalne w połowie funkcji. Ale tak to nie działa. LangGraph nie zamraża interpretera Pythona w miejscu. Stan jest zapisywany tylko na granicach między node'ami. Durable execution w LangGraph oznacza, że system śledzi twój postęp, utrwalając stan grafu po tym, jak node skończy swoją pracę i zwróci wynik. Jeśli node wywali się w połowie swojej logiki, system nie ma żadnego zapisu jego częściowego postępu. Ostatni znany dobry stan to ten, który został przekazany do node'a przy jego starcie. Kiedy restartujesz lub robisz retry grafu, wykonywanie wznawia się poprzez ponowne odpalenie całego wywalonego node'a od jego pierwszej linijki. Pomyśl o scenariuszu z płatnością. Załóżmy, że piszesz pojedynczy node, który wykonuje dwie akcje. Po pierwsze, wywołuje zewnętrzne API, żeby obciążyć kartę kredytową. Po drugie, aktualizuje zdalną bazę danych, żeby zapisać transakcję. Obciążenie karty się udaje, ale połączenie z bazą danych łapie timeout, przez co node crashuje. Stan grafu nie idzie do przodu. Kiedy workflow jest wznawiany, przekazuje stary stan z powrotem do tego samego node'a. Node startuje od nowa. Uderza do zewnętrznego API i obciąża kartę po raz drugi. I tu jest kluczowa sprawa. Ponieważ node'y mogą restartować się od początku, każdy side effect wewnątrz node'a musi być idempotentny. Idempotentność to właściwość, gdzie wielokrotne wykonanie operacji daje dokładnie taki sam wynik, jak jej jednorazowe wykonanie. Jeśli twój node wchodzi w interakcję ze światem zewnętrznym, musisz napisać kod przy założeniu, że odpali się on wielokrotnie dla tego samego kroku. Jak zapewnić to bezpieczeństwo? Masz dwa praktyczne podejścia. Pierwsze to wykorzystanie kluczy idempotentności w twoich zewnętrznych serwisach. Kiedy wywołujesz API płatności, przekazujesz unikalny identyfikator wygenerowany na podstawie obecnego stanu grafu. Jeśli node crashuje i odpala się ponownie, wysyła ten sam unikalny identyfikator. Zewnętrzny serwis rozpoznaje zduplikowany request i zwraca odpowiedź o sukcesie, bez ponownego transferu pieniędzy. Drugie podejście to strukturalne projektowanie grafu. Jeśli konkretna operacja nie jest natywnie idempotentna, nie grupuj jej z innymi krokami, które mogą się wywalić. Wrzuć tę niebezpieczną operację do jej własnego, dedykowanego node'a. Niech to będzie jedyna rzecz, którą ten node robi. Jeśli wrzucisz obciążenie karty do node'a A, a update bazy danych do node'a B, to timeout bazy danych wywali tylko node B. Graf wznowi działanie od node'a B. Obciążenie płatności w nodzie A jest całkowicie bezpieczne, ponieważ node A zakończył działanie, a graf zapisał swój stan przed pójściem dalej. To ty kontrolujesz, gdzie system zapisuje swój postęp, poprzez to, jak wyznaczasz granice swoich node'ów. Nigdy nie wrzucaj nieodwracalnej, nieidempotentnej akcji do tego samego node'a, co coś, co może losowo się wywalić. To wszystko w tym odcinku. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 11 z 18. Czasami AI nie powinno mieć ostatniego słowa. Możesz chcieć zatrzymać agenta w połowie myśli, poprosić człowieka o zatwierdzenie i wstrzyknąć jego odpowiedź bezpośrednio do działającej logiki. Właśnie to robią interrupty typu Human-in-the-Loop. Kiedy potrzebujesz, żeby człowiek podjął decyzję wewnątrz workflow LangGraph, używasz specjalnej funkcji o nazwie interrupt. Ważne jest, aby zrozumieć, co ona właściwie robi pod maską. Słuchacze mogą to pomylić ze standardowym input promptem w Pythonie. Ale tak nie jest. Standardowy input prompt blokuje aktywny thread, zajmując pamięć systemową podczas oczekiwania, aż użytkownik wciśnie klawisz. W LangGraph wywołanie interrupt działa zupełnie inaczej. W pełni serializuje stan grafu, zapisuje go w bazie danych checkpointera i całkowicie zawiesza egzekucję. Graf idzie spać. Może czekać w nieskończoność na odpowiedź, nie zużywając żadnych aktywnych zasobów obliczeniowych. Ten flow odbywa się w dwóch odrębnych fazach: pauzowania i wznawiania. Najpierw przyjrzyjmy się pauzowaniu. Wewnątrz jednego z twoich node'ów, twój agent dociera do punktu, w którym potrzebuje autoryzacji człowieka. Dokładnie w tej linijce kodu wywołujesz funkcję interrupt. Przekazujesz do tej funkcji payload, którym zazwyczaj jest obiekt JSON zawierający kontekst potrzebny człowiekowi. Wyobraź sobie agenta, który zajmuje się zautomatyzowanym customer supportem. Decyduje się przygotować zwrot na pięćset dolarów. Przed przetworzeniem płatności, node agenta wywołuje interrupt. Przekazuje payload określający, że proponowaną akcją jest zwrot, a kwota to pięćset. W momencie wywołania tej funkcji, graf się zatrzymuje. Runtime LangGraph przechwytuje to zdarzenie i przekazuje payload JSON w górę, do twojej aplikacji klienckiej. Proces grafu się wyłącza, zostawiając payload w oczekiwaniu na weryfikację przez człowieka w web UI. Teraz druga faza: wybudzenie grafu. Zewnętrzny proces, na przykład twój serwer backendowy odbierający API call z web UI, odpowiada za zrestartowanie grafu. Menedżer klika zatwierdź w swoim dashboardzie. Twój backend odbiera to zatwierdzenie i ponownie uruchamia graf za pomocą specjalnej instrukcji o nazwie Command. Wysyłając ten Command, dołączasz argument resume zawierający odpowiedź człowieka. W naszym scenariuszu ta odpowiedź to prosta wartość boolean równa true. I tu jest kluczowa sprawa. Kiedy graf się budzi, nie uruchamia ponownie zapauzowanego node'a od samego początku. Wznawia egzekucję dokładnie w tej samej linijce kodu, w której się zatrzymał. Funkcja interrupt, która pierwotnie zapauzowała graf, kończy się wykonywać i zwraca wartość, którą wysłałeś przez Command resume. Odpowiedź boolean od człowieka jest wstrzykiwana bezpośrednio do zmiennej czekającej na wynik funkcji interrupt. Następnie agent odczytuje tę wartość true, przechodzi swój warunek i finalizuje zwrot pięciuset dolarów. Ta architektura tworzy czystą granicę. Logika grafu nie musi obsługiwać webhooków, maili ani interfejsów użytkownika. Po prostu wywołuje funkcję, która przerzuca payload przez ścianę i czeka na wartość zwrotną. Zewnętrzny system obsługuje całą interakcję z użytkownikiem i po prostu wpycha odpowiedź z powrotem. Wstrzykując odpowiedź człowieka bezpośrednio do returna funkcji, unikasz zaśmiecania głównego stanu grafu tymczasowymi danymi z interakcji. Siła funkcji interrupt tkwi w traktowaniu feedbacku od człowieka nie jako skomplikowanego architektonicznego objazdu, ale jako standardowego wywołania funkcji, które może bezpiecznie zapauzować wszechświat, dopóki nie otrzyma odpowiedzi. To wszystko w tym odcinku. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 12 z 18. Twój agent zaczyna wariować, wykonując akcję, której nie chciałeś, lub generując fatalną odpowiedź. Zazwyczaj musisz zrestartować cały proces od zera i mieć nadzieję, że za drugim razem zachowa się lepiej. A co, gdybyś mógł dosłownie cofnąć execution do dokładnego momentu przed błędem, ręcznie zmienić state i pozwolić mu działać w alternatywnej linii czasu? Właśnie to omówimy dzisiaj w temacie Debugowanie przeszłości: Time Travel i Forking. Aby manipulować przeszłością, najpierw musisz ją zobaczyć. Robisz to za pomocą metody get state history, przekazując swój thread ID. Ta metoda zwraca iterator zawierający każdy state, przez który przeszedł graf podczas execution tego threada. Każdy z tych historycznych state'ów posiada unikalny identyfikator, tak zwany checkpoint ID. Możesz myśleć o tym ID jako o swoich dokładnych współrzędnych w czasie. Jeśli chcesz po prostu zrobić replay grafu od określonego punktu, pobierasz docelowy checkpoint ID z tej historii. Następnie wywołujesz metodę invoke swojego grafu, przekazując obiekt konfiguracyjny, który zawiera zarówno thread ID, jak i ten konkretny checkpoint ID. Graf natychmiast wznawia execution od tego konkretnego state'u. Nie uruchamia ponownie żadnych wcześniejszych node'ów, oszczędzając moc obliczeniową i czas. Replay jest przydatny, ale prawdziwa siła tkwi w zmianie przeszłości, aby sforkować execution. Spójrzmy na praktyczny scenariusz. Załóżmy, że twój agent miał za zadanie napisać żart i wygenerował okropny żart o psie. Sprawdzasz state history i znajdujesz checkpoint ID dla state'u tuż przed krokiem generowania. Zamiast po prostu robić replay od tego punktu, używasz metody update state. Podajesz thread ID, ten konkretny historyczny checkpoint ID oraz nowe wartości state'u, które chcesz wstrzyknąć. W tym przypadku ręcznie aktualizujesz zmienną topic, zmieniając ją z psa na kurczaki. I tu jest kluczowa sprawa. Developerzy często myślą, że aktualizacja przeszłego state'u cofa execution, nadpisując lub usuwając oryginalną historię. Wcale tak nie jest. LangGraph działa w oparciu o architekturę append-only. Kiedy wywołujesz update state na historycznym checkpoincie, system bezpiecznie tworzy zupełnie nowy checkpoint, odgałęziający się od tego starego. Twoja oryginalna oś czasu, wraz z tym okropnym żartem o psie, pozostaje w pełni nienaruszona i dostępna. Nie wymazałeś przeszłości; sforkowałeś nową rzeczywistość. Po zastosowaniu tego update'u, graf znajduje się w nowo utworzonym checkpoincie ze zmienionym state'em. Aby kontynuować działanie na tej nowej osi czasu, po prostu ponownie wywołujesz metodę invoke na grafie z tym thread ID, pomijając jakikolwiek konkretny checkpoint ID. Graf domyślnie przechodzi do najnowszego state'u w tym nowo sforkowanym branchu i wznawia execution. Twój agent odczytuje zaktualizowany state i zamiast tego generuje żart o kurczakach. Jeśli uważasz te techniczne analizy za pomocne i chcesz wesprzeć podcast, możesz wyszukać DevStoriesEU na Patreonie. Time travel zmienia debugowanie ze zgadywania, co poszło nie tak, w precyzyjne manipulowanie historią grafu, aby badać alternatywne wyniki, bez utraty ani jednego śladu oryginalnego runu. To wszystko w tym odcinku. Dzięki za wysłuchanie i twórz dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 13 z 18. Tworzysz agenta, a użytkownik mówi mu, że zawsze chce mieć kod w Pythonie 3.11. Następnego dnia zaczyna nową konwersację, a agent całkowicie o tym zapomina i zamiast tego generuje kod w Pythonie 3.9. Pamięć threadu jest izolowana do pojedynczej konwersacji. Kiedy twój agent musi zachować fakty pomiędzy zupełnie oddzielnymi sesjami, potrzebujesz Long-Term Memory z wykorzystaniem Store'ów pomiędzy threadami. Częstym błędem jest próba rozwiązania tego przez upychanie długoterminowych faktów w state checkpointera. Checkpointery to pamięć krótkoterminowa. Są to snapshoty state'u ściśle przypisane do pojedynczego threadu, zaprojektowane, by pauzować, wznawiać lub odtwarzać jedną konwersację. Jeśli użytkownik poda swoje preferencje w threadzie A, thread B nie ma absolutnie żadnego sposobu, żeby je zobaczyć. Aby współdzielić wiedzę między wieloma threadami, LangGraph dostarcza interfejs Store. Store to warstwa pamięci key-value, która znajduje się poza state'ami poszczególnych threadów. Konfigurujesz go, przekazując obiekt store, na przykład PostgresStore, jako argument, kiedy kompilujesz swój graf. Po skompilowaniu, ten store jest podpięty do środowiska wykonawczego grafu. Wewnątrz twojego grafu, node'y uzyskują dostęp do tej warstwy pamięci przez obiekt Runtime. Kiedy definiujesz node'a, masz dostęp do kontekstu runtime, który wystawia store. Po prostu odwołujesz się do runtime dot store, żeby wejść w interakcję ze swoją pamięcią długoterminową. Oto kluczowa sprawa. Dane w store są zorganizowane przy użyciu namespace'ów. Namespace to hierarchiczna lista stringów, która partycjonuje twoje dane, zupełnie jak ścieżka do folderu na twoim komputerze. Dla aplikacji multi-tenant, możesz zdefiniować namespace, który zaczyna się od stringa users, po którym następuje konkretne user ID, a kończy się na preferences. Pomyśl o tym scenariuszu z asystentem kodowania. Użytkownik zaczyna sesję w poniedziałek. Podczas czatu wspomina, że preferuje Pythona 3.11 i dark mode. Node w twoim grafie rozpoznaje to jako stałą preferencję. Wywołuje metodę put na runtime dot store. Przekazuje namespace dla tego konkretnego użytkownika, unikalny key dla elementu oraz dictionary zawierający preferencje. Dane są teraz zapisane poza threadem. W piątek ten sam użytkownik otwiera twoją aplikację i zaczyna zupełnie nowy thread. State checkpointera dla tego nowego threadu jest całkowicie pusty. Jednak twój graf zawiera setup node, który uruchamia się jako pierwszy. Ten node wywołuje metodę search na runtime dot store, podając prefix namespace'u użytkownika. Store zwraca zapisane preferencje. Następnie node umieszcza te preferencje w obecnym state threadu. Od tego momentu agent wie, żeby używać Pythona 3.11 i dark mode dla tej nowej konwersacji. Interfejs store dostarcza trzy główne operacje. Używasz put, żeby zapisać lub nadpisać element. Używasz get, żeby pobrać pojedynczy element, kiedy znasz jego dokładny namespace i key. Używasz search, żeby pobrać wiele elementów, które współdzielą prefix namespace'u. Search jest szczególnie przydatny, kiedy z czasem zapisałeś kilka różnych fragmentów pamięci dla użytkownika i musisz zaciągnąć je wszystkie do obecnego kontekstu. Oddzielając krótkoterminowy state od cross-threadowego store'a, odsprzęgasz cykl życia wiedzy swojego agenta od cyklu życia pojedynczej konwersacji. Dzięki za wysłuchanie — do usłyszenia następnym razem.

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 14 z 18. Użytkownicy nienawidzą gapić się na statyczny spinner przez trzydzieści sekund, podczas gdy twój system działa w tle. Chcesz pokazać im proces myślowy systemu w czasie rzeczywistym, ale przechwytywanie tych wewnętrznych sygnałów często wymaga podpinania złożonych, customowych callbacków. Streaming i format v2 rozwiązują ten problem, unifikując każde wewnętrzne zdarzenie w jeden, przewidywalny flow. Najpierw wyjaśnijmy powszechne nieporozumienie. Inżynierowie często mylą streaming tokenów modelu językowego ze streamingiem stanu aplikacji. To zupełnie różne warstwy informacji. Stream tokenów to po prostu tekst pojawiający się słowo po słowie. Stream stanu śledzi szerszy postęp twojego workflow, przechodząc od jednego zadania do drugiego. LangGraph obsługuje oba te streamy jednocześnie. Uzyskujesz dostęp do tego zachowania, żądając streamu i przekazując argument version v2 do twojej metody egzekucji. To standaryzuje output. Zamiast użerać się z mieszanymi typami danych, każde pojedyncze zdarzenie, które opuszcza twój graf, staje się zunifikowanym słownikiem zawierającym dokładnie trzy pola: type, ns oraz data. Pole type definiuje kategorię zdarzenia. Pole ns to skrót od namespace, wskazujący dokładną ścieżkę w hierarchii twojego grafu, z której pochodzi zdarzenie. Staje się to krytyczne, gdy masz zagnieżdżone subgrafy i musisz dokładnie wiedzieć, który subkomponent wywołał zdarzenie. I wreszcie, pole data przechowuje właściwy payload. Możesz dokładnie kontrolować, co trafia do tego streamu, wybierając jeden lub więcej stream modes. Tryb values wypycha do ciebie pełny, zaktualizowany stan grafu za każdym razem, gdy dowolny node zakończy swoją pracę. Jest to przydatne, jeśli twoja aplikacja wymaga pełnego obrazu na każdym kroku. Tryb updates jest znacznie lżejszy. Streamuje tylko konkretne dane zwrócone przez node, reprezentując jedynie deltę, czyli zmianę wprowadzoną do ogólnego stanu. Tryb messages działa na bardziej szczegółowym poziomie, streamując poszczególne chunki wygenerowanej wiadomości czatu w miarę ich produkowania przez działający pod spodem model językowy. Wyobraź sobie interfejs na frontendzie. Chcesz mieć świecący wskaźnik statusu, który podświetla aktualnie aktywny krok — na przykład pobieranie kontekstu, potem ewaluację dokumentów, a następnie draftowanie — jednocześnie wyświetlając draftowany tekst token po tokenie. Aby to zbudować, odpalasz egzekucję grafu ze stream modes ustawionymi na updates i messages, upewniając się, że przekazujesz flagę version v2. Twój frontend zaczyna odbierać ciągły, zunifikowany stream tych słowników. Gdy przychodzi słownik z polem type ustawionym na updates, odczytujesz pole namespace. To mówi ci dokładnie, który node właśnie zakończył swoją pracę. Używasz tego sygnału, aby przesunąć swój świecący wskaźnik statusu do następnego kroku w interfejsie użytkownika. Milisekundy później stream dostarcza nowy słownik z polem type ustawionym na messages. Wyciągasz surowy token tekstu z pola data i dołączasz go bezpośrednio do akapitu, który czyta twój użytkownik. Zarówno wysokopoziomowe zmiany stanu, jak i niskopoziomowe generowanie tekstu przychodzą dokładnie przez ten sam pipe. Oto kluczowy wniosek. Wrzucając tokeny, zmiany stanu i postęp node'ów do jednej, trzypolowej struktury słownika, format v2 całkowicie eliminuje potrzebę pisania oddzielnej logiki obsługi lub złożonych, asynchronicznych callbacków dla różnych typów zdarzeń w czasie rzeczywistym. To tyle na ten odcinek. Do usłyszenia następnym razem!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 15 z 18. Kiedy twój agent AI staje się zbyt złożony, często kończysz z jednym ogromnym, nieczytelnym mega-grafem, w którym pojedyncza zmiana psuje wszystko. Wcale nie musisz tak budować — zamiast tego możesz tworzyć wyspecjalizowane mini-grafy i łączyć je ze sobą. Dziś mówimy o komponowaniu złożoności: Subgraphs. Subgraphs pozwalają na reużywanie logiki i rozdzielenie pracy między różne zespoły. Zamiast wrzucać każdy pojedynczy krok twojej aplikacji do jednego pliku, tworzysz mniejsze, niezależne grafy. Po skompilowaniu, graf zachowuje się dokładnie tak, jak standardowa funkcja typu callable. To oznacza, że możesz wziąć cały skompilowany graf i wrzucić go bezpośrednio do innego grafu jako pojedynczy node. Wyobraź sobie główny system routingu dla asystenta korporacyjnego. Główny graf obsługuje user input, sprawdza bezpieczeństwo i decyduje, co zrobić dalej. Kiedy użytkownik zadaje trudne techniczne pytanie, router musi przeprowadzić złożone zbieranie danych. Zamiast kodować tę logikę bezpośrednio w routerze, delegujesz ją do dedykowanego subgraphu Research. Zupełnie inny zespół inżynierów może budować, testować i dopracowywać ten graf Research w pełnej izolacji. Parent graph w ogóle nie przejmuje się tym, jak te badania są robione. On po prostu wywołuje node. Mamy częstą tendencję do przekombinowywania sposobu, w jaki dane przechodzą między tymi grafami. Jeśli parent graph i subgraph używają dokładnie tego samego state schema — co oznacza, że współdzielą dokładnie te same state keys — nie potrzebujesz żadnych specjalnych adapterów. Po prostu przekazujesz skompilowany subgraph Research bezpośrednio do funkcji add node w twoim głównym grafie. Silnik automatycznie przekazuje parent state do subgraphu, odpala logikę i po zakończeniu scala wyniki z powrotem do parent state. A co się dzieje, gdy zespoły nie zgrają się idealnie? Załóżmy, że twój parent router używa state key o nazwie user query, ale ten drugi zespół inżynierów zbudował subgraph Research tak, by oczekiwał klucza o nazwie search term. Nie możesz wrzucić skompilowanego subgraphu bezpośrednio do parent graphu. Klucze nie będą pasować i wykonanie zakończy się błędem. I tu pojawia się kluczowa sprawa. Rozwiązujesz tę niezgodność używając prostej funkcji typu wrapper. W swoim parent grafie definiujesz standardową funkcję node'a, która przyjmuje parent state. Wewnątrz tej funkcji wyciągasz wartość user query. Następnie ręcznie wywołujesz skompilowany subgraph Research, przekazując mu payload, w którym mapujesz to user query na klucz search term. Subgraph odpala swoją wewnętrzną logikę i zwraca swój final state. Twój wrapper bierze ten output, tłumaczy wyniki z powrotem na konkretne klucze, których oczekuje parent, i je zwraca. Z perspektywy parent graphu, ten wrapper wygląda jak każdy normalny node. Nie ma on pojęcia, że w jego środku właśnie wykonał się ogromny, złożony subgraph. Po prostu przekazał dane na wejściu i otrzymał zaktualizowany state na wyjściu. Ten wzorzec daje ci ścisłą modularność bez utraty kontroli. Traktowanie skompilowanego grafu po prostu jako kolejnej funkcji callable za prostym wrapperem to najpotężniejszy sposób na skalowanie architektury AI, tak aby nie zapaść się pod ciężarem własnego kodu. To wszystko w tym odcinku. Dzięki za wysłuchanie i koduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 16 z 18. Jeśli ekspercki subagent zostanie wywołany dwa razy w jednej konwersacji, czy powinien pamiętać pierwsze wywołanie, czy zacząć od nowa z całkowitą amnezją? Ten wybór zmienia wszystko w działaniu systemów multi-agentowych i jest w całości kontrolowany przez Subgraph Persistence i Multi-Agent Patterns. Kiedy budujesz parent graph, który routuje zadania do subgrafów, state management staje się skomplikowany. Wyobraź sobie głównego bota obsługi klienta, który ogarnia ogólny czat. Gdy user zada złożone pytanie o fakturę, główny bot routuje request do dedykowanego subgrafu Billing Expert. Domyślnie subgrafy są całkowicie stateless. Kiedy kompilujesz tego Billing Experta bez podania checkpointera, działa on ściśle w modelu per-invocation. Główny bot przekazuje mu wymagane inputy, ekspert wykonuje swoje wewnętrzne kroki, zwraca wynik i natychmiast odrzuca swój wewnętrzny state. Jeśli pięć minut później user zada pytanie uzupełniające o rozliczenia, główny bot ponownie wywołuje eksperta. Ekspert nie ma w pamięci poprzedniej wymiany zdań. Zaczyna zupełnie od nowa. Dla prostego subgrafu do ekstrakcji danych, taka amnezja jest jak najbardziej okej. Dla interaktywnego, wyspecjalizowanego agenta, jest to niesamowicie frustrujące dla usera. Aby to naprawić, ekspert potrzebuje własnej pamięci pomiędzy turami. Bardzo częstym błędem jest tutaj przekazywanie zupełnie nowej instancji checkpointera, na przykład obiektu memory saver, bezpośrednio do metody compile subgrafu. Nie rób tego, chyba że chcesz, aby subgraf współdzielił dokładnie ten sam state pomiędzy zupełnie różnymi userami i sesjami. Jeśli user A i user B gadają z systemem w tym samym czasie, przekazanie jawnej instancji checkpointera do subgrafu oznacza, że ich dane zostaną zmiksowane w jeden globalny state. To tworzy ogromny cross-talk między odizolowanymi parent threads. Zamiast tego, po prostu przekazujesz wartość boolean True do argumentu checkpointer, kiedy kompilujesz subgraf. I tu robi się ciekawie. Ustawienie go na True mówi subgrafowi, żeby polegał na mechanizmie checkpointera z parent graph, ale utrzymywał całkowicie odizolowaną historię multi-turn specjalnie dla siebie. Pod spodem framework ogarnia namespacing. Automatycznie tworzy unikalny thread ID dla subgrafu, który jest na stałe powiązany z thread ID parenta. Spójrz teraz ponownie na scenariusz z Billing Expertem w tej konfiguracji. User zadaje pytanie o fakturę. Główny bot routuje je do eksperta. Ekspert odpowiada i przechodzi w stan uśpienia. Później, w tej samej konwersacji, user zadaje pytanie uzupełniające. Główny bot routuje je z powrotem do eksperta. Ponieważ został skompilowany z checkpointerem ustawionym na True, ekspert budzi się, sprawdza swój dedykowany sub-thread i ładuje kontekst faktury z wcześniejszej tury. Działa jak stały uczestnik konwersacji. A ponieważ ten sub-thread jest ściśle zescopowany do threadu parenta, inny user gadający z systemem dostaje swoją własną, całkowicie czystą instancję Billing Experta. To, jak skonfigurujesz checkpointer subgrafu, dyktuje całą jego tożsamość w twoim systemie: zostawienie go pustym tworzy jednorazową, stateless utility function, podczas gdy ustawienie go na True tworzy ciągłego, świadomego kontekstu współpracownika. Dzięki za spędzenie ze mną tych kilku minut. Do usłyszenia następnym razem, trzymaj się.

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 17 z 18. Skrypt Pythona działający poprawnie na twoim laptopie to nie jest aplikacja produkcyjna. Jeśli spróbujesz uruchamiać stateful agenty odpalając samodzielne skrypty, nieuchronnie uderzysz w ścianę, gdy przyjdzie czas na skalowanie. Aby to rozwiązać, przyjrzymy się strukturze aplikacji i gotowości do deployu. Kiedy po raz pierwszy budujesz agenta w LangGraph, prawdopodobnie prototypujesz go w Jupyter notebooku albo w pojedynczym pliku Pythona. Definiujesz node'y, łączysz krawędzie, kompilujesz graf i wywołujesz metodę invoke bezpośrednio w tym samym pliku, żeby sprawdzić, czy działa. Do testów to w zupełności wystarczy. Jednak serwer produkcyjny nie czyta w twoich myślach. Potrzebuje ustandaryzowanego sposobu, żeby serwować ten graf jako API, zainstalować wymagane pakiety i wstrzyknąć zmienne środowiskowe. Aby twój prototyp był gotowy do deployu, musisz zorganizować swój kod w czystą strukturę katalogów. Załóżmy, że tworzysz nowy folder o nazwie my-app. Przenosisz swój kod w Pythonie z notebooka do czystego pliku wewnątrz tego folderu. Następnie dodajesz plik z zależnościami, zazwyczaj requirements kropka txt. Na koniec tworzysz plik konfiguracyjny o nazwie langgraph kropka json w głównym katalogu folderu my-app. Plik langgraph kropka json to główny blueprint twojej aplikacji. Kiedy używasz CLI LangGrapha albo robisz deploy na środowisko produkcyjne, ten plik konfiguracyjny mówi systemowi pod spodem, jak dokładnie zbudować i uruchomić twój projekt. Wymaga trzech głównych informacji dotyczących zależności, zmiennych środowiskowych i entry pointów grafu. Po pierwsze, deklarujesz swoje zależności. To po prostu string ze ścieżką w pliku JSON, wskazujący na twój plik requirements. Dzięki temu serwer deploymentowy zainstaluje dokładnie te pakiety Pythona, na których polega twój agent, zapobiegając błędom brakujących modułów na produkcji. Następnie definiujesz string ze środowiskiem. Wskazuje on na twój plik kropka env. Stateful agenty zawsze potrzebują sekretów, takich jak dane logowania do bazy danych czy klucze API do modeli. Wskazanie pliku środowiskowego daje pewność, że runtime bezpiecznie załaduje te klucze przed próbą wystartowania grafu. I to jest ta część, która ma największe znaczenie. Trzecim wymaganiem w pliku konfiguracyjnym jest mapowanie grafów. Mówi to serwerowi dokładnie, gdzie w kodzie źródłowym znajduje się twój skompilowany graf. Działa to jak słownik. Przypisujesz swojemu grafowi ID, które staje się jego oficjalną nazwą w wygenerowanym API. Następnie mapujesz to ID na konkretny moduł w Pythonie i nazwę zmiennej. Na przykład, możesz zmapować ID customer-support-agent na string agent kropka py dwukropek compiled-graph. Serwer zagląda do pliku agent kropka py, znajduje zmienną o nazwie compiled-graph i ładuje ją do pamięci. Taka struktura wymaga celowej zmiany w tym, jak piszesz swój kod. Początkujący często uruchamiają grafy przez samodzielne skrypty w Pythonie, które wykonują akcje od razu po uruchomieniu. Ale runtime LangGrapha polega na pliku langgraph kropka json, żeby dynamicznie wystawić graf jako web service. Nie uruchamia on twojego skryptu z góry na dół. Importuje tylko skompilowany obiekt grafu, który określiłeś w pliku konfiguracyjnym. Z tego powodu twój plik w Pythonie powinien tylko definiować node'y, łączyć je i przypisywać skompilowany graf do zmiennej. Musisz usunąć wszelkie pozostałości kodu testowego na dole, które ręcznie wywołują graf. Jeśli zostawisz kod testowy w pliku, wykona się on podczas fazy importu na serwerze, powodując błędy deployu albo niechciane calle do API już podczas samego startu usługi. Jawnie deklarując swoje zależności, środowisko i ścieżki do grafu w jednym centralnym pliku JSON, oddzielasz definicję swojego agenta od jego wykonania, zamieniając lokalny skrypt w solidną usługę gotową do deployu. To wszystko w tym odcinku. Dzięki za wysłuchanie i buduj dalej!

Cześć, tu Alex z DEV STORIES DOT EU. LangGraph, odcinek 18 z 18. Masz złożony multi-agentowy workflow i musisz przetestować jeden konkretny edge case routingu w kroku czwartym. Nie powinieneś musieć uruchamiać całego systemu od początku do końca, tylko po to, żeby wywołać ten warunek. Testowanie wykonania grafu end-to-end pozwala skupić się dokładnie na tej logice, której potrzebujesz. Kiedy developerzy próbują wyizolować fragmenty grafu do testów, często sięgają po skomplikowane mocki. Próbują mockować otaczającą strukturę grafu lub stubować wszystkie poprzedzające node'y. W LangGraph nie musisz tego robić. Architektura kręci się całkowicie wokół state'u. Ponieważ node'y to po prostu funkcje, które odczytują i zapisują state, możesz ręcznie wstrzyknąć konkretny payload state'u i natywnie przetestować wyizolowane fragmenty node'ów. To właśnie tutaj wstrzykiwanie state'u i breakpointy stają się niezwykle przydatne w twoim test suite. Potrzebujesz tylko dwóch narzędzi, żeby wskoczyć bezpośrednio w środek grafu. Pierwszym z nich jest metoda update state. Drugim jest parametr konfiguracyjny o nazwie interrupt after. Użycie ich w standardowym frameworku testowym, takim jak pytest, pozwala zasymulować dokładne warunki bez uruchamiania całej aplikacji. Zastosujmy to do konkretnego scenariusza. Załóżmy, że masz graf, w którym node trzeci robi zewnętrzne API call, a node czwarty sprawdza wynik. Chcesz zweryfikować, czy jeśli payload API zawiera konkretny kod błędu, node czwarty poprawnie wykona routing execution flow do twojego error handler node'a. Zamiast uruchamiać node pierwszy i drugi, żeby to wywołać, izolujesz problem. Inicjalizujesz graf z identyfikatorem threadu. Następnie używasz update state, żeby wstawić zasymulowany, błędny payload API bezpośrednio do state'u threadu. Działasz tak, jakby node trzeci miał się właśnie uruchomić z tymi konkretnymi danymi. Następnie robisz invoke na grafie, ale przekazujesz dictionary konfiguracyjny ustawiający interrupt after na node czwarty. Kiedy startujesz graf, wykonanie zaczyna się natychmiast w node'zie trzecim, używając twojego wstrzykniętego failure state'u. Node trzeci przetwarza błędny payload i przekazuje wynikowy state do node'a czwartego. Node czwarty ewaluuje logikę i decyduje o routingu do error handlera. Ponieważ ustawiłeś breakpoint, graf pauzuje wykonanie w momencie, gdy node czwarty kończy działanie. Teraz twój test może ocenić wynik. Pobierasz aktualny state z grafu. Możesz napisać swoje asercje, żeby upewnić się, że node czwarty poprawnie zaktualizował zmienne state'u. Co ważniejsze, możesz sprawdzić plan wykonania grafu. Patrząc na kolejny oczekujący node w metadanych state'u, możesz potwierdzić, że logika routingu zadziałała idealnie, a error handler jest w kolejce. Oto kluczowa sprawa. Manipulując state'em bezpośrednio, zamieniasz mocno powiązany, nieprzewidywalny chain agentów w deterministyczny test case krok po kroku. Weryfikujesz dokładnie, jak graf przechodzi z jednego node'a do drugiego, bez czekania na modele językowe czy network calle w poprzednich krokach. Ponieważ to ostatni odcinek serii, zachęcam cię do przejrzenia oficjalnej dokumentacji i spróbowania zbudowania tych workflowów w praktyce. Jeśli masz pomysły na to, co powinniśmy omówić w następnej kolejności, odwiedź devstories dot eu i zaproponuj temat. To wszystko w tym odcinku. Dzięki za słuchanie i buduj dalej!