Atgal į žurnalą
•7 min. skaitymo

Autonominių Daugiaagentių Darbo Eigų Kūrimas Su LangGraph Ir Next.js

Būsenos turinčių daugiaagentių DI grafų architektūra su pritaikytais būsenos kanalais, sąlyginiu nukreipimu, žmogaus patvirtinimu ir srautiniu atsaku Next.js karkase.

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
Autonominių Daugiaagentių Darbo Eigų Kūrimas Su LangGraph Ir Next.js

Kuriant gamybines AI programas, viršijančias paprastas pokalbių sąsajas, vienos užklausos LLM grandinės greitai pasiekia architektūrines ribas. Sudėtingoms inžinerinėms užduotims—pavyzdžiui, automatizuotam kodo peržiūrėjimui, kelių šaltinių tyrimų sintezei ar autonominiam sistemų refaktorizavimui—reikia užduočių delegavimo specializuotiems agentams, iteracinių peržiūros ciklų ir griežto būsenos išsaugojimo.

Tiesinės grandinės (A -> B -> C) nepavyksta, kai agentas B pateikia netikslų rezultatą arba kai žingsniui C reikia žmogaus patvirtinimo prieš modifikuojant gamybines duomenų bazes. Norint sukurti atsparius agentų tinklus, sistemos turi palaikyti ciklinius būsenos grafus, sąlyginį nukreipimą ir deterministinį būsenos išsaugojimą.

Šiame vadove įgyvendinsime būseną palaikančią daugiaagentę grafo architektūrą, naudodami LangGraph (TypeScript), ir integruosime ją į Next.js App Router programą su tiesioginiu SSE srautu.

Architektūra: Būseną Palaikantys Daugiaagentiniai Grafai

Užuot vertus vieną LLM atlikti kiekvieną vaidmenį, daugiaagentė sistema padalija atsakomybę į atskirus mazgus, sujungtus bendru būsenos kontekstu.

                  +-----------------------+
                  |     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    |
      +-------------------+    +-------------------+

Pagrindiniai LangGraph Primityvai:

  1. Būsenos Kanalai (State Channels): Tipais saugi bendra atmintis, perduodama tarp mazgų.
  2. Mazgai (Nodes): Atskiros TypeScript funkcijos, reprezentuojančios agentus arba įrankių vykdytojus.
  3. Kraštai (Edges): Nukreiptos jungtys, perduodančios vykdymą tarp mazgų.
  4. Sąlyginiai Kraštai (Conditional Edges): Dinaminė nukreipimo logika, pagrįsta įvertinimo kriterijais arba klaidų ribomis.
  5. Tikrinimo Taškai (Checkpointers): Išsaugota būsena, įgalinanti pristabdymą / atnaujinimą ir žmogaus įsikišimą.

1. Tipais Saugių Būsenos Kanalų Apibrėžimas

LangGraph karkase būsena apibrėžiama naudojant anotuotus kanalus. Kiekvienas mazgas gauna esamą grafo būseną ir grąžina dalinį būsenos atnaujinimą. Redukcijos funkcijos apibrėžia, kaip atnaujinimai sujungiami su esama būsena.

Sukurkite 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. Specializuotų Agentų Mazgų Įgyvendinimas

Mazgai yra asinchroninės funkcijos, vykdančios specifinę logiką. Čia apibrėžiame tris mazgus: Researcher, Writer ir Reviewer.

Sukurkite 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 Kūrimas Su Sąlyginiais Kraštais

Sujungiame mazgus į ciklinį grafą, naudodami sąlyginio krašto funkciją, kad nuspręstume, ar grįžti pas Writer taisymams, ar užbaigti vykdymą.

Sukurkite 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. Next.js API Maršruto Srautas Su Server-Sent Events

Norėdami užtikrinti momentinius sąsajos atnaujinimus agentų vykdymo metu, transliuojame būsenos įvykius iš Next.js App Router naudodami ReadableStream.

Sukurkite 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. Kompromisai Ir Architektūriniai Aspektai

Nors daugiaagentės grafų topologijos suteikia struktūrą ir atsparumą, inžinerinės komandos turi atsižvelgti į esminius operacinius kompromisus:

Aspektas Vienos LLM Grandinė Daugiaagentis Grafas
Vėlavimas (Latency) Mažas (Vienas užklausos ciklas) Didesnis (Keli nuoseklūs / lygiagretūs šaukimai)
Kaina (Cost) Minimalus žetonų panaudojimas Didesnė dėl konteksto dubliavimo tarp mazgų
Determiniškumas Mažas (Didelė paklaida daugiažingsnėse instrukcijose) Didelis (Griežti būsenų perėjimai ir ribų patikrinimai)
Žmogaus Įsitraukimas Sudėtinga tvarkingai sustabdyti būseną Integruoti būsenos išsaugojimo taškai (interrupt())

Gamybinis Inžinerinis Patikros Sąrašas:

  1. Pakartotinis Įėjimas Ir Serijavimas: Užtikrinkite, kad kanalo būsenos objektas gali būti serijuojamas į JSON be simbolių praradimo, kai išsaugoma Redis arba PostgreSQL duomenų bazėse.
  2. Iteracijų Saugikliai: Visada nustatykite maksimalius kartojimosi limitus (state.iterationCount >= N) sąlyginėse kraštų funkcijose, kad išvengtumėte begalinio žetonų naudojimo klaidų ciklų metu.
  3. Stebimumas (Observability): Stebėkite grafo perėjimus naudodami įrankius, tokius kaip LangSmith arba OpenTelemetry, kad nustatytumėte mazgų vėlavimo kliūtis.

Išvada

Daugiaagentės architektūros perkelia LLM inžineriją iš neapriboto teksto generavimo į nuspėjamas paskirstytas sistemas. Naudodamos LangGraph su Next.js, komandos gali kurti būseną palaikančius, savarankiškai pasitaisančius agentų grafus, kurie tiesiogiai transliuoja progresą, išlaikydami griežtą žmogaus priežiūrą.

Share this article