LangGraph w praktyce: stany, węzły, krawędzie oraz pięć wzorców agentów

Podsumowanie: Gdzie skończyliśmy

W poprzednim rozdziale przedstawiliśmy podstawy koncepcyjne: agent AI łączy mózg (LLM) z orkiestratorem, który koordynuje narzędzia i pamięć, a wszystko to jest objęte jakąś formą nadzoru. Zobaczyliśmy również, że LangGraph przekształca to w graf — węzły przechowują jednostki pracy, krawędzie kontrolują przepływ między nimi, a wspólny obiekt stanu przemieszcza się przez cały system, łącząc się na każdym kroku za pomocą reduktorów.

Tym razem przechodzimy od teorii do implementacji. Po przeczytaniu tego posta będziesz mógł:

• Deklarować stan za pomocą Annotation, włączając w to dostosowane reduktory
• Łączyć węzły za pomocą zarówno zwykłych, jak i warunkowych krawędzi

Zaimplementuj wszystkie pięć wzorców przepływu pracy z poprzedniego posta — łączenie zapytań, routowanie, paralelizacja, orkiestrator-pracownik oraz ewaluator-optymalizator — w działającym kodzie JavaScript

Dodaj do każdego wzorca diagram pokazujący dokładnie, w jaki sposób elementy się łączą

Cały poniższy kod opiera się na @langchain/langgraph oraz zwykłym JavaScript (te same wzorce działają w TypeScript, wystarczy dodać parametry typu do Annotation<T>).

npm install @langchain/langgraph @langchain/core @langchain/openai

1. Stan i adnotacje — Wspólny notatnik

Zanim jakikolwiek węzeł będzie mógł zostać uruchomiony, LangGraph musi zrozumieć strukturę danych przepływających przez graf. Annotation służy do opisania tej struktury.

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

const WorkflowState = Annotation.Root({
  query: Annotation(),
  searchResults: Annotation(),
  draft: Annotation(),
  finalAnswer: Annotation(),
});

Każda kluczowa wartość, którą zdefiniujesz wewnątrz Annotation.Root({...}), staje się polem w twoim obiekcie stanu współdzielonego. Każdy węzeł może odczytywać te pola, a po powrocie musi zawierać jedynie te pola, które faktycznie aktualizuje — LangGraph sam łączy ten częściowy wynik z pełnym stanem.

Reducerzy: Decydowanie o sposobie łączenia aktualizacji

Chyba że zaznaczysz inaczej, wartość zwracana przez węzeł zastępuje całkowicie odpowiadające jej pole. To działa dobrze w przypadku czegoś takiego jak draft, gdzie interesuje nas tylko najnowsza wersja. Jednak nie sprawdza się to przy kumulujących się listach wiadomości, gdzie nadpisywanie usunęłoby wszystko, co pojawiło się wcześniej. W takich sytuacjach pomocny jest reducer — to funkcja, która określa, jak połączyć istniejącą wartość z nową aktualizacją.

const AgentState = Annotation.Root({
  // Default behavior: last write wins
  draft: Annotation(),

  // Custom reducer: append instead of overwrite
  messages: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
});

Gdy to jest wdrożone, jeśli nodeA zwraca { messages: ["hi"] }, a nodeB zwraca { messages: ["hello"] }, reducer łączy je w ["hi", "hello"], zamiast pozwolić, by jeden przepisał drugi. Ten mechanizm jest niezbędny do bezpiecznego uruchamiania równoległych gałęzi (zobacz wzorzec 3 poniżej) — dwa węzły mogą jednocześnie zapisywać do tej samej klucza stanu bez konfliktów.

2. Węzły — jednostki pracy

Węzeł to nic innego jak funkcja — synchroniczna lub asynchroniczna — która otrzymuje bieżący stan i zwraca jego częściową aktualizację.

async function draftNode(state) {
  const response = await llm.invoke(`Write a short draft about: ${state.query}`);
  return { draft: response.content };
}

Zarejestrujesz go w grafie za pomocą addNode:

const graph = new StateGraph(WorkflowState)
  .addNode("draft", draftNode);

Zauważ, że sam węzeł nie ma pojęcia o strukturze otaczającego go grafu. Po prostu konsumuje stan i wytwarza aktualizację. Łączenie węzłów jest wyłącznie zadaniem krawędzi.

3. Krawędzie — łączenie elementów

Krawędzie określają kolejność wykonywania. LangGraph obsługuje trzy rodzaje:

Krawędzie normalne — stała, bezwarunkowa ścieżka od jednego węzła do drugiego.

graph.addEdge(START, "draft");   // entry point: where the graph starts
graph.addEdge("draft", "polish");
graph.addEdge("polish", END);    // exit point: where the graph stops

Krawędzie warunkowe — funkcja routingu sprawdza aktualny stan i decyduje, który węzeł ma zostać uruchomiony dalej.

graph.addConditionalEdges("classify", (state) => {
  if (state.category === "billing") return "handleBilling";
  if (state.category === "technical") return "handleTechnical";
  return "handleGeneral";
});

Krawędzie równoległe (fan-out) — gdy jeden węzeł ma kilka krawędzi normalnych wychodzących, wszystkie ich cele są wykonywane jednocześnie.

graph.addEdge("orchestrator", "workerA");
graph.addEdge("orchestrator", "workerB");
graph.addEdge("orchestrator", "workerC");

Każdy graf musi zawierać co najmniej jedną krawędź zaczynającą się od START oraz trasę, która ostatecznie dociera do END; w przeciwnym razie LangGraph nie będzie wiedział, gdzie powinna rozpocząć się lub zakończyć eksploatacja.

Ponieważ teraz omówiono stan, węzły i krawędzie, masz wszystkie niezbędne elementy do skonstruowania pięciu wzorców przepływu pracy przedstawionych wcześniej.

4. Wzorzec 1 — Łańcuchowanie zapytań

Koncepcja: podzielenie dużej zadania na liniową sekwencję wywołań modeli językowych, przy czym każdy krok opiera się na tym, co było wcześniej.

START → outline → draft → polish → END

const ChainState = Annotation.Root({
  topic: Annotation(),
  outline: Annotation(),
  draft: Annotation(),
  final: Annotation(),
});

async function outlineNode(state) {
  const res = await llm.invoke(`Create a short outline for: ${state.topic}`);
  return { outline: res.content };
}async function draftNode(state) {
  const res = await llm.invoke(`Write a draft based on this outline:\n${state.outline}`);
  return { draft: res.content };
}async function polishNode(state) {
  const res = await llm.invoke(`Polish and tighten this draft:\n${state.draft}`);
  return { final: res.content };
}const chainGraph = new StateGraph(ChainState)
  .addNode("outline", outlineNode)
  .addNode("draft", draftNode)
  .addNode("polish", polishNode)
  .addEdge(START, "outline")
  .addEdge("outline", "draft")
  .addEdge("draft", "polish")
  .addEdge("polish", END)
  .compile();const result = await chainGraph.invoke({ topic: "the history of tea" });
console.log(result.final);

Każdy węzeł tutaj odczytuje tylko pole zapisane przez węzeł poprzedzający go — nie ma żadnego rozgałęziania ani pętli, jedynie prosta ścieżka od początku do końca.

5. Wzorzec 2 — Ukierunkowywanie

Koncepcja: najpierw sklasyfikuj wprowadzony dane, a następnie przekazaj je do jednej z kilku dedykowanych ścieżek obsługi.

┌──> handleBilling   ──┐
START → classify ───┼──> handleTechnical ─┼──> END
                    └──> handleGeneral   ──┘

const RouteState = Annotation.Root({
  message: Annotation(),
  category: Annotation(),
  response: Annotation(),
});

async function classifyNode(state) {
  const res = await llm.invoke(
    `Classify this message as "billing", "technical", or "general". Only reply with one word.\n${state.message}`
  );
  return { category: res.content.trim().toLowerCase() };
}async function handleBilling(state) {
  return { response: `[Billing team] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleTechnical(state) {
  return { response: `[Tech support] ${await llm.invoke(state.message).then(r => r.content)}` };
}
async function handleGeneral(state) {
  return { response: `[General desk] ${await llm.invoke(state.message).then(r => r.content)}` };
}const routeGraph = new StateGraph(RouteState)
  .addNode("classify", classifyNode)
  .addNode("handleBilling", handleBilling)
  .addNode("handleTechnical", handleTechnical)
  .addNode("handleGeneral", handleGeneral)
  .addEdge(START, "classify")
  .addConditionalEdges("classify", (state) => {
    if (state.category === "billing") return "handleBilling";
    if (state.category === "technical") return "handleTechnical";
    return "handleGeneral";
  })
  .addEdge("handleBilling", END)
  .addEdge("handleTechnical", END)
  .addEdge("handleGeneral", END)
  .compile();

Funkcja przekazywana do addConditionalEdges musi zwracać nazwę następnego węzła do odwiedzenia — to właśnie ten mechanizm pozwala grafowi dynamicznie wybierać ścieżkę zamiast zawsze podążać tą samą trasą.

6. Wzorzec 3 — Parallelizacja

Koncepcja: uruchom kilka jednoczesnych wywołań LLM na różnych fragmentach zadania, a następnie połącz uzyskane wyniki.

┌──> summarizeDocA ──┐
START → split ──┼──> summarizeDocB ──┼──> merge → END
                └──> summarizeDocC ──┘

const ParallelState = Annotation.Root({
  documents: Annotation(),
  summaries: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  finalSummary: Annotation(),
});

function splitNode(state) {
  return {}; // no-op entry point, just fans out to the summarizers below
}async function summarizeDocA(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[0]}`);
  return { summaries: [res.content] };
}
async function summarizeDocB(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[1]}`);
  return { summaries: [res.content] };
}
async function summarizeDocC(state) {
  const res = await llm.invoke(`Summarize:\n${state.documents[2]}`);
  return { summaries: [res.content] };
}async function mergeNode(state) {
  const res = await llm.invoke(`Combine these summaries into one:\n${state.summaries.join("\n---\n")}`);
  return { finalSummary: res.content };
}const parallelGraph = new StateGraph(ParallelState)
  .addNode("split", splitNode)
  .addNode("summarizeDocA", summarizeDocA)
  .addNode("summarizeDocB", summarizeDocB)
  .addNode("summarizeDocC", summarizeDocC)
  .addNode("merge", mergeNode)
  .addEdge(START, "split")
  .addEdge("split", "summarizeDocA")
  .addEdge("split", "summarizeDocB")
  .addEdge("split", "summarizeDocC")
  .addEdge("summarizeDocA", "merge")
  .addEdge("summarizeDocB", "merge")
  .addEdge("summarizeDocC", "merge")
  .addEdge("merge", END)
  .compile();

Zwróć uwagę na reduktora przypisanego do summaries: bez niego ostatnia zakończona równoległa gałąź po prostu nadpisze poprzednie wyniki, zamiast umożliwić bezpieczne gromadzenie wszystkich trzech wyników w tablicy.

7. Wzorzec 4 — Orchestrator-Worker

Koncepcja: jedna wywołanie LLM dzieli całkowite zadanie na mniejsze podzadania i przekazuje każde z nich węadowi robociemu, a następnie ostatni krok łączy wszystko w jeden wynik. Wygląda to bardzo podobnie do paralelizacji, ale tutaj orchestrator decyduje w czasie rzeczywistym, ilu robotników uruchomić i co każdy z nich powinien robić, zamiast ustalać to z góry.

START → orchestrator ──> worker(s), one per subtask ──> compile → END

const OrchestratorState = Annotation.Root({
  task: Annotation(),
  subtasks: Annotation(),
  sections: Annotation({
    reducer: (current, update) => [...current, ...update],
    default: () => [],
  }),
  report: Annotation(),
});

async function orchestratorNode(state) {
  const res = await llm.invoke(
    `Break this task into 3-5 short subtasks, one per line:\n${state.task}`
  );
  const subtasks = res.content.split("\n").filter(Boolean);
  return { subtasks };
}async function workerNode(state) {
  // In a real app you'd invoke one worker per subtask (e.g. via Send / map-reduce);
  // here's the simple per-subtask logic each worker runs:
  const results = await Promise.all(
    state.subtasks.map((subtask) => llm.invoke(`Write a section about: ${subtask}`))
  );
  return { sections: results.map((r) => r.content) };
}async function compileNode(state) {
  const res = await llm.invoke(`Compile these sections into one report:\n${state.sections.join("\n\n")}`);
  return { report: res.content };
}const orchestratorGraph = new StateGraph(OrchestratorState)
  .addNode("orchestrator", orchestratorNode)
  .addNode("worker", workerNode)
  .addNode("compile", compileNode)
  .addEdge(START, "orchestrator")
  .addEdge("orchestrator", "worker")
  .addEdge("worker", "compile")
  .addEdge("compile", END)
  .compile();

Jeśli potrzebujesz prawdziwie dynamicznego rozgałęzienia — to znaczy liczba gałęzi roboczych faktycznie zmienia się przy każdym uruchomieniu — LangGraph.js oferuje API Send, które umożliwia uruchomienie jednej inicjatywy roboczej na każde podzadanie w momencie wykonywania. Ten mechanizm zasługuje na osobny, szczegółowy opis, ale przedstawiona tutaj konfiguracja już obejmuje zdecydowaną większość codziennych potrzeb, takich jak tworzenie raportów czy budowanie agentów badawczych.

8. Wzorzec 5 — Evaluator-Optimizer

Koncepcja: jeden węzeł generuje wynik, drugi go sprawdza, a warunkowa krawędź kieruje wykonywanie z powrotem do generatora, dopóki wynik nie zostanie zaakceptowany. To właśnie w tym wzorcu wykorzystywana jest obsługa cykli w LangGraph.

START → generate → evaluate ──(fail)──> generate  (loop)
                       │
                     (pass)
                       ↓
                      END

const LoopState = Annotation.Root({
  prompt: Annotation(),
  code: Annotation(),
  feedback: Annotation(),
  passed: Annotation(),
  attempts: Annotation({
    reducer: (current, update) => current + update,
    default: () => 0,
  }),
});

async function generateNode(state) {
  const context = state.feedback ? `\nPrevious feedback: ${state.feedback}` : "";
  const res = await llm.invoke(`Write code for: ${state.prompt}${context}`);
  return { code: res.content, attempts: 1 };
}async function evaluateNode(state) {
  const res = await llm.invoke(
    `Review this code for bugs. Reply "PASS" if it's correct, otherwise explain what's wrong:\n${state.code}`
  );
  const passed = res.content.trim().startsWith("PASS");
  return { passed, feedback: passed ? null : res.content };
}const loopGraph = new StateGraph(LoopState)
  .addNode("generate", generateNode)
  .addNode("evaluate", evaluateNode)
  .addEdge(START, "generate")
  .addEdge("generate", "evaluate")
  .addConditionalEdges("evaluate", (state) => {
    if (state.passed || state.attempts >= 4) return END; // safety cap on retries
    return "generate";
  })
  .compile();

Zauważ, że pole attempts używa reduktora, który dodaje do bieżącej sumy zamiast ją zastępować — to również pełni rolę ograniczenia. Bez jakiegoś limitu prób, ewaluator, który nigdy nie zatwierdza wyniku, mógłby utrzymywać pętlę w działaniu nieograniczenie.

9. Kompilowanie i uruchamianie którychkolwiek z nich

Niezależnie od wybranego wzorca, kończysz w ten sam sposób:

const app = graph.compile();
const finalState = await app.invoke({ /* initial state fields */ });

Wywołanie .compile() sprawdza, czy do każdego węzła można faktycznie dotrzeć, potwierdza, że START i END są prawidłowo połączone, oraz przekształca opis grafu w obiekt wykonywalny. Następnie .invoke() uruchamia ten graf od początku do końca i zwraca uzyskany stan — to ten sam cykl wykonywania, który został opisany koncepcyjnie wcześniej w tej serii.

Podsumowanie

W trakcie tego przewodnika pomysły z teorii przekształciły się w działający kod: Annotation.Root służy do definiowania stanu, zwykłe funkcje pełnią rolę węzłów, addEdge i addConditionalEdges kontrolują przepływ, a reduktory zapobiegają wzajemnemu niszczeniu się aktualizacji wykonywanych równocześnie lub powtarzająco. Dzięki tym czterem elementom każdy z pięciu omawianych wzorców pracy — łańcuchowanie, routowanie, paralelizacja, orkiestrator-pracownik oraz ewaluator-optymalizator — okazuje się być różną kombinacją tych samych podstawowych narzędzi.

Następna część tej serii wprowadzi narzędzia zewnętrzne, dodaje pamięć umożliwiającą przechowywanie stanu między sesjami oraz wprowadzi etap, w którym osoba sprawdza i zatwierdza działania przed ich wykonaniem — elementy niezbędne do przekształcenia tych grafów w prawdziwego agenta.

Literatura pokrewna

• Budowanie agenta AI od zera: paterny, ReAct i LangGraph — Poznaj podstawowe koncepcje leżące u podstaw agentów AI — planowanie, używanie narzędzi, refleksja oraz patern ReAct — oraz to, jak LangChain i LangGraph wpisują się w proces ręcznego budowania takiego agenta.
• Co właściwie automatyzuje LangChain po stworzeniu pętli agenta — Wyjaśnia, w jaki sposób LangChain, LangGraph oraz podobne SDK-y otaczają tę samą podstawową pętlę agenta stworzoną od zera, oraz kiedy korzystanie z frameworka jest pomocne, a kiedy szkodliwe.

Eskalacja węzłów, a nie zadań: szczeblowa struktura sześciu poziomów dla kosztów procesów LLM — Dlaczego wybór między DAG a agentem na każde zadanie zwiększa koszty LLM oraz jak szczeblowa struktura eskalacji węzłów z umowami, zakresem i budżetem pomaga je kontrolować.