Torna al Diario
•8 min di lettura

Sviluppo di Workflow Multi-Agente Autonomi con LangGraph e Next.js

Architettura di grafi LLM multi-agente con stato, canali di stato personalizzati, routing condizionale, approvazione umana e streaming in Next.js.

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
Sviluppo di Workflow Multi-Agente Autonomi con LangGraph e Next.js

Quando si sviluppano applicazioni IA in produzione che superano le semplici interfacce di chat, le catene LLM a singolo prompt raggiungono rapidamente i loro limiti architetturali. Compiti complessi di ingegneria del software — come la revisione automatica del codice, la sintesi di ricerche da più fonti o il refactoring autonomo dei sistemi — richiedono la delega tra agenti specializzati, cicli di revisione iterativi e una rigorosa persistenza dello stato.

Le catene lineari (A -> B -> C) falliscono quando l'agente B produce risultati imprecisi o quando il passaggio C richiede la verifica umana prima di modificare database in produzione. Per costruire reti di agenti resilienti, i sistemi devono supportare grafi di stato ciclici, routing condizionale e checkpointing deterministico.

In questa guida implementeremo un'architettura di grafo multi-agente con stato utilizzando LangGraph (TypeScript) e la integreremo in un'applicazione Next.js App Router con streaming SSE in tempo reale.

L'Architettura: Grafi Multi-Agente con Stato

Invece di forzare un singolo LLM a ricoprire ogni ruolo, un sistema multi-agente separa le responsabilità in nodi distinti collegati da un contesto di stato condiviso.

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

Primitivi Fondamentali in LangGraph:

  1. Canali di Stato (State Channels): Memoria condivisa fortemente tipizzata trasmessa tra i nodi.
  2. Nodi (Nodes): Funzioni TypeScript isolate che rappresentano singoli agenti o esecutori di strumenti.
  3. Archi (Edges): Connessioni dirette che trasferiscono l'esecuzione tra i nodi.
  4. Archi Condizionali (Conditional Edges): Logica di routing dinamica basata su criteri di valutazione o soglie di errore.
  5. Checkpointer: Archivi di stato persistenti che consentono di mettere in pausa, riprendere e interrompere per l'intervento umano.

1. Definizione di Canali di Stato Tipizzati

In LangGraph lo stato viene definito tramite canali annotati. Ogni nodo riceve lo stato corrente del grafo e restituisce un aggiornamento parziale dello stato. Le funzioni di riduzione determinano come gli aggiornamenti si uniscono allo stato esistente.

Crea 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. Implementazione dei Nodi Agente Specializzati

I nodi sono funzioni asincrone che eseguono logica di dominio specifica. Qui definiamo tre nodi: Researcher, Writer e Reviewer.

Crea 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. Costruzione dello StateGraph con Archi Condizionali

Colleghiamo i nodi in un grafo ciclico utilizzando una funzione di arco condizionale per decidere se reindirizzare al Writer per la revisione o completare l'esecuzione.

Crea 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. Streaming delle API Route in Next.js con Server-Sent Events

Per offrire aggiornamenti immediati dell'interfaccia durante l'esecuzione degli agenti, inviamo gli eventi di stato in streaming dall'App Router di Next.js tramite ReadableStream.

Crea 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. Compromessi e Considerazioni Architetturali

Sebbene le topologie di grafi multi-agente offrano struttura e resilienza, i team di ingegneria devono considerare importanti compromessi operativi:

Aspetto Catena LLM Singola Grafo Multi-Agente
Latenza Bassa (Singola richiesta) Maggiore (Molteplici chiamate sequenziali / parallele)
Costo Consumo minimo di token Maggiore a causa della duplicazione del contesto tra i nodi
Determinisco Basso (Elevata varianza su istruzioni complesse) Elevato (Transizioni di stato rigorose e controllo dei limiti)
Intervento Umano Difficile da sospendere in modo pulito Checkpointer integrati (interrupt())

Lista di Controllo per la Produzione:

  1. Reentrancy e Serializzazione: Assicurati che l'oggetto di stato del canale possa essere serializzato in JSON senza perdita di riferimenti durante il salvataggio su Redis o PostgreSQL.
  2. Interruttori per i Cicli: Imposta sempre limiti massimi di iterazione (state.iterationCount >= N) negli archi condizionali per evitare il consumo illimitato di token in caso di cicli di revisione non riusciti.
  3. Osservabilità: Traccia le transizioni del grafo con strumenti come LangSmith o OpenTelemetry per individuare i colli di bottiglia della latenza nei nodi.

Conclusione

Le architetture multi-agente evolvono l'ingegneria dei modelli LLM dalla generazione di testo non vincolata a sistemi distribuiti prevedibili. Sfruttando LangGraph con Next.js, i team possono costruire grafi di agenti con stato in grado di correggersi autonomamente, trasmettendo i progressi in tempo reale e mantenendo al contempo una rigorosa supervisione umana.

Share this article