Volver al Diario
•8 min de lectura

Construcción de Flujos de Trabajo Multi-Agente Autónomos con LangGraph y Next.js

Arquitectura de grafos de LLM multi-agente con estado, canales de estado personalizados, enrutamiento condicional, aprobación humana y respuestas en tiempo real en Next.js.

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
Construcción de Flujos de Trabajo Multi-Agente Autónomos con LangGraph y Next.js

Al construir aplicaciones de IA en producción más allá de simples interfaces de chat, las cadenas de LLM de un solo prompt alcanzan rápidamente sus límites arquitectónicos. Las tareas de ingeniería complejas, como las revisiones automáticas de código, la síntesis de investigación de múltiples fuentes o la reestructuración autónoma de sistemas, requieren delegación entre agentes especializados, bucles de revisión iterativos y persistencia estricta del estado.

Las cadenas lineales (A -> B -> C) fallan cuando el agente B produce resultados inexactos o cuando el paso C requiere verificación humana antes de modificar bases de datos de producción. Para construir redes de agentes resilientes, los sistemas deben admitir grafos de estado cíclicos, enrutamiento condicional y puntos de control deterministas.

En esta guía, implementaremos una arquitectura de grafo multi-agente con estado utilizando LangGraph (TypeScript) y la integraremos en una aplicación Next.js App Router con transmisión SSE en tiempo real.

La Arquitectura: Grafos Multi-Agente con Estado

En lugar de forzar a un solo LLM a asumir cada rol, un sistema multi-agente desacopla las responsabilidades en nodos individuales conectados por un contexto de estado compartido.

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

Primitivas Principales en LangGraph:

  1. Canales de Estado (State Channels): Memoria compartida con tipos seguros transmitida entre nodos.
  2. Nodos (Nodes): Funciones de TypeScript aisladas que representan agentes individuales o ejecutores de herramientas.
  3. Aristas (Edges): Conexiones dirigidas que transfieren la ejecución entre nodos.
  4. Aristas Condicionales (Conditional Edges): Lógica de enrutamiento dinámico basada en criterios de evaluación o umbrales de error.
  5. Puntos de Control (Checkpointers): Almacenamiento de estado persistente que permite pausar, reanudar e interrumpir para intervención humana.

1. Definición de Canales de Estado Seguros por Tipo

En LangGraph, el estado se define mediante canales anotados. Cada nodo recibe el estado actual del grafo y devuelve una actualización parcial. Las funciones reductoras determinan cómo se combinan las actualizaciones con el estado existente.

Cree 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. Implementación de Nodos de Agente Especializados

Los nodos son funciones asíncronas que ejecutan lógica de dominio específico. Aquí definimos tres nodos: Researcher, Writer y Reviewer.

Cree 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. Construcción del StateGraph con Aristas Condicionales

Conectamos los nodos en un grafo cíclico, utilizando una función de arista condicional para decidir si regresar al Writer para revisiones o finalizar la ejecución.

Cree 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. Transmisión en Rutas de API de Next.js con Server-Sent Events

Para proporcionar actualizaciones inmediatas en la interfaz durante la ejecución de los agentes, transmitimos eventos de estado desde el App Router de Next.js usando ReadableStream.

Cree 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. Compromisos y Consideraciones Arquitectónicas

Aunque las topologías de grafos multi-agente proporcionan estructura y flexibilidad, los equipos de ingeniería deben considerar importantes compromisos operativos:

Aspecto Cadena de LLM Único Grafo Multi-Agente
Latencia Baja (Una sola solicitud) Mayor (Múltiples llamadas secuenciales o paralelas)
Costo Consumo mínimo de tokens Mayor debido a la duplicación de contexto entre nodos
Determinismo Bajo (Alta varianza en instrucciones complejas) Alto (Transiciones de estado estrictas y control de límites)
Aprobación Humana Difícil de pausar limpiamente Puntos de control integrados (interrupt())

Lista de Verificación para Producción:

  1. Re-entrada y Serialización: Asegúrese de que el objeto de estado del canal pueda serializarse a JSON sin perder referencias al guardarlo en Redis o PostgreSQL.
  2. Interruptores de Bucles: Defina siempre límites máximos de iteración (state.iterationCount >= N) en aristas condicionales para evitar un consumo ilimitado de tokens en bucles de revisión fallidos.
  3. Observabilidad: Realice un seguimiento de las transiciones del grafo con herramientas como LangSmith u OpenTelemetry para detectar cuellos de botella de latencia.

Conclusión

Las arquitecturas multi-agente transforman la ingeniería de LLM de una generación de texto no restringida a sistemas distribuidos predecibles. Al combinar LangGraph con Next.js, los equipos pueden desarrollar grafos de agentes con estado y capacidad de autocorregirse que transmiten avances en tiempo real manteniendo un control humano riguroso.

Share this article