Назад в журнал
•8 мин чтения

Создание автономных мультиагентных рабочих процессов с LangGraph и Next.js

Архитектура мультиагентных LLM-графов с сохранением состояния, пользовательскими каналами состояния, условной маршрутизацией, проверкой человеком и потоковым ответом в Next.js.

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
Создание автономных мультиагентных рабочих процессов с LangGraph и Next.js

При разработке продакшен-решений ИИ, выходящих за рамки простых интерфейсов чата, линейные цепочки LLM быстро упираются в архитектурные ограничения. Сложные инженерные задачи — такие как автоматическое проведение ревью кода, синтез исследований из нескольких источников или автономный рефакторинг систем — требуют делегирования задач между специализированными агентами, итеративных циклов проверки и строгого сохранения состояния.

Линейные цепочки (A -> B -> C) дают сбой, когда агент B выдает неточный результат или когда шаг C требует подтверждения человеком перед изменением базовых данных в продакшене. Чтобы создавать отказоустойчивые агентные сети, системы должны поддерживать циклические графы состояния, условную маршрутизацию и детерминированные контрольные точки (checkpointing).

В этом руководстве мы реализуем архитектуру графа мультиагентов с сохранением состояния с использованием LangGraph (TypeScript) и интегрируем ее в приложение Next.js App Router с потоковой передачей SSE в реальном времени.

Архитектура: Мультиагентные графы с сохранением состояния

Вместо того чтобы заставлять одну модель LLM выполнять все роли, мультиагентная система разделяет обязанности между отдельными узлами, связанными общим контекстом состояния.

                  +-----------------------+
                  |     User Prompt       |
                  +-----------------------+
                              |
                              v
                  +-----------------------+
                  |    Supervisor Node    |
                  +-----------------------+
                     /        |        \
       (Research)   /         |         \  (Drafting)
                   v          |          v
          +------------+      |      +------------+
          | Researcher |      |      |   Writer   |
          |   Agent    |      |      |   Agent    |
          +------------+      |      +------------+
                   \          |          /
                    \         v         /
                  +-----------------------+
                  |     Reviewer Node     |
                  +-----------------------+
                              |
                     (Needs Revision?)
                     /                 \
             [YES / Retry]         [NO / Complete]
                  /                     \
                 v                       v
      +-------------------+    +-------------------+
      | Interrupt / Edit  |    |   Final Output    |
      +-------------------+    +-------------------+

Ключевые примитивы в LangGraph:

  1. Каналы состояния (State Channels): Строго типизированная общая память, передаваемая между узлами.
  2. Узлы (Nodes): Изолированные функции TypeScript, представляющие отдельных агентов или исполнителей инструментов.
  3. Ребра (Edges): Направленные связи, передающие выполнение между узлами.
  4. Условные ребра (Conditional Edges): Динамическая логика маршрутизации на основе критериев оценки или пороговых значений ошибок.
  5. Контрольные точки (Checkpointers): Персистентное хранилище состояния, позволяющее приостанавливать, возобновлять и запрашивать подтверждение человека.

1. Определение строго типизированных каналов состояния

В LangGraph состояние определяется с использованием аннотированных каналов. Каждый узел получает текущее состояние графа и возвращает частичное обновление состояния. Редукторы определяют, как обновления объединяются с существующим состоянием.

Создайте src/lib/agents/state.ts:

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

export interface AgentTask {
  id: string;
  description: string;
  assignedTo: "researcher" | "writer" | "reviewer";
  status: "pending" | "in_progress" | "completed" | "failed";
  result?: string;
}

// Define the central state annotation schema
export const AgentGraphAnnotation = Annotation.Root({
  // Append new messages to conversation history
  messages: Annotation<BaseMessage[]>({
    reducer: (x, y) => x.concat(y),
    default: () => [],
  }),
  // Active task context
  tasks: Annotation<AgentTask[]>({
    reducer: (x, y) => y, // Overwrite with latest task list
    default: () => [],
  }),
  // Quality rating produced by the Reviewer Agent (0 to 10)
  reviewScore: Annotation<number>({
    reducer: (_, y) => y,
    default: () => 0,
  }),
  // Re-evaluation loop counter to prevent infinite recursion
  iterationCount: Annotation<number>({
    reducer: (x, y) => x + y,
    default: () => 0,
  }),
  // Flag indicating human approval requirement
  requiresApproval: Annotation<boolean>({
    reducer: (_, y) => y,
    default: () => false,
  }),
});

export type AgentGraphState = typeof AgentGraphAnnotation.State;

2. Реализация специализированных узлов агентов

Узлы — это асинхронные функции, выполняющие предметно-ориентированную логику. Здесь мы определяем три узла: Researcher, Writer и Reviewer.

Создайте src/lib/agents/nodes.ts:

import { SystemMessage, HumanMessage, AIMessage } from "@langchain/core/messages";
import { ChatOpenAI } from "@langchain/openai";
import type { AgentGraphState } from "./state";

const model = new ChatOpenAI({
  modelName: "gpt-4o-mini",
  temperature: 0.2,
});

// Node 1: Researcher Agent
export async function researcherNode(state: AgentGraphState) {
  const lastMessage = state.messages[state.messages.length - 1];
  
  const response = await model.invoke([
    new SystemMessage(
      "You are a Senior Technical Researcher. Gather structural facts, API constraints, and performance benchmarks."
    ),
    new HumanMessage(lastMessage.content as string),
  ]);

  return {
    messages: [new AIMessage({ content: `[Researcher]: ${response.content}` })],
    iterationCount: 1,
  };
}

// Node 2: Writer Agent
export async function writerNode(state: AgentGraphState) {
  const researchData = state.messages
    .filter((m) => typeof m.content === "string" && m.content.startsWith("[Researcher]"))
    .map((m) => m.content)
    .join("\n");

  const response = await model.invoke([
    new SystemMessage(
      "You are a Technical Writer. Synthesize research data into structured, concrete TypeScript guides."
    ),
    new HumanMessage(`Synthesize the following research:\n${researchData}`),
  ]);

  return {
    messages: [new AIMessage({ content: `[Writer]: ${response.content}` })],
  };
}

// Node 3: Reviewer Node (Evaluates output quality)
export async function reviewerNode(state: AgentGraphState) {
  const draftMessage = state.messages.find(
    (m) => typeof m.content === "string" && m.content.startsWith("[Writer]")
  );

  const evaluation = await model.invoke([
    new SystemMessage(
      "You are a Code Reviewer. Score the technical accuracy from 1 to 10. Respond ONLY with JSON: {\"score\": number, \"feedback\": string}"
    ),
    new HumanMessage((draftMessage?.content as string) || ""),
  ]);

  let score = 5;
  try {
    const parsed = JSON.parse(evaluation.content as string);
    score = parsed.score;
  } catch {
    score = 6;
  }

  return {
    reviewScore: score,
    requiresApproval: score >= 8,
  };
}

3. Построение StateGraph с условными ребрами

Мы объединяем узлы в циклический граф с помощью функции условного ребра, чтобы решить, перенаправлять ли запрос обратно Writer для доработки или завершить выполнение.

Создайте src/lib/agents/graph.ts:

import { StateGraph, START, END, MemorySaver } from "@langchain/langgraph";
import { AgentGraphAnnotation } from "./state";
import { researcherNode, writerNode, reviewerNode } from "./nodes";

// Router function determining graph execution path
function routeReviewOutcome(state: typeof AgentGraphAnnotation.State) {
  // Prevent infinite iteration loops
  if (state.iterationCount >= 3) {
    return "finalize";
  }

  // If score meets threshold, proceed to human approval or completion
  if (state.reviewScore >= 8) {
    return "finalize";
  }

  // Otherwise, route back to Writer for revision
  return "revision";
}

export function buildAgentGraph() {
  const workflow = new StateGraph(AgentGraphAnnotation)
    .addNode("researcher", researcherNode)
    .addNode("writer", writerNode)
    .addNode("reviewer", reviewerNode)
    // Primary execution path
    .addEdge(START, "researcher")
    .addEdge("researcher", "writer")
    .addEdge("writer", "reviewer")
    // Conditional edge after evaluation
    .addConditionalEdges("reviewer", routeReviewOutcome, {
      revision: "writer",
      finalize: END,
    });

  // Attach memory checkpointer for state retention across HTTP requests
  const checkpointer = new MemorySaver();
  
  return workflow.compile({
    checkpointer,
  });
}

4. Потоковая передача API маршрута Next.js с помощью Server-Sent Events

Чтобы обеспечивать мгновенное обновление интерфейса во время работы агентов, мы передаем события состояния в потоковом режиме из Next.js App Router с помощью ReadableStream.

Создайте src/app/api/agent/stream/route.ts:

import { NextRequest, NextResponse } from "next/server";
import { HumanMessage } from "@langchain/core/messages";
import { buildAgentGraph } from "@/lib/agents/graph";

export const runtime = "nodejs";

export async function POST(req: NextRequest) {
  const { prompt, threadId } = await req.json();

  if (!prompt || !threadId) {
    return NextResponse.json({ error: "Missing prompt or threadId" }, { status: 400 });
  }

  const app = buildAgentGraph();
  const encoder = new TextEncoder();

  const stream = new ReadableStream({
    async start(controller) {
      try {
        const eventStream = await app.streamEvents(
          {
            messages: [new HumanMessage(prompt)],
          },
          {
            version: "v2",
            configurable: { thread_id: threadId },
          }
        );

        for await (const event of eventStream) {
          if (event.event === "on_chain_start") {
            controller.enqueue(
              encoder.encode(`data: ${JSON.stringify({ type: "node_start", node: event.name })}\n\n`)
            );
          } else if (event.event === "on_chat_model_stream") {
            const chunk = event.data?.chunk?.content;
            if (chunk) {
              controller.enqueue(
                encoder.encode(`data: ${JSON.stringify({ type: "token", text: chunk })}\n\n`)
              );
            }
          } else if (event.event === "on_chain_end" && event.name === "LangGraph") {
            controller.enqueue(
              encoder.encode(`data: ${JSON.stringify({ type: "done", state: event.data?.output })}\n\n`)
            );
          }
        }
      } catch (err: any) {
        controller.enqueue(
          encoder.encode(`data: ${JSON.stringify({ type: "error", error: err.message })}\n\n`)
        );
      } finally {
        controller.close();
      }
    },
  });

  return new NextResponse(stream, {
    headers: {
      "Content-Type": "text/event-stream",
      "Cache-Control": "no-cache",
      Connection: "keep-alive",
    },
  });
}

5. Компромиссы и архитектурные соображения

Хотя мультиагентные топологии графов обеспечивают четкую структуру и высокую надежность, инженерным командам необходимо учитывать важные эксплуатационные компромиссы:

Аспект Одиночная цепочка LLM Мультиагентный граф
Задержка (Latency) Низкая (Одиночный запрос) Выше (Множественные последовательные / параллельные вызовы)
Стоимость (Cost) Минимальный расход токенов Выше из-за дублирования контекста между узлами
Детерминированность Низкая (Высокая вариативность в сложных инструкциях) Высокая (Строгие переходы состояний и контроль ограничений)
Проверка человеком Трудно корректно приостановить Встроенные контрольные точки (interrupt())

Чек-лист для продакшена:

  1. Повторный вход и сериализация: Убедитесь, что объект состояния канала может сериализоваться в JSON без потери ссылок на символы при сохранении в Redis или PostgreSQL.
  2. Ограничители циклов: Всегда задавайте максимальные лимиты итераций (state.iterationCount >= N) в условных ребрах, чтобы предотвратить бесконечный расход токенов в ошибочных циклах повтора.
  3. Наблюдаемость (Observability): Отслеживайте переходы графа с помощью таких инструментов, как LangSmith или OpenTelemetry, для выявления узких мест задержки на узлах.

Заключение

Мультиагентные архитектуры переводят инженерию LLM от неконтролируемой генерации текста к предсказуемым распределенным системам. Используя LangGraph вместе с Next.js, команды могут создавать самокорректирующиеся графы агентов с сохранением состояния, которые передают прогресс в реальном времени под строгим контролем человека.

Share this article