Zurück zum Journal
•7 Min. Lesezeit

Erstellung Autonomer Multi-Agenten-Workflows mit LangGraph und Next.js

Architektur zustandsbehafteter Multi-Agenten-LLM-Graphen mit benutzerdefinierten Zustandskanälen, bedingtem Routing, Human-in-the-Loop-Freigabe und Streaming-Antworten in Next.js.

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
Erstellung Autonomer Multi-Agenten-Workflows mit LangGraph und Next.js

Beim Bau von KI-Produktionsanwendungen, die über einfache Chat-Schnittstellen hinausgehen, stoßen einfache Einzel-Prompt-LLM-Ketten schnell an ihre architektonischen Grenzen. Komplexe Engineering-Aufgaben – wie automatisierte Code-Reviews, recherchebasierte Synthesen aus mehreren Quellen oder autonomes System-Refactoring – erfordern die Delegierung an spezialisierte Agenten, iterative Überarbeitungsschleifen und strikte Zustandspersistenz.

Lineare Ketten (A -> B -> C) scheitern, wenn Agent B ungenaue Ergebnisse liefert oder wenn Schritt C eine menschliche Überprüfung erfordert, bevor Produktionsdatenbanken verändert werden. Um widerstandsfähige Agentennetzwerke aufzubauen, müssen Systeme zyklische Zustandsgraphen, bedingtes Routing und deterministisches Checkpointing unterstützen.

In diesem Leitfaden implementieren wir eine zustandsbehaftete Multi-Agenten-Graph-Architektur mit LangGraph (TypeScript) und integrieren sie in eine Next.js App Router-Anwendung mit Echtzeit-SSE-Streaming.

Die Architektur: Zustandsbehaftete Multi-Agenten-Graphen

Anstatt ein einzelnes LLM zu zwingen, jede Rolle zu übernehmen, entkoppelt ein Multi-Agenten-System die Verantwortlichkeiten in separate Knoten, die über einen gemeinsamen Zustandskontext verbunden sind.

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

Kern-Primitiven in LangGraph:

  1. Zustandskanäle (State Channels): Typsicherer gemeinsamer Speicher, der zwischen Knoten übergeben wird.
  2. Knoten (Nodes): Isolierte TypeScript-Funktionen, die einzelne Agenten oder Werkzeugausführer repräsentieren.
  3. Kanten (Edges): Gerichtete Verbindungen, die die Ausführung zwischen Knoten weiterleiten.
  4. Bedingte Kanten (Conditional Edges): Dynamische Routing-Logik basierend auf Evaluierungskriterien oder Fehlerschwellen.
  5. Checkpointer: Persistente Zustands-Speicher, die das Anhalten/Fortsetzen und Human-in-the-Loop-Unterbrechungen ermöglichen.

1. Definition Typsicherer Zustandskanäle

In LangGraph wird der Zustand über annotierte Kanäle definiert. Jeder Knoten empfängt den aktuellen Graphzustand und gibt ein partielles Zustands-Update zurück. Reduzierfunktionen bestimmen, wie Updates mit dem bestehenden Zustand zusammengeführt werden.

Erstellen Sie 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. Implementierung Spezialisierter Agenten-Knoten

Knoten sind asynchrone Funktionen, die domänenspezifische Logik ausführen. Hier definieren wir drei Knoten: Researcher, Writer und Reviewer.

Erstellen Sie 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. Aufbau des StateGraph mit Bedingten Kanten

Wir verbinden die Knoten zu einem zyklischen Graphen und verwenden eine bedingte Kantenfunktion, um zu entscheiden, ob zur Überarbeitung an den Writer zurückgeroutet oder der Vorgang abgeschlossen wird.

Erstellen Sie 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-Route Streaming mit Server-Sent Events

Um sofortige UI-Updates während der Agentenausführung zu ermöglichen, streamen wir Zustandsereignisse vom Next.js App Router über ReadableStream.

Erstellen Sie 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. Kompromisse und Architektonische Überlegungen

Während Multi-Agenten-Graphentopologien Struktur und Belastbarkeit bieten, müssen Engineering-Teams wichtige betriebliche Kompromisse berücksichtigen:

Aspekt Einzel-LLM-Kette Multi-Agenten-Graph
Latenz Niedrig (Einzelner Aufruf) Höher (Mehrere sequenzielle / parallele Aufrufe)
Kosten Minimaler Token-Verbrauch Höher durch Kontextduplizierung über Knoten hinweg
Determinisierung Niedrig (Hohe Varianz bei mehrschrittigen Anweisungen) Hoch (Strikte Zustandsübergänge und Grenzprüfungen)
Human-in-the-Loop Schwer sauber anzuhalten Integrierte Checkpointer (interrupt())

Checkliste für die Produktion:

  1. Re-Entranz & Serialisierung: Stellen Sie sicher, dass das Kanalzustandsobjekt ohne Verlust von Symbolreferenzen in JSON serialisiert werden kann, wenn es in Redis- oder PostgreSQL-Checkpointern gespeichert wird.
  2. Schleifen-Schutzschalter: Erzwingen Sie immer maximale Iterationszahlen (state.iterationCount >= N) in bedingten Kantenfunktionen, um einen unbegrenzten Token-Verbrauch bei fehlgeschlagenen Überarbeitungsschleifen zu vermeiden.
  3. Observability: Verfolgen Sie Graphübergänge mit Tools wie LangSmith oder OpenTelemetry, um Latenzengpässe an den Knoten zu erkennen.

Fazit

Multi-Agenten-Architekturen verlagern das LLM-Engineering von uneingeschränkter Textgenerierung hin zu vorhersagbaren verteilten Systemen. Durch die Nutzung von LangGraph mit Next.js können Teams zustandsbehaftete, selbstkorrigierende Agentengraphen aufbauen, die den Fortschritt in Echtzeit streamen und gleichzeitig eine strikte menschliche Steuerung aufrechterhalten.

Share this article