LangGraph

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 1 din 18. Majoritatea scripturilor pentru Large Language Models funcționează perfect pentru un prompt scurt și un răspuns rapid. Dar când un task durează douăzeci de minute să ruleze și rețeaua pică la jumătate, totul se duce de râpă. Pierzi progresul, contextul și bugetul API. Problema orchestrației: De ce LangGraph? este exact despre rezolvarea acestei fragilități. LangGraph este un framework de orchestrare construit pentru a gestiona aplicații stateful, multi-actor. S-ar putea să auzi numele și să presupui că este doar un feature din LangChain. Nu este. LangGraph este un engine de orchestrare low-level și nu ai nevoie deloc de LangChain ca să-l folosești. Există special pentru a modela workflow-urile agenților ca grafuri stateful, în loc de simple scripturi liniare. Scripturile standard se execută in-memory. Dacă un script se oprește neașteptat, toate acele date de runtime dispar. Imaginează-ți un scenariu în care ai un agent în background care analizează un document de o sută de pagini. Agentul a citit, a extras date și a făcut cross-referencing timp de douăzeci de minute neîntrerupte. Dacă apare un timeout de server la minutul nouăsprezece, un script standard pierde tot acel state. Trebuie să iei tot jobul de la capăt. LangGraph rezolvă această problemă de orchestrare prin durable execution. Modelând workflow-ul tău ca un graf, fiecare pas distinct devine un nod, iar conexiunile logice dintre ele sunt muchii. Pe măsură ce aplicația trece de la un nod la altul, LangGraph îi salvează automat progresul. Tratează procesele long-running ca pe o serie de checkpoint-uri sigure. Dacă sistemul dă crash, LangGraph reia execuția exact de unde a rămas. Acest mecanism de checkpointing se bazează pe o memorie completă. Memoria în LangGraph nu este doar o listă continuă de mesaje de chat. Este întregul state al grafului. Când un nod își termină procesarea, actualizează un obiect de shared state. Următorul nod din secvență își citește inputul direct din acel state. Asta înseamnă că memoria persistă pe întregul lifecycle al aplicației. Agentul din background care îți analizează documentul nu uită datele critice pe care le-a găsit la pagina cinci când ajunge, în sfârșit, la pagina nouăzeci, pentru că acel graph state le păstrează în siguranță. Iată ideea cheie. Pentru că acel graph state este pus pe pauză și salvat curat între pași, ai posibilitatea să pui un human in the loop. Uneori, un agent autonom ajunge la un punct de decizie unde are nevoie de permisiune înainte să continue, cum ar fi trimiterea unui e-mail final sau executarea unei tranzacții financiare. Într-un script standard, pauza pentru ca un utilizator să dea click pe un buton duce adesea la connection timeouts. În LangGraph, pur și simplu configurezi un anumit nod să oprească execuția. Sistemul intră în sleep și păstrează perfect acel state curent. Un operator uman poate apoi să revizuiască datele colectate, să aprobe următoarea acțiune sau chiar să modifice manual acel agent state înainte de a da continue. Odată aprobat, graful se trezește și își reia treaba cu contextul actualizat. Ideea de bază este că dezvoltarea de agenți complecși se bazează enorm pe gestionarea de state și failure. LangGraph îți mută arhitectura de la scripturi fragile, memory-bound, către grafuri rezistente care supraviețuiesc întreruperilor, își amintesc trecutul și așteaptă cu răbdare ghidarea umană. Dacă vrei să susții podcastul, poți căuta DevStoriesEU pe Patreon. Asta e tot pentru acest episod. Mulțumesc că ai ascultat și keep building!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 2 din 18. Construiești un agent de inteligență artificială îndesând instrucțiuni, edge cases și exemple într-un singur prompt masiv, apoi îl trimiți mai departe și speri să facă ce trebuie. Funcționează până când pică, iar debugging-ul rezultatului este frustrant. Pentru a repara asta, trebuie să te oprești din a scrie prompt-uri monolit și să începi să gândești sisteme. Asta ne aduce la Gândirea în LangGraph: Modelul Mental. LangGraph te forțează să te îndepărtezi de scripturile liniare. Îți cere să te gândești la aplicația ta ca la un state machine. Primul pas este pur și simplu să-ți definești procesul. Înainte să scrii orice cod, uită-te la ce vrei să obțină sistemul și împarte-l în acțiuni discrete. Gândește-te la un sistem de triaj pentru customer support. Un utilizator trimite un mesaj. Un operator uman ar citi emailul, ar decide dacă este o problemă de facturare sau de suport tehnic, și apoi ar redacta un răspuns potrivit. Această secvență este procesul tău. Mapezi acel proces pe un workflow LangGraph folosind trei componente de bază: State, Nodes și Edges. Să începem cu State. State este memoria partajată a grafului tău. Este o structură care ține contextul întregii tale operațiuni în orice moment dat. Fiecare pas din workflow-ul tău va citi din acest State și va scrie update-uri înapoi în el. Ascultătorii fac adesea o greșeală specifică aici. Ei încearcă să stocheze string-uri de prompt complet formatate în interiorul State-ului. Nu face asta. State-ul ar trebui să țină doar date brute. Ține textul original al emailului de la client, categoria extrasă, sau o listă brută de mesaje anterioare. Formatarea se face on-demand, mai târziu, exact în pasul în care este nevoie de ea. Apoi, avem Nodes. Dacă State-ul este memoria, Nodes sunt workerii care fac task-urile propriu-zise. Un Node este doar o funcție Python care execută un singur pas logic din procesul tău. Primește State-ul curent, execută o acțiune și returnează un update. În exemplul nostru de triaj, ai crea trei Nodes separate. Primul este un Node de Read. Preia emailul primit și salvează textul brut în State. Al doilea este un Node de Classify. Se uită la textul brut din State, cere unui language model să îl categorizeze ca facturare sau tehnic, și salvează categoria rezultată înapoi în State. Al treilea este un Node de Draft. Citește atât emailul, cât și categoria din State, le formatează local într-un prompt și generează un răspuns. Fiecare Node face exact un singur job. În cele din urmă, ai nevoie de o modalitate de a conecta acești workeri. Acesta este rolul Edge-urilor. Edge-urile reprezintă logica de rutare. Ele dictează ce se întâmplă după ce un Node își termină treaba. Un Edge standard spune pur și simplu: odată ce Node-ul de Read termină, mergi mereu la Node-ul de Classify. Dar LangGraph folosește și conditional edges. Aici devine interesant. După Node-ul de Classify, poți folosi un conditional edge pentru a inspecta State-ul. Dacă categoria este facturare, Edge-ul rutează flow-ul către un Node specific de Draft pentru facturare. Dacă este tehnic, îl rutează către un Node de Draft tehnic. Edge-urile iau decizii de trafic pe baza datelor pe care Node-urile tale tocmai le-au produs. Începi cu procesul, izolezi datele într-un State partajat, definești workerii ca Nodes și dictezi flow-ul cu Edges. Spărgând problema în bucăți, izolezi erorile. Tratează-ți aplicația nu ca pe un singur generator de text, ci ca pe o linie de asamblare coordonată, unde datele brute se mută sistematic de la un worker specializat la altul. Mersi de ascultare, happy coding tuturor!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 3 din 18. Două funcții paralele își termină execuția în exact aceeași milisecundă și încearcă să scrie în exact aceeași listă partajată. De obicei, una îi face overwrite celeilalte, iar datele dispar. Pentru a preveni asta, LangGraph se bazează pe The Graph API: State and Reducers. Fundamentul oricărui workflow LangGraph este state-ul său. Îl definești folosind un TypedDict standard din Python. Acest dictionary stabilește exact ce keys va urmări graph-ul tău și data types pentru fiecare key. Gândește-te la el ca la o schemă care este pasată de la un node la altul pe măsură ce graph-ul tău rulează. Iată ideea de bază. Node-urile din LangGraph nu fac mutate direct pe state. Un node primește o copie a state-ului curent, își face treaba și returnează un dictionary de update-uri. O greșeală comună este să presupui că returnarea unui dictionary înlocuiește întregul obiect de state. Nu se întâmplă asta. Dacă state-ul tău are cinci keys, iar node-ul tău returnează un dictionary cu un singur key, LangGraph le lasă pe celelalte patru neatinse și aplică doar update-ul tău specific. Cum aplică LangGraph acest update? Aici intervin reducer-ii. Un reducer este pur și simplu o funcție care dictează modul în care o valoare returnată face merge cu valoarea existentă pentru un anumit key. By default, LangGraph folosește un overwrite reducer. Dacă node-ul tău returnează un string nou pentru un key de status, string-ul vechi dispare, fiind înlocuit complet de cel nou. Uneori, un overwrite este exact ceea ce vrei să eviți. Ia în considerare un workflow paralel de data fetching. Ai un key de state partajat numit results, care este o listă. Pornești două node-uri care rulează în același timp pentru a face fetch la batch-uri diferite de date. Dacă ambele node-uri returnează un dictionary care actualizează key-ul results, comportamentul default de overwrite face ca node-ul care termină ultimul să șteargă munca celuilalt. Pentru a repara asta, adnotezi key-ul results din TypedDict-ul tău cu un reducer specific, cum ar fi operatorul built-in din Python, operator dot add. Acum, când cele două node-uri își returnează listele, reducer-ul acționează ca un agent de circulație. Ia lista existentă și face append în siguranță la output-urile de la ambele node-uri. Nimic nu se pierde. Există un edge case. Ce se întâmplă dacă ai un reducer de tip append pe key-ul results, dar ajungi la un punct în graph-ul tău unde chiar trebuie să golești lista și să o iei de la capăt? Dacă node-ul tău returnează o listă goală, reducer-ul face pur și simplu append cu o listă goală la cea existentă, lăsând datele vechi intacte. Pentru acest scenariu, LangGraph oferă un type special, Overwrite. Când node-ul tău își împachetează update-ul într-un obiect Overwrite, LangGraph îl detectează și dă bypass complet la reducer. Aruncă lista veche și forțează un hard reset. State-ul într-un graph complex nu este o variabilă globală fragilă care primește mutate constant, ci un log append-only de update-uri controlate, guvernat de reguli clare de reducere. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 4 din 18. Uneori vrei doar să scrii un script Python standard, cu if-statements și for-loops normale, dar tot ai nevoie de state persistence enterprise-grade. Nu vrei să construiești un state machine explicit doar pentru a rula în secvență câteva apeluri către un language model. Aici intervine Functional API, mai exact decoratorii entrypoint și task, care rezolvă această problemă. Construirea de aplicații cu nodes și edges explicite te obligă să definești manual modul în care datele sunt rutate de la un pas la altul. Această structură oferă un control imens, dar poate părea greoaie atunci când logica ta este extrem de secvențială sau se bazează pe loops standard de programare. Functional API îți permite să scrii cod Python normal, de sus în jos, păstrând în același timp funcționalitățile built-in de streaming și recovery. În loc să instanțiezi un obiect de tip graph, aplici decoratori pe funcțiile tale Python existente. Începi cu decoratorul task. Îl aplici unităților individuale de lucru din aplicația ta. Gândește-te la un task ca la un pas discret care face ceva specific, cum ar fi interogarea unei baze de date, calcularea unei metrici sau trimiterea unui prompt către un model. Când o funcție are aplicat decoratorul task, framework-ul o pune într-un tracking layer pentru a-i monitoriza execuția. Apoi, folosești decoratorul entrypoint. Îl pui pe funcția principală de orchestrare care direcționează întregul flow. În interiorul acestei funcții entrypoint, apelezi task-urile decorate folosind control flow-ul standard din Python. Atribui output-ul unui task unei variabile, apoi pasezi acea variabilă către următorul task. Poți folosi blocuri try-except, list comprehensions sau while-loops. Logica de orchestrare se comportă exact așa cum se comportă Python-ul nativ. Deoarece codul arată complet standard, ai putea presupune că îi lipsește memoria unei structuri formale de state. Aceasta este o concepție greșită des întâlnită. Functional API îți face totuși automat checkpoint la progres, în culise. De fiecare dată când un task se finalizează, LangGraph interceptează return value-ul și îl salvează într-un persistent store. Framework-ul înregistrează în siguranță input-urile și output-urile fiecărei funcții decorate, pe măsură ce au loc. Ia în considerare un script automat pentru scrierea unui eseu. Definești trei task-uri decorate: o funcție pentru a genera o schiță, o funcție pentru a scrie un paragraf și o funcție pentru a revizui draft-ul. În interiorul funcției tale principale entrypoint, apelezi mai întâi generatorul de schițe. Apoi, scrii un for-loop standard care iterează prin secțiunile acelei schițe, apelând task-ul de scriere a paragrafului pentru fiecare în parte. Adaugi rezultatele la o listă locală. În final, rulezi task-ul de revizuire. Folosești un simplu if-statement pentru a verifica scorul rezultat. Dacă scorul este slab, codul tău declanșează pur și simplu un while-loop pentru a rescrie anumite paragrafe până când scorul se îmbunătățește. Iată ideea cheie. Datorită checkpointing-ului ascuns, dacă scriptul tău întâmpină un network timeout în timp ce scrie al treilea paragraf, nu îți pierzi munca. Când repornești procesul cu același thread identifier, LangGraph știe că schița și primele două paragrafe sunt deja complete. Sare complet peste executarea acelor task-uri, le preia output-urile cached din state store și reia execuția exact de la al treilea paragraf. Functional API mută încărcătura cognitivă de la vizualizarea de routing topologies abstracte înapoi la citirea codului de sus în jos, oferindu-ți reziliența unui state machine cu simplitatea unui plain script. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 5 din 18. Construiești o aplicație de chat, iar un utilizator observă o greșeală de scriere în prompt-ul său. Dă click pe edit, corectează și apasă send. Dar, în loc să înlocuiască mesajul vechi, backend-ul tău pur și simplu adaugă versiunea corectată la finalul istoricului, lăsând greșeala originală exact acolo unde era, ca o fantomă duplicată. Gestionarea istoricului conversației cu MessagesState este modul prin care previi asta. Când developerii construiesc primul lor graph, de obicei definesc un dictionary de state custom pentru a ține istoricul de chat. O greșeală comună este folosirea de list appending standard pentru a gestiona acest istoric. Ei atașează funcția standard operator dot add la lista lor de mesaje. Asta îi spune graph-ului să ia pur și simplu orice mesaje noi și să le lipească la finalul array-ului existent. Această abordare de tip append-only funcționează bine pentru un simplu ping-pong bot în care un utilizator vorbește, AI-ul răspunde, iar istoricul crește secvențial. Dar se strică complet atunci când state-ul trebuie să fie mutable. Dacă un om editează un prompt anterior, sau un agent decide să regenereze ultimul răspuns, adăugarea standard nu poate gestiona asta. Te alegi cu duplicate. LangGraph oferă o structură de state built-in pentru a rezolva asta, numită MessagesState. Conține o singură cheie numită messages. Iată ideea principală. Puterea MessagesState nu stă în cheia în sine, ci în funcția specifică de reducer atașată ei, numită add messages. Reducer-ul add messages nu dă doar un append orb la date. El urmărește ID-urile mesajelor. De fiecare dată când un mesaj nou intră în state, reducer-ul îi verifică ID-ul unic. Dacă acel ID există deja oriunde în istoricul conversației, reducer-ul suprascrie mesajul vechi cu cel nou. Dacă ID-ul este nou, sau dacă mesajul nu are încă un ID, reducer-ul îi dă append la finalul listei. Gândește-te înapoi la scenariul nostru cu greșeala de scriere. Utilizatorul uman trimite un prompt. Sistemul îi dă acelui mesaj ID-ul 123. Utilizatorul își dă seama de greșeală, dă edit la text și trimite corecția. Frontend-ul trimite noul text, etichetându-l explicit cu ID-ul 123. Când acele date ajung în graph, reducer-ul add messages scanează istoricul, găsește mesajul original la ID-ul 123 și înlocuiește textul in place. Fantoma duplicată a dispărut. Conversația curge exact așa cum a fost intenționat. Pe lângă gestionarea ID-urilor, reducer-ul add messages se ocupă și de deserializarea datelor. Într-o aplicație de producție, mesajele tale ajung adesea în formate diferite. Frontend-ul tău ar putea trimite dictionary-uri JSON raw care conțin string-uri de role și content. Node-urile interne ale graph-ului tău ar putea genera obiecte de tip mesaj LangChain native. Reducer-ul acționează ca un traducător universal pentru aceste input-uri. Dacă pasezi o listă de dictionary-uri Python simple în state, funcția add messages le convertește automat în clasele corecte de mesaje LangChain. Nu trebuie să scrii cod boilerplate pentru a da parse unui dictionary într-un HumanMessage sau AIMessage înainte de a actualiza state-ul. Normalizează datele pentru tine. Când construiești agenți de chat, istoricul de state nu este un log append-only, ci este un document viu, iar legarea update-urilor de ID-uri unice de mesaje este ceea ce menține acel document precis. Mersi că ai ascultat. Până data viitoare!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 6 din 18. Alege paradigma de design greșită de la început și fie vei scrie o sută de linii de boilerplate pentru un script simplu, fie vei crea un coșmar spaghetti din funcții Python de bază. Astăzi ne concentrăm pe alegerea abstracției: Graph versus Functional. E ușor să presupui că unul dintre aceste API-uri este inerent mai puternic sau mai production-ready decât celălalt. Asta este fals. În spate, atât Graph API, cât și Functional API se compilează în exact același runtime engine. Ambele suportă persistence, streaming și execution control în exact același mod. Alegerea dintre ele ține pur și simplu de modelul tău mental și de cum vrei să-ți exprimi logica. Hai să ne uităm mai întâi la Functional API. Acesta se bazează pe control flow standard, imperativ, din Python. Scrii funcții Python normale, le marchezi cu un decorator și faci routing execuției folosind if-statements și loop-uri standard. State management-ul aici este în întregime function-scoped. Datele circulă strict de la valoarea de return a unei funcții în argumentele următoarei. Nu există niciun obiect de memorie globală shared care să plutească în background. Dacă workflow-ul tău este liniar sau dacă are o logică predictibilă, tightly scoped, Functional API îți menține codul lean și familiar. Eviți complet overhead-ul definirii structurilor de tip graph. Graph API necesită un mindset diferit. În loc să apelezi funcțiile direct, definești o schemă de state globală, shared. Apoi scrii node-uri, care sunt funcții mici al căror singur job este să citească și să facă mutate pe acel shared state. În final, conectezi explicit acele node-uri între ele folosind edge-uri. Routing-ul nu este gestionat de un conditional statement ascuns adânc în body-ul unei funcții. În schimb, logica ce dictează unde merge aplicația mai departe este extrasă în conditional edges explicite, declarate la top level-ul graph-ului. Aici este ideea principală. Alegi între ele în funcție de cum gestionează sistemul tău state-ul și routing-ul în timp. Imaginează-ți un dev care construiește un tool de bază pentru data extraction. Rulează un singur language model, face parse la output și îl salvează. Functional API este perfect pentru asta. Este rapid de scris și ușor de citit. Dar dăm fast forward trei luni. Acel script simplu este refactorizat într-un sistem multi-agent complex. Acum ai un researcher agent care face hand-off la date către un writer agent, un critic agent care face push back cu corecturi și o pauză de execuție care așteaptă ca un manager uman să aprobe draftul final. Dacă încerci să construiești acel workflow asincron multi-agent cu Functional API, abordarea imperativă cedează. Ajungi să pasezi data payloads masive în sus și în jos prin call stack-uri adânci de funcții. Logica ta de routing ajunge îngropată în condiționale deeply nested. Ăsta e exact momentul în care migrezi către Graph API. Abstracția de tip Graph strălucește aici pentru că decuplează state-ul de execuție. Pentru că state-ul este global și shared, node-urile individuale ale agenților nu trebuie să-și paseze structuri de date grele între ele. Un node pur și simplu citește shared state-ul, face update la cheia specifică de care este responsabil și se termină. Edge-urile explicite preiau controlul, făcând routing-ul foarte vizibil. Te poți uita la definiția graph-ului și poți face map out imediat la întregul workflow, fără să citești o singură linie de business logic. Folosești Functional API când control flow-ul este suficient de simplu încât să-l citești top-to-bottom, dar treci la Graph API când routing-ul devine suficient de complex încât trebuie să-l desenezi pe un whiteboard. Mulțumesc pentru audiție. Aveți grijă de voi, tuturor.

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 7 din 18. Condițiile hardcoded te ajută doar până la un punct când construiești un agent. Dacă un utilizator pune o întrebare complexă, un simplu keyword search nu poate decide cu precizie ce ar trebui să facă aplicația ta mai departe. Cum ar fi dacă AI-ul însuși ar putea dicta traseul workflow-ului tău? Aici intervin dynamic routing și conditional edges. Într-un setup standard de graph, conectezi node A la node B. Este un path static, garantat. Dar când construiești un mecanism inteligent de routing, path-ul trebuie să se schimbe în funcție de datele primite. Ai putea presupune că edge-urile din LangGraph acceptă doar conexiuni hardcoded de tip string. Nu este cazul. Un edge poate fi o funcție Python care citește state-ul curent al graph-ului tău și calculează dinamic numele următorului node. Atașezi această logică la graph-ul tău folosind metoda add conditional edges. Această metodă necesită trei componente. În primul rând, node-ul de pornire. În al doilea rând, o funcție de routing. În al treilea rând, un dictionary care mapează posibilele output-uri de tip string ale funcției tale de routing către node-urile de destinație reale din graph-ul tău. Iată ideea cheie. Cea mai sigură metodă de a controla un conditional edge este să-l combini cu un large language model care generează structured data. Nu vrei ca funcția de routing în sine să facă evaluări complexe sau natural language processing. În schimb, ai un upstream node unde modelul este forțat să returneze o structură strictă, cum ar fi un model Pydantic. Gândește-te la un router de customer service. Un utilizator trimite un mesaj. Primul node din graph-ul tău este un intent classifier. În interiorul acestui node, pasezi mesajul utilizatorului către un language model și îi ceri să returneze un structured output cu un singur field numit intent. Modelul evaluează textul și populează acel field cu o valoare specifică, cum ar fi billing, tech support sau sales. Acest răspuns structurat este apoi salvat în graph state. Acum, conditional edge-ul preia controlul. Edge-ul este atașat de classifier node. Când classifier node-ul se termină, conditional edge-ul declanșează o scurtă funcție Python. Această funcție ia graph state-ul ca input, se uită în state și extrage valoarea intent-ului pe care modelul tocmai a generat-o. Dacă intent-ul este billing, funcția returnează string-ul billing. Conditional edge-ul se uită în mapping dictionary-ul său, vede că string-ul billing corespunde billing node-ului tău și predă execuția acelui node specific. Dacă intent-ul este tech support, returnează un string diferit, direcționând flow-ul către tech support node. Folosești language model-ul pentru capacitățile sale de reasoning ca să categorizezi input-ul, dar păstrezi logica efectivă de routing deterministă. Funcția Python din conditional edge doar citește o variabilă și returnează un string. Este extrem de predictibilă și ușor de testat. Cea mai utilă concluzie aici este că ar trebui mereu să decuplezi decizia de direcție. Lasă language model-ul să decidă intent-ul și să-l scrie în state, apoi folosește un conditional edge pur Python pentru a citi acel state și a direcționa graph-ul. Înainte să încheiem, dacă găsești aceste episoade utile și vrei să ajuți la susținerea emisiunii, poți căuta DevStoriesEU pe Patreon — ne ajută enorm. Asta e tot pentru acest episod. Mulțumesc că asculți și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 8 din 18. Nu poți să dai hardcode căilor tale de execuție când nu ai nicio idee câte sub-task-uri va decide agentul tău să creeze până când rulează efectiv. Dacă sistemul tău decide din mers că trebuie să proceseze trei itemi, sau treizeci de itemi, rutarea statică standard va eșua. Pentru a repara asta, ai nevoie de workflow-uri Map-Reduce cu Send API. O greșeală foarte comună când construiești în LangGraph este să încerci să folosești conditional edges standard pentru un fan-out dinamic. Conditional edges sunt perfecte atunci când vrei să alegi între căi cunoscute, predeterminate, pe baza unei verificări logice. Totuși, ele nu mai fac față atunci când trebuie să dai spawn unui număr necunoscut de task-uri paralele identice la runtime. Paralelizarea standard îți permite să rutezi către mai multe noduri fixe. Numești nodurile, iar graful le declanșează. Dar ce se întâmplă când trebuie să rulezi exact același nod de mai multe ori simultan, fiecare cu o altă bucată de date? Nu poți face asta cu rutarea de bază. Asta ne aduce la pattern-ul Orchestrator-worker. În această arhitectură, un nod central se uită la datele primite, calculează câte task-uri separate sunt necesare și face dispatch unor workeri dinamici pentru a le gestiona concurent. LangGraph permite acest pattern în mod specific prin Send API. Ia în considerare un agent care are ca task scrierea unui raport de research cuprinzător. Primul nod acționează ca orchestrator. Citește prompt-ul utilizatorului și generează un outline. În funcție de complexitatea subiectului, acest outline poate conține trei secțiuni, sau poate conține douăsprezece. Vrei ca un nod worker separat să redacteze fiecare secțiune exact în același timp. Pentru a realiza asta, definești o funcție conditional edge imediat după nodul tău orchestrator. În loc să returnezi un simplu string care indică următorul nod static din graf, această funcție edge returnează o listă de obiecte Send. Aici este ideea cheie. Un obiect Send împachetează o destinație și datele ei împreună. Ia două argumente. Primul argument este numele nodului worker căruia vrei să-i dai trigger. Al doilea argument este payload-ul specific pentru acel worker izolat. În scenariul raportului nostru, funcția orchestrator iterează prin outline-ul generat. Pentru fiecare subiect de secțiune pe care îl găsește, creează un nou obiect Send care indică spre un singur nod numit draft_section, pasând string-ul individual al subiectului ca payload. Când LangGraph evaluează această funcție edge, primește lista de obiecte Send. Apoi, pornește dinamic o instanță paralelă a nodului draft_section pentru fiecare item din acea listă. Dacă orchestratorul a generat un outline cu șapte secțiuni, LangGraph lansează șapte noduri de redactare paralele. Fiecare nod rulează cod identic, dar operează pe propriul său payload unic. Generarea acestor workeri dinamici este faza de map. Colectarea output-urilor lor independente este faza de reduce. Pentru că aceste noduri worker rulează concurent, ele nu pot să suprascrie în siguranță un singur string în state-ul grafului tău. State-ul tău general trebuie configurat pentru a colecta simultan mai multe update-uri primite. Gestionezi asta atașând o funcție reducer la câmpul specific din state care va conține secțiunile redactate, instruindu-l să dea append noilor itemi într-o listă, în loc să suprascrie valoarea anterioară. Pe măsură ce fiecare worker paralel de redactare termină de scris, acesta returnează blocul său de text. LangGraph prinde aceste răspunsuri și folosește reducer-ul tău pentru a stivui în siguranță fiecare bloc de text în array-ul partajat. Odată ce fiecare worker dinamic își finalizează execuția, întregul pas paralel se rezolvă. Workflow-ul merge apoi mai departe, purtând o listă completă și populată cu toate secțiunile redactate. Send API scoate execuția paralelă din definiția statică a grafului tău și o pune direct în mâinile datelor tale de runtime. Mulțumesc că ai stat cu noi. Sper că ai învățat ceva nou.

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 9 din 18. Serverul tău dă crash în mijlocul procesării în timp ce un agent procesează un dataset masiv. Nu ar trebui să fie nevoie să o ia de la zero. Ar trebui să reia exact de unde a rămas. Această reziliență este ceea ce discutăm astăzi legat de Persistence, mai exact Threads și Checkpoints. Pentru a adăuga persistence unei aplicații LangGraph, trebuie să înțelegi conceptul de thread. Un thread reprezintă o singură secvență de execuție izolată sau o conversație specifică a utilizatorului. Acesta păstrează state-ul de lucru al graph-ului pe măsură ce avansează de la un node la altul. Dă-mi voie să clarific ceva imediat. Oamenii confundă adesea memoria unui thread cu memoria pe termen lung cross-session. Un thread nu este o bază de date globală în care agentul tău își amintește pentru totdeauna fapte din diferite task-uri. Este memoria de lucru pe termen scurt, strict legată de o secvență în desfășurare. Activezi această memorie furnizând un checkpointer atunci când compilezi graph-ul. Un checkpointer este un obiect care gestionează salvarea și încărcarea state-ului graph-ului într-un storage backend. Odată ce graph-ul tău este compilat cu un checkpointer, declanșezi persistence-ul pasând un obiect de configurare care conține un thread ID de fiecare dată când invoci graph-ul. Acest ID este cheia unică pe care o folosește checkpointer-ul pentru a urmări istoricul acelui run specific. Când rulezi graph-ul cu acel thread ID, checkpointer-ul salvează automat state-ul. Dar nu salvează continuu. Salvează la limite specifice numite super-steps. Un super-step este un ciclu de execuție distinct în graph. Dacă graph-ul tău rulează node-ul A urmat de node-ul B, asta înseamnă două super-steps. Dacă graph-ul face branch și rulează node-ul C și node-ul D în același timp, acea execuție paralelă este grupată într-un singur super-step. Checkpointer-ul nu întrerupe un node în timp ce lucrează. Așteaptă limita de super-step. Odată ce toate node-urile programate pentru acel super-step termină execuția și returnează update-urile lor, LangGraph creează un checkpoint. Acest checkpoint conține un state snapshot, captând exact cum arată variabilele de state ale graph-ului în acel moment. Să vedem cum se comportă asta în practică. Să presupunem că ai un agent care analizează un dataset masiv. Graph-ul are patru pași. Pasul unu face fetch la date. Pasul doi le curăță. Pasul trei rulează o analiză costisitoare. Pasul patru formatează rezumatul. Începi run-ul, pasând o configurație cu thread ID-ul unu doi trei. Graph-ul finalizează cu succes pașii de fetch, curățare și analiză. La sfârșitul pasului trei, checkpointer-ul salvează un state snapshot. Apoi, înainte ca pasul patru să se poată termina, serverul tău dă crash. Pentru că ai folosit un checkpointer și un thread ID, state-ul este în siguranță. Când serverul tău repornește, pur și simplu invoci din nou graph-ul, pasând exact același thread ID unu doi trei. Checkpointer-ul caută cel mai recent checkpoint. Găsește state snapshot-ul salvat imediat după pasul de analiză. Graph-ul încarcă acel state și reia execuția imediat la pasul patru. Node-urile de fetch, curățare și analiză sunt complet omise deoarece output-urile lor sunt deja stocate în siguranță în checkpoint-ul acelui thread. Iată ideea principală. Prin compilarea graph-ului cu un checkpointer și legarea execuției de un thread ID, transformi operațiunile in-memory fragile în workflow-uri durabile care supraviețuiesc automat întreruperilor. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 10 din 18. Workflow-ul tău procesează o plată, dă de un rate limit la pasul următor și dă crash. Când sistemul își revine și reia workflow-ul, clientul tău este facturat a doua oară. Codul tău nu s-a schimbat, dar presupunerea pe care ai făcut-o despre cum se reia graful a fost greșită. Rezolvarea necesită înțelegerea conceptelor de durable execution și idempotență. Mulți developeri presupun că atunci când un long-running process se pune pe pauză sau dă fail, reluarea lui se face exact de la linia de cod Python unde s-a oprit. Ei se așteaptă ca runtime-ul să țină minte în mod magic variabilele locale la mijlocul funcției. Nu asta se întâmplă. LangGraph nu îngheață interpretorul Python pe loc. State-ul este salvat doar la granițele dintre noduri. Durable execution în LangGraph înseamnă că sistemul îți urmărește progresul persistând state-ul grafului după ce un nod își termină treaba și dă return. Dacă un nod dă fail la jumătatea logicii sale, sistemul nu are nicio înregistrare a progresului său parțial. Ultimul state bun cunoscut este cel pasat nodului atunci când a pornit. Când dai restart sau retry la graf, execuția se reia rulând din nou întregul nod care a dat fail, chiar de la prima linie. Gândește-te la scenariul cu plata. Să presupunem că scrii un singur nod care face două acțiuni. În primul rând, apelează un API extern pentru a taxa un card de credit. În al doilea rând, face un update într-o bază de date remote pentru a înregistra tranzacția. Taxarea cardului are succes, dar conexiunea la baza de date dă timeout, ceea ce face ca nodul să dea crash. State-ul grafului nu avansează. Când workflow-ul se reia, pasează vechiul state înapoi în același nod. Nodul o ia de la capăt. Apelează API-ul extern și taxează cardul de credit a doua oară. Iată ideea cheie. Pentru că nodurile pot da restart de la început, orice side effect din interiorul unui nod trebuie să fie idempotent. Idempotența este o proprietate prin care executarea unei operațiuni de mai multe ori produce exact același rezultat ca executarea ei o singură dată. Dacă nodul tău interacționează cu lumea exterioară, trebuie să scrii codul presupunând că va rula de mai multe ori pentru același pas. Cum asiguri această siguranță? Ai două abordări practice. Prima este folosirea de idempotency keys cu serviciile tale externe. Când apelezi API-ul de plată, pasezi un identificator unic derivat din state-ul curent al grafului. Dacă nodul dă crash și rulează din nou, trimite același identificator unic. Serviciul extern recunoaște request-ul duplicat și returnează un răspuns de succes fără să mute efectiv banii din nou. A doua abordare ține de designul structural al grafului. Dacă o operațiune specifică nu este nativ idempotentă, nu o grupa cu alți pași care ar putea da fail. Pune operațiunea periculoasă în propriul ei nod dedicat. Fă-o să fie singurul lucru pe care îl face acel nod. Dacă pui taxarea plății în nodul A și update-ul bazei de date în nodul B, un timeout la baza de date dă crash doar nodului B. Graful se reia de la nodul B. Taxarea din nodul A este complet sigură, pentru că nodul A a terminat, iar graful și-a salvat state-ul înainte să meargă mai departe. Tu controlezi unde își salvează sistemul progresul prin felul în care desenezi granițele nodurilor. Nu pune niciodată o acțiune ireversibilă, non-idempotentă, în același nod cu ceva care ar putea da fail aleatoriu. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 11 din 18. Uneori, un AI nu ar trebui să aibă ultimul cuvânt. S-ar putea să vrei să blochezi un agent în mijlocul procesării, să ceri aprobarea unui om și să injectezi răspunsul său direct înapoi în logica de execuție. Exact asta fac interrupt-urile Human-in-the-Loop. Când ai nevoie de un om pentru a lua o decizie în cadrul unui workflow LangGraph, folosești o funcție specifică numită interrupt. Este vital să înțelegi ce face aceasta de fapt în spate. Ascultătorii ar putea confunda acest lucru cu un input prompt standard din Python. Nu este. Un input prompt standard blochează un thread activ, ocupând memoria sistemului în timp ce așteaptă ca un utilizator să apese o tastă. În LangGraph, apelarea interrupt se comportă foarte diferit. Serializează complet state-ul grafului, îl salvează în baza ta de date de checkpointer și suspendă complet execuția. Graful intră în repaus. Poate aștepta la nesfârșit un răspuns fără a consuma resurse de calcul active. Flow-ul se desfășoară în două faze distincte: pausing și resuming. Mai întâi, să ne uităm la pausing. Într-unul din nodurile tale, agentul tău ajunge la un punct în care are nevoie de autorizare umană. Exact la acea linie de cod, apelezi funcția interrupt. Pasezi un payload în această funcție, care este de obicei un obiect JSON ce conține contextul de care are nevoie omul. Imaginează-ți un agent care gestionează customer support automatizat. Decide să pregătească o rambursare de cinci sute de dolari. Înainte de a procesa plata, nodul agentului apelează interrupt. Acesta predă un payload care specifică faptul că acțiunea propusă este o rambursare, iar suma este de cinci sute. În momentul în care acea funcție este apelată, graful se oprește. Runtime-ul LangGraph prinde acest eveniment și propagă payload-ul JSON către aplicația ta client. Procesul grafului se închide, lăsând payload-ul în așteptarea revizuirii umane pe un web UI. Acum, a doua fază: trezirea grafului. Un proces extern, cum ar fi serverul tău de backend care primește un apel API de la web UI, este responsabil pentru repornirea grafului. Managerul uman dă click pe aprobare în dashboard-ul său. Backend-ul tău preia acea aprobare și repornește graful folosind o instrucțiune specială numită Command. Când trimiți acest Command, incluzi un argument resume care conține răspunsul uman. În scenariul nostru, acest răspuns este o simplă valoare boolean de true. Iată ideea cheie. Când graful se trezește, nu re-rulează nodul pus în pauză de la început. Reia execuția exact la linia de cod unde s-a oprit. Funcția interrupt care a pus inițial graful în pauză își termină execuția și returnează orice valoare ai trimis prin comanda resume. Răspunsul boolean al omului este injectat direct în variabila care așteaptă rezultatul interrupt-ului. Agentul citește apoi acea valoare true, trece de conditional check-ul său și finalizează rambursarea de cinci sute de dolari. Această arhitectură creează o limită clară. Logica grafului nu trebuie să gestioneze webhook-uri, e-mailuri sau UI-uri. Apelează doar o funcție care aruncă un payload peste zid și așteaptă un return value. Sistemul extern gestionează toată interacțiunea cu utilizatorul și pur și simplu împinge răspunsul înapoi. Prin injectarea răspunsului uman direct în return-ul funcției, eviți poluarea state-ului principal al grafului cu date de interacțiune temporare. Puterea funcției interrupt constă în tratarea feedback-ului uman nu ca pe un ocol arhitectural complex, ci ca pe un function call standard care poate pune universul în pauză în siguranță până când primește un răspuns. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 12 din 18. Agentul tău o ia razna, executând o acțiune pe care nu ți-ai dorit-o sau generând un răspuns groaznic. În mod normal, trebuie să repornești întregul proces de la zero și să speri că se comportă mai bine a doua oară. Ce-ar fi dacă ai putea literalmente să derulezi înapoi execuția până la momentul exact dinaintea greșelii, să modifici manual state-ul și să o lași să ruleze pe un timeline alternativ? Exact asta acoperim astăzi cu Debugging the Past: Time Travel și Forking. Pentru a manipula trecutul, trebuie mai întâi să-l vezi. Faci acest lucru folosind o metodă numită get state history, pasând identificatorul thread-ului tău. Această metodă returnează un iterator care conține fiecare state prin care a trecut graful în timpul execuției acelui thread. Fiecare dintre aceste state-uri istorice posedă un identificator unic numit checkpoint ID. Poți considera acest ID ca fiind coordonatele tale exacte în timp. Dacă vrei pur și simplu să dai replay la graf dintr-un anumit punct, iei checkpoint ID-ul țintă din acel istoric. Apoi apelezi metoda invoke a grafului, pasând un obiect de configurare care include atât thread ID-ul, cât și acel checkpoint ID specific. Graful reia imediat execuția din acel state exact. Nu dă re-run la niciunul dintre node-urile anterioare, economisind compute și timp. Replay-ul este util, dar adevărata putere constă în schimbarea trecutului pentru a da fork la execuție. Să analizăm un scenariu practic. Să presupunem că agentul tău a avut task-ul de a scrie o glumă și a generat o glumă groaznică despre un câine. Verifici state history-ul și găsești checkpoint ID-ul pentru state-ul de dinaintea pasului de generare. În loc să dai doar replay din acel punct, folosești metoda update state. Pasezi thread ID-ul, checkpoint ID-ul istoric specific și noile valori de state pe care vrei să le injectezi. În acest caz, faci update manual la variabila topic, schimbând-o de la un câine la găini. Iată ideea cheie. Developerii cred adesea că un update pe un state anterior dă rollback la execuție, suprascriind sau ștergând istoricul original. Nu este așa. LangGraph funcționează pe o arhitectură append-only. Când apelezi update state pe un checkpoint istoric, sistemul creează în siguranță un checkpoint complet nou, făcând branch din cel vechi. Timeline-ul original, complet cu gluma groaznică despre câine, rămâne complet intact și accesibil. Nu ai șters trecutul; ai dat fork la o nouă realitate. Odată ce aplici acel update, graful se află la un checkpoint nou creat cu state-ul modificat. Pentru a continua pe acest nou timeline, pur și simplu apelezi invoke pe graf din nou cu thread ID-ul, omițând orice checkpoint ID specific. Graful ia ca default cel mai nou state de pe acest branch nou forkat și reia execuția. Agentul tău citește state-ul updatat și generează în schimb o glumă despre găini. Dacă găsești aceste detalieri tehnice utile și vrei să susții emisiunea, poți căuta DevStoriesEU pe Patreon. Time travel-ul transformă debugging-ul din a ghici ce a mers prost, în manipularea precisă a istoricului grafului pentru a explora rezultate alternative, fără a pierde nicio urmă a run-ului original. Asta e tot pentru acest episod. Mulțumesc pentru audiție și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 13 din 18. Construiești un agent, iar un utilizator îi spune că vrea mereu codul în Python 3.11. Mâine, începe o conversație nouă, iar agentul uită complet, returnând în schimb Python 3.9. Memoria de thread este izolată la o singură conversație. Când agentul tău trebuie să rețină informații de-a lungul unor sesiuni complet separate, ai nevoie de memorie pe termen lung folosind Store-uri cross-thread. O greșeală comună este să încerci să rezolvi asta îndesând informații pe termen lung în state-ul de checkpointer. Checkpointerele reprezintă memoria pe termen scurt. Ele sunt strict snapshot-uri de state per thread, concepute pentru a pune pe pauză, a relua sau a da replay la o singură conversație. Dacă un utilizator declară o preferință în thread-ul A, thread-ul B nu are absolut nicio modalitate să o vadă. Pentru a partaja cunoștințe între mai multe thread-uri, LangGraph oferă interfața Store. Un Store este un layer de memorie key-value care stă în afara state-urilor individuale ale thread-urilor. Îl configurezi pasând un obiect de tip store, cum ar fi un PostgresStore, ca argument atunci când compilezi graful. Odată compilat, acel store este atașat la mediul de execuție al grafului. În interiorul grafului tău, nodurile accesează acest layer de memorie prin intermediul obiectului Runtime. Când definești un nod, poți accesa contextul de runtime, care expune store-ul. Pur și simplu accesezi runtime dot store pentru a interacționa cu memoria ta pe termen lung. Iată ideea de bază. Datele dintr-un store sunt organizate folosind namespace-uri. Un namespace este o listă ierarhică de string-uri care îți partiționează datele, la fel ca un path de folder de pe computerul tău. Pentru o aplicație multi-tenant, ai putea defini un namespace care începe cu string-ul users, urmat de un user ID specific, și se termină cu preferences. Gândește-te la scenariul cu asistentul de programare. Un utilizator începe o sesiune luni. În timpul chat-ului, menționează că preferă Python 3.11 și dark mode. Un nod din graful tău recunoaște acest lucru ca pe o preferință permanentă. Apelează metoda put pe runtime dot store. Pasează namespace-ul pentru acel utilizator specific, o cheie unică pentru element și un dicționar care conține preferințele. Datele sunt acum salvate în afara thread-ului. Vineri, același utilizator deschide aplicația ta și pornește un thread complet nou. State-ul de checkpointer pentru acest nou thread este complet gol. Cu toate acestea, graful tău include un nod de setup care rulează primul. Acest nod apelează metoda search pe runtime dot store, furnizând prefixul de namespace al utilizatorului. Store-ul returnează preferințele salvate. Nodul plasează apoi aceste preferințe în state-ul curent al thread-ului. Din acel moment, agentul știe să folosească Python 3.11 și dark mode pentru această nouă conversație. Interfața store oferă trei operațiuni principale. Folosești put pentru a salva sau a suprascrie un element. Folosești get pentru a recupera un singur element atunci când îi cunoști exact namespace-ul și cheia. Folosești search pentru a recupera mai multe elemente care împart un prefix de namespace. Căutarea este deosebit de utilă atunci când ai salvat mai multe fragmente distincte de memorie pentru un utilizator de-a lungul timpului și trebuie să le aduci pe toate în contextul curent. Prin separarea state-ului pe termen scurt de un store cross-thread, decuplezi durata de viață a cunoștințelor agentului tău de durata de viață a unei singure conversații. Mersi că ai ascultat — ne auzim data viitoare.

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 14 din 18. Utilizatorii urăsc să se holbeze la un loading spinner static timp de treizeci de secunde în timp ce sistemul tău lucrează în culise. Vrei să le arăți procesul de gândire al sistemului în timp real, dar capturarea acelor semnale interne necesită adesea să legi custom callbacks complexe. Execuția de tip streaming și formatul v2 rezolvă acest lucru prin unificarea fiecărui eveniment intern într-un singur flow previzibil. Mai întâi, să clarificăm o neînțelegere comună. Inginerii confundă adesea streaming-ul de tokens dintr-un language model cu streaming-ul de state al aplicației. Sunt layere de informații complet diferite. Un token stream este doar text care apare cuvânt cu cuvânt. Un state stream urmărește progresul mai amplu al workflow-ului tău, trecând de la un task la altul. LangGraph le gestionează pe ambele simultan. Accesezi acest comportament cerând un stream și pasând argumentul version v2 către metoda ta de execuție. Asta standardizează output-ul. În loc să lucrezi cu tipuri de date mixte, fiecare eveniment care iese din graful tău devine un dicționar unificat care conține exact trei câmpuri: type, ns și data. Câmpul type definește categoria evenimentului. Câmpul ns vine de la namespace, indicând calea exactă din ierarhia grafului tău de unde a provenit evenimentul. Asta devine critic atunci când ai subgrafuri nested și trebuie să știi exact ce subcomponentă a declanșat evenimentul. În cele din urmă, câmpul data conține payload-ul propriu-zis. Controlezi exact ce intră în acest stream selectând unul sau mai multe stream modes. Modul values îți trimite state-ul complet și actualizat al grafului, de fiecare dată când orice nod își termină treaba. Asta e util dacă aplicația ta are nevoie de imaginea completă la fiecare pas. Modul updates este mult mai light. El face stream doar la datele specifice returnate de un nod, reprezentând doar delta-ul sau schimbarea adusă state-ului general. Modul messages operează la un nivel mai granular, făcând stream la chunk-urile individuale ale unui chat message generat, pe măsură ce sunt produse de un language model din spate. Imaginează-ți o interfață de frontend. Vrei un status indicator luminos care evidențiază ce pas este activ în prezent — poate fetching de context, apoi evaluarea documentelor, apoi drafting — în timp ce afișezi simultan textul draft-ului, token cu token. Ca să construiești asta, începi execuția grafului cu stream modes setate pe updates și messages, asigurându-te că pasezi flag-ul version v2. Frontend-ul tău începe să primească un stream continuu și unificat al acestor dicționare. Când un dicționar sosește cu type setat pe updates, citești câmpul namespace. Asta îți spune exact ce nod tocmai și-a terminat treaba. Folosești acel semnal ca să muți status indicator-ul luminos la pasul următor pe user interface. Milisecunde mai târziu, stream-ul livrează un nou dicționar cu type setat pe messages. Extragi token-ul de raw text din câmpul data și îi dai append direct la paragraful pe care îl citește user-ul tău. Atât schimbările de state high-level, cât și generarea de text low-level ajung prin exact același pipe. Iată ideea cheie. Prin forțarea de tokens, schimbări de state și progres al nodurilor într-o singură structură de dicționar cu trei câmpuri, formatul v2 elimină complet nevoia de a scrie handling logic separat sau callbacks asincrone complexe pentru diferite tipuri de evenimente în timp real. Asta e tot pentru episodul ăsta. Ne auzim data viitoare!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 15 din 18. Când agentul tău AI devine complex, adesea ajungi la un mega-graph masiv, greu de citit, unde o singură modificare strică totul. Nu trebuie să construiești așa — poți construi mini-graphs specializate pe care să le conectezi între ele. Vorbim despre Compunerea Complexității: Subgraphs. Subgraphs îți permit să refolosești logica și să distribui dezvoltarea între echipe diferite. În loc să pui fiecare pas al aplicației tale într-un singur fișier, creezi graphs mai mici, independente. Odată ce un graph este compilat, se comportă exact ca o funcție callable standard. Asta înseamnă că poți lua un întreg graph compilat și să-l pui direct într-un alt graph, ca un singur node. Gândește-te la un sistem master de routing pentru un asistent enterprise. Graph-ul principal gestionează user input-ul, verifică securitatea și decide ce să facă mai departe. Când un utilizator pune o întrebare tehnică profundă, router-ul trebuie să facă un data gathering complex. În loc să scrii acea logică direct în router, o delegi unui subgraph dedicat de Research. O echipă de ingineri complet diferită poate să construiască, să testeze și să rafineze acest graph de Research în mod izolat. Graph-ului părinte nu îi pasă cum se face research-ul. El doar apelează node-ul. Există o tendință comună de a complica excesiv modul în care datele circulă între aceste graphs. Dacă graph-ul părinte și subgraph-ul folosesc exact aceeași state schema — adică împart exact aceleași state keys — nu ai nevoie de adaptoare speciale. Pur și simplu pasezi subgraph-ul de Research compilat direct în funcția add node a master graph-ului tău. Engine-ul introduce automat parent state-ul în subgraph, rulează logica și face merge la rezultate înapoi în parent state când termină. Acum, ce se întâmplă când echipele nu se coordonează perfect? Să presupunem că router-ul tău părinte folosește un state key numit user query, dar echipa separată de ingineri a construit subgraph-ul de Research să aștepte un key numit search term. Nu poți pune subgraph-ul compilat direct în graph-ul părinte. Key-urile nu se vor potrivi, iar execuția va eșua. Iată ideea cheie. Rezolvi această nepotrivire folosind o funcție wrapper simplă. În graph-ul părinte, definești o funcție node standard care acceptă parent state-ul. În interiorul acestei funcții, extragi valoarea user query. Apoi, apelezi manual subgraph-ul de Research compilat, pasându-i un payload unde mapezi acel user query la key-ul search term. Subgraph-ul își rulează logica internă și returnează state-ul final. Funcția ta wrapper preia acel output, traduce rezultatele înapoi în key-urile specifice pe care le așteaptă părintele și le returnează. Pentru graph-ul părinte, acest wrapper arată ca orice node normal. Habar nu are că un subgraph masiv și complex tocmai s-a executat în interiorul său. Pur și simplu a trimis date și a primit înapoi state-ul updatat. Acest pattern îți oferă modularitate strictă fără a sacrifica controlul. Tratarea unui graph compilat ca pe o altă funcție callable în spatele unui wrapper simplu este cel mai puternic mod de a scala o arhitectură AI fără a te prăbuși sub propriul cod. Asta e tot pentru acest episod. Mulțumesc că asculți și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 16 din 18. Dacă un subagent expert este apelat de două ori într-o singură conversație, ar trebui să-și amintească primul apel sau să o ia de la capăt cu amnezie totală? Alegerea asta schimbă totul despre modul în care se comportă sistemele multi-agent și este controlată în întregime de Subgraph Persistence și Multi-Agent Patterns. Când construiești un parent graph care rutează task-uri către subgraphs, state management-ul devine complicat. Gândește-te la un bot principal de customer service care se ocupă de chat-ul general. Când un user pune o întrebare complexă despre o factură, bot-ul principal rutează request-ul către un subgraph dedicat, un Billing Expert. By default, aceste subgraphs sunt complet stateless. Când compilezi acel Billing Expert fără să specifici un checkpointer, funcționează strict per-invocation. Bot-ul principal îi pasează input-urile necesare, expertul execută pașii interni, returnează un rezultat și elimină imediat state-ul intern. Dacă user-ul pune o întrebare de follow-up despre facturare cinci minute mai târziu, bot-ul principal apelează expertul din nou. Expertul nu are nicio memorie a discuției anterioare. Începe complet de la zero. Pentru un subgraph simplu de data-extraction, amnezia asta este perfect în regulă. Dar pentru un agent interactiv și specializat, este incredibil de frustrant pentru user. Ca să repari asta, expertul are nevoie de propria memorie pe parcursul mai multor turn-uri. O greșeală foarte comună aici este să pasezi o instanță nouă de checkpointer, cum ar fi un obiect memory saver, direct în metoda compile a subgraph-ului. Nu face asta decât dacă vrei ca subgraph-ul să facă share la exact același state între useri și sesiuni complet diferite. Dacă user-ul A și user-ul B vorbesc cu sistemul în același timp, pasarea unei instanțe explicite de checkpointer către subgraph înseamnă că datele lor se amestecă într-un singur global state. Asta creează un cross-talk masiv între parent threads izolate. În schimb, pur și simplu pasezi valoarea boolean True argumentului checkpointer atunci când compilezi subgraph-ul. Aici devine interesant. Setarea pe True îi spune subgraph-ului să se bazeze pe mecanismul de checkpointer al parent graph-ului, dar să mențină un istoric multi-turn complet izolat, special pentru el. În spate, framework-ul gestionează namespacing-ul. Creează automat un thread ID unic pentru subgraph, care este legat permanent de thread ID-ul părintelui. Acum, uită-te din nou la scenariul cu Billing Expert cu configurația asta. User-ul pune o întrebare despre factură. Bot-ul principal o rutează către expert. Expertul răspunde și devine inactiv. Mai târziu, în aceeași conversație, user-ul pune o întrebare de follow-up. Bot-ul principal rutează înapoi către expert. Pentru că a fost compilat cu checkpointer-ul setat pe True, expertul se trezește, își verifică sub-thread-ul dedicat și încarcă contextul facturii din turn-ul anterior. Se comportă ca un participant persistent la conversație. Și pentru că acel sub-thread este strict scoped la thread-ul părintelui, un alt user care vorbește cu sistemul primește propria lui instanță complet curată de Billing Expert. Modul în care configurezi checkpointer-ul unui subgraph îi dictează întreaga identitate în sistemul tău: dacă îl lași gol, creezi o utility function de unică folosință, stateless, în timp ce setarea lui pe True creează un colaborator continuu, context-aware. Îți mulțumesc că ai petrecut câteva minute cu mine. Până data viitoare, numai bine.

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 17 din 18. Un script Python care rulează cu succes pe laptopul tău nu este o aplicație de producție. Dacă încerci să rulezi agenți stateful executând scripturi standalone, inevitabil te vei lovi de un zid atunci când va fi timpul să scalezi. Pentru a rezolva asta, ne uităm la structura aplicației și la pregătirea pentru deploy. Când construiești pentru prima dată un agent LangGraph, probabil îl prototipezi într-un Jupyter notebook sau într-un singur fișier Python. Definești nodurile, conectezi edge-urile, compilezi graful și apelezi metoda invoke chiar acolo, în același fișier, ca să vezi dacă funcționează. Asta e în regulă pentru testare. Totuși, un server de producție nu îți poate citi gândurile. Are nevoie de o modalitate standardizată de a servi acel graf ca un API, de a instala pachetele necesare și de a injecta variabile de mediu. Pentru a-ți face prototipul gata pentru deploy, trebuie să îți organizezi codul într-o structură de directoare curată. Să presupunem că creezi un folder nou numit my-app. Îți muți codul Python din notebook într-un fișier curat în interiorul acestui folder. Apoi, adaugi un fișier de dependențe, de obicei requirements dot txt. În cele din urmă, creezi un fișier de configurare numit langgraph dot json în rădăcina folderului my-app. Fișierul langgraph dot json este blueprint-ul principal pentru aplicația ta. Când folosești Command Line Interface-ul LangGraph sau faci deploy într-un mediu de producție, acest fișier de configurare îi spune sistemului de bază exact cum să construiască și să ruleze proiectul tău. Necesită trei informații principale privind dependențele, variabilele de mediu și entry point-urile grafului. Mai întâi, îți declari dependențele. Acesta este doar un string cu calea în fișierul JSON, care indică spre fișierul tău requirements. Se asigură că serverul de deploy instalează exact pachetele Python pe care se bazează agentul tău, prevenind erorile de tip missing module în producție. Apoi, definești string-ul de environment. Acesta indică spre fișierul tău dot env. Agenții stateful au întotdeauna nevoie de secrete, cum ar fi credențialele bazei de date sau API key-urile modelului. Indicarea spre fișierul de environment asigură că runtime-ul încarcă în siguranță aceste chei înainte de a încerca să pornească graful. Asta e partea care contează. A treia cerință din fișierul de configurare este mapping-ul grafurilor. Asta îi spune serverului exact unde se află graful tău compilat în codul sursă. Funcționează ca un dicționar. Îi atribui grafului tău un ID, care devine numele său oficial în API-ul generat. Apoi, mapezi acel ID la un anumit modul Python și la un nume de variabilă. De exemplu, poți mapa ID-ul customer-support-agent la string-ul agent dot py colon compiled-graph. Serverul se uită în fișierul agent dot py, găsește variabila numită compiled-graph și o încarcă în memorie. Această structură necesită o schimbare deliberată în modul în care îți scrii codul. Începătorii rulează adesea grafuri prin scripturi Python standalone care execută acțiuni imediat ce sunt rulate. Dar runtime-ul LangGraph se bazează pe langgraph dot json pentru a expune graful dinamic ca un web service. Nu îți rulează scriptul de sus în jos. Importă doar obiectul graf compilat pe care l-ai specificat în fișierul de configurare. Din această cauză, fișierul tău Python ar trebui doar să definească nodurile, să le conecteze și să atribuie graful compilat unei variabile. Trebuie să elimini orice cod de testare rămas în partea de jos care invocă manual graful. Dacă lași codul de testare în fișier, acesta se va executa în timpul fazei de import pe server, provocând erori de deploy sau apeluri API nedorite doar din pornirea serviciului. Prin declararea explicită a dependențelor, a environment-ului și a path-urilor grafului într-un singur fișier JSON central, separi definiția agentului tău de execuția sa, transformând un script local într-un serviciu robust, deployable. Asta e tot pentru acest episod. Mulțumesc că m-ai ascultat și continuă să construiești!

Salut, sunt Alex de la DEV STORIES DOT EU. LangGraph, episodul 18 din 18. Ai un workflow complex multi-agent și trebuie să testezi un edge case specific de routing în pasul patru. Nu ar trebui să fii nevoit să rulezi întregul sistem de la început până la sfârșit doar pentru a atinge acea condiție. Testarea execuției de graph end-to-end este modul prin care țintești exact logica de care ai nevoie. Când developerii încearcă să izoleze părți dintr-un graph pentru testare, adesea apelează la mock objects complexe. Încearcă să facă mock la structura de graph din jur sau să facă stub la toate nodurile precedente. În LangGraph, nu trebuie să faci asta. Arhitectura se învârte în întregime în jurul state-ului. Deoarece nodurile sunt doar funcții care citesc și scriu state, poți injecta manual un payload specific de state și poți testa fragmente izolate de noduri în mod nativ. Aici devin incredibil de utile state injection și breakpoints în test suite-ul tău. Ai nevoie doar de două tool-uri pentru a sări direct în mijlocul unui graph. Primul este metoda update state. Al doilea este un parametru de configurare numit interrupt after. Folosirea acestora într-un testing framework standard, cum ar fi pytest, îți permite să simulezi condiții exacte fără a executa întreaga aplicație. Să aplicăm asta pe un scenariu concret. Să presupunem că ai un graph în care nodul trei face un API call extern, iar nodul patru verifică rezultatul. Vrei să verifici că, dacă payload-ul API conține un failure code specific, nodul patru rutează corect flow-ul de execuție către nodul tău de error handler. În loc să rulezi nodurile unu și doi pentru a declanșa asta, izolezi problema. Inițializezi graph-ul cu un thread identifier. Apoi, folosești update state pentru a insera un payload API simulat, eșuat, direct în thread state. Acționezi ca și cum nodul trei este pe cale să ruleze cu acele date specifice. Mai departe, invoci graph-ul, dar pasezi un configuration dictionary care setează interrupt after la nodul patru. Când pornești graph-ul, execuția începe imediat la nodul trei folosind failure state-ul injectat de tine. Nodul trei procesează payload-ul greșit și pasează state-ul rezultat către nodul patru. Nodul patru evaluează logica și decide să ruteze către error handler. Pentru că ai setat un breakpoint, graph-ul pune pe pauză execuția în momentul în care nodul patru termină. Acum testul tău poate evalua rezultatul. Extragi state-ul curent din graph. Poți scrie aserțiunile tale pentru a te asigura că nodul patru a actualizat corect variabilele de state. Mai important, poți inspecta planul de execuție al graph-ului. Uitându-te la următorul nod pending din state metadata, poți confirma că logica de routing a funcționat perfect și că error handler-ul este pus în coadă. Iată ideea cheie. Prin manipularea directă a state-ului, transformi un chain de agenți extrem de interconectat și imprevizibil într-un test case determinist, pas cu pas. Verifici exact cum tranziționează graph-ul de la un nod la altul, fără a aștepta language models sau network calls în pașii precedenți. Deoarece acesta este episodul final al seriei, te încurajez să explorezi documentația oficială și să încerci să construiești aceste workflow-uri hands-on. Dacă ai idei despre ce ar trebui să acoperim în continuare, vizitează devstories dot eu și sugerează un subiect. Asta e tot pentru acest episod. Mulțumesc că m-ai ascultat și continuă să construiești!