LangGraph

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 1 de 18. A maioria dos scripts de large language models funciona perfeitamente para um prompt rápido e uma resposta rápida. Mas quando uma tarefa demora vinte minutos a correr e a rede vai abaixo a meio, tudo desmorona. Perdes o teu progresso, o teu contexto e o teu orçamento de API. O Problema da Orquestração: Porquê o LangGraph? é exatamente sobre resolver esta fragilidade. O LangGraph é uma framework de orquestração criada para lidar com aplicações stateful e multi-actor. Podes ouvir o nome e assumir que é apenas uma feature do LangChain. Não é. O LangGraph é um motor de orquestração de mais baixo nível, e não precisas de usar o LangChain de todo para o utilizares. Existe especificamente para modelar workflows de agentes como grafos stateful, em vez de simples scripts lineares. Os scripts standard executam em memória. Se um script parar inesperadamente, todos esses dados de runtime desaparecem. Considera um cenário em que tens um agente em background a pesquisar um documento de cem páginas. O agente esteve a ler, a extrair factos e a cruzar informação durante vinte minutos ininterruptos. Se ocorrer um timeout do servidor no minuto dezanove, um script standard perde todo esse state. Tens de recomeçar o trabalho todo do zero. O LangGraph resolve este problema de orquestração através de durable execution. Ao modelar o teu workflow como um grafo, cada passo distinto torna-se um node, e as ligações lógicas entre eles são edges. À medida que a aplicação avança de um node para o seguinte, o LangGraph guarda automaticamente o seu progresso. Trata os processos long-running como uma série de checkpoints seguros. Se o sistema crashar, o LangGraph retoma a execução exatamente onde parou. Este mecanismo de checkpointing depende de uma memória abrangente. A memória no LangGraph não é apenas uma lista contínua de mensagens de chat. É o state completo do grafo. Quando um node termina o seu processamento, atualiza um objeto de state partilhado. O node seguinte na sequência lê o seu input diretamente desse state. Isto significa que a memória persiste ao longo de todo o lifecycle da aplicação. O agente em background a pesquisar o teu documento não se esquece dos dados críticos que encontrou na página cinco quando finalmente chega à página noventa, porque o graph state guarda-os com segurança. Aqui está o insight principal. Como o graph state é pausado e guardado de forma limpa entre passos, ganhas a capacidade de colocar um human in the loop. Às vezes, um agente autónomo chega a um ponto de decisão onde precisa de permissão antes de prosseguir, como enviar um email final ou executar uma transação financeira. Num script standard, pausar para que um utilizador clique num botão leva frequentemente a timeouts de ligação. No LangGraph, simplesmente configuras um node específico para interromper a execução. O sistema entra em sleep e preserva o state atual perfeitamente. Um operador humano pode então rever os dados recolhidos, aprovar a ação seguinte, ou até mesmo modificar manualmente o agent state antes de clicar em continue. Uma vez aprovado, o grafo acorda e retoma o seu trabalho com o contexto atualizado. A conclusão principal é que construir agentes complexos depende muito da gestão de state e de falhas. O LangGraph afasta a tua arquitetura de scripts frágeis e limitados pela memória, em direção a grafos resilientes que sobrevivem a interrupções, lembram-se do seu passado e aguardam pacientemente por orientação humana. Se quiseres ajudar a apoiar o programa, podes procurar por DevStoriesEU no Patreon. É tudo por este episódio. Obrigado por ouvires, e continua a desenvolver!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 2 de 18. Crias um agente de inteligência artificial enfiando instruções, edge cases e exemplos num único prompt gigantesco, depois envias tudo e esperas que ele faça a coisa certa. Funciona até falhar, e fazer o debug do resultado é frustrante. Para resolver isto, tens de parar de escrever prompts monolíticos e começar a desenhar sistemas. Isto leva-nos a Pensar em LangGraph: O Modelo Mental. O LangGraph obriga-te a afastares-te dos scripts lineares. Pede-te que penses na tua aplicação como uma máquina de estados. O primeiro passo é simplesmente definires o teu processo. Antes de escreveres qualquer código, olha para o que queres que o sistema alcance e divide isso em ações discretas. Imagina um sistema de triagem de suporte ao cliente. Um utilizador envia uma mensagem. Um operador humano leria o email, decidiria se é um problema de faturação ou de suporte técnico, e depois faria o draft de uma resposta apropriada. Essa sequência é o teu processo. Mapeias esse processo para um workflow de LangGraph usando três componentes principais que são State, Nodes e Edges. Vamos começar pelo State. O State é a memória partilhada do teu graph. É uma estrutura que guarda o contexto de toda a tua operação num dado momento. Cada passo no teu workflow vai ler deste State e escrever atualizações de volta nele. Os ouvintes costumam cometer um erro específico aqui. Tentam guardar strings de prompt totalmente formatadas dentro do State. Não faças isto. O State deve guardar apenas dados brutos. Guarda o texto original do email do cliente, a categoria extraída ou uma lista bruta de mensagens anteriores. A formatação acontece on-demand, mais tarde, exatamente dentro do passo onde é necessária. A seguir, temos os Nodes. Se o State é a memória, os Nodes são os workers que executam as tarefas reais. Um node é apenas uma função Python que executa um único passo lógico do teu processo. Recebe o State atual, executa uma ação e devolve uma atualização. No nosso exemplo de triagem, criarias três nodes separados. O primeiro é um node de Read. Pega no email recebido e guarda o texto bruto no State. O segundo é um node de Classify. Olha para o texto bruto no State, pede a um language model para o categorizar como faturação ou técnico, e guarda a categoria resultante de volta no State. O terceiro é um node de Draft. Lê tanto o email como a categoria do State, formata-os num prompt localmente e gera uma resposta. Cada node faz exatamente um trabalho. Finalmente, precisas de uma forma de ligar estes workers. Esse é o papel das Edges. As Edges representam a lógica de routing. Ditam o que acontece depois de um node terminar o seu trabalho. Uma edge standard diz simplesmente: assim que o node de Read terminar, vai sempre para o node de Classify. Mas o LangGraph também usa conditional edges. É aqui que a coisa fica interessante. Depois do node de Classify, podes usar uma conditional edge para inspecionar o State. Se a categoria for faturação, a edge encaminha o fluxo para um node de Draft específico de faturação. Se for técnica, encaminha para um node de Draft técnico. As Edges tomam decisões de tráfego com base nos dados que os teus nodes acabaram de produzir. Começas com o processo, isolas os dados num State partilhado, defines os workers como Nodes, e ditas o fluxo com Edges. Ao decompor o problema, isolas as falhas. Trata a tua aplicação não como um simples gerador de texto, mas como uma linha de montagem coordenada, onde os dados brutos se movem sistematicamente de um worker especializado para o próximo. Obrigado por ouvirem, happy coding a todos!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 3 de 18. Duas funções paralelas terminam de executar exatamente no mesmo milissegundo e tentam escrever na mesma lista partilhada. Normalmente, uma faz overwrite à outra, e os dados desaparecem. Para evitar isto, o LangGraph apoia-se na Graph API: State e Reducers. A base de qualquer workflow do LangGraph é o seu state. Tu defines isto usando um TypedDict standard de Python. Este dictionary define as keys exatas que o teu graph vai fazer track e os data types para cada key. Pensa nisto como o schema que é passado de node para node à medida que o teu graph corre. Aqui está a ideia principal. Os nodes no LangGraph não mutam o state diretamente. Um node recebe uma cópia do state atual, faz o seu trabalho e retorna um dictionary de updates. Um erro comum é assumir que retornar um dictionary substitui todo o object do state. Não substitui. Se o teu state tiver cinco keys e o teu node retornar um dictionary com apenas uma key, o LangGraph deixa as outras quatro intactas e aplica apenas o teu update específico. Como é que o LangGraph aplica esse update? É aí que entram os reducers. Um reducer é simplesmente uma função que dita como um valor retornado faz merge com o valor existente para uma key específica. Por default, o LangGraph usa um overwrite reducer. Se o teu node retornar uma nova string para uma status key, a string antiga desaparece, substituída completamente pela nova. Às vezes, fazer overwrite é exatamente o que tu queres evitar. Considera um workflow paralelo de data fetching. Tu tens uma state key partilhada chamada results, que é uma lista. Tu inicias dois nodes a correr ao mesmo tempo para fazer fetch de diferentes batches de dados. Se ambos os nodes retornarem um dictionary a atualizar a key results, o comportamento default de overwrite faz com que o node que terminar em último apague o trabalho do outro. Para corrigir isto, tu anotas a key results no teu TypedDict com um reducer específico, como o built-in operator dot add do Python. Agora, quando os dois nodes retornam as suas listas, o reducer age como um polícia sinaleiro. Ele pega na lista existente e faz append em segurança dos outputs de ambos os nodes. Nada se perde. Há um edge case. E se tu tiveres um reducer do tipo append na tua results key, mas chegares a um ponto no teu graph onde precisas genuinamente de limpar a lista e começar do zero? Se o teu node retornar uma lista vazia, o reducer simplesmente adiciona uma lista vazia à existente, deixando os dados antigos intactos. Para este cenário, o LangGraph fornece um type especial Overwrite. Quando o teu node encapsula o seu update num object Overwrite, o LangGraph deteta-o e faz bypass ao reducer por completo. Ele deita fora a lista antiga e força um hard reset. O state num graph complexo não é uma variável global frágil a ser constantemente mutada, mas sim um append-only log de updates controlados, governado por regras claras de redução. É tudo por este episódio. Obrigado por ouvires, e continua a desenvolver!

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 4 de 18. Às vezes, só queres escrever um script Python standard com if-statements e for-loops normais, mas continuas a precisar de state persistence de nível empresarial. Não queres construir uma state machine explícita só para correr algumas chamadas de modelos de linguagem em sequência. É aqui que a Functional API, especificamente os decorators entrypoint e task, resolve o problema. Construir aplicações com nodes e edges explícitos exige que definas manualmente como os dados são encaminhados de um passo para o seguinte. Essa estrutura oferece um controlo imenso, mas pode parecer pesada quando a tua lógica é altamente sequencial ou depende de loops de programação standard. A Functional API permite-te escrever código Python normal, de cima para baixo, mantendo as funcionalidades integradas de streaming e recuperação. Em vez de instanciares um objeto graph, aplicas decorators às tuas funções Python existentes. Começas com o decorator task. Aplicas isto às unidades de trabalho individuais na tua aplicação. Pensa numa task como um passo discreto que faz algo específico, como consultar uma base de dados, calcular uma métrica ou fazer um prompt a um modelo. Quando uma função tem o decorator task, a framework envolve-a numa tracking layer para monitorizar a sua execução. A seguir, usas o decorator entrypoint. Colocas isto na função principal de orquestração que direciona o flow geral. Dentro desta função entrypoint, chamas as tuas tasks decoradas usando o control flow standard do Python. Atribuis o output de uma task a uma variável e, a seguir, passas essa variável para a task seguinte. Podes usar blocos try-except, list comprehensions ou while-loops. A lógica de orquestração comporta-se exatamente como o Python nativo. Como o código parece totalmente standard, podes presumir que lhe falta a memória de uma estrutura formal de state. Isso é um equívoco comum. A Functional API continua a fazer o checkpointing automático do teu progresso nos bastidores. Sempre que uma task é concluída, o LangGraph interceta o valor de retorno e guarda-o numa persistent store. A framework regista com segurança os inputs e outputs de cada função decorada à medida que acontecem. Imagina um script automatizado de escrita de ensaios. Defines três tasks decoradas: uma função para gerar um esboço, uma função para escrever um parágrafo e uma função para rever o rascunho. Dentro da tua função entrypoint principal, chamas primeiro o gerador de esboços. A seguir, escreves um for-loop standard que itera sobre as secções desse esboço, chamando a task de escrever parágrafo para cada uma delas. Fazes append dos resultados a uma lista local. Finalmente, corres a task de revisão. Usas um simples if-statement para verificar a pontuação resultante. Se a pontuação for fraca, o teu código simplesmente aciona um while-loop para reescrever parágrafos específicos até que a pontuação melhore. Aqui está o ponto-chave. Devido ao checkpointing oculto, se o teu script encontrar um network timeout enquanto escreve o terceiro parágrafo, não perdes o teu trabalho. Quando reinicias o processo com o mesmo identificador de thread, o LangGraph sabe que o esboço e os dois primeiros parágrafos já estão completos. Ele ignora completamente a execução dessas tasks, recupera os seus cached outputs da state store, e retoma a execução precisamente no terceiro parágrafo. A Functional API transfere a carga cognitiva da visualização de topologias de routing abstratas de volta para a leitura de código de cima para baixo, dando-te a resiliência de uma state machine com a simplicidade de um script normal. É tudo por este episódio. Obrigado por ouvires, e continua a desenvolver!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 5 de 18. Crias uma app de chat, e um utilizador deteta uma gralha no seu prompt. Clica em editar, corrige e envia. Mas em vez de substituir a message antiga, o teu backend simplesmente acrescenta a versão corrigida ao fim do histórico, deixando a gralha original exatamente onde estava, como um fantasma duplicado. Gerir o histórico da conversa com o MessagesState é como evitas isto. Quando os developers constroem o seu primeiro graph, normalmente definem um dictionary de state personalizado para guardar o histórico de chat. Um erro comum é usar o append normal de listas para gerir este histórico. Eles associam a função standard operator dot add à sua lista de messages. Isto diz ao graph para simplesmente pegar em quaisquer novas messages e colá-las ao fim do array existente. Esta abordagem append-only funciona bem para um bot simples de ping-pong, onde o utilizador fala, a IA responde, e o histórico cresce sequencialmente. Mas quebra completamente quando o state precisa de ser mutável. Se um humano editar um prompt passado, ou um agent decidir regenerar a sua última resposta, a adição standard não consegue lidar com isso. Acabas com duplicados. O LangGraph fornece uma estrutura de state built-in para resolver isto, chamada MessagesState. Contém uma única key chamada messages. Aqui está o ponto chave. O poder do MessagesState não está na key em si, mas na função reducer específica que lhe está associada, chamada add messages. O reducer add messages não faz apenas um append cego de dados. Ele rastreia os IDs das messages. Sempre que uma nova message entra no state, o reducer verifica o seu ID único. Se esse ID já existir em algum lugar no histórico da conversa, o reducer faz overwrite da message antiga com a nova. Se o ID for novo, ou se a message ainda não tiver um ID, o reducer faz append da mesma ao fim da lista. Lembra-te do nosso cenário da gralha. O utilizador humano envia um prompt. O sistema atribui a essa message o ID 123. O utilizador apercebe-se do erro, edita o texto e submete a correção. O frontend envia o novo texto, marcando-o explicitamente com o ID 123. Quando esses dados chegam ao graph, o reducer add messages examina o histórico, encontra a message original no ID 123, e troca o texto in place. O fantasma duplicado desaparece. A conversa flui exatamente como pretendido. Para além de gerir IDs, o reducer add messages também lida com a deserialization de dados. Numa aplicação em produção, as tuas messages chegam frequentemente em formatos diferentes. O teu frontend pode enviar dictionaries JSON raw contendo strings de role e content. Os teus nodes internos do graph podem gerar objetos de message nativos do LangChain. O reducer atua como um tradutor universal para estes inputs. Se passares uma lista de dictionaries Python plain para o state, a função add messages converte-os automaticamente nas classes de message corretas do LangChain. Não precisas de escrever código boilerplate para fazer parse de um dictionary num HumanMessage ou AIMessage antes de atualizar o state. Ele normaliza os dados por ti. Ao construir agents de chat, o histórico de state não é um log append-only, é um documento vivo, e ligar as tuas atualizações a IDs de message únicos é o que mantém esse documento preciso. Obrigado por ouvires. Até à próxima!

Olá, daqui fala o Alex do DEV STORIES DOT EU. LangGraph, episódio 6 de 18. Escolhe o paradigma de design errado logo de início, e vais acabar a escrever centenas de linhas de boilerplate para um script simples, ou a criar um autêntico pesadelo de esparguete com funções básicas de Python. Hoje, vamos focar-nos em Escolher a Tua Abstração: Graph versus Functional. É fácil assumir que uma destas APIs é inerentemente mais poderosa ou mais production-ready do que a outra. Isso é falso. Nos bastidores, tanto a Graph API como a Functional API compilam para o exato mesmo runtime engine. Ambas suportam persistence, streaming e controlo de execução exatamente da mesma maneira. A escolha entre elas resume-se puramente ao teu modelo mental e à forma como queres expressar a tua lógica. Vamos olhar primeiro para a Functional API. Esta baseia-se no control flow imperativo standard do Python. Escreves funções Python normais, marca-las com um decorator, e fazes o routing da tua execução usando if-statements e loops standard. A gestão de state aqui é inteiramente function-scoped. Os dados fluem estritamente do return value de uma função para os arguments da próxima. Não há nenhum objeto de memória global partilhada a flutuar em background. Se o teu workflow for linear, ou se tiver uma lógica previsível e tightly scoped, a Functional API mantém o teu código lean e familiar. Evitas completamente o overhead de definir estruturas de graph. A Graph API exige uma mentalidade diferente. Em vez de chamares funções diretamente, defines um schema de state global partilhado. A seguir, escreves nodes, que são pequenas funções cujo único trabalho é ler e mutar esse state partilhado. Finalmente, ligas explicitamente esses nodes uns aos outros usando edges. O routing não é tratado por um conditional statement escondido nas profundezas do body de uma função. Em vez disso, a lógica que dita para onde a aplicação vai a seguir é extraída para conditional edges explícitas, declaradas no top level do graph. Aqui está o insight principal. Escolhes entre elas com base na forma como o teu sistema lida com o state e o routing ao longo do tempo. Imagina um developer a construir uma ferramenta básica de extração de dados. Ela corre um único language model, faz o parse do output, e guarda-o. A Functional API é perfeita para isto. É rápida de escrever e fácil de ler. Mas avança três meses no tempo. Esse script simples está a ser refactored para um sistema multi-agent complexo. Agora tens um agent researcher a passar dados para um agent writer, um agent critic a devolver correções, e uma pausa na execução à espera que um manager humano aprove o draft final. Se tentares construir esse workflow multi-agent assíncrono com a Functional API, a abordagem imperativa colapsa. Acabas a passar payloads de dados massivos para cima e para baixo em call stacks de funções profundas. A tua lógica de routing fica enterrada dentro de conditionals profundamente nested. Este é o momento exato em que migras para a Graph API. A abstração de Graph brilha aqui porque faz o decouple do state da execução. Como o state é global e partilhado, os teus nodes de agents individuais não precisam de passar estruturas de dados pesadas uns aos outros. Um node simplesmente lê o state partilhado, atualiza a key específica pela qual é responsável, e termina. As edges explícitas assumem o controlo, tornando o routing altamente visível. Podes olhar para a definição do graph e mapear imediatamente todo o workflow sem leres uma única linha de business logic. Usas a Functional API quando o control flow é simples o suficiente para ser lido de cima para baixo, mas mudas para a Graph API quando o routing se torna complexo o suficiente para precisares de o desenhar num quadro branco. Obrigado por ouvirem. Fiquem bem, pessoal.

Olá, daqui é o Alex do DEV STORIES DOT EU. LangGraph, episódio 7 de 18. Condicionais hardcoded têm os seus limites quando estás a construir um agent. Se um utilizador fizer uma pergunta complexa, uma simples pesquisa por keywords não consegue decidir de forma fiável o que a tua aplicação deve fazer a seguir. E se a própria IA pudesse ditar o caminho do teu workflow? É aqui que entram o routing dinâmico e as conditional edges. Numa configuração standard de um graph, tu ligas o node A ao node B. É um caminho estático e garantido. Mas quando estás a construir um mecanismo de routing inteligente, o caminho precisa de mudar com base nos dados recebidos. Podes assumir que as edges no LangGraph só aceitam ligações de strings hardcoded. Mas não é o caso. Uma edge pode ser uma função Python que lê o state atual do teu graph e calcula dinamicamente o nome do próximo node. Tu adicionas esta lógica ao teu graph usando o método add conditional edges. Este método requer três componentes. Primeiro, o node inicial. Segundo, uma função de routing. Terceiro, um dictionary que mapeia as possíveis strings de output da tua função de routing para os nodes de destino reais no teu graph. Aqui está a ideia principal. A forma mais fiável de conduzir uma conditional edge é combiná-la com um large language model a gerar dados estruturados. Tu não queres que a própria função de routing faça avaliações complexas ou natural language processing. Em vez disso, tens um node upstream onde o modelo é forçado a devolver uma estrutura rígida, como um modelo Pydantic. Considera um router de apoio ao cliente. Um utilizador envia uma mensagem. O primeiro node no teu graph é um classificador de intent. Dentro deste node, tu passas a mensagem do utilizador para um language model e exiges que ele devolva um output estruturado com um único campo chamado intent. O modelo avalia o texto e preenche esse campo com um valor específico, como billing, tech support ou sales. Esta resposta estruturada é depois guardada no state do graph. Agora a conditional edge entra em ação. A edge está ligada ao node classificador. Quando o node classificador termina, a conditional edge aciona uma pequena função em Python. Esta função recebe o state do graph como input, olha para dentro do state, e extrai o valor de intent que o modelo acabou de gerar. Se o intent for billing, a função devolve a string billing. A conditional edge olha para o seu dictionary de mapeamento, vê que a string billing corresponde ao teu node de billing, e passa a execução para esse node específico. Se o intent for tech support, devolve uma string diferente, fazendo o routing do flow para o node de tech support. Estás a usar o language model pelas suas capacidades de raciocínio para categorizar o input, mas estás a manter a lógica de routing determinística. A função Python na conditional edge está apenas a ler uma variável e a devolver uma string. É altamente previsível e fácil de testar. A principal lição aqui é que deves sempre desacoplar a decisão da direção. Deixa o language model decidir o intent e escrevê-lo no state, e depois usa uma conditional edge em Python puro para ler esse state e guiar o graph. Antes de terminarmos, se achas estes episódios úteis e queres apoiar o programa, podes procurar por DevStoriesEU no Patreon — ajuda-nos imenso. É tudo por este episódio. Obrigado por ouvires, e continua a construir!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 8 de 18. Não podes fazer hardcode dos teus execution paths quando não fazes ideia de quantas sub-tasks o teu agent vai decidir criar até ele correr de facto. Se o teu sistema decidir on the fly que precisa de processar três itens, ou trinta itens, o routing estático standard vai falhar. Para resolver isto, precisas de Map-Reduce Workflows com a Send API. Um erro muito comum ao desenvolver em LangGraph é tentar usar conditional edges standard para fan-out dinâmico. As conditional edges são perfeitas quando queres escolher entre caminhos conhecidos e predeterminados com base numa verificação lógica. No entanto, elas falham quando precisas de fazer spawn de um número desconhecido de tasks paralelas idênticas em runtime. A paralelização standard permite-te fazer routing para múltiplos nodes fixos. Tu dás o nome aos nodes, e o graph faz trigger deles. Mas o que acontece quando precisas de correr exatamente o mesmo node várias vezes em simultâneo, cada uma com dados diferentes? Não consegues fazer isto com routing básico. Isto leva-nos ao pattern Orchestrator-worker. Nesta arquitetura, um node central analisa os dados recebidos, calcula quantas tasks separadas são necessárias e despacha workers dinâmicos para as processar em simultâneo. O LangGraph viabiliza este pattern especificamente através da Send API. Imagina um agent encarregue de escrever um research report abrangente. O primeiro node atua como orchestrator. Ele lê o prompt do utilizador e gera um outline. Dependendo da complexidade do tópico, este outline pode conter três secções, ou pode conter doze. Tu queres que um worker node separado faça o draft de cada secção exatamente ao mesmo tempo. Para conseguir isto, defines uma função de conditional edge imediatamente a seguir ao teu orchestrator node. Em vez de devolver uma simples string que aponta para o próximo node estático no graph, esta função de edge devolve uma lista de objetos Send. Aqui está o ponto chave. Um objeto Send empacota um destino e os seus dados em conjunto. Ele recebe dois argumentos. O primeiro argumento é o nome do worker node que queres fazer trigger. O segundo argumento é o payload específico para esse worker isolado. No nosso cenário do report, a função do orchestrator itera pelo outline gerado. Por cada tópico de secção que encontra, cria um novo objeto Send a apontar para um único node chamado draft_section, passando a string individual do tópico como payload. Quando o LangGraph avalia esta função de edge, recebe a lista de objetos Send. A seguir, faz spin up dinamicamente de uma instância paralela do node draft_section para cada item dessa lista. Se o orchestrator gerou um outline com sete secções, o LangGraph lança sete drafting nodes paralelos. Cada node corre código idêntico, mas opera no seu próprio payload único. A geração destes workers dinâmicos é a fase de map. A recolha dos seus outputs independentes é a fase de reduce. Como estes worker nodes correm em simultâneo, não podem fazer overwrite com segurança de uma única string no teu graph state. O teu state geral tem de ser configurado para recolher múltiplos updates recebidos em simultâneo. Tu tratas disto anexando uma função reducer ao state field específico que vai guardar as tuas secções em draft, instruindo-a a fazer append de novos itens a uma lista em vez de fazer overwrite do valor anterior. À medida que cada draft worker paralelo termina de escrever, devolve o seu bloco de texto. O LangGraph apanha estas respostas e usa o teu reducer para empilhar com segurança cada bloco de texto no array partilhado. Assim que cada worker dinâmico conclui a sua execução, todo o step paralelo é resolvido. O workflow avança então, carregando uma lista completa e preenchida de todas as secções em draft. A Send API tira a execução paralela da tua definição de graph estático e coloca-a diretamente nas mãos dos teus dados em runtime. Obrigado por estares aí. Espero que tenhas aprendido algo novo.

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 9 de 18. O teu servidor crasha a meio de um raciocínio enquanto um agent está a processar um dataset massivo. Não devia ter de recomeçar do zero. Devia continuar exatamente de onde parou. Essa resiliência é o que vamos abordar hoje com Persistência, especificamente Threads e Checkpoints. Para adicionar persistência a uma aplicação LangGraph, tens de perceber o conceito de thread. Uma thread representa uma única sequência de execução isolada ou uma conversa específica do utilizador. Ela guarda o state de trabalho do graph à medida que passa de node em node. Deixa-me esclarecer uma coisa desde já. As pessoas costumam confundir a memória da thread com a memória de longo prazo cross-session. Uma thread não é uma base de dados global onde o teu agent se lembra de factos de diferentes tarefas para sempre. É a memória de trabalho de curto prazo estritamente ligada a uma sequência em curso. Ativas esta memória fornecendo um checkpointer quando compilas o teu graph. Um checkpointer é um objeto que trata de guardar e carregar o state do graph para um storage backend. Assim que o teu graph estiver compilado com um checkpointer, acionas a persistência passando um objeto de configuração que contém um thread ID sempre que invocas o graph. Este ID é a chave única que o checkpointer usa para rastrear o histórico dessa run específica. Quando corres o graph com esse thread ID, o checkpointer guarda o state automaticamente. Mas não guarda continuamente. Guarda em limites específicos chamados super-steps. Um super-step é um ciclo de execução distinto no graph. Se o teu graph corre o node A seguido do node B, isso são dois super-steps. Se o teu graph ramifica e corre o node C e o node D ao mesmo tempo, essa execução paralela é agrupada num único super-step. O checkpointer não interrompe um node enquanto este está a trabalhar. Ele espera pelo limite do super-step. Assim que todos os nodes agendados para esse super-step terminam a execução e devolvem os seus updates, o LangGraph cria um checkpoint. Este checkpoint contém um state snapshot, capturando exatamente como as variáveis de state do graph estão naquele exato momento. Vamos ver como isto se comporta na prática. Supõe que tens um agent a analisar um dataset massivo. O graph tem quatro steps. O step um faz o fetch dos dados. O step dois limpa-os. O step três corre uma análise pesada. O step quatro formata o resumo. Inicias a run, passando uma configuração com o thread ID um dois três. O graph conclui com sucesso os steps de fetch, limpeza e análise. No final do step três, o checkpointer guarda um state snapshot. Depois, antes que o step quatro consiga terminar, o teu servidor crasha. Como usaste um checkpointer e um thread ID, o state está seguro. Quando o teu servidor reinicia, simplesmente invocas o graph novamente, passando exatamente o mesmo thread ID um dois três. O checkpointer procura o último checkpoint. Encontra o state snapshot guardado logo após o step de análise. O graph carrega esse state e retoma a execução imediatamente no step quatro. Os nodes de fetch, limpeza e análise são completamente ignorados porque os seus outputs já estão guardados com segurança no checkpoint da thread. Aqui está o ponto-chave. Ao compilar o teu graph com um checkpointer e ligar a tua execução a um thread ID, transformas operações frágeis em memória em workflows duradouros que sobrevivem a interrupções automaticamente. É tudo por este episódio. Obrigado por ouvires, e continua a desenvolver!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 10 de 18. O teu workflow processa um pagamento, atinge um rate limit no passo seguinte e crasha. Quando o sistema recupera e retoma o workflow, o teu cliente é cobrado uma segunda vez. O teu código não mudou, mas a suposição que fizeste sobre como o graph retoma estava errada. A solução requer compreender a durable execution e a idempotência. Muitos developers assumem que, quando um long-running process pausa ou falha, ao ser retomado, continua exatamente da linha de código Python onde parou. Eles esperam que o runtime se lembre magicamente das variáveis locais a meio da função. Não é isso que acontece. O LangGraph não congela o interpretador de Python. O state é guardado apenas nas fronteiras entre os nodes. Durable execution no LangGraph significa que o sistema regista o teu progresso persistindo o graph state depois de um node terminar o seu trabalho e fazer return. Se um node falhar a meio da sua lógica, o sistema não tem registo do seu progresso parcial. O último state válido conhecido é aquele que foi passado ao node quando este começou. Quando reinicias ou fazes retry do graph, a execução retoma e corre novamente todo esse node que falhou, desde a sua primeira linha. Pensa no cenário do pagamento. Supõe que escreves um único node que executa duas ações. Primeiro, chama uma API externa para cobrar um cartão de crédito. Segundo, atualiza uma base de dados remota para registar a transação. A cobrança do cartão de crédito é bem-sucedida, mas a ligação à base de dados dá timeout, fazendo com que o node crashe. O graph state não avança. Quando o workflow é retomado, ele passa o state antigo de volta para esse mesmo node. O node recomeça. Chama a API externa e cobra o cartão de crédito uma segunda vez. Aqui está a ideia principal. Como os nodes podem reiniciar do início, qualquer side effect dentro de um node tem de ser idempotente. A idempotência é uma propriedade em que executar uma operação várias vezes produz exatamente o mesmo resultado que executá-la apenas uma vez. Se o teu node interage com o mundo exterior, tens de escrever o código assumindo que ele será executado várias vezes para o mesmo step. Como garantes esta segurança? Tens duas abordagens práticas. A primeira é utilizar idempotency keys com os teus serviços externos. Quando chamas a API de pagamento, passas um identificador único derivado do graph state atual. Se o node crashar e correr novamente, envia o mesmo identificador único. O serviço externo reconhece a request duplicada e devolve uma response de sucesso sem realmente movimentar dinheiro novamente. A segunda abordagem é o design estrutural do graph. Se uma operação específica não for nativamente idempotente, não a agrupes com outros steps que possam falhar. Coloca a operação perigosa num node dedicado. Faz com que seja a única coisa que esse node executa. Se colocares a cobrança do pagamento no node A e a atualização da base de dados no node B, um timeout da base de dados faz crashar apenas o node B. O graph retoma no node B. A cobrança do pagamento no node A está completamente segura, porque o node A terminou e o graph guardou o seu state antes de avançar. Tu controlas onde o sistema guarda o seu progresso pela forma como defines os limites dos teus nodes. Nunca coloques uma ação irreversível e não idempotente no mesmo node que algo que possa falhar aleatoriamente. É tudo por este episódio. Obrigado por ouvires, e continua a construir!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 11 de 18. Às vezes, uma IA não deve ter a palavra final. Podes querer congelar um agent a meio de um raciocínio, pedir a aprovação de um humano e injetar a sua resposta diretamente na lógica em execução. É exatamente isso que os interrupts de Human-in-the-Loop fazem. Quando precisas que um humano tome uma decisão dentro de um workflow do LangGraph, usas uma função específica chamada interrupt. É vital perceber o que isto realmente faz under the hood. Os ouvintes podem confundir isto com um input prompt standard de Python. Não é. Um input prompt standard bloqueia uma thread ativa, prendendo a memória do sistema enquanto espera que o utilizador prima uma tecla. No LangGraph, chamar o interrupt comporta-se de maneira muito diferente. Ele serializa completamente o state do graph, guarda-o na tua base de dados de checkpointer, e suspende a execução por completo. O graph adormece. Pode esperar indefinidamente por uma resposta sem consumir quaisquer recursos computacionais ativos. O flow acontece em duas fases distintas: pausar e retomar. Primeiro, vamos olhar para a pausa. Dentro de um dos teus nodes, o teu agent chega a um ponto em que precisa de autorização humana. Nessa exata linha de código, chamas a função interrupt. Passas um payload para esta função, que geralmente é um objeto JSON contendo o contexto de que o humano precisa. Considera um agent a lidar com suporte automatizado ao cliente. Ele decide preparar um reembolso de quinhentos dólares. Antes de processar o pagamento, o node do agent chama o interrupt. Ele entrega um payload especificando que a ação proposta é um reembolso e o valor é quinhentos. No momento em que essa função é chamada, o graph para. O runtime do LangGraph apanha este evento e faz subir o payload JSON até à tua aplicação client. O processo do graph encerra, deixando o payload à espera de revisão humana numa web UI. Agora para a segunda fase: acordar o graph novamente. Um processo externo, como o teu servidor backend a receber uma API call da web UI, é responsável por reiniciar o graph. O gestor humano clica em aprovar no seu dashboard. O teu backend pega nessa aprovação e inicia o graph novamente usando uma instrução especial chamada Command. Ao enviar este Command, incluis um argumento resume contendo a resposta do humano. No nosso cenário, esta resposta é um simples valor boolean de true. Aqui está a ideia principal. Quando o graph acorda, não volta a correr o node pausado desde o início. Retoma a execução na exata linha de código onde parou. A função interrupt que originalmente pausou o graph termina a execução, e devolve o valor que enviaste através do comando resume. A resposta boolean do humano é injetada diretamente na variável à espera do resultado do interrupt. O agent depois lê esse valor true, passa a sua verificação condicional, e finaliza o reembolso de quinhentos dólares. Esta arquitetura cria uma fronteira limpa. A lógica do graph não precisa de lidar com webhooks, emails ou user interfaces. Apenas chama uma função que atira um payload por cima do muro e espera por um return value. O sistema externo lida com toda a user interaction e simplesmente envia a resposta de volta. Ao injetar a resposta do humano diretamente no function return, evitas poluir o state principal do teu graph com dados de interação temporários. O poder da função interrupt reside em tratar o feedback humano não como um desvio arquitetónico complexo, mas como uma function call standard que pode pausar o universo em segurança até obter uma resposta. É tudo por este episódio. Obrigado por ouvires, e continua a construir!

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 12 de 18. O teu agent descarrila, executando uma ação que não querias ou gerando uma resposta terrível. Normalmente, tens de reiniciar todo o processo do zero e esperar que se comporte melhor à segunda vez. E se pudesses literalmente rebobinar a execução até ao momento exato antes do erro, alterar manualmente o state, e deixá-lo correr numa linha do tempo alternativa? É exatamente isso que vamos abordar hoje com Debugging do Passado: Time Travel e Forking. Para manipular o passado, primeiro tens de o ver. Fazes isso usando um método chamado get state history, passando o identificador da tua thread. Este método devolve um iterador que contém todos os states pelos quais o graph passou durante a execução dessa thread. Cada um destes states históricos possui um identificador único chamado checkpoint ID. Podes pensar neste ID como as tuas coordenadas exatas no tempo. Se simplesmente quiseres fazer replay do graph a partir de um ponto específico, pegas no checkpoint ID de destino desse histórico. A seguir, chamas o método invoke do teu graph, passando um objeto de configuração que inclui tanto a thread ID como esse checkpoint ID específico. O graph retoma imediatamente a execução a partir desse state exato. Não volta a correr nenhum dos nodes anteriores, poupando compute e tempo. Fazer replay é útil, mas o verdadeiro poder está em alterar o passado para fazer fork da execução. Vejamos um cenário prático. Imagina que o teu agent tinha a tarefa de escrever uma piada, e gerou uma piada terrível sobre um cão. Verificas o histórico de states e encontras o checkpoint ID do state imediatamente antes da etapa de geração ter ocorrido. Em vez de simplesmente fazeres replay a partir desse ponto, usas o método update state. Forneces a thread ID, o checkpoint ID histórico específico, e os novos valores de state que queres injetar. Neste caso, atualizas manualmente a variável topic, mudando-a de cão para galinhas. Aqui está o ponto chave. Os developers costumam pensar que atualizar um state passado faz rollback da execução, substituindo ou apagando o histórico original. Mas não faz. O LangGraph opera numa arquitetura append-only. Quando chamas o update state num checkpoint histórico, o sistema cria com segurança um checkpoint totalmente novo, fazendo branch a partir desse antigo. A tua linha do tempo original, completa com a piada terrível sobre o cão, permanece totalmente intacta e acessível. Não apagaste o passado; fizeste fork para uma nova realidade. Assim que aplicas essa atualização, o graph fica num checkpoint recém-criado com o state alterado. Para continuares por esta nova linha do tempo, simplesmente invocas o graph novamente com a thread ID, omitindo qualquer checkpoint ID específico. O graph assume por defeito o state mais recente neste branch recém-criado e retoma a execução. O teu agent lê o state atualizado e, em vez disso, gera uma piada sobre galinhas. Se estás a achar estas explicações técnicas úteis e queres apoiar o programa, podes procurar por DevStoriesEU no Patreon. O time travel transforma o debugging, de adivinhar o que correu mal, numa manipulação precisa do histórico do graph para explorar resultados alternativos, sem perder um único rasto da run original. É tudo por este episódio. Obrigado por ouvires, e continua a construir!

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 13 de 18. Crias um agent e um utilizador diz-lhe que quer sempre o seu código em Python 3.11. No dia seguinte, ele começa uma nova conversa e o agent esquece-se completamente, fazendo output de Python 3.9 em vez disso. A memória da thread é isolada a uma única conversa. Quando o teu agent precisa de reter factos entre sessões completamente separadas, precisas de Long-Term Memory usando Stores entre threads. Um erro comum é tentar resolver isto metendo factos de longo prazo no state do checkpointer. Os checkpointers são memória de curto prazo. São estritamente snapshots de state por thread, desenhados para fazer pause, resume ou replay de uma única conversa. Se um utilizador declara uma preferência na thread A, a thread B não tem absolutamente nenhuma forma de a ver. Para partilhar conhecimento entre várias threads, o LangGraph fornece a interface Store. Uma Store é uma camada de memória key-value que fica fora dos states individuais das threads. Configuras isto passando um objeto store, como uma PostgresStore, como argumento quando compilas o teu graph. Uma vez compilado, essa store fica anexada ao ambiente de execução do graph. Dentro do teu graph, os nodes acedem a esta camada de memória através do objeto Runtime. Quando defines um node, podes aceder ao contexto de runtime, que expõe a store. Basta acederes a runtime dot store para interagires com a tua memória de longo prazo. Aqui está a ideia principal. Os dados numa store são organizados usando namespaces. Um namespace é uma lista hierárquica de strings que particiona os teus dados, muito parecido com o caminho de uma pasta no teu computador. Para uma aplicação multi-tenant, podes definir um namespace que começa com a string users, seguida por um ID de utilizador específico, e que termina com preferences. Pensa naquele cenário do assistente de código. Um utilizador começa uma sessão na segunda-feira. Durante o chat, menciona que prefere Python 3.11 e dark mode. Um node no teu graph reconhece isto como uma preferência permanente. Ele chama o método put em runtime dot store. Passa o namespace para esse utilizador específico, uma key única para o item, e um dictionary que contém as preferências. Os dados ficam agora guardados fora da thread. Na sexta-feira, o mesmo utilizador abre a tua aplicação e começa uma thread completamente nova. O state do checkpointer para esta nova thread está completamente vazio. No entanto, o teu graph inclui um setup node que corre primeiro. Este node chama o método search em runtime dot store, fornecendo o prefixo do namespace do utilizador. A store devolve as preferências guardadas. O node depois coloca essas preferências no state da thread atual. A partir desse ponto, o agent sabe que deve usar Python 3.11 e dark mode para esta nova conversa. A interface da store fornece três operações principais. Usas o put para guardar ou fazer overwrite de um item. Usas o get para ir buscar um único item quando sabes o seu namespace e key exatos. Usas o search para ir buscar múltiplos itens que partilham um prefixo de namespace. O search é particularmente útil quando guardaste vários fragmentos de memória distintos para um utilizador ao longo do tempo e precisas de os puxar a todos para o contexto atual. Ao separar o state de curto prazo de uma store cross-thread, desacoplas o tempo de vida do conhecimento do teu agent do tempo de vida de uma única conversa. Obrigado por ouvires — até à próxima.

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 14 de 18. Os utilizadores detestam ficar a olhar para um loading spinner estático durante trinta segundos enquanto o teu sistema trabalha em background. Tu queres mostrar-lhes o processo de pensamento do sistema em tempo real, mas capturar esses sinais internos muitas vezes exige a criação de custom callbacks complexos. A execução em streaming e o formato v2 resolvem isto ao unificar cada evento interno num único fluxo previsível. Primeiro, vamos esclarecer um mal-entendido comum. Os engenheiros costumam confundir o streaming de tokens de um language model com o streaming do state da aplicação. São camadas de informação completamente diferentes. Uma stream de tokens é apenas texto a aparecer palavra a palavra. Uma stream de state monitoriza o progresso geral do teu workflow, a passar de uma task para a seguinte. O LangGraph lida com ambos em simultâneo. Tu acedes a este comportamento ao pedir uma stream e ao passar o argumento version v2 para o teu método de execução. Isto padroniza o output. Em vez de lidares com tipos de dados mistos, cada evento que sai do teu graph torna-se num dicionário unificado que contém exatamente três campos: type, ns e data. O campo type define a categoria do evento. O campo ns significa namespace, a indicar o caminho exato na hierarquia do teu graph onde o evento teve origem. Isto torna-se crítico quando tens subgraphs aninhados e precisas de saber exatamente qual subcomponente disparou o evento. Por fim, o campo data contém o payload propriamente dito. Tu controlas exatamente o que é inserido nesta stream ao selecionar um ou mais stream modes. O modo values envia-te o state completo e atualizado do graph sempre que qualquer node conclui o seu trabalho. Isto revela-se útil se a tua aplicação precisar da visão completa a cada passo. O modo updates é muito mais leve. Ele faz o streaming apenas dos dados específicos devolvidos por um node, a representar apenas o delta ou a alteração feita ao state geral. O modo messages opera a um nível mais granular, a fazer o streaming dos chunks individuais de uma mensagem de chat gerada, à medida que são produzidos por um language model subjacente. Imagina uma interface de frontend. Tu queres um indicador de status brilhante que destaque qual é o passo que está ativo no momento — talvez a fazer o fetch de contexto, depois a avaliar documentos, e depois a fazer o draft — enquanto exibe em simultâneo o texto do draft, token a token. Para construíres isto, tu inicias a execução do teu graph com os stream modes definidos tanto para updates como para messages, a garantir que passas a flag da versão v2. O teu frontend começa a receber uma stream contínua e unificada destes dicionários. Quando um dicionário chega com o type definido como updates, tu lês o campo namespace. Isto diz-te exatamente qual node acabou de concluir o seu trabalho. Tu usas esse sinal para mover o teu indicador de status brilhante para o passo seguinte na user interface. Milissegundos depois, a stream entrega um novo dicionário com o type definido como messages. Tu extrais o token de texto raw do campo data e fazes o append diretamente ao parágrafo que o teu utilizador está a ler. Tanto as mudanças de state de alto nível como a geração de texto de baixo nível chegam exatamente pelo mesmo canal. Aqui está a ideia principal. Ao forçar tokens, mudanças de state e o progresso dos nodes para uma única estrutura de dicionário de três campos, o formato v2 elimina completamente a necessidade de escrever lógica de handling separada ou callbacks assíncronos complexos para diferentes tipos de eventos em tempo real. É tudo por agora. Vemo-nos na próxima!

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 15 de 18. Quando o teu agente de IA se torna complexo, muitas vezes acabas com um mega-graph enorme e ilegível, onde uma única alteração estraga tudo. Não tens de construir as coisas dessa forma — em vez disso, podes construir mini-graphs especializados e encaixá-los uns nos outros. Estamos a falar sobre Compor Complexidade: Subgraphs. Os subgraphs permitem-te reutilizar lógica e distribuir o desenvolvimento por diferentes equipas. Em vez de colocares cada passo da tua aplicação num único ficheiro, crias graphs mais pequenos e independentes. Assim que um graph é compilado, comporta-se exatamente como uma callable function normal. Isto significa que podes pegar num graph compilado inteiro e inseri-lo diretamente noutro graph como um único node. Pensa num sistema de master routing para um assistente corporativo. O graph principal lida com o input do utilizador, verifica a segurança e decide o que fazer a seguir. Quando um utilizador faz uma pergunta técnica complexa, o router precisa de fazer uma recolha de dados complexa. Em vez de programares essa lógica diretamente no router, delegas isso num subgraph de Research dedicado. Uma equipa de engenharia completamente diferente pode construir, testar e refinar este graph de Research de forma isolada. O parent graph não quer saber como a pesquisa é feita. Apenas chama o node. Há uma tendência comum para complicar demasiado a forma como os dados passam entre estes graphs. Se o parent graph e o subgraph usarem exatamente o mesmo state schema — ou seja, se partilharem exatamente as mesmas state keys — não precisas de nenhum adaptador especial. Basta passares o subgraph de Research compilado diretamente para a função add node do teu master graph. O engine alimenta automaticamente o parent state para o subgraph, corre a lógica e faz o merge dos resultados de volta no parent state quando termina. Agora, o que acontece quando as equipas não se coordenam na perfeição? Imagina que o teu parent router usa uma state key chamada user query, mas a outra equipa de engenharia construiu o subgraph de Research à espera de uma key chamada search term. Não podes simplesmente inserir o subgraph compilado diretamente no parent graph. As keys não vão corresponder, e a execução vai falhar. Aqui está o ponto chave. Resolves esta incompatibilidade usando uma wrapper function simples. No teu parent graph, defines uma node function normal que aceita o parent state. Dentro desta função, extrais o valor da user query. A seguir, chamas manualmente o subgraph de Research compilado, passando-lhe um payload onde mapeias essa user query para a key search term. O subgraph corre a sua lógica interna e devolve o seu state final. A tua wrapper function pega nesse output, traduz os resultados de volta para as keys específicas que o parent espera, e devolve-os. Para o parent graph, este wrapper parece um node normal. Não faz ideia de que um subgraph enorme e complexo acabou de ser executado lá dentro. Simplesmente passou dados para dentro e recebeu o state atualizado de volta. Este padrão dá-te uma modularidade estrita sem sacrificar o controlo. Tratar um graph compilado como apenas mais uma callable function por trás de um wrapper simples é a forma mais poderosa de escalar uma arquitetura de IA sem colapsares sob o teu próprio código. É tudo por este episódio. Obrigado por ouvires, e continua a programar!

Olá, daqui fala o Alex da DEV STORIES DOT EU. LangGraph, episódio 16 de 18. Se um sub-agent especialista for chamado duas vezes numa única conversa, deve lembrar-se da primeira chamada ou começar do zero com amnésia total? Esta escolha muda tudo na forma como os sistemas multi-agent se comportam, e é controlada inteiramente pela Subgraph Persistence e pelos Multi-Agent Patterns. Quando constróis um parent graph que encaminha tarefas para subgraphs, a gestão de state torna-se complexa. Imagina um bot principal de apoio ao cliente que lida com o chat geral. Quando um utilizador faz uma pergunta complexa sobre uma fatura, o bot principal encaminha o pedido para um subgraph dedicado de Billing Expert. Por default, os subgraphs são totalmente stateless. Quando compilas esse Billing Expert sem especificar um checkpointer, ele opera estritamente por invocação. O bot principal passa-lhe os inputs necessários, o especialista executa os seus passos internos, devolve um resultado e descarta imediatamente o seu state interno. Se o utilizador fizer uma pergunta de seguimento sobre faturação cinco minutos depois, o bot principal chama o especialista novamente. O especialista não tem memória da interação anterior. Começa completamente do zero. Para um subgraph simples de extração de dados, esta amnésia é perfeitamente aceitável. Para um agent interativo e especializado, é incrivelmente frustrante para o utilizador. Para corrigir isto, o especialista precisa da sua própria memória entre interações. Um erro muito comum aqui é passar uma instância de checkpointer completamente nova, como um objeto memory saver, diretamente para o método compile do subgraph. Não faças isto, a não ser que queiras que o subgraph partilhe exatamente o mesmo state entre utilizadores e sessões completamente diferentes. Se o utilizador A e o utilizador B falarem com o sistema ao mesmo tempo, passar uma instância explícita de checkpointer para o subgraph significa que os seus dados vão ser misturados num único state global. Isto cria um cross-talk enorme entre parent threads isoladas. Em vez disso, basta passares o valor booleano True para o argumento checkpointer quando compilas o subgraph. É aqui que a coisa fica interessante. Defini-lo como True diz ao subgraph para depender do mecanismo de checkpointer do parent graph, mas para manter um histórico multi-turn completamente isolado, especificamente para si próprio. Nos bastidores, a framework gere o namespacing. Cria automaticamente um thread ID único para o subgraph, que fica permanentemente ligado ao thread ID da parent thread. Agora, olha novamente para o cenário do Billing Expert com esta configuração. O utilizador faz uma pergunta sobre uma fatura. O bot principal encaminha-a para o especialista. O especialista responde e fica inativo. Mais tarde, na mesma conversa, o utilizador faz uma pergunta de seguimento. O bot principal encaminha de volta para o especialista. Como foi compilado com o checkpointer definido como True, o especialista acorda, verifica a sua sub-thread dedicada e carrega o context da fatura do turn anterior. Age como um participante persistente na conversa. E como essa sub-thread tem um scope estritamente limitado à parent thread, um utilizador diferente que fale com o sistema obtém a sua própria instância completamente limpa do Billing Expert. A forma como configuras o checkpointer de um subgraph dita toda a sua identidade no teu sistema: deixá-lo em branco cria uma utility function descartável e stateless, enquanto defini-lo como True cria um colaborador contínuo e context-aware. Obrigado por passares uns minutos comigo. Até à próxima, fica bem.

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 17 de 18. Um script Python a correr com sucesso no teu portátil não é uma aplicação de produção. Se tentares correr stateful agents através da execução de scripts standalone, vais inevitavelmente bater numa parede quando chegar a altura de escalar. Para resolver isto, vamos olhar para a Estrutura da Aplicação e para a Preparação para Deploy. Quando constróis um agent LangGraph pela primeira vez, provavelmente fazes o protótipo num Jupyter notebook ou num único ficheiro Python. Defines os nodes, ligas as edges, compilas o graph e chamas o método invoke logo ali no mesmo ficheiro para ver se funciona. Isso serve perfeitamente para testar. No entanto, um servidor de produção não consegue ler a tua mente. Ele precisa de uma forma standard para servir esse graph como uma API, instalar os packages necessários e injetar environment variables. Para deixares o teu protótipo pronto para deploy, tens de organizar o teu código numa estrutura de diretórios limpa. Imagina que crias uma nova pasta chamada my-app. Moves o teu código Python do notebook para um ficheiro limpo dentro desta pasta. A seguir, adicionas um ficheiro de dependências, tipicamente requirements dot txt. Finalmente, crias um ficheiro de configuração chamado langgraph dot json na raiz da pasta my-app. O ficheiro langgraph dot json é o blueprint central da tua aplicação. Quando usas a Command Line Interface do LangGraph, ou fazes deploy para um ambiente de produção, este ficheiro de configuração diz ao sistema subjacente exatamente como fazer o build e correr o teu projeto. Ele requer três informações principais sobre dependências, environment variables e os entry points do graph. Primeiro, declaras as tuas dependências. Isto é apenas uma string com o path no ficheiro JSON a apontar para o teu ficheiro de requirements. Isto garante que o servidor de deploy instala os packages Python exatos dos quais o teu agent depende, evitando erros de missing module em produção. A seguir, defines a string de environment. Esta aponta para o teu ficheiro dot env. Stateful agents precisam sempre de secrets, como credenciais de base de dados ou API keys de modelos. Apontar para o ficheiro de environment garante que o runtime carrega estas keys com segurança antes de tentar iniciar o graph. Esta é a parte que interessa. O terceiro requisito no ficheiro de configuração é o mapping dos graphs. Isto diz ao servidor exatamente onde o teu compiled graph vive no source code. Funciona como um dicionário. Atribuis um ID ao teu graph, que se torna o seu nome oficial na API gerada. Depois, fazes o map desse ID para um módulo Python específico e um nome de variável. Por exemplo, podes fazer o map do ID customer-support-agent para a string agent dot py dois pontos compiled-graph. O servidor olha para o ficheiro agent dot py, encontra a variável chamada compiled-graph, e carrega-a para a memória. Esta estrutura exige uma mudança deliberada na forma como escreves o teu código. Os iniciantes muitas vezes correm graphs através de scripts Python standalone que executam ações assim que são corridos. Mas o runtime do LangGraph depende do langgraph dot json para expor o graph dinamicamente como um web service. Ele não corre o teu script de cima a baixo. Apenas importa o objeto do compiled graph que especificaste no ficheiro de configuração. Por causa disto, o teu ficheiro Python deve apenas definir os nodes, ligá-los, e atribuir o compiled graph a uma variável. Tens de remover qualquer código de teste que tenha sobrado no final e que invoque o graph manualmente. Se deixares código de teste no ficheiro, ele vai ser executado durante a fase de import no servidor, causando falhas de deploy ou chamadas de API indesejadas só por iniciar o serviço. Ao declarares explicitamente as tuas dependências, environment e paths do graph num ficheiro JSON central, separas a definição do teu agent da sua execução, transformando um script local num serviço robusto e pronto para deploy. É tudo por este episódio. Obrigado por ouvires, e continua a desenvolver!

Olá, daqui é o Alex da DEV STORIES DOT EU. LangGraph, episódio 18 de 18. Tens um workflow multi-agent complexo e precisas de testar um edge case de routing específico no passo quatro. Não devias ter de correr o sistema inteiro do início ao fim só para atingir essa condição. Testar a execução do graph end-to-end é a forma de atingires exatamente a lógica de que precisas. Quando os programadores tentam isolar partes de um graph para testes, muitas vezes recorrem a objetos mock complexos. Tentam fazer mock da estrutura do graph à volta ou fazer stub de todos os nodes anteriores. No LangGraph, não precisas de fazer isso. A arquitetura gira inteiramente à volta do state. Como os nodes são apenas funções que leem e escrevem state, podes injetar manualmente um payload de state específico e testar fragmentos de nodes isolados de forma nativa. É aqui que a injeção de state e os breakpoints se tornam incrivelmente úteis na tua test suite. Só precisas de duas ferramentas para saltar diretamente para o meio de um graph. A primeira é o método update state. A segunda é um parâmetro de configuração chamado interrupt after. Usar isto dentro de uma testing framework standard como o pytest permite-te simular condições exatas sem executar a aplicação inteira. Vamos aplicar isto a um cenário concreto. Imagina que tens um graph onde o node três faz uma API call externa, e o node quatro verifica o resultado. Queres verificar se, caso o payload da API contenha um código de falha específico, o node quatro faz o routing correto do fluxo de execução para o teu error handler node. Em vez de correres os nodes um e dois para acionar isto, isolas o problema. Inicializas o graph com um identificador de thread. A seguir, usas o update state para inserir um payload de API simulado e com falha diretamente no state da thread. Ages como se o node três estivesse prestes a correr com esses dados específicos. Depois, invocas o graph, mas passas um dicionário de configuração a definir o interrupt after para o node quatro. Quando inicias o graph, a execução começa imediatamente no node três usando o teu state de falha injetado. O node três processa o bad payload e passa o state resultante para o node quatro. O node quatro avalia a lógica e decide fazer o routing para o error handler. Como definiste um breakpoint, o graph pausa a execução no momento em que o node quatro termina. Agora o teu teste pode avaliar o resultado. Puxas o state atual do graph. Podes escrever as tuas assertions para garantir que o node quatro atualizou as variáveis de state corretamente. Mais importante ainda, podes inspecionar o plano de execução do graph. Ao olhares para o próximo node pendente nos metadados do state, podes confirmar que a lógica de routing funcionou perfeitamente e que o error handler está na queue. Aqui está o ponto chave. Ao manipulares o state diretamente, transformas uma chain de agents altamente interligada e imprevisível num test case determinístico, passo a passo. Verificas exatamente como o graph transita de um node para o próximo sem esperares por language models ou network calls nos passos anteriores. Como este é o último episódio da série, encorajo-te a explorares a documentação oficial e a tentares construir estes workflows de forma hands-on. Se tiveres ideias sobre o que devemos abordar a seguir, visita devstories dot eu e sugere um tópico. É tudo por agora. Obrigado por ouvires, e continua a programar!