返回日志
•6 分钟阅读

使用 LangGraph 和 Next.js 构建自主多智能体工作流

具有状态保持、自定义状态通道、条件路由、人工审批和 Next.js 实时流式响应的多智能体 LLM 图架构。

LangGraphNext.jsTypeScriptMulti-Agent SystemsAI Architecture
使用 LangGraph 和 Next.js 构建自主多智能体工作流

在构建超越简单聊天界面的生产级 AI 应用时，单 Prompt 的 LLM 链很快就会遇到架构瓶颈。复杂的工程任务（例如自动化代码审查、多源研究合成或自主系统重构）需要将任务分发给专门的智能体、进行迭代修改循环以及严格的状态持久化。

当智能体 B 生成不准确的输出，或者步骤 C 在修改生产数据库前需要人工验证时，线性链（A -> B -> C）就会失效。为了构建具有韧性的智能体网络，系统必须支持循环状态图、条件路由和确定性检查点（Checkpointing）。

在指南中，我们将使用 LangGraph (TypeScript) 实现一个带状态的多智能体图架构，并将其集成到支持实时 SSE 流式传输的 Next.js App Router 应用中。

架构：带状态的多智能体图

多智能体系统没有强制单个 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 中，状态是通过带注解的通道定义的。每个节点接收当前图状态并返回局部状态更新。Reducer 函数决定更新如何与现有状态合并。

创建 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. Next.js API 路由与 Server-Sent Events 流式传输

为了在智能体执行期间提供即时 UI 更新，我们使用 ReadableStream 从 Next.js App Router 流式传输状态事件。

创建 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) 最小 Token 消耗 较高（因节点间上下文重复）
确定性 (Determinism) 低（复杂指令下变异大） 高（严格的状态转移与边界控制）
人工干预 难以干净利落地暂停 内置检查点（interrupt()）

生产工程检查清单：

  1. 重入性与序列化：确保通道状态对象在保存到 Redis 或 PostgreSQL 时可序列化为 JSON 且不会丢失符号引用。
  2. 循环熔断器：务必在条件边函数中设置最大迭代限制（state.iterationCount >= N），防止失败修改循环导致无休止消耗 Token。
  3. 可观测性 (Observability)：使用 LangSmith 或 OpenTelemetry 追踪图状态转移，以便识别节点的延迟瓶颈。

总结

多智能体架构将 LLM 工程从无约束的文本生成转变为可预测的分布式系统。通过结合 LangGraph 和 Next.js，团队能够构建具备状态保持、自愈能力且能在严密人工监督下实时流式传输进度的智能体图系统。

Share this article