Назад до журналу
•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