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 сам справляется с объединением этого частичного результата в полное состояние.

Редьюсеры: определение способа объединения обновлений

Если вы не указали иное, значение, возвращаемое узлом, заменяет соответствующее поле полностью. Это хорошо работает для таких данных, как draft, где важна только последняя версия. Однако это приводит к проблемам с такими структурами, как накопительный список сообщений, где перезапись уничтожит всё, что было до нее. Именно здесь на помощь приходит редьюсер — это функция, которая определяет, как объединить существующее значение с поступившим обновлением.

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 — цепочка запросов

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

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 — Оркестратор-рабочий узел

Концепция: один вызов большой языковой модели разделяет общую задачу на более мелкие подзадачи и передаёт каждую из них отдельному рабочему узлу, после чего финальный этап объединяет всё в один результат. Это сильно напоминает параллелизм, но здесь оркестратор динамически решает, сколько рабочих узлов запустить и что каждый из них должен делать, вместо того чтобы всё было задано заранее.

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 для управления потоком, а также редьюсеры для предотвращения взаимного сброса конкурирующих или повторяющихся обновлений. Благодаря всего лишь этим четырем элементам каждый из пяти рассмотренных шаблонов рабочего процесса — цепочечная обработка, маршрутизация, параллелизм, модель оркестратор-рабочие процессы и модель оценщик-оптимизатор — оказывается различной комбинацией тех же основных инструментов.

В следующей части этой серии будут введены внешние инструменты, добавится память для сохранения состояния между сессиями, а также появится этап, на котором человек проверяет и утверждает действия перед их выполнением — это необходимые элементы для превращения этих графов в настоящего агента.