LangChain & LangGraph
Guia Completo

Origem

LangChain foi criado por Harrison Chase em outubro de 2022 como side project na Robust Intelligence. A ideia: eliminar o boilerplate de conectar LLMs com ferramentas externas, memória e pipelines de dados. O projeto viralizou no GitHub, Harrison largou o emprego e fundou a LangChain, Inc. em 2023 — US$25M seed (Benchmark) + US$35M Série A.

Evolução de Versões

O Ecossistema

┌─────────────────────────────────────────────────────────┐
│                    ECOSSISTEMA LANGCHAIN                 │
├──────────────┬──────────────┬──────────────┬────────────┤
│  LangChain   │  LangGraph   │  LangSmith   │ LangServe  │
│  (framework) │ (orquestrador│(observabilid.│(depreciado)│
│  chains,LCEL │  de agentes) │  e evals)    │            │
└──────────────┴──────────────┴──────────────┴────────────┘

┌─────────────────────────────────────┐
│              SEU CÓDIGO             │
├─────────────────────────────────────┤
│       LCEL / Chains / Agents        │
├──────────┬──────────┬───────────────┤
│  Model   │ Prompts  │  Tools/Memory │
│   I/O    │          │               │
├──────────┴──────────┴───────────────┤
│      @langchain/core (Runnable API) │
├──────────┬──────────┬───────────────┤
│  OpenAI  │Anthropic │  ...outros    │
└──────────┴──────────┴───────────────┘

Os 6 Componentes Core

import { ChatOpenAI } from "@langchain/openai";
import { ChatAnthropic } from "@langchain/anthropic";

// Mesma interface, provider diferente
const model = new ChatOpenAI({ model: "gpt-4o" });
// const model = new ChatAnthropic({ model: "claude-3-5-sonnet-20241022" });

const response = await model.invoke("Olá, mundo!");

import { ChatPromptTemplate } from "@langchain/core/prompts";

const prompt = ChatPromptTemplate.fromMessages([
  ["system", "Você é um especialista em {dominio}."],
  ["human", "{pergunta}"],
]);

const formatted = await prompt.format({
  dominio: "marketing digital",
  pergunta: "Como calcular CAC?",
});

import { StructuredOutputParser } from "@langchain/core/output_parsers";
import { z } from "zod";

const parser = StructuredOutputParser.fromZodSchema(
  z.object({
    nome: z.string(),
    score: z.number().min(0).max(10),
    justificativa: z.string(),
  })
);

const formatInstructions = parser.getFormatInstructions();

import { tool } from "@langchain/core/tools";
import { z } from "zod";

const buscarProdutos = tool(
  async ({ categoria, limite }) => {
    const produtos = await db.produtos.findMany({
      where: { categoria }, take: limite
    });
    return JSON.stringify(produtos);
  },
  {
    name: "buscar_produtos",
    description: "Busca produtos por categoria no catálogo",
    schema: z.object({
      categoria: z.string().describe("Categoria do produto"),
      limite: z.number().default(10),
    }),
  }
);

Loop de Execução LCEL

invoke(input)
    │
    ▼
[Runnable 1].invoke(input)    → output_1
    │
    ▼
[Runnable 2].invoke(output_1) → output_2
    │
    ▼
[Runnable 3].invoke(output_2) → final_output

Como Tool Calling Funciona

1. Você envia: mensagens + definições de tools (schemas JSON)
   │
   ▼
2. LLM retorna: AIMessage com tool_calls[] (não texto)
   { tool_calls: [{ name: "buscar_produtos", args: {...} }] }
   │
   ▼
3. Seu código executa a tool e captura o resultado
   │
   ▼
4. Você envia: ToolMessage com o resultado
   │
   ▼
5. LLM processa e gera resposta final (ou chama mais tools)

Streaming

const chain = prompt.pipe(model).pipe(parser);

for await (const chunk of await chain.stream({ pergunta: "O que é CAC?" })) {
  process.stdout.write(chunk); // imprime token por token
}

Callbacks

const handler = {
  handleLLMStart: async (llm, messages) => {
    console.log("LLM starting with", messages);
  },
  handleLLMEnd: async (output) => {
    console.log("LLM finished:", output);
  },
  handleToolStart: async (tool, input) => {
    console.log(`Tool ${tool.name} called with:`, input);
  },
};

await chain.invoke(input, { callbacks: [handler] });

Introduzido em agosto 2023. Substitui as classes pesadas (LLMChain, SequentialChain, RetrievalQA) por um sistema de pipe operator. Ideia central: qualquer dois componentes que implementem Runnable podem ser conectados com .pipe().

Pipe Syntax

import { ChatPromptTemplate } from "@langchain/core/prompts";
import { ChatOpenAI } from "@langchain/openai";
import { StringOutputParser } from "@langchain/core/output_parsers";

const prompt = ChatPromptTemplate.fromTemplate(
  "Explique {conceito} em uma frase para um CEO"
);
const model = new ChatOpenAI({ model: "gpt-4o-mini" });
const parser = new StringOutputParser();

const chain = prompt.pipe(model).pipe(parser);
const resultado = await chain.invoke({ conceito: "NRR" });

A Interface Runnable

interface Runnable<Input, Output> {
  invoke(input: Input, config?: RunnableConfig): Promise<Output>;
  stream(input: Input, config?: RunnableConfig): AsyncGenerator<Output>;
  batch(inputs: Input[], config?: RunnableConfig): Promise<Output[]>;
  pipe<NewOutput>(next: Runnable<Output, NewOutput>): Runnable<Input, NewOutput>;
}

Componentes Utilitários

import { RunnableLambda } from "@langchain/core/runnables";

const formataContexto = RunnableLambda.from(async (docs: Document[]) => {
  return docs.map(d => d.pageContent).join("\n\n");
});

const chain = retriever.pipe(formataContexto).pipe(prompt).pipe(model);

import { RunnablePassthrough } from "@langchain/core/runnables";

const chain = RunnablePassthrough.assign({
  context: retriever.pipe(formatDocs),
}).pipe(prompt).pipe(model).pipe(parser);

// Input: { question: "O que é CAC?" }
// Antes do prompt: { question: "...", context: "...docs..." }

import { RunnableParallel } from "@langchain/core/runnables";

const parallelChain = new RunnableParallel({
  resumo: chainDeResumo,
  sentimento: chainDeSentimento,
  topicos: chainDeTopicos,
});

const resultado = await parallelChain.invoke(texto);
// → { resumo: "...", sentimento: "positivo", topicos: [...] }

Quando Usar vs. Não Usar LCEL

Exemplo Completo: RAG com LCEL

import { ChatOpenAI, OpenAIEmbeddings } from "@langchain/openai";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";
import { RunnablePassthrough } from "@langchain/core/runnables";
import { MemoryVectorStore } from "langchain/vectorstores/memory";

const embeddings = new OpenAIEmbeddings();
const vectorStore = await MemoryVectorStore.fromTexts(
  ["CAC é o Custo de Aquisição de Cliente...", "NRR mede retenção..."],
  [{}, {}],
  embeddings
);
const retriever = vectorStore.asRetriever();

const prompt = ChatPromptTemplate.fromMessages([
  ["system", "Use o contexto abaixo para responder.\nContexto:\n{context}"],
  ["human", "{question}"],
]);

const model = new ChatOpenAI({ model: "gpt-4o-mini" });
const parser = new StringOutputParser();

const ragChain = RunnablePassthrough.assign({
  context: ({ question }: { question: string }) =>
    retriever.invoke(question).then(docs => docs.map(d => d.pageContent).join("\n")),
}).pipe(prompt).pipe(model).pipe(parser);

const resposta = await ragChain.invoke({ question: "O que é CAC?" });

// Streaming automático
for await (const token of await ragChain.stream({ question: "O que é NRR?" })) {
  process.stdout.write(token);
}

Estrutura de Pacotes

# Core — obrigatório sempre
npm install @langchain/core

# Providers — instale só os que usar
npm install @langchain/openai
npm install @langchain/anthropic
npm install @langchain/google-genai

# Utilitários, loaders, splitters
npm install langchain

# LangGraph — para agentes
npm install @langchain/langgraph

# LangSmith — tracing
npm install langsmith

Python vs. TypeScript

Integração com Fastify (Streaming)

import Fastify from "fastify";
import { ChatOpenAI } from "@langchain/openai";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { StringOutputParser } from "@langchain/core/output_parsers";

const fastify = Fastify();

// Instancia uma vez — modelos são stateless, seguro compartilhar
const model = new ChatOpenAI({ model: "gpt-4o-mini", temperature: 0 });

const analisarCampanha = ChatPromptTemplate.fromMessages([
  ["system", "Você é especialista em marketing. Analise campanhas com metodologia TOC."],
  ["human", "Analise: {dados_campanha}"],
]).pipe(model).pipe(new StringOutputParser());

fastify.post("/analisar-campanha", async (request, reply) => {
  const { dadosCampanha } = request.body as { dadosCampanha: string };

  reply.raw.setHeader("Content-Type", "text/event-stream");
  reply.raw.setHeader("Cache-Control", "no-cache");

  for await (const token of await analisarCampanha.stream({
    dados_campanha: dadosCampanha,
  })) {
    reply.raw.write(`data: ${JSON.stringify({ token })}\n\n`);
  }

  reply.raw.end();
});

Variáveis de Ambiente

OPENAI_API_KEY=sk-...
ANTHROPIC_API_KEY=sk-ant-...

# LangSmith (tracing automático)
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__...
LANGCHAIN_PROJECT=v4mos-ai-prod

# LangGraph Cloud (se usar)
LANGGRAPH_API_KEY=...

Guia de Decisão Rápida

Tarefa simples (prompt + resposta)?
    → OpenAI SDK direto ou LCEL simples

RAG com pipeline padronizado?
    → LangChain (LCEL + Retrieval)

Agente com tools e estado?
    → LangGraph (não AgentExecutor)

Multi-agentes ou workflows complexos?
    → LangGraph

Precisa de observabilidade completa?
    → LangSmith + LangGraph

Microserviço com latência crítica?
    → Considere SDK direto + instrumentação manual

O Problema que o LangGraph Resolve

• Caixa preta: Não havia como observar ou modificar o estado do agente durante execução
• Sem controle de fluxo: Impossível criar lógica condicional complexa ou loops controlados
• Sem persistência nativa: Cada invocação era independente — conversas multi-turn exigiam gambiarras
• Single-agent apenas: Não suportava múltiplos agentes cooperando
• Sem human-in-the-loop: Impossível pausar e esperar aprovação humana no meio de uma execução

A Ideia Central

Agentes são grafos dirigidos onde:

• Nós (nodes) são funções de processamento
• Arestas (edges) definem o fluxo de execução
• Estado é compartilhado entre todos os nós
• Checkpoints persistem o estado a qualquer momento

Empresas usando em Produção

StateGraph — O Coração

import { StateGraph, START, END } from "@langchain/langgraph";

const grafo = new StateGraph(MeuSchema)
  .addNode("processar", funcaoProcessar)
  .addNode("validar", funcaoValidar)
  .addEdge(START, "processar")
  .addEdge("processar", "validar")
  .addEdge("validar", END)
  .compile();

const resultado = await grafo.invoke({ input: "dados" });

Nodes — O Trabalho Real

// Assinatura de um node
async function meuNode(
  state: typeof MeuSchema.State,
  config?: RunnableConfig
): Promise<Partial<typeof MeuSchema.State>> {
  const resultado = await model.invoke(state.messages);
  // Retorna apenas as keys que mudaram
  return { messages: [resultado] };
}

Edges — Tipos

// Edge normal — transição fixa
graph.addEdge("nodeA", "nodeB");

// Conditional edge — decide com base no estado
function roteador(state: typeof MeuSchema.State): string {
  if (state.precisaAprovacao) return "solicitar_aprovacao";
  if (state.erros.length > 0) return "corrigir_erros";
  return "finalizar";
}
graph.addConditionalEdges("processar", roteador);

// Send — paralelismo dinâmico (map-reduce)
import { Send } from "@langchain/langgraph";

function distribuirTarefas(state: typeof GraphState.State) {
  return state.itens.map(item =>
    new Send("processar_item", { item })
  );
}
graph.addConditionalEdges("gerar_itens", distribuirTarefas, ["processar_item"]);

O estado é composto de channels. Cada channel tem um tipo TypeScript, um reducer (define como novos valores se fundem) e um valor padrão.

Definindo Estado com Annotation.Root (padrão atual)

import { Annotation } from "@langchain/langgraph";
import { BaseMessage } from "@langchain/core/messages";

const EstadoAgente = Annotation.Root({
  messages: Annotation<BaseMessage[]>({
    reducer: (x, y) => x.concat(y), // APPEND — cada node adiciona ao array
    default: () => [],
  }),
  etapa: Annotation<string>({
    reducer: (_, y) => y, // LAST WRITE WINS — substitui
    default: () => "inicio",
  }),
  aprovado: Annotation<boolean>({
    reducer: (_, y) => y,
    default: () => false,
  }),
});

Forma Moderna com StateSchema + Zod

import { StateSchema, MessagesValue, ReducedValue } from "@langchain/langgraph";
import { z } from "zod";

const EstadoCampanha = new StateSchema({
  messages: MessagesValue, // reducer built-in para mensagens

  errosEncontrados: new ReducedValue(z.array(z.string()), {
    reducer: (atual, novo) => [...atual, ...novo], // append
    default: () => [],
  }),

  etapaAtual: z.string().default("inicio"),  // last write wins
  aprovado: z.boolean().default(false),
});

Como o Estado Flui

Node A executa → retorna { messages: [nova_msg] }
    │
    ▼
Reducer: estado.messages = concat(atual, [nova_msg])
    │
    ▼
Node B executa → retorna { etapa: "processando" }
    │
    ▼
Reducer: estado.etapa = "processando"  (last write wins)
    │
    ▼
... e assim por diante

Super-Steps (Modelo Pregel)

LangGraph usa o modelo Pregel do Google. Cada "super-step" é uma rodada onde múltiplos nós podem executar em paralelo. Após o super-step, o estado é consolidado via reducers — isso habilita o paralelismo com Send().

No LangGraph, checkpointing não é feature opcional — é a base de human-in-the-loop, resiliência, conversas multi-turn e time travel para debug.

Escolhendo o Checkpointer

import { PostgresSaver } from "@langchain/langgraph-checkpoint-postgres";
import pg from "pg";

const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
});

const checkpointer = new PostgresSaver(pool);
await checkpointer.setup(); // cria as tabelas necessárias

const grafo = workflow.compile({ checkpointer });

// Listar checkpoints de um thread (histórico completo)
const historico = grafo.getStateHistory(config);
for await (const checkpoint of historico) {
  console.log(checkpoint.metadata.step, checkpoint.values);
}

// Pegar o estado atual
const estadoAtual = await grafo.getState(config);

// "Time travel" — retomar de checkpoint anterior
const configAntigo = {
  configurable: {
    thread_id: "user-123",
    checkpoint_id: "checkpoint-abc123",
  },
};
const resultadoRetomado = await grafo.invoke(null, configAntigo);

Pacotes e Instalação

npm install @langchain/langgraph @langchain/core
npm install @langchain/openai @langchain/anthropic
npm install @langchain/langgraph-checkpoint-sqlite
npm install @langchain/langgraph-checkpoint-postgres

Streaming com LangGraph em Fastify

fastify.post("/chat/:threadId", async (request, reply) => {
  const { threadId } = request.params as { threadId: string };
  const { message } = request.body as { message: string };
  const config = { configurable: { thread_id: threadId } };

  reply.raw.setHeader("Content-Type", "text/event-stream");
  reply.raw.setHeader("Cache-Control", "no-cache");
  reply.raw.setHeader("Connection", "keep-alive");

  const stream = await grafo.streamEvents(
    { messages: [{ role: "user", content: message }] },
    { ...config, version: "v2" }
  );

  for await (const event of stream) {
    if (event.event === "on_chat_model_stream") {
      const chunk = event.data?.chunk;
      if (chunk?.content) {
        reply.raw.write(
          `data: ${JSON.stringify({ type: "token", content: chunk.content })}\n\n`
        );
      }
    }
    if (event.event === "on_tool_start") {
      reply.raw.write(
        `data: ${JSON.stringify({ type: "tool_start", tool: event.name })}\n\n`
      );
    }
  }

  reply.raw.write(`data: ${JSON.stringify({ type: "done" })}\n\n`);
  reply.raw.end();
});

Python vs TypeScript no LangGraph

LangGraph Platform resolve problemas de infraestrutura: escalabilidade horizontal, checkpoints gerenciados, filas para agentes long-running, Studio para debug visual e APIs REST automáticas.

LangGraph CLI

npm install -g @langchain/langgraph-cli

langgraph new meu-agente   # cria projeto
langgraph dev              # desenvolvimento com hot reload
langgraph build            # build para produção
langgraph deploy           # deploy para LangGraph Cloud

Self-Hosted com Docker

docker run -p 8123:8123 \
  -e DATABASE_URL=postgresql://... \
  -e LANGCHAIN_API_KEY=... \
  langchain/langgraph-api

LangGraph Studio

IDE visual (desktop app) para: visualizar o grafo, inspecionar estado a cada step, "time travel" (voltar checkpoints anteriores), editar estado manualmente e debugar com breakpoints visuais.

Setup em 2 Minutos

# .env — com essas 3 vars, TODO código LangChain/LangGraph
# envia traces ao LangSmith automaticamente
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__seu_api_key_aqui
LANGCHAIN_PROJECT=v4mos-ai-producao

import { traceable } from "langsmith/traceable";
import { wrapOpenAI } from "langsmith/wrappers";
import OpenAI from "openai";

const openai = wrapOpenAI(new OpenAI()); // wrapa para tracing automático

const minhaFuncao = traceable(
  async (input: string) => {
    const response = await openai.chat.completions.create({
      model: "gpt-4o",
      messages: [{ role: "user", content: input }],
    });
    return response.choices[0].message.content;
  },
  { name: "minha_funcao", project_name: "v4mos-ai" }
);

import { evaluate } from "langsmith/evaluation";

await evaluate(
  async (input) => ragChain.invoke({ question: input.question }),
  {
    data: "dataset-perguntas-marketing",
    evaluators: [
      {
        evaluatorType: "labeled_criteria",
        criteria: "relevance",
        feedbackKey: "relevancia",
      },
      async (run, example) => {
        const score = calcularSeuScore(
          run.outputs?.output, example.outputs?.answer
        );
        return { key: "accuracy", score };
      },
    ],
    experimentPrefix: "rag-v2",
  }
);

Antes vs Depois

import { AgentExecutor, createOpenAIToolsAgent } from "langchain/agents";

// Caixa preta — você não controla o loop
const agent = await createOpenAIToolsAgent({ llm: model, tools, prompt });
const agentExecutor = new AgentExecutor({ agent, tools });
const resultado = await agentExecutor.invoke({ input: "Analise o ROAS" });

import { StateGraph, START, END, MessagesAnnotation } from "@langchain/langgraph";
import { ToolNode } from "@langchain/langgraph/prebuilt";

const model = new ChatOpenAI({ model: "gpt-4o" }).bindTools(tools);

async function chamarModelo(state: typeof MessagesAnnotation.State) {
  const resposta = await model.invoke(state.messages);
  return { messages: [resposta] };
}

function decidirProximoPasso(state: typeof MessagesAnnotation.State) {
  const ultima = state.messages[state.messages.length - 1] as AIMessage;
  return ultima.tool_calls?.length ? "tools" : END;
}

const agente = new StateGraph(MessagesAnnotation)
  .addNode("agente", chamarModelo)
  .addNode("tools", new ToolNode(tools))
  .addEdge(START, "agente")
  .addConditionalEdges("agente", decidirProximoPasso)
  .addEdge("tools", "agente")
  .compile({ checkpointer: new MemorySaver() }); // agora tem persistência!

Atalho: createReactAgent

import { createReactAgent } from "@langchain/langgraph/prebuilt";

// Drop-in replacement para AgentExecutor
const agente = createReactAgent({
  llm: model,
  tools: [buscarMetricas, calcularROAS],
  checkpointSaver: new MemorySaver(),
});

const resultado = await agente.invoke({
  messages: [{ role: "user", content: "Analise o ROAS" }],
});

Checklist de Migração

• Identificar todos os AgentExecutor no codebase
• Definir schema de estado para cada agente
• Extrair tools existentes (compatíveis sem mudança)
• Criar nó "agente" com o modelo + bindTools
• Usar ToolNode para execução das tools
• Implementar função de roteamento (continua vs. termina)
• Configurar checkpointer (MemorySaver dev, Postgres prod)
• Compilar e testar com os mesmos inputs
• Habilitar LANGCHAIN_TRACING_V2=true e comparar traces
• Canary deploy (10% do tráfego) e comparar latência/qualidade

Guia de Decisão Final

Sua task:
    │
    ├── "Prompt + resposta simples"
    │   → LCEL: prompt.pipe(model).pipe(parser)
    │
    ├── "RAG com retrieval + geração"
    │   → LCEL com Retrieval
    │
    ├── "Extração de dados estruturados"
    │   → LCEL com model.withStructuredOutput()
    │
    ├── "Agente que decide quais tools chamar"
    │   ├── Simples, sem persistência?
    │   │   → createReactAgent (LangGraph prebuilt)
    │   └── Com estado, HITL, ou multi-turn?
    │       → StateGraph customizado
    │
    ├── "Múltiplos agentes cooperando"
    │   → LangGraph + createSupervisor
    │
    ├── "Pausar e esperar aprovação humana"
    │   → LangGraph com interrupt() + checkpointing
    │
    ├── "Processar 100 documentos em paralelo e agregar"
    │   → LangGraph com Send() (map-reduce)
    │
    └── "Tenho AgentExecutor legado"
        → Migrar: use createReactAgent como passo 1