LangGraph

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 1 di 18. La maggior parte degli script per large language model funziona perfettamente per un prompt veloce e una risposta immediata. Ma quando un task ci mette venti minuti a girare e la rete cade a metà, salta tutto. Perdi i progressi, il context e il tuo budget API. The Orchestration Problem: Why LangGraph? parla esattamente di come risolvere questa fragilità. LangGraph è un framework di orchestrazione creato per gestire applicazioni stateful e multi-actor. Potresti sentire il nome e pensare che sia solo una feature di LangChain. Non lo è. LangGraph è un motore di orchestrazione di più basso livello, e non hai affatto bisogno di usare LangChain per poterlo utilizzare. Esiste specificamente per modellare i workflow degli agent come grafi stateful, anziché come semplici script lineari. Gli script standard vengono eseguiti in memoria. Se uno script si interrompe inaspettatamente, tutti i dati di runtime scompaiono. Immagina uno scenario in cui hai un agent in background che analizza un documento di cento pagine. L'agent ha letto, estratto dati e incrociato informazioni per venti minuti ininterrotti. Se si verifica un timeout del server al diciannovesimo minuto, uno script standard perde tutto quello state. Devi ricominciare l'intero job da capo. LangGraph risolve questo problema di orchestrazione attraverso la durable execution. Modellando il tuo workflow come un grafo, ogni singolo passaggio diventa un nodo, e le connessioni logiche tra di essi sono gli edge. Man mano che l'applicazione passa da un nodo all'altro, LangGraph salva automaticamente i progressi. Tratta i processi long-running come una serie di checkpoint sicuri. Se il sistema crasha, LangGraph riprende l'esecuzione esattamente da dove si era interrotta. Questo meccanismo di checkpointing si basa su una memoria completa. La memoria in LangGraph non è solo una lista continua di messaggi di chat. È l'intero state del grafo. Quando un nodo termina la sua elaborazione, aggiorna un oggetto di state condiviso. Il nodo successivo nella sequenza legge il suo input direttamente da quello state. Questo significa che la memoria persiste per l'intero lifecycle dell'applicazione. L'agent in background che analizza il tuo documento non dimentica i dati cruciali trovati a pagina cinque quando finalmente arriva a pagina novanta, perché lo state del grafo li conserva in modo sicuro. Ecco il punto chiave. Dato che lo state del grafo viene messo in pausa e salvato in modo pulito tra un passaggio e l'altro, hai la possibilità di inserire un human in the loop. A volte un agent autonomo raggiunge un punto decisionale in cui ha bisogno di un'autorizzazione prima di procedere, come inviare un'email finale o eseguire una transazione finanziaria. In uno script standard, mettersi in pausa per far cliccare un pulsante all'utente spesso causa dei timeout di connessione. In LangGraph, ti basta configurare un nodo specifico per fermare l'esecuzione. Il sistema va in sleep e preserva perfettamente lo state corrente. Un operatore umano può quindi esaminare i dati raccolti, approvare l'azione successiva, o persino modificare manualmente lo state dell'agent prima di premere continue. Una volta approvato, il grafo si sveglia e riprende il suo lavoro con il context aggiornato. Il concetto fondamentale è che la creazione di agent complessi si basa fortemente sulla gestione dello state e dei fallimenti. LangGraph sposta la tua architettura da script fragili e memory-bound verso grafi resilienti che sopravvivono alle interruzioni, ricordano il loro passato e attendono pazientemente la guida umana. Se desideri supportare il podcast, puoi cercare DevStoriesEU su Patreon. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 2 di 18. Costruisci un agente di intelligenza artificiale infilando istruzioni, edge case ed esempi in un unico, enorme prompt, poi lo invii e speri che faccia la cosa giusta. Funziona finché non fallisce, e fare il debug del risultato è frustrante. Per risolvere questo problema, devi smettere di scrivere prompt monolitici e iniziare a progettare sistemi. Questo ci porta a Thinking in LangGraph: The Mental Model. LangGraph ti costringe ad allontanarti dagli script lineari. Ti chiede di pensare alla tua applicazione come a una state machine. Il primissimo passo è semplicemente definire il tuo processo. Prima di scrivere codice, guarda cosa vuoi che il sistema ottenga e scomponilo in azioni distinte. Considera un sistema di triage per il supporto clienti. Un utente invia un messaggio. Un operatore umano leggerebbe l'email, deciderebbe se è un problema di fatturazione o di supporto tecnico, e poi scriverebbe una risposta appropriata. Quella sequenza è il tuo processo. Mappi quel processo su un workflow di LangGraph usando tre componenti principali, che sono State, Node ed Edge. Iniziamo dallo State. Lo State è la memoria condivisa del tuo graph. È una struttura che contiene il contesto dell'intera operazione in qualsiasi momento. Ogni passaggio nel tuo workflow leggerà da questo State e ci scriverà sopra degli aggiornamenti. Gli ascoltatori spesso fanno un errore specifico qui. Cercano di salvare delle string di prompt completamente formattate all'interno dello State. Non farlo. Lo State dovrebbe contenere solo dati grezzi. Contiene il testo originale dell'email del cliente, la categoria estratta o una lista grezza dei messaggi precedenti. La formattazione avviene on-demand, più tardi, esattamente all'interno del passaggio in cui è necessaria. Poi, abbiamo i Node. Se lo State è la memoria, i Node sono i worker che svolgono i task effettivi. Un Node è semplicemente una funzione Python che esegue un singolo passaggio logico del tuo processo. Riceve lo State corrente, esegue un'azione e restituisce un aggiornamento. Nel nostro esempio di triage, creeresti tre Node separati. Il primo è un Read Node. Prende l'email in arrivo e salva il testo grezzo nello State. Il secondo è un Classify Node. Guarda il testo grezzo nello State, chiede a un language model di categorizzarlo come fatturazione o tecnico, e salva la categoria risultante di nuovo nello State. Il terzo è un Draft Node. Legge sia l'email che la categoria dallo State, le formatta localmente in un prompt e genera una risposta. Ogni Node fa esattamente un lavoro. Infine, ti serve un modo per connettere questi worker. Questo è il ruolo degli Edge. Gli Edge rappresentano la logica di routing. Dettano cosa succede dopo che un Node finisce il suo lavoro. Un Edge standard dice semplicemente: una volta che il Read Node finisce, vai sempre al Classify Node. Ma LangGraph usa anche degli Edge condizionali. Ed è qui che diventa interessante. Dopo il Classify Node, puoi usare un Edge condizionale per ispezionare lo State. Se la categoria è fatturazione, l'Edge fa il routing del flusso verso uno specifico Draft Node per la fatturazione. Se è tecnico, fa il routing verso un Draft Node tecnico. Gli Edge prendono decisioni di traffico in base ai dati che i tuoi Node hanno appena prodotto. Inizi con il processo, isoli i dati in uno State condiviso, definisci i worker come Node, e detti il flusso con gli Edge. Scomponendo il problema, isoli i fallimenti. Tratta la tua applicazione non come un singolo generatore di testo, ma come una catena di montaggio coordinata in cui i dati grezzi si muovono sistematicamente da un worker specializzato al successivo. Grazie per l'ascolto, happy coding a tutti!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 3 di 18. Due funzioni parallele terminano l'esecuzione nello stesso esatto millisecondo e provano a scrivere nella stessa identica list condivisa. Di solito, una sovrascrive l'altra e i dati svaniscono. Per evitare questo, LangGraph si affida alla Graph API: State e Reducers. Le fondamenta di qualsiasi workflow LangGraph sono il suo state. Lo definisci usando un TypedDict standard di Python. Questo dictionary definisce le key esatte che il tuo graph terrà tracciate e i tipi di dato per ogni key. Pensalo come lo schema che viene passato da node a node durante l'esecuzione del tuo graph. Ecco il concetto chiave. I node in LangGraph non mutano lo state direttamente. Un node riceve una copia dello state corrente, fa il suo lavoro e restituisce un dictionary di update. Un errore comune è dare per scontato che restituire un dictionary sostituisca l'intero oggetto state. Non è così. Se il tuo state ha cinque key e il tuo node restituisce un dictionary con una sola key, LangGraph lascia le altre quattro intatte e applica solo il tuo update specifico. Come fa LangGraph ad applicare quell'update? È qui che entrano in gioco i reducer. Un reducer è semplicemente una funzione che stabilisce come un valore restituito fa il merge con il valore esistente per una key specifica. Di default, LangGraph usa un overwrite reducer. Se il tuo node restituisce una nuova string per una key di status, la vecchia string sparisce, rimpiazzata completamente da quella nuova. A volte l'overwrite è esattamente ciò che vuoi evitare. Prendi ad esempio un workflow di data fetching parallelo. Hai una key di state condivisa chiamata results, che è una list. Avvii due node in esecuzione contemporaneamente per fare il fetch di batch di dati diversi. Se entrambi i node restituiscono un dictionary che aggiorna la key results, il comportamento di overwrite di default fa sì che il node che finisce per ultimo cancelli il lavoro dell'altro. Per risolvere questo problema, annoti la key results nel tuo TypedDict con un reducer specifico, come la funzione built-in di Python operator dot add. Ora, quando i due node restituiscono le loro list, il reducer agisce come un vigile urbano. Prende la list esistente e fa un append in modo sicuro degli output di entrambi i node. Non viene perso nulla. C'è un edge case. Cosa succede se hai un reducer in stile append sulla tua key results, ma raggiungi un punto nel tuo graph in cui hai davvero bisogno di svuotare la list e ricominciare da capo? Se il tuo node restituisce una list vuota, il reducer si limita a fare l'append di una list vuota a quella esistente, lasciando intatti i vecchi dati. Per questo scenario, LangGraph fornisce un tipo speciale Overwrite. Quando il tuo node wrappa il suo update in un oggetto Overwrite, LangGraph lo rileva e bypassa completamente il reducer. Butta via la vecchia list e forza un hard reset. Lo state in un graph complesso non è una variabile globale fragile che viene costantemente mutata, ma un log append-only di update controllati, governati da chiare regole di riduzione. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 4 di 18. A volte vuoi semplicemente scrivere uno script Python standard con normali if-statement e for-loop, ma hai comunque bisogno di una state persistence di livello enterprise. Non vuoi costruire una state machine esplicita solo per eseguire in sequenza qualche chiamata al language model. È qui che la Functional API, in particolare i decorator entrypoint e task, risolve il problema. Creare applicazioni con node ed edge espliciti ti richiede di definire manualmente come i dati fanno routing da uno step all'altro. Questa struttura offre un controllo immenso, ma può risultare pesante quando la tua logica è altamente sequenziale o si basa su loop di programmazione standard. La Functional API ti permette di scrivere normale codice Python top-to-bottom, mantenendo le funzionalità integrate di streaming e recovery. Invece di istanziare un oggetto graph, applichi i decorator alle tue funzioni Python esistenti. Inizi con il decorator task. Lo applichi alle singole unità di lavoro nella tua applicazione. Pensa a un task come a uno step discreto che fa qualcosa di specifico, come fare una query a un database, calcolare una metrica o fare prompting su un modello. Quando una funzione ha il decorator task, il framework la avvolge in un tracking layer per monitorare la sua esecuzione. Successivamente, usi il decorator entrypoint. Lo posizioni sulla funzione di orchestrazione principale che dirige il flow complessivo. All'interno di questa funzione entrypoint, chiami i tuoi task decorati usando il control flow standard di Python. Assegni l'output di un task a una variabile, e poi passi quella variabile al task successivo. Puoi usare blocchi try-except, list comprehension o while-loop. La logica di orchestrazione si comporta esattamente come il Python nativo. Dato che il codice sembra del tutto standard, potresti pensare che gli manchi la memoria di una struttura di stato formale. Questo è un malinteso comune. La Functional API fa comunque il checkpointing automatico dei tuoi progressi dietro le quinte. Ogni volta che un task viene completato, LangGraph intercetta il return value e lo salva in un persistent store. Il framework registra in modo sicuro gli input e gli output di ogni funzione decorata man mano che avvengono. Considera uno script per la scrittura automatica di un saggio. Definisci tre task decorati: una funzione per generare una scaletta, una funzione per scrivere un paragrafo e una funzione per fare la review della bozza. All'interno della tua funzione entrypoint principale, chiami prima il generatore della scaletta. Successivamente, scrivi un for-loop standard che itera sulle sezioni di quella scaletta, chiamando il task di scrittura del paragrafo per ciascuna di esse. Fai l'append dei risultati a una lista locale. Infine, esegui il task di review. Usi un semplice if-statement per controllare lo score risultante. Se lo score è basso, il tuo codice triggera semplicemente un while-loop per riscrivere paragrafi specifici finché lo score non migliora. Ecco il punto chiave. Grazie al checkpointing nascosto, se il tuo script incontra un network timeout durante la scrittura del terzo paragrafo, non perdi il tuo lavoro. Quando riavvii il processo con lo stesso thread identifier, LangGraph sa che la scaletta e i primi due paragrafi sono già completi. Salta completamente l'esecuzione di quei task, recupera i loro output in cache dallo state store, e riprende l'esecuzione esattamente dal terzo paragrafo. La Functional API sposta il carico cognitivo dalla visualizzazione di topologie di routing astratte di nuovo alla lettura del codice top-to-bottom, dandoti la resilienza di una state machine con la semplicità di un normale script. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 5 di 18. Sviluppi un'app di chat e un utente nota un errore di battitura nel suo prompt. Clicca su modifica, lo corregge e preme invio. Ma invece di sostituire il vecchio messaggio, il tuo backend si limita ad accodare la versione corretta alla fine della history, lasciando l'errore originale esattamente dov'era, come un duplicato fantasma. Gestire la history della conversazione con MessagesState è il modo per evitare tutto questo. Quando gli sviluppatori creano il loro primo graph, di solito definiscono uno state dictionary personalizzato per contenere la history della chat. Un errore comune è usare il classico list appending per gestire questa history. Collegano la funzione standard operator dot add alla loro lista di messages. Questo dice al graph di prendere semplicemente i nuovi messages e incollarli alla fine dell'array esistente. Questo approccio append-only funziona bene per un semplice bot ping-pong in cui un utente parla, l'AI risponde e la history cresce in modo sequenziale. Ma si rompe completamente quando lo state deve essere mutabile. Se un umano modifica un prompt passato, o un agent decide di rigenerare la sua ultima response, la standard addition non può gestirlo. Finisci per avere dei duplicati. LangGraph fornisce una struttura di state built-in per risolvere questo problema, chiamata MessagesState. Contiene una singola key chiamata messages. Ecco l'intuizione chiave. La potenza di MessagesState non è la key in sé, ma la specifica funzione reducer collegata ad essa, chiamata add messages. Il reducer add messages non si limita ad appendere dati alla cieca. Tiene traccia degli ID dei messages. Ogni volta che un nuovo message entra nello state, il reducer controlla il suo ID univoco. Se quell'ID esiste già da qualche parte nella history della conversazione, il reducer sovrascrive il vecchio message con quello nuovo. Se l'ID è nuovo, o se il message non ha ancora un ID, il reducer lo appende alla fine della lista. Ripensa al nostro scenario dell'errore di battitura. L'utente umano invia un prompt. Il sistema assegna a quel message un ID 123. L'utente si accorge dell'errore, modifica il testo e invia la correzione. Il tuo frontend invia il nuovo testo, taggandolo esplicitamente con l'ID 123. Quando quei dati arrivano al graph, il reducer add messages scansiona la history, trova il message originale all'ID 123 e sostituisce il testo in place. Il duplicato fantasma è sparito. La conversazione scorre esattamente come previsto. Oltre a gestire gli ID, il reducer add messages gestisce anche la deserializzazione dei dati. In un'applicazione in produzione, i tuoi messages spesso arrivano in formati diversi. Il tuo frontend potrebbe inviare dizionari JSON raw contenenti le stringhe role e content. I tuoi nodi interni del graph potrebbero generare oggetti message nativi di LangChain. Il reducer agisce come un traduttore universale per questi input. Se passi una lista di semplici dizionari Python nello state, la funzione add messages li converte automaticamente nelle classi message corrette di LangChain. Non devi scrivere codice boilerplate per fare il parsing di un dizionario in un HumanMessage o AIMessage prima di aggiornare lo state. Normalizza i dati per te. Quando costruisci chat agents, la state history non è un log append-only, è un documento vivo, e legare i tuoi aggiornamenti a ID univoci dei messages è ciò che mantiene quel documento accurato. Grazie per l'ascolto. Alla prossima!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 6 di 18. Scegli il paradigma di design sbagliato fin dall'inizio, e finirai per scrivere cento righe di boilerplate per un semplice script, oppure per intrecciare un incubo di spaghetti code partendo da funzioni Python di base. Oggi ci concentriamo su Scegliere la Tua Astrazione: Graph o Functional. È facile supporre che una di queste API sia intrinsecamente più potente o più production-ready dell'altra. Questo è falso. Dietro le quinte, sia la Graph API che la Functional API vengono compilate nello stesso identico runtime engine. Entrambe supportano la persistence, lo streaming e l'execution control esattamente allo stesso modo. La scelta tra le due riguarda puramente il tuo modello mentale e come vuoi esprimere la tua logica. Guardiamo prima la Functional API. Questa si basa sul control flow imperativo standard di Python. Scrivi normali funzioni Python, le annoti con un decorator e fai il routing della tua esecuzione usando statement if e loop standard. Lo state management qui è interamente function-scoped. I dati fluiscono rigorosamente dal return value di una funzione agli argomenti della successiva. Non c'è nessun oggetto di memoria globale condiviso che fluttua in background. Se il tuo workflow è lineare, o se ha una logica prevedibile e strettamente limitata, la Functional API mantiene il tuo codice snello e familiare. Eviti completamente l'overhead di definire le strutture del graph. La Graph API richiede una mentalità diversa. Invece di chiamare direttamente le funzioni, definisci uno state schema globale e condiviso. Poi scrivi i nodi, che sono piccole funzioni il cui unico compito è leggere e mutare quello state condiviso. Infine, colleghi esplicitamente questi nodi tra loro usando degli edge. Il routing non è gestito da uno statement condizionale nascosto in profondità nel body di una funzione. Invece, la logica che detta dove andrà l'applicazione successivamente viene tirata fuori in conditional edge espliciti, dichiarati al top level del graph. Ecco l'intuizione chiave. Scegli tra le due in base a come il tuo sistema gestisce lo state e il routing nel tempo. Immagina uno sviluppatore che crea un tool di data extraction di base. Fa girare un singolo language model, parsa l'output e lo salva. La Functional API è perfetta per questo. È veloce da scrivere e facile da leggere. Ma andiamo avanti veloce di tre mesi. Quel semplice script sta subendo un refactoring per diventare un complesso sistema multi-agent. Ora hai un agent researcher che passa i dati a un agent writer, un agent critic che risponde con delle correzioni, e una execution pause in attesa che un manager umano approvi la bozza finale. Se provi a costruire quel workflow multi-agent asincrono con la Functional API, l'approccio imperativo crolla. Finisci per passare enormi payload di dati su e giù per call stack di funzioni profondissimi. La tua logica di routing finisce sepolta all'interno di statement condizionali profondamente annidati. Questo è il momento esatto in cui migri alla Graph API. L'astrazione Graph brilla in questo caso perché disaccoppia lo state dall'esecuzione. Poiché lo state è globale e condiviso, i tuoi singoli nodi agent non hanno bisogno di passarsi strutture dati pesanti. Un nodo legge semplicemente lo state condiviso, aggiorna la key specifica di cui è responsabile e termina. Gli edge espliciti prendono il sopravvento, rendendo il routing altamente visibile. Puoi guardare la definizione del graph e mappare immediatamente l'intero workflow senza leggere una singola riga di business logic. Usi la Functional API quando il control flow è abbastanza semplice da essere letto dall'alto verso il basso, ma passi alla Graph API quando il routing diventa così complesso da doverlo disegnare su una lavagna. Grazie per l'ascolto. Statemi bene, tutti.

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 7 di 18. I conditional hardcoded hanno dei limiti quando costruisci un agent. Se un utente fa una domanda complessa, una semplice ricerca per keyword non può decidere in modo affidabile cosa dovrebbe fare la tua applicazione dopo. E se l'IA stessa potesse dettare il percorso del tuo workflow? È qui che entrano in gioco il routing dinamico e i conditional edge. In una configurazione standard di un graph, colleghi il node A al node B. È un percorso statico e garantito. Ma quando costruisci un meccanismo di routing intelligente, il percorso deve cambiare in base ai dati in arrivo. Potresti pensare che gli edge in LangGraph accettino solo connessioni con string hardcoded. Non è così. Un edge può essere una funzione Python che legge lo state corrente del tuo graph e calcola dinamicamente il nome del node successivo. Aggiungi questa logica al tuo graph usando il method add_conditional_edges. Questo method richiede tre componenti. Primo, il node di partenza. Secondo, una funzione di routing. Terzo, un dictionary che mappa i possibili output string della tua funzione di routing ai node di destinazione effettivi nel tuo graph. Ecco il punto chiave. Il modo più affidabile per guidare un conditional edge è combinarlo con un Large Language Model che genera dati strutturati. Non vuoi che la funzione di routing stessa esegua valutazioni complesse o Natural Language Processing. Invece, hai un node a monte in cui il modello è forzato a restituire una struttura rigorosa, come un modello Pydantic. Prendi come esempio un router per l'assistenza clienti. Un utente invia un messaggio. Il primo node nel tuo graph è un classifier di intent. All'interno di questo node, passi il messaggio dell'utente a un language model e gli richiedi di restituire un output strutturato con un singolo campo chiamato intent. Il modello valuta il testo e popola quel campo con un valore specifico, come billing, tech support o sales. Questa risposta strutturata viene quindi salvata nello state del graph. A questo punto entra in gioco il conditional edge. L'edge è collegato al node del classifier. Quando il node del classifier termina, il conditional edge attiva una breve funzione Python. Questa funzione prende in input lo state del graph, guarda all'interno dello state ed estrae il valore dell'intent appena generato dal modello. Se l'intent è billing, la funzione restituisce la string billing. Il conditional edge consulta il suo dictionary di mappatura, vede che la string billing corrisponde al tuo node di billing e passa l'esecuzione a quel node specifico. Se l'intent è tech support, restituisce una string diversa, facendo il routing del flusso al node di tech support. Stai usando il language model per le sue capacità di ragionamento per categorizzare l'input, ma stai mantenendo la logica di routing effettiva deterministica. La funzione Python nel conditional edge si limita a leggere una variabile e a restituire una string. È altamente prevedibile e facile da testare. La cosa più utile da portarsi a casa qui è che dovresti sempre separare la decisione dalla direzione. Lascia che il language model decida l'intent e lo scriva nello state, poi usa un conditional edge in puro Python per leggere quello state e guidare il graph. Prima di concludere, se trovi utili questi episodi e vuoi supportare lo show, puoi cercare DevStoriesEU su Patreon: ci aiuta davvero molto. Questo è tutto per questo episodio. Grazie per l'ascolto e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 8 di 18. Non puoi fare l'hardcode dei tuoi execution path quando non hai idea di quanti sub-task il tuo agent deciderà di creare finché non viene effettivamente eseguito. Se il tuo sistema decide al volo di dover elaborare tre elementi, o trenta, il routing statico standard fallirà. Per risolvere questo problema, hai bisogno dei workflow Map-Reduce con la Send API. Un errore molto comune quando sviluppi in LangGraph è cercare di usare i conditional edge standard per il fan-out dinamico. I conditional edge sono perfetti quando vuoi scegliere tra percorsi noti e predeterminati in base a un controllo logico. Tuttavia, falliscono quando devi generare un numero sconosciuto di task paralleli identici a runtime. La parallelizzazione standard ti permette di fare routing verso nodi multipli fissi. Dai un nome ai nodi, e il graph li triggera. Ma cosa succede quando devi eseguire l'esatto stesso nodo più volte in simultanea, ognuna con un dato diverso? Non puoi farlo con il routing di base. Questo ci porta al pattern Orchestrator-worker. In questa architettura, un nodo centrale analizza i dati in ingresso, calcola quanti task separati sono necessari e invia worker dinamici per gestirli in simultanea. LangGraph abilita questo pattern nello specifico tramite la Send API. Considera un agent incaricato di scrivere un report di ricerca completo. Il primo nodo fa da orchestrator. Legge il prompt dell'utente e genera una scaletta. A seconda della complessità dell'argomento, questa scaletta potrebbe contenere tre sezioni, o magari dodici. Vuoi che un nodo worker separato scriva la bozza di ogni sezione esattamente nello stesso momento. Per ottenere questo risultato, definisci una funzione di conditional edge subito dopo il tuo nodo orchestrator. Invece di restituire una semplice string che punta al nodo statico successivo nel graph, questa funzione di edge restituisce una lista di oggetti Send. Ecco il punto chiave. Un oggetto Send impacchetta insieme una destinazione e i suoi dati. Prende due argomenti. Il primo argomento è il nome del nodo worker che vuoi triggerare. Il secondo argomento è il payload specifico per quel worker isolato. Nel nostro scenario del report, la funzione orchestrator itera attraverso la scaletta generata. Per ogni argomento di sezione che trova, crea un nuovo oggetto Send che punta a un singolo nodo chiamato draft_section, passando la string del singolo argomento come payload. Quando LangGraph valuta questa funzione di edge, riceve la lista di oggetti Send. Quindi avvia dinamicamente un'istanza parallela del nodo draft_section per ogni singolo elemento in quella lista. Se l'orchestrator ha generato una scaletta con sette sezioni, LangGraph lancia sette nodi di drafting paralleli. Ogni nodo esegue codice identico, ma opera sul proprio payload univoco. Generare questi worker dinamici è la fase di map. Raccogliere i loro output indipendenti è la fase di reduce. Dato che questi nodi worker vengono eseguiti in simultanea, non possono sovrascrivere in modo sicuro una singola string nel tuo graph state. Il tuo state complessivo deve essere configurato per raccogliere simultaneamente aggiornamenti multipli in entrata. Gestisci questo aspetto attaccando una funzione reducer allo specifico campo dello state che conterrà le tue sezioni in bozza, istruendola ad appendere i nuovi elementi a una lista anziché sovrascrivere il valore precedente. Man mano che ogni draft worker parallelo finisce di scrivere, restituisce il suo blocco di testo. LangGraph intercetta queste risposte e usa il tuo reducer per impilare in modo sicuro ogni blocco di testo nell'array condiviso. Una volta che ogni worker dinamico completa la sua esecuzione, l'intero step parallelo si risolve. Il workflow quindi va avanti, portando con sé una lista completa e popolata di tutte le sezioni in bozza. La Send API toglie l'esecuzione parallela dalla definizione statica del tuo graph e la mette direttamente nelle mani dei tuoi dati a runtime. Grazie per avermi fatto compagnia. Spero tu abbia imparato qualcosa di nuovo.

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 9 di 18. Il tuo server crasha a metà dell'elaborazione mentre un agent sta processando un enorme dataset. Non dovrebbe dover ricominciare da zero. Dovrebbe riprendere esattamente da dove si era interrotto. Questa resilienza è ciò di cui parliamo oggi con la persistence, in particolare i thread e i checkpoint. Per aggiungere la persistence a un'applicazione LangGraph, devi capire il concetto di thread. Un thread rappresenta una singola sequenza di esecuzione isolata o una specifica conversazione con l'utente. Mantiene il working state del graph mentre si sposta da un node all'altro. Fammi chiarire subito una cosa. Spesso si confonde la memory del thread con la memory a lungo termine cross-session. Un thread non è un database globale in cui il tuo agent ricorda i fatti attraverso task diversi per sempre. È la working memory a breve termine strettamente legata a una singola sequenza in corso. Abiliti questa memory fornendo un checkpointer quando compili il tuo graph. Un checkpointer è un oggetto che gestisce il salvataggio e il caricamento dello state del graph su uno storage backend. Una volta che il tuo graph è compilato con un checkpointer, attivi la persistence passando un oggetto di configurazione che contiene un thread ID ogni volta che fai l'invoke del graph. Questo ID è la chiave univoca che il checkpointer usa per tracciare la history di quella specifica run. Quando fai girare il graph con quel thread ID, il checkpointer salva automaticamente lo state. Ma non salva in continuazione. Salva a specifici boundary chiamati super-step. Un super-step è un ciclo di esecuzione distinto all'interno del graph. Se il tuo graph esegue il node A seguito dal node B, questi sono due super-step. Se il tuo graph si ramifica ed esegue i node C e D allo stesso tempo, quell'esecuzione parallela viene raggruppata in un unico super-step. Il checkpointer non interrompe un node mentre sta lavorando. Aspetta il boundary del super-step. Una volta che tutti i node programmati per quel super-step finiscono l'esecuzione e restituiscono i loro update, LangGraph crea un checkpoint. Questo checkpoint contiene uno state snapshot, che cattura esattamente l'aspetto delle variabili di state del graph in quel preciso momento. Vediamo come si comporta nella pratica. Supponi di avere un agent che analizza un enorme dataset. Il graph ha quattro step. Lo step uno fa il fetch dei dati. Lo step due li pulisce. Lo step tre fa girare un'analisi pesante. Lo step quattro formatta il summary. Fai partire la run, passando una configurazione con thread ID uno due tre. Il graph completa con successo gli step di fetch, clean e analisi. Alla fine dello step tre, il checkpointer salva uno state snapshot. Poi, prima che lo step quattro possa finire, il tuo server crasha. Dato che hai usato un checkpointer e un thread ID, lo state è al sicuro. Quando il tuo server si riavvia, fai semplicemente di nuovo l'invoke del graph, passando esattamente lo stesso thread ID uno due tre. Il checkpointer cerca l'ultimo checkpoint. Trova lo state snapshot salvato subito dopo lo step di analisi. Il graph carica quello state e riprende l'esecuzione immediatamente allo step quattro. I node di fetch, clean e analisi vengono completamente saltati perché i loro output sono già salvati in modo sicuro nel checkpoint del thread. Ecco il punto chiave. Compilando il tuo graph con un checkpointer e legando la tua esecuzione a un thread ID, trasformi delle fragili operazioni in-memory in workflow durevoli che sopravvivono automaticamente alle interruzioni. Questo è tutto per questo episodio. Grazie per aver ascoltato, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 10 di 18. Il tuo workflow elabora un pagamento, raggiunge un rate limit nello step successivo e va in crash. Quando il sistema si riprende e fa ripartire il workflow, al tuo cliente viene addebitato il pagamento una seconda volta. Il tuo codice non è cambiato, ma l'assunto che hai fatto su come il grafo riparte era sbagliato. Il fix richiede di capire la durable execution e l'idempotenza. Molti sviluppatori danno per scontato che quando un processo long-running va in pausa o fallisce, riprenderlo significhi ripartire dall'esatta riga di codice Python in cui si era fermato. Si aspettano che il runtime ricordi magicamente le variabili locali a metà funzione. Ma non è quello che succede. LangGraph non congela l'interprete Python in quello stato. Lo stato viene salvato solo ai confini tra i nodi. La durable execution in LangGraph significa che il sistema tiene traccia dei tuoi progressi persistendo lo stato del grafo dopo che un nodo ha finito il suo lavoro e fa return. Se un nodo fallisce a metà della sua logica, il sistema non ha alcuna traccia del suo progresso parziale. L'ultimo stato valido noto è quello passato al nodo quando è partito. Quando fai restart o retry del grafo, l'esecuzione riprende rieseguendo l'intero nodo fallito dalla sua primissima riga. Pensa allo scenario del pagamento. Supponi di scrivere un singolo nodo che esegue due azioni. Per prima cosa, chiama un'API esterna per addebitare una carta di credito. Secondo, aggiorna un database remoto per registrare la transazione. L'addebito sulla carta di credito va a buon fine, ma la connessione al database va in timeout, causando il crash del nodo. Lo stato del grafo non avanza. Quando il workflow riprende, passa di nuovo il vecchio stato a quello stesso nodo. Il nodo ricomincia da capo. Chiama l'API esterna e addebita la carta di credito una seconda volta. Ecco il punto chiave. Dato che i nodi possono ripartire dall'inizio, qualsiasi side effect all'interno di un nodo deve essere idempotente. L'idempotenza è una proprietà per cui eseguire un'operazione più volte produce esattamente lo stesso risultato di eseguirla una volta sola. Se il tuo nodo interagisce con il mondo esterno, devi scrivere il codice dando per scontato che girerà più volte per lo stesso step. Come garantisci questa sicurezza? Hai due approcci pratici. Il primo è sfruttare le idempotency keys con i tuoi servizi esterni. Quando chiami l'API di pagamento, passi un identificatore univoco derivato dallo stato corrente del grafo. Se il nodo crasha e viene rieseguito, invia lo stesso identificatore univoco. Il servizio esterno riconosce la richiesta duplicata e restituisce una risposta di successo senza spostare di nuovo denaro. Il secondo approccio è il design strutturale del grafo. Se una specifica operazione non è nativamente idempotente, non raggrupparla con altri step che potrebbero fallire. Metti l'operazione pericolosa all'interno di un suo nodo dedicato. Fai in modo che sia l'unica cosa che fa quel nodo. Se metti l'addebito del pagamento nel nodo A e l'aggiornamento del database nel nodo B, un timeout del database fa crashare solo il nodo B. Il grafo riprende dal nodo B. L'addebito del pagamento nel nodo A è completamente al sicuro, perché il nodo A ha finito, e il grafo ha salvato il suo stato prima di andare avanti. Sei tu a controllare dove il sistema salva i suoi progressi in base a come disegni i confini dei tuoi nodi. Non mettere mai un'azione irreversibile e non idempotente nello stesso nodo di qualcosa che potrebbe fallire in modo casuale. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 11 di 18. A volte un'IA non dovrebbe avere l'ultima parola. Magari vuoi bloccare un agent a metà di un ragionamento, chiedere l'approvazione a un umano e iniettare la sua risposta direttamente nella logica in esecuzione. Questo è esattamente ciò che fanno gli interrupt Human-in-the-Loop. Quando hai bisogno che un umano prenda una decisione all'interno di un workflow di LangGraph, usi una funzione specifica chiamata interrupt. È fondamentale capire cosa fa effettivamente sotto il cofano. Potresti confonderla con un normale input prompt di Python. Ma non lo è. Un input prompt standard blocca un thread attivo, occupando la memoria di sistema in attesa che l'utente prema un tasto. In LangGraph, chiamare interrupt si comporta in modo molto diverso. Serializza completamente lo state del graph, lo salva nel tuo database checkpointer e sospende del tutto l'esecuzione. Il graph si addormenta. Può aspettare all'infinito una risposta senza consumare risorse di calcolo attive. Il flow avviene in due fasi distinte: pausa e ripresa. Per prima cosa, diamo un'occhiata alla pausa. All'interno di uno dei tuoi node, il tuo agent arriva a un punto in cui ha bisogno di un'autorizzazione umana. In quell'esatta riga di codice, chiami la funzione interrupt. Passi un payload a questa funzione, che di solito è un oggetto JSON contenente il context di cui l'umano ha bisogno. Immagina un agent che gestisce l'assistenza clienti automatizzata. Decide di preparare un rimborso di cinquecento dollari. Prima di processare il pagamento, il node dell'agent chiama interrupt. Passa un payload specificando che l'azione proposta è un rimborso e l'importo è cinquecento. Nel momento in cui chiami quella funzione, il graph si ferma. Il runtime di LangGraph intercetta questo evento e fa risalire il payload JSON fino alla tua applicazione client. Il processo del graph si spegne, lasciando il payload in attesa della revisione umana su una web UI. Ora passiamo alla seconda fase: risvegliare il graph. Un processo esterno, come il tuo server backend che riceve una chiamata API dalla web UI, si occupa di far ripartire il graph. Il manager umano clicca approva sulla sua dashboard. Il tuo backend prende quell'approvazione e fa ripartire il graph usando un'istruzione speciale chiamata Command. Quando invii questo Command, includi un argomento resume che contiene la risposta dell'umano. Nel nostro scenario, questa risposta è un semplice valore boolean true. Ecco il punto chiave. Quando il graph si sveglia, non riesegue il node in pausa dall'inizio. Riprende l'esecuzione nell'esatta riga di codice in cui si era fermato. La funzione interrupt che originariamente aveva messo in pausa il graph finisce di essere eseguita, e restituisce qualsiasi valore tu abbia inviato tramite il comando resume. La risposta boolean dell'umano viene iniettata direttamente nella variabile in attesa del risultato di interrupt. L'agent quindi legge quel valore true, passa il suo conditional check e finalizza il rimborso di cinquecento dollari. Questa architettura crea un confine netto. La logica del graph non deve gestire webhook, email o user interface. Chiama semplicemente una funzione che lancia un payload dall'altra parte e aspetta un return value. Il sistema esterno gestisce tutta l'interazione con l'utente e inserisce semplicemente la risposta di nuovo dentro. Iniettando la risposta umana direttamente nel return della funzione, eviti di inquinare lo state principale del tuo graph con dati di interazione temporanei. La potenza della funzione interrupt sta nel trattare il feedback umano non come una complessa deviazione architetturale, ma come una normale function call che può mettere in pausa l'universo in tutta sicurezza finché non riceve una risposta. Questo è tutto per questo episodio. Grazie per aver ascoltato, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 12 di 18. Il tuo agent va fuori controllo, compiendo un'azione indesiderata o generando una risposta pessima. Normalmente, devi riavviare l'intero processo da zero e sperare che si comporti meglio la seconda volta. E se potessi letteralmente riavvolgere l'esecuzione fino al momento esatto prima dell'errore, modificare manualmente lo state e farla proseguire lungo una timeline alternativa? È proprio di questo che parleremo oggi con Debugging del passato: viaggi nel tempo e forking. Per manipolare il passato, devi prima visualizzarlo. Lo fai usando un metodo chiamato get state history, passando come parametro il tuo thread ID. Questo metodo restituisce un iteratore contenente ogni state attraversato dal graph durante l'esecuzione di quel thread. Ognuno di questi state storici possiede un identificatore univoco chiamato checkpoint ID. Puoi pensare a questo ID come alle tue coordinate temporali esatte. Se vuoi semplicemente fare il replay del graph da un punto specifico, recuperi il checkpoint ID di destinazione da quella history. A questo punto, chiami il metodo invoke del graph, passando un oggetto di configurazione che include sia il thread ID che quel checkpoint ID specifico. Il graph riprende immediatamente l'esecuzione da quello state preciso. Non riesegue nessuno dei node precedenti, risparmiando compute e tempo. Il replay è utile, ma la vera potenza risiede nel modificare il passato per fare il fork dell'esecuzione. Consideriamo uno scenario pratico. Supponiamo che al tuo agent sia stato assegnato il compito di scrivere una barzelletta e che ne abbia generata una pessima su un cane. Controlli la state history e trovi il checkpoint ID per lo state immediatamente precedente alla fase di generazione. Invece di fare semplicemente il replay da quel punto, usi il metodo update state. Fornisci il thread ID, il checkpoint ID storico specifico e i nuovi valori dello state che vuoi iniettare. In questo caso, aggiorni manualmente la variabile topic, cambiandola da cane a polli. Ecco il punto chiave. Gli sviluppatori spesso pensano che l'aggiornamento di uno state passato faccia il rollback dell'esecuzione, sovrascrivendo o eliminando la history originale. Non è così. LangGraph opera con un'architettura append-only. Quando chiami update state su un checkpoint storico, il sistema crea in modo sicuro un nuovo checkpoint che fa il branch da quello precedente. La tua timeline originale, completa della pessima battuta sul cane, rimane perfettamente intatta e accessibile. Non hai cancellato il passato; hai fatto il fork di una nuova realtà. Una volta applicato l'aggiornamento, il graph si trova in un checkpoint appena creato con lo state modificato. Per proseguire lungo questa nuova timeline, ti basta fare di nuovo l'invoke del graph con il thread ID, omettendo qualsiasi checkpoint ID specifico. Il graph passa di default allo state più recente su questo nuovo branch appena creato tramite fork e riprende l'esecuzione. Il tuo agent legge lo state aggiornato e genera invece una battuta sui polli. Se trovi utili queste analisi tecniche e vuoi supportare il podcast, puoi cercare DevStoriesEU su Patreon. Il time travel trasforma il debugging: invece di indovinare cosa è andato storto, puoi manipolare con precisione la history del graph per esplorare risultati alternativi, senza perdere una singola traccia della run originale. Questo è tutto per questo episodio. Grazie per l'ascolto e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 13 di 18. Costruisci un agent e un utente gli dice che vuole sempre il suo codice in Python 3.11. Il giorno dopo, inizia una nuova conversazione e l'agent se ne dimentica completamente, restituendo invece Python 3.9. La memoria del thread è isolata a una singola conversazione. Quando il tuo agent ha bisogno di conservare informazioni tra sessioni completamente separate, ti serve la Long-Term Memory usando gli Store Across Threads. Un errore comune è cercare di risolvere questo problema infilando informazioni a lungo termine nello state del checkpointer. I checkpointer sono memoria a breve termine. Sono rigorosamente snapshot dello state per singolo thread, progettati per mettere in pausa, riprendere o fare il replay di una singola conversazione. Se un utente esprime una preferenza nel thread A, il thread B non ha assolutamente modo di vederla. Per condividere la conoscenza tra più thread, LangGraph fornisce l'interfaccia Store. Uno Store è un livello di memoria key-value che si trova al di fuori dei singoli state dei thread. Lo configuri passando un oggetto store, come un PostgresStore, come argomento quando compili il tuo graph. Una volta compilato, quello store viene collegato all'ambiente di esecuzione del graph. All'interno del tuo graph, i node accedono a questo livello di memoria tramite l'oggetto Runtime. Quando definisci un node, puoi accedere al runtime context, che espone lo store. Ti basta accedere a runtime punto store per interagire con la tua memoria a lungo termine. Ecco il punto chiave. I dati in uno store sono organizzati usando i namespace. Un namespace è una lista gerarchica di string che partiziona i tuoi dati, un po' come il percorso di una cartella sul tuo computer. Per un'applicazione multi-tenant, potresti definire un namespace che inizia con la string users, seguita da uno specifico user ID, e che termina con preferences. Pensa a quello scenario dell'assistente di programmazione. Un utente inizia una sessione lunedì. Durante la chat, dice di preferire Python 3.11 e la dark mode. Un node nel tuo graph riconosce questa preferenza come permanente. Chiama il metodo put su runtime punto store. Passa il namespace per quello specifico utente, una key univoca per l'elemento, e un dictionary che contiene le preferenze. I dati ora sono salvati fuori dal thread. Venerdì, lo stesso utente apre la tua applicazione e inizia un thread completamente nuovo. Lo state del checkpointer per questo nuovo thread è completamente vuoto. Tuttavia, il tuo graph include un node di setup che viene eseguito per primo. Questo node chiama il metodo search su runtime punto store, fornendo il prefisso del namespace dell'utente. Lo store restituisce le preferenze salvate. Il node poi inserisce quelle preferenze nello state del thread corrente. Da quel momento in poi, l'agent sa di dover usare Python 3.11 e la dark mode per questa nuova conversazione. L'interfaccia dello store fornisce tre operazioni principali. Usi put per salvare o sovrascrivere un elemento. Usi get per recuperare un singolo elemento quando conosci il suo namespace e la sua key esatti. Usi search per recuperare più elementi che condividono un prefisso del namespace. Il search è particolarmente utile quando hai salvato diversi frammenti di memoria distinti per un utente nel tempo e hai bisogno di portarli tutti nel context corrente. Separando lo state a breve termine da uno store cross-thread, disaccoppi il ciclo di vita della conoscenza del tuo agent dal ciclo di vita di una singola conversazione. Grazie per aver ascoltato, alla prossima.

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 14 di 18. Gli utenti odiano dover fissare un loading spinner statico per trenta secondi mentre il tuo sistema lavora in background. Vuoi mostrare loro il processo decisionale del sistema in tempo reale, ma catturare quei segnali interni spesso richiede la configurazione di custom callback complesse. L'esecuzione in streaming e il formato v2 risolvono questo problema unificando ogni evento interno in un unico stream prevedibile. Innanzitutto, chiariamo un malinteso comune. Gli ingegneri spesso confondono lo streaming dei token del language model con lo streaming dello state dell'applicazione. Si tratta di livelli di informazione completamente diversi. Uno stream di token è semplicemente testo che appare parola per parola. Uno stream di state tiene traccia dell'avanzamento generale del tuo workflow, passando da un task all'altro. LangGraph gestisce entrambi simultaneamente. Accedi a questo comportamento richiedendo uno stream e passando l'argomento version v2 al tuo metodo di esecuzione. Questo standardizza l'output. Invece di gestire tipi di dati misti, ogni singolo evento che esce dal tuo graph diventa un dictionary unificato contenente esattamente tre campi: type, ns e data. Il campo type definisce la categoria dell'evento. Il campo ns sta per namespace e indica il percorso esatto nella gerarchia del tuo graph da cui ha avuto origine l'evento. Questo diventa fondamentale quando hai subgraph annidati e hai bisogno di sapere esattamente quale sub-component ha generato l'evento. Infine, il campo data contiene il payload effettivo. Controlli esattamente cosa viene inserito in questo stream selezionando una o più stream mode. La mode values ti invia lo state completo e aggiornato del graph ogni volta che un node completa il suo lavoro. Questo si rivela utile se la tua applicazione richiede un quadro completo in ogni step. La mode updates è molto più leggera. Trasmette in streaming solo i dati specifici restituiti da un node, rappresentando solo il delta o la modifica apportata allo state complessivo. La mode messages opera a un livello più granulare, trasmettendo in streaming i singoli chunk di un messaggio di chat generato man mano che vengono prodotti da un language model sottostante. Immagina un'interfaccia frontend. Vuoi un indicatore di status luminoso che evidenzi quale step è attualmente attivo, ad esempio il fetching del context, poi la valutazione dei documenti e infine il drafting, visualizzando contemporaneamente il testo del draft token per token. Per costruire questo, avvii l'esecuzione del tuo graph con le stream mode impostate sia su updates che su messages, assicurandoti di passare il flag di versione v2. Il tuo frontend inizia a ricevere uno stream continuo e unificato di questi dictionary. Quando arriva un dictionary con il type impostato su updates, leggi il campo namespace. Questo ti indica esattamente quale node ha appena terminato il suo lavoro. Usi quel segnale per spostare il tuo indicatore di status luminoso allo step successivo sulla user interface. Pochi millisecondi dopo, lo stream fornisce un nuovo dictionary con il type impostato su messages. Estrai il token di testo grezzo dal campo data e lo aggiungi direttamente al paragrafo che il tuo utente sta leggendo. Sia le modifiche di state di alto livello che la generazione di testo di basso livello arrivano attraverso la stessa identica pipe. Ecco il punto chiave. Forzando token, modifiche di state e progresso del node in un'unica struttura a dictionary a tre campi, il formato v2 elimina completamente la necessità di scrivere logiche di handling separate o complesse callback asincrone per diversi tipi di eventi in tempo reale. Questo è tutto per oggi. Alla prossima!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 15 di 18. Quando il tuo AI agent diventa complesso, spesso ti ritrovi con un mega-graph enorme e illeggibile, dove una singola modifica manda tutto in tilt. Non devi per forza sviluppare in questo modo: puoi creare dei mini-graph specializzati e collegarli tra loro. Stiamo parlando di Composing Complexity: Subgraphs. I subgraph ti permettono di riutilizzare la logica e distribuire lo sviluppo su team diversi. Invece di inserire ogni singolo step della tua applicazione in un unico file, crei dei graph più piccoli e autonomi. Una volta compilato, un graph si comporta esattamente come una normale callable function. Questo significa che puoi prendere un intero graph compilato e inserirlo direttamente in un altro graph come un singolo node. Pensa a un sistema di routing master per un assistente enterprise. Il graph principale gestisce l'input dell'utente, controlla la sicurezza e decide cosa fare dopo. Quando un utente fa una domanda tecnica approfondita, il router deve eseguire una raccolta dati complessa. Invece di scrivere questa logica direttamente nel router, la deleghi a un subgraph di Research dedicato. Un team di engineering completamente diverso può sviluppare, testare e perfezionare questo graph di Research in totale isolamento. Al parent graph non interessa come viene fatta la ricerca. Si limita a chiamare il node. C'è la tendenza comune a complicare troppo il modo in cui i dati passano tra questi graph. Se il parent graph e il subgraph usano esattamente lo stesso state schema, ovvero condividono le stesse state keys, non ti servono adattatori speciali. Ti basta passare il subgraph di Research compilato direttamente nella funzione add node del tuo master graph. L'engine passa automaticamente il parent state al subgraph, esegue la logica e, quando finisce, fa il merge dei risultati di nuovo nel parent state. Ora, cosa succede quando i team non si coordinano perfettamente? Supponiamo che il tuo parent router usi una state key chiamata user query, ma che l'altro team di engineering abbia sviluppato il subgraph di Research per aspettarsi una key chiamata search term. Non puoi inserire direttamente il subgraph compilato nel parent graph. Le key non corrisponderanno e l'esecuzione fallirà. Ecco il punto chiave. Puoi risolvere questa discrepanza usando una semplice wrapper function. Nel tuo parent graph, definisci una node function standard che accetta il parent state. All'interno di questa funzione, estrai il valore della user query. A quel punto chiami manualmente il subgraph di Research compilato, passandogli un payload in cui mappi quella user query alla key search term. Il subgraph esegue la sua logica interna e restituisce il suo final state. La tua wrapper function prende quell'output, traduce i risultati di nuovo nelle key specifiche che il parent si aspetta, e li restituisce. Per il parent graph, questo wrapper sembra un normalissimo node. Non ha idea che un subgraph enorme e complesso sia appena stato eseguito al suo interno. Ha semplicemente passato dei dati in ingresso e ha ricevuto in cambio uno state aggiornato. Questo pattern ti offre una modularità rigorosa senza sacrificare il controllo. Trattare un graph compilato come un'altra callable function dietro un semplice wrapper è il modo in assoluto più potente per scalare un'architettura AI senza collassare sotto il tuo stesso codice. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 16 di 18. Se un subagent esperto viene chiamato due volte in una singola conversazione, dovrebbe ricordare la prima chiamata o ricominciare da capo con un'amnesia totale? Questa scelta cambia tutto nel modo in cui si comportano i sistemi multi-agent, ed è controllata interamente dalla Subgraph Persistence e dai Multi-Agent Patterns. Quando costruisci un parent graph che fa il routing dei task verso i subgraph, lo state management diventa complicato. Pensa a un primary bot di customer service che gestisce la chat generale. Quando un utente fa una domanda complessa su una fattura, il primary bot fa il routing della richiesta a un subgraph dedicato, il Billing Expert. Di default, i subgraph sono completamente stateless. Quando compili quel Billing Expert senza specificare un checkpointer, opera strettamente per singola invocazione. Il primary bot gli passa gli input necessari, l'esperto esegue i suoi step interni, restituisce un risultato e scarta immediatamente il suo state interno. Se l'utente fa una domanda di follow-up sulla fatturazione cinque minuti dopo, il primary bot chiama di nuovo l'esperto. L'esperto non ha memoria dello scambio precedente. Ricomincia completamente da zero. Per un semplice subgraph di data-extraction, questa amnesia va benissimo. Per un agent interattivo e specializzato, è incredibilmente frustrante per l'utente. Per risolvere la cosa, l'esperto ha bisogno della sua memoria tra un turno e l'altro. Un errore molto comune qui è passare una nuova istanza del checkpointer, come un oggetto memory saver, direttamente nel metodo compile del subgraph. Non farlo, a meno che tu non voglia che il subgraph condivida l'esatto stesso state tra utenti e sessioni completamente diversi. Se l'utente A e l'utente B parlano entrambi con il sistema allo stesso tempo, passare un'istanza esplicita del checkpointer al subgraph significa che i loro dati vengono mescolati insieme in un unico state globale. Questo crea un enorme cross-talk tra parent thread isolati. Invece, ti basta passare il valore booleano True all'argomento checkpointer quando compili il subgraph. Ed è qui che diventa interessante. Impostarlo su True dice al subgraph di affidarsi al meccanismo di checkpointer del parent graph, ma di mantenere una history multi-turn completamente isolata e specifica per se stesso. Dietro le quinte, il framework gestisce il namespacing. Crea automaticamente un thread ID univoco per il subgraph che è permanentemente legato al thread ID del parent. Ora guarda di nuovo lo scenario del Billing Expert con questa configurazione. L'utente fa una domanda su una fattura. Il primary bot ne fa il routing verso l'esperto. L'esperto risponde e diventa dormiente. Più tardi, nella stessa conversazione, l'utente fa un follow-up. Il primary bot fa di nuovo il routing verso l'esperto. Poiché è stato compilato con il checkpointer impostato su True, l'esperto si sveglia, controlla il suo sub-thread dedicato e carica il context della fattura dal turno precedente. Si comporta come un partecipante persistente nella conversazione. E poiché quel sub-thread ha uno scope strettamente legato al thread del parent, un utente diverso che parla con il sistema ottiene la sua istanza completamente pulita del Billing Expert. Il modo in cui configuri il checkpointer di un subgraph detta la sua intera identità nel tuo sistema: lasciarlo vuoto crea una utility function usa e getta e stateless, mentre impostarlo su True crea un collaboratore continuo e context-aware. Grazie per aver passato qualche minuto con me. Alla prossima, stammi bene.

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 17 di 18. Uno script Python che gira correttamente sul tuo laptop non è un'applicazione in produzione. Se provi a eseguire agent stateful tramite script standalone, inevitabilmente andrai a sbattere contro un muro quando sarà il momento di scalare. Per risolvere questo problema, analizzeremo la struttura dell'applicazione e la preparazione per il deploy. Quando crei per la prima volta un agent LangGraph, probabilmente ne fai un prototipo in un notebook Jupyter o in un singolo file Python. Definisci i node, colleghi gli edge, compili il graph e chiami il metodo invoke direttamente nello stesso file per vedere se funziona. Questo va benissimo per i test. Tuttavia, un server in produzione non può leggerti nel pensiero. Ha bisogno di un modo standardizzato per servire quel graph come API, installare i pacchetti richiesti e iniettare le variabili d'ambiente. Per rendere il tuo prototipo pronto per il deploy, devi organizzare il codice in una directory structure pulita. Diciamo che crei una nuova cartella chiamata my-app. Sposti il tuo codice Python fuori dal notebook e in un file pulito all'interno di questa cartella. Poi, aggiungi un file per le dependency, in genere requirements punto txt. Infine, crei un file di configurazione chiamato langgraph punto json nella root della cartella my-app. Il file langgraph punto json è il blueprint principale della tua applicazione. Quando usi la Command Line Interface di LangGraph, o fai il deploy in un ambiente di produzione, questo file di configurazione dice al sistema sottostante esattamente come fare la build e far girare il tuo progetto. Richiede tre informazioni principali che riguardano le dependency, le variabili d'ambiente e gli entry point del graph. Per prima cosa, dichiari le tue dependency. Questa è solo una string nel file JSON con il path che punta al tuo file requirements. Assicura che il server di deploy installi esattamente i pacchetti Python da cui dipende il tuo agent, prevenendo errori di module mancanti in produzione. Poi, definisci la string dell'environment. Questa punta al tuo file punto env. Gli agent stateful hanno sempre bisogno di secret, come le credenziali del database o le API key del modello. Puntare al file di environment assicura che il runtime carichi in modo sicuro queste key prima di tentare di avviare il graph. Questa è la parte che conta. Il terzo requisito nel file di configurazione è il mapping dei graph. Questo dice al server esattamente dove si trova il tuo graph compilato nel codice sorgente. Funziona come un dictionary. Assegni al tuo graph un ID, che diventa il suo nome ufficiale nell'API generata. Poi, mappi quell'ID a uno specifico module Python e a un nome di variabile. Per esempio, potresti mappare l'ID customer-support-agent alla string agent punto py due punti compiled-graph. Il server guarda il file agent punto py, trova la variabile chiamata compiled-graph e la carica in memoria. Questa struttura richiede un cambiamento intenzionale nel modo in cui scrivi il tuo codice. I principianti spesso fanno girare i graph tramite script Python standalone che eseguono azioni non appena vengono lanciati. Ma il runtime di LangGraph si affida a langgraph punto json per esporre il graph dinamicamente come web service. Non fa girare il tuo script dall'alto verso il basso. Importa solo l'oggetto graph compilato che hai specificato nel file di configurazione. Per questo motivo, il tuo file Python dovrebbe solo definire i node, collegarli e assegnare il graph compilato a una variabile. Devi rimuovere qualsiasi codice di test rimasto in fondo che invoca manualmente il graph. Se lasci del codice di test nel file, questo verrà eseguito durante la fase di import sul server, causando fallimenti nel deploy o chiamate API indesiderate solo per aver avviato il servizio. Dichiarando esplicitamente le tue dependency, l'environment e i path del graph in un unico file JSON centrale, separi la definizione del tuo agent dalla sua esecuzione, trasformando uno script locale in un servizio robusto e pronto per il deploy. Questo è tutto per questo episodio. Grazie per l'ascolto, e continua a sviluppare!

Ciao, sono Alex di DEV STORIES DOT EU. LangGraph, episodio 18 di 18. Hai un workflow multi-agente complesso e devi testare uno specifico edge case di routing nello step quattro. Non dovresti dover eseguire l'intero sistema dall'inizio alla fine solo per raggiungere quella condizione. Testare l'esecuzione del graph end-to-end è il modo per puntare esattamente alla logica che ti serve. Quando gli sviluppatori cercano di isolare parti di un graph per i test, spesso ricorrono a mock complessi. Cercano di mockare la struttura del graph circostante o di fare lo stub di tutti i nodi precedenti. In LangGraph, non hai bisogno di farlo. L'architettura ruota interamente attorno allo state. Dato che i nodi sono semplicemente funzioni che leggono e scrivono lo state, puoi iniettare manualmente uno specifico payload di state e testare frammenti di nodo isolati in modo nativo. È qui che la state injection e i breakpoint diventano incredibilmente utili nella tua test suite. Ti servono solo due strumenti per saltare direttamente nel mezzo di un graph. Il primo è il metodo update state. Il secondo è un parametro di configurazione chiamato interrupt after. Usando questi strumenti all'interno di un testing framework standard come pytest, puoi simulare condizioni esatte senza eseguire l'intera applicazione. Applichiamo questo concetto a uno scenario concreto. Supponi di avere un graph in cui il nodo tre fa una chiamata API esterna e il nodo quattro ne verifica il risultato. Vuoi verificare che, se il payload dell'API contiene uno specifico codice di errore, il nodo quattro instradi correttamente l'execution flow al tuo error handler node. Invece di eseguire i nodi uno e due per triggerare questa condizione, isoli il problema. Inizializzi il graph con un thread identifier. Poi, usi update state per inserire un payload API simulato e fallito direttamente nel thread state. Agisci come se il nodo tre stesse per essere eseguito con quei dati specifici. Successivamente, invochi il graph, ma passi un dizionario di configurazione che imposta interrupt after sul nodo quattro. Quando avvii il graph, l'esecuzione inizia immediatamente dal nodo tre usando il tuo failure state iniettato. Il nodo tre elabora il payload errato e passa lo state risultante al nodo quattro. Il nodo quattro valuta la logica e decide di fare routing verso l'error handler. Dato che hai impostato un breakpoint, il graph mette in pausa l'esecuzione nel momento in cui il nodo quattro finisce. Ora il tuo test può valutare il risultato. Estrai lo state corrente dal graph. Puoi scrivere le tue asserzioni per assicurarti che il nodo quattro abbia aggiornato correttamente le variabili di state. Cosa ancora più importante, puoi ispezionare l'execution plan del graph. Guardando il prossimo nodo in sospeso nei metadati dello state, puoi confermare che la logica di routing ha funzionato perfettamente e che l'error handler è in coda. Ecco il punto chiave. Manipolando direttamente lo state, trasformi una chain di agenti altamente interconnessa e imprevedibile in un test case deterministico e step-by-step. Verifichi esattamente come il graph transita da un nodo al successivo senza aspettare i language model o le chiamate di rete negli step precedenti. Dato che questo è l'ultimo episodio della serie, ti incoraggio a esplorare la documentazione ufficiale e a provare a costruire questi workflow hands-on. Se hai idee su cosa dovremmo trattare in futuro, visita devstories dot eu e suggerisci un argomento. Questo è tutto per questo episodio. Grazie per l'ascolto e continua a sviluppare!