LangGraph入门实战：AI Agent工作流可视化编排

LangGraph 的必要性

从 Chain 到 Graph：技术演进的内在驱动力

如果接触过之前关于 LangChain 的文章，你对链式调用的模式应该不陌生。它的逻辑很直观——A 传入 B，B 传给 C，一条直线贯穿到底。

但实际业务复杂度攀升后，这种线性结构的局限会迅速暴露。核心痛点集中在三个方面：

LangGraph 正是为解决这三座大山而设计的。

LangGraph 核心定位

通俗地讲，LangGraph 将有向图作为 Agent 工作流的编排引擎。从宏观上看，它究竟解决了哪些痛点？我们可以从以下几个维度进行对比：

实际上，80% 的常规任务用 Chain 即可，但一旦涉及状态机、多轮决策这类逻辑，就必须借助 LangGraph。

LangGraph 核心概念

LangGraph 的底层原理是状态图（StateGraph）——将整个 Agent 的运行抽象为一张图，图中包含节点、边，以及一块在所有节点间流转的“公共黑板”。

┌──────────────────────────────────────────────────────────────┐
│                    LangGraph 核心架构                        │
│                                                              │
│  ┌─────────┐            ┌─────────────────┐                  │
│  │  START   │ ───────────→ │   StateGraph    │                  │
│  └─────────┘            └────────┬────────┘                  │
│                                   │                           │
│               ┌──────────┼──────────┐                        │
│               ▼          ▼          ▼                        │
│          ┌────────┐ ┌───────┐ ┌───────┐                      │
│          │ Nodes  │ │ Edges │ │ State │                      │
│          │ (节点) │ │ (边)  │ │ (状态) │                      │
│          └────────┘ └───────┘ └───────┘                      │
└──────────────────────────────────────────────────────────────┘

1. 状态 (State) - 公共黑板

State 是 LangGraph 中最核心的概念，你可以把它想象成一块在各节点之间共享的“数据容器”。所有节点都能读写它，从而形成动态的上下文。

// 定义状态类型（类似 TypeScript interface）
interface AgentState {
  messages: BaseMessage[];  // 对话历史记录
  userQuery: string;        // 用户输入内容
  toolResults: string[];    // 工具调用返回结果
  currentStep: string;      // 当前执行步骤标识
  retryCount: number;       // 已重试次数
}

2. 节点 (Node) - 执行单元

节点是图中的“动作”。每个节点接收当前状态，执行业务逻辑，然后返回状态的更新。

// 节点函数签名
async function myNode(state: AgentState): Promise> {
  // 1. 读取当前状态
  const { userQuery, messages } = state;
  // 2. 执行具体逻辑
  const result = await doSomething(userQuery);
  // 3. 返回状态更新（LangGraph 自动合并）
  return {
    messages: [...messages, new AIMessage(result)],
    currentStep: "completed"
  };
}

3. 边 (Edge) - 流转规则

边定义了节点之间的流转路径，分为两种类型：

• 普通边：固定跳转，A → B
• 条件边：根据状态动态选择，A → (B 或 C)

// 普通边
graph.addEdge("node_a", "node_b");

// 条件边
graph.addConditionalEdges("agent", (state) => {
  if (state.needsTool) return "tools";
  return END;
});

基础工作流实战

最简单的 LangGraph 图

我们先从最基础的“提问 → 回答”图开始，建立直观感受：

import { StateGraph, MessagesAnnotation, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, AIMessage } from "@langchain/core/messages";
import dotenv from "dotenv";
dotenv.config();

async function basicGraph() {
  // 1. 初始化语言模型
  const model = new ChatOpenAI({
    apiKey: process.env.DASHSCOPE_API_KEY,
    configuration: { baseURL: process.env.DASHSCOPE_API_URL, },
    model: "qwen-plus",
    temperature: 0.7,
  });

  // 2. 定义节点处理函数
  async function chatNode(state: typeof MessagesAnnotation.State) {
    const response = await model.invoke(state.messages);
    return { messages: [response] };
  }

  // 3. 构建状态图
  const workflow = new StateGraph(MessagesAnnotation)
    .addNode("chat", chatNode)  // 注册聊天节点
    .addEdge(START, "chat")     // 设定入口边
    .addEdge("chat", END);      // 设定出口边

  // 4. 编译并调用
  const app = workflow.compile();
  const result = await app.invoke({
    messages: [new HumanMessage("你好，请介绍一下自己")]
  });
  console.log(result.messages[result.messages.length - 1]?.content);
}
basicGraph();

带条件分支的图

接下来我们实现一个“智能客服路由”——根据用户问题类型，路由到不同的处理节点：

import { StateGraph, MessagesAnnotation, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { HumanMessage, BaseMessage } from "@langchain/core/messages";
import dotenv from "dotenv";
dotenv.config();

type SupportState = {
  messages: BaseMessage[];
  intent?: "technical" | "billing" | "general";
  resolution?: string;
};

async function conditionalGraph() {
  const model = new ChatOpenAI({
    apiKey: process.env.DASHSCOPE_API_KEY,
    configuration: { baseURL: process.env.DASHSCOPE_API_URL, },
    model: "qwen-plus",
    temperature: 0.3,
  });

  // 意图识别节点
  async function intentClassifier(state: SupportState) {
    const lastMessage = state.messages[state.messages.length - 1];
    const prompt = `分析用户问题的意图，只返回一个词：technical（技术问题）、billing（计费问题）、general（一般问题）。用户问题：${lastMessage.content}`;
    const response = await model.invoke([new HumanMessage(prompt)]);
    const intent = response.content.toString().toLowerCase().trim();
    return {
      intent: intent as "technical" | "billing" | "general",
      messages: [...state.messages, response]
    };
  }

  // 技术支持节点
  async function technicalHandler(state: SupportState) {
    const userQuery = state.messages[0].content;
    const response = await model.invoke([new HumanMessage(`用户遇到技术问题，请给出专业的技术支持建议。问题：${userQuery}`)]);
    return { messages: [...state.messages, response], resolution: "technical_solved" };
  }

  // 计费处理节点
  async function billingHandler(state: SupportState) {
    const userQuery = state.messages[0].content;
    const response = await model.invoke([new HumanMessage(`用户遇到计费问题，请引导用户联系财务部门。问题：${userQuery}`)]);
    return { messages: [...state.messages, response], resolution: "billing_redirected" };
  }

  // 通用咨询节点
  async function generalHandler(state: SupportState) {
    const userQuery = state.messages[0].content;
    const response = await model.invoke([new HumanMessage(`用户咨询一般问题，请友好回答。问题：${userQuery}`)]);
    return { messages: [...state.messages, response], resolution: "general_answered" };
  }

  // StateGraph 配置
  const workflow = new StateGraph({
    channels: {
      messages: { value: (x: BaseMessage[], y: BaseMessage[]) => [...x, ...y], default: () => [], },
      intent: { value: (_, b) => b },
      resolution: { value: (_, b) => b },
    },
  })
    .addNode("classify", intentClassifier)
    .addNode("technical", technicalHandler)
    .addNode("billing", billingHandler)
    .addNode("general", generalHandler)
    .addEdge(START, "classify")
    .addConditionalEdges("classify", (state) => {
      if (state.intent === "technical") return "technical";
      if (state.intent === "billing") return "billing";
      return "general";
    })
    .addEdge("technical", END)
    .addEdge("billing", END)
    .addEdge("general", END);

  const app = workflow.compile();

  // 测试用例
  const testCases = [
    "我的代码报错了，怎么办？",
    "这个月为什么多扣了钱？",
    "今天天气怎么样？"
  ];

  for (const query of testCases) {
    console.log(`\n? 用户：${query}`);
    const result = await app.invoke({ messages: [new HumanMessage(query)] });
    console.log(`? 意图：${result.intent}`);
    console.log(`? 回复：${result.messages[result.messages.length - 1].content}`);
  }
}
conditionalGraph();

实现 ReAct Agent（核心模式）

这是 LangGraph 最经典的应用——实现带循环的 ReAct Agent。它让 Agent 可以反复思考、调用工具、观察结果，直到任务完成。

import { StateGraph, MessagesAnnotation, START, END } from "@langchain/langgraph";
import { ChatOpenAI } from "@langchain/openai";
import { tool } from "@langchain/core/tools";
import { z } from "zod";
import { HumanMessage, AIMessage, ToolMessage } from "@langchain/core/messages";
import { ToolNode } from "@langchain/langgraph/prebuilt";
import dotenv from "dotenv";
dotenv.config();

async function reactAgent() {
  // 1. 定义工具
  const calculator = tool(async ({ expression }) => {
    const result = Function('"use strict";return (' + expression + ')')();
    return result.toString();
  }, {
    name: "calculator",
    description: "数学计算",
    schema: z.object({ expression: z.string() }),
  });

  const getWeather = tool(async ({ city }) => {
    const weathers: Record = {
      "北京": "晴天，25°C",
      "上海": "多云，28°C",
      "深圳": "阵雨，30°C",
    };
    return weathers[city] || `${city}：晴，22°C`;
  }, {
    name: "get_weather",
    description: "查询天气",
    schema: z.object({ city: z.string() }),
  });

  const tools = [calculator, getWeather];

  // 2. 初始化模型并绑定工具
  const model = new ChatOpenAI({
    apiKey: process.env.DASHSCOPE_API_KEY,
    configuration: { baseURL: process.env.DASHSCOPE_API_URL, },
    model: "qwen-plus",
    temperature: 0.3,
  }).bindTools(tools);

  // 3. 定义节点
  async function agentNode(state: typeof MessagesAnnotation.State) {
    const response = await model.invoke(state.messages);
    return { messages: [response] };
  }

  // 使用预置的 ToolNode
  const toolNode = new ToolNode(tools);

  // 4. 定义条件路由（核心：决定是否继续循环）
  function shouldContinue(state: typeof MessagesAnnotation.State) {
    const lastMessage = state.messages[state.messages.length - 1] as AIMessage;
    if (lastMessage.tool_calls && lastMessage.tool_calls.length > 0) {
      return "tools";
    }
    return END;
  }

  // 5. 构建工作流（形成循环）
  const workflow = new StateGraph(MessagesAnnotation)
    .addNode("agent", agentNode)
    .addNode("tools", toolNode)
    .addEdge(START, "agent")
    .addConditionalEdges("agent", shouldContinue)
    .addEdge("tools", "agent");  // 关键：工具执行完回到 agent

  const app = workflow.compile();

  // 6. 执行
  const userInput = "北京今天天气怎么样？另外帮我算一下 25 * 4 + 10";
  console.log(`? 用户：${userInput}\n`);
  const result = await app.invoke({
    messages: [new HumanMessage(userInput)]
  });
  console.log("? 最终回复：");
  console.log(result.messages[result.messages.length - 1]?.content);
}
reactAgent();

整个工作流的运行逻辑可以参考下面的示意图：

START → agent (思考) → should_continue (条件判断)
                              ↓
                    ┌─────────┴─────────┐
                    ↓                    ↓
              有工具调用            无工具调用
                    ↓                    ↓
               tools (执行)            END
                    ↓
               ┌─────┴─────┐
               ↓            ↓
           回到 agent 继续思考

LangGraph 与传统 Chain 的区别

核心差异对比

为什么说 LangGraph 更适合复杂 Agent？

来看一个真实案例。某电商平台的智能推荐系统，初期因 Chain 设计导致 30% 的推荐结果因上下文断裂而失效。迁移到 LangGraph 之后，问题才彻底解决。根本原因在于三个方面：

1. 循环能力：Agent 可以反复“思考-行动-观察”，直至任务完成
2. 状态持久化：利用 Checkpointer，对话中断后也能恢复现场
3. 条件路由：根据中间结果动态决策，而非硬编码一条路径走到底

前端应用场景展望

表单流程编排

对于复杂表单，LangGraph 能让流程更加清晰：

// 多步骤表单工作流
const formWorkflow = new StateGraph(FormState)
  .addNode("step1", collectPersonalInfo)
  .addNode("step2", collectAddress)
  .addNode("step3", validateForm)
  .addEdge("step1", "step2")
  .addEdge("step2", "step3")
  .addConditionalEdges("step3", (state) => state.isValid ? END : "step1");

AI 对话流程

用户咨询 → 意图识别 → 知识库检索 → 生成回复 → 满意度收集

代码审查流水线

代码提交 → 静态分析 → AI 审查 → 人工审批 → 合并/拒绝

入门常见问题

问题 1：状态类型定义报错

现象：自定义状态时 TypeScript 报错。

解决：确保状态字段有正确的归约器定义。

// ✅ 正确写法
const workflow = new StateGraph({
  channels: {
    messages: { value: (a, b) => [...a, ...b] },  // 追加模式
    count: { value: (_, b) => b },                 // 覆盖模式
  }
});

问题 2：ToolNode 找不到工具

现象：ToolNode 未定义。

解决：确认导入路径正确。

import { ToolNode } from "@langchain/langgraph/prebuilt";

问题 3：循环无法终止

现象：Agent 陷入无限循环。

解决：在条件边中加入最大迭代次数限制。

function shouldContinue(state) {
  if (state.iterationCount > 5) return END;
  // ... 其他逻辑
}

问题 4：状态更新不生效

现象：节点返回的字段没有合并到状态中。

解决：节点必须返回一个包含需要更新字段的对象，而不是直接修改 state。

// ✅ 正确：返回需要更新的字段
return { messages: [response] };

// ❌ 错误：直接修改 state
state.messages.push(response);
return state;

结语

通过本教程，我们基本掌握了 LangGraph 的核心概念和实战用法。其“节点-边-状态”图式结构，为构建复杂 AI Agent 提供了一套标准化、工程化的方案。如有疏漏，欢迎交流指正！

来源：互联网