LangGraph на практиці: стани, вузли, ребра та п’ять шаблонів агентів

Підсумок: де ми зупинилися

У попередній частині ми розглянули концептуальну основу: штучний інтелект-агент поєднує „мозок“ (LLM) із оркестратором, який координує інструменти та пам’ять, причому все це знаходиться під певним наглядом. Ми також бачили, що LangGraph перетворює це на граф — вузли містять одиниці роботи, ребра контролюють потік між ними, а спільний об’єкт стану переміщується по всій системі, об’єднуючись на кожному кроці за допомогою редукторів.

Цього разу ми переходимо від теорії до практичної реалізації. До кінця цієї статті ви зможете:

• Оголошувати стан за допомогою Annotation, включаючи користувацькі редуктори
• З’єднувати вузли за допомогою як звичайних, так і умовних ребер

Реалізуйте усі п’ять шаблонів робочого процесу з попередньої статті — ланцюгове формулювання запитів, маршрутизація, паралелізація, оркестрування між оператором та працівником, а також оцінювач та оптимізатор — у функціональному JavaScript.

Наведіть діаграму для кожного шаблону, яка чітко покаже, як пов’язані його елементи.

Увесь код нижче ґрунтується на @langchain/langgraph та звичайному JavaScript (ті самі шаблони працюють у TypeScript, достатньо додати параметри типу до Annotation<T>).

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

1. Стан та анотації — спільний блокнот

Перш ніж будь-який вузол зможе запрацювати, LangGraph має зрозуміти структуру даних, які проходять через ваш граф. Annotation — це засіб для опису такої структури.

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

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

Кожен ключ, який ви визначаєте всередині Annotation.Root({...}), стає полем у вашому об’єкті спільного стану. Будь-який вузол може читати ці поля, а коли повертається, йому потрібно включати лише ті поля, які він фактично оновлює — LangGraph самостійно об’єднує цей частковий результат у повний стан.

Reducerи: визначення способу об’єднання оновлень

Якщо ви не вказуєте інше, значення, яке повертає вузол, повністю замінює відповідне поле. Це підходить для чогось на кшталт draft, де важлива лише остання версія. Але це не спрацьовує у випадках, як-от накопичувальний список повідомлень, де перезапис призведе до видалення всього, що було раніше. Саме тут допомагає reducer — це функція, яка визначає, як об’єднати існуюче значення з новим оновленням.

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: () => [],
  }),
});

Завдяки цьому, якщо nodeA повертає { messages: ["hi"] }, а nodeB — { messages: ["hello"] }, редюсер об’єднує їх у ["hi", "hello"], замість того щоб дозволити одному з них перезаписати інший. Цей механізм є необхідним для безпечної роботи з паралельними гілками (див. приклад 3 нижче) — два вузли можуть одночасно записувати у ту саму ключ-стан, не виникаючи конфліктів.

2. Вузли — одиниці роботи

Вузол — це просто функція, синхронна або асинхронна, яка отримує поточний стан та повертає його часткове оновлення.

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

Ви реєструєте його у графі за допомогою addNode:

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

Зауважте, що сам вузол не має уявлення про структуру навколишньої графіки. Він просто споживає стан та генерує оновлення. З’єднання вузлів — це суто завдання ребер.

3. Ребра — організація потоку

Ребра визначають порядок виконання. LangGraph підтримує три типи:

Звичайні ребра — фіксований, безумовний шлях від одного вузла до іншого.

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

Умовні ребра — функція маршрутизації перевіряє поточний стан та вирішує, який вузол буде виконуватися далі.

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

Паралельні (розгалужені) ребра — коли один вузол має кілька зовнішніх звичайних ребер, усі їхні цілі виконуються одночасно.

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

Кожна графіка має містити принаймні одну ребро, яке починається від START, та маршрут, який зрештою доходить до END; інакше LangGraph не буде знати, де слід розпочати чи завершити виконання.

Тепер, коли розглянуто стани, вузли та ребра, у вас є всі необхідні елементи для створення п’яти шаблонів робочих процесів, представлених раніше.

4. Шаблон 1 — ланцюг запитів

Концепція: розділити велике завдання на лінійний ланцюг викликів LLM, де кожен крок ґрунтується на попередньому.

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);

Кожен вузол тут читає лише поле, яке записав попередній вузол — немає розгалужень чи циклів, лише пряма послідовність від початку до кінця.

5. Шаблон 2 — маршрутизація

Концепція: спочатку класифікуйте надходження даних, а потім передайте їх одному з кількох спеціалізованих шляхів обробки.

┌──> 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();

Функція, яку ви передаєте до addConditionalEdges, має повертати назву наступного вузла для обробки — саме цей механізм дозволяє графу динамічно обирати шлях замість постійного слідування одному й тому ж маршруту.

6. Паттерн 3 — Паралелізація

Концепція: одночасно викликати кілька моделей LLM для обробки різних частин завдання, а потім поєднати отримані результати.

┌──> 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();

Зверніть увагу на функцію-редуктор, приєднану до summaries: без неї остання паралельна гілка, яка завершиться, просто перезапише попередні результати, замість того щоб дозволити всім трьом результатам безпечно накопичуватися в масиві.

7. Шаблон 4 — Оркестратор-Робітник

Концепція: один виклик LLM ділить загальне завдання на менші підзавдання та передає кожне з них робочому вузлу, а на останньому етапі все об’єднується в один результат. Це дуже схоже на паралелізацію, але тут оркестратор на льоту вирішує, скільки робітників запустити та що кожен з них має робити, замість того щоб це було визначено заздалегідь.

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();

Якщо вам потрібна справжня динамічна розгалуженість — тобто кількість робочих вузлів фактично змінюється при кожному виконанні — LangGraph.js пропонує API Send, який дозволяє створювати по одному виклику робітника на кожне підзавдання під час виконання. Цей механізм заслуговує на окремий детальний опис, але наведена тут налаштування вже покриває більшість повсякденних потреб, таких як створення звітів чи розробка дослідницьких агентів.

8. Шаблон 5 — Оцінювач-оптимізатор

Концепція: один вузол генерує результат, другий вузол його перевіряє, а умовний ребро направляє виконання назад до генератора, поки результат не буде схвалений. Саме в цьому шаблоні фактично використовується підтримка циклів у 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();

Зверніть увагу, що поле attempts використовує функцію-зменшувач, яка додає значення до поточної суми замість того, щоб її замінювати — це також виступає як захисний механізм. Без певного ліміту на кількість спроб оцінювач, який ніколи не схвалює результат, може тримати цикл у роботі нескінченно.

9. Компіляція та запуск будь-якого з цих варіантів

Незалежно від того, який шаблон ви створюєте, процес завершується однаково:

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

Виклик методу .compile() перевіряє, чи можна фактично досягти кожного вузла, підтверджує, що точки START та END належним чином з’єднані, та перетворює опис графу на виконуваний об’єкт. Потім метод .invoke() запускає цей граф від початку до кінця та повертає результативний стан — це той самий цикл виконання, про який концептуально йшлося раніше в цій серії.

Підсумок

Протягом цього пошагового огляду ідеї з теорії перетворилися на функціональний код: Annotation.Root для визначення стану, звичайні функції як вузли, addEdge та addConditionalEdges для керування потоком, а також редюсери для запобігання тому, щоб одночасні або повторні оновлення впливали одне на одного. За допомогою лише цих чотирьох елементів кожен із п’яти розглянутих шаблонів робочого процесу — ланцюгова обробка, маршрутизація, паралелізація, модель оркестратор-працівник та модель оцінювач-оптимізатор — виявляється різною комбінацією тих самих основних інструментів.

Наступна частина цієї серії введе зовнішні інструменти, додасть пам’ять для збереження стану між сеансами та запровадить етап, під час якого людина переглядає та схвалює дії перед їх виконанням — це елементи, необхідні для перетворення цих графів на справжнього агента.