LangGraph 实战指南：由单 Agent 走向多 Agent 工作流编排

当智能体需要同时处理搜索、计算、代码执行等多类任务时，把全部工具和决策逻辑集中在一个 Agent 中，往往会带来上下文冗余、执行串行和维护困难。LangGraph 通过图结构组织状态与任务节点，为条件分支、流程恢复和多 Agent 协作提供了统一的编排方式。下面将结合代码逐步理解这些关键机制。

基于实际项目代码，深入理解 LangGraph 的核心概念与实战应用

前言

在构建复杂的 AI Agent 应用时，单 Agent 架构往往面临以下挑战：

• Token 消耗高：所有工具描述和 Prompt 都放在 system prompt 中，每次调用都携带大量无关信息
• 决策效率低：单个 LLM 需要处理所有任务，容易出错
• 无法并行：所有任务串行执行，整体效率受限

LangGraph 是 LangChain 团队推出的工作流编排框架，专门用于构建多 Agent 协作系统。它支持：

• ✅ 网状工作流：支持分支、循环、条件路由
• ✅ 状态管理：持久化状态，支持中断和恢复
• ✅ 多 Agent 协作：主 Agent 分发任务，子 Agent 并行处理
• ✅ 可视化：自动生成 Mermaid 流程图

---

一、为什么需要多 Agent？

1.1 单 Agent 的局限

单 Agent 架构：
┌─────────────────────────────┐
│  Agent                      │
│  ├─ LLM (大脑)              │
│  ├─ Tool 1: 搜索            │
│  ├─ Tool 2: 计算            │
│  ├─ Tool 3: 代码执行        │
│  └─ Tool 4: 数据库查询      │
└─────────────────────────────┘

问题：
- 所有 Tool 描述都在 Prompt 中，Token 消耗高
- 执行搜索时，计算和代码的描述是干扰信息
- 无法并行处理多个任务

1.2 多 Agent 的优势

多 Agent 架构：
┌──────────────┐
│  主 Agent    │  ← 任务分发
└──────┬───────┘
       │
  ┌────┴────┬────────┐
  ↓         ↓        ↓
┌─────┐  ┌─────┐  ┌─────┐
│搜索 │  │计算 │  │代码 │  ← 子 Agent 并行处理
│Agent│  │Agent│  │Agent│
└─────┘  └─────┘  └─────┘

优势：
✅ 每个 Agent 只保留必要的 Prompt，Token 消耗更低
✅ 多个 Agent 并行思考，整体效率更高
✅ 多角色互相讨论，纠错能力更强

1.3 Agent 的本质

Agent = LLM (大脑) + Harness (工具集)

Harness 包括：
- Tool: 外部工具调用
- MCP: Model Context Protocol
- RAG: 检索增强生成
- Skill: 技能模块
- Memory: 记忆管理

---

二、LangGraph 核心概念

2.1 从 LangChain 到 LangGraph

特性	LangChain	LangGraph
工作流类型	线性工作流	网状工作流
编排方式	Chain（链式）	Graph（图状）
支持分支	❌ 不支持	✅ 支持
支持循环	❌ 不支持	✅ 支持
状态管理	简单	强大（持久化）
适用场景	简单流程	复杂多 Agent

2.2 核心组件

LangGraph 工作流由以下组件构成：

1. State (状态)
   - 工作流的数据载体
   - 在节点间传递和更新

2. Node (节点)
   - 工作单元
   - 接收状态，处理逻辑，返回新状态

3. Edge (边)
   - 连接节点
   - 定义执行顺序

4. Conditional Edge (条件边)
   - 根据状态动态选择下一个节点
   - 实现分支逻辑

5. START / END
   - 特殊节点
   - 标记工作流的开始和结束

---

三、基础图：线性工作流

3.1 完整代码示例

// basic-graph.mjs
import {
    Annotation,    // 状态字段声明
    END,           // 结束节点
    START,         // 开始节点
    StateGraph     // 状态图编排器
} from '@langchain/langgraph';

// 1. 定义状态结构
const StateAnnotation = Annotation.Root({
  text: Annotation({
    reducer: (_prev, next) => next,  // 状态更新策略：直接覆盖
    default: () => "",                // 默认值
  })
});

// 2. 定义节点函数
const step1 = (state) => ({ text: `${state.text}->step1` });
const step2 = (state) => ({ text: `${state.text}->step2` });

// 3. 构建工作流图
const graph = new StateGraph(StateAnnotation)
  .addNode("step1", step1)           // 添加节点
  .addNode("step2", step2)
  .addEdge(START, "step1")           // 连接边
  .addEdge("step1", "step2")
  .addEdge("step2", END)
  .compile();                        // 编译工作流

// 4. 可视化流程图
const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);

// 5. 执行工作流
const result = await graph.invoke({ text: "hello" });
console.log(result);
// 输出: { text: "hello->step1->step2" }

3.2 执行流程

输入: { text: "hello" }
  ↓
START
  ↓
step1: text = "hello->step1"
  ↓
step2: text = "hello->step1->step2"
  ↓
END

最终状态: { text: "hello->step1->step2" }

3.3 关键概念解析

State Annotation（状态声明）

const StateAnnotation = Annotation.Root({
  text: Annotation({
    reducer: (_prev, next) => next,  // 状态合并策略
    default: () => "",                // 初始值
  })
});

• reducer: 决定如何合并状态

  • (_prev, next) => next: 新值直接覆盖旧值
  • (_prev, next) => _prev + next: 追加拼接
  • (prev, next) => [...prev, ...next]: 数组合并

• default: 状态的初始值

Node（节点）

const step1 = (state) => ({ text: `${state.text}->step1` });

• 节点是一个函数
• 接收当前状态作为参数
• 返回新状态（部分更新）

Edge（边）

graph
  .addEdge(START, "step1")      // 固定边：START → step1
  .addEdge("step1", "step2")    // 固定边：step1 → step2
  .addEdge("step2", END);       // 固定边：step2 → END

• START 和 END 是特殊标记
• addEdge 定义固定的执行顺序

---

四、条件路由：分支逻辑

4.1 完整代码示例

// conditional-routing.mjs
import {
    Annotation,
    END,
    START,
    StateGraph
} from '@langchain/langgraph';

// 1. 定义状态结构
const StateAnnotation = Annotation.Root({
    query: Annotation({
        reducer: (_prev, next) => next,
        default: () => ""
    }),
    route: Annotation({
        reducer: (_prev, next) => next,
        default: () => ""
    }),
    answer: Annotation({
        reducer: (_prev, next) => next,
        default: () => ""
    })
});

// 2. 路由节点：判断走向
const router = (state) => {
    const isMath = /[+-*]/.test(state.query);
    return {
        route: isMath ? "math" : "chat"
    };
};

// 3. 数学计算节点
const mathNode = (state) => {
    try {
        return { answer: String(eval(state.query)) };
    } catch {
        return { answer: "表达式无法计算" };
    }
};

// 4. 聊天节点
const chatNode = (state) => ({
    answer: `你说的是: ${state.query}`
});

// 5. 构建工作流
const graph = new StateGraph(StateAnnotation)
    .addNode("router", router)
    .addNode("math", mathNode)
    .addNode("chat", chatNode)
    .addEdge(START, "router")
    // 条件边：根据 route 字段动态选择
    .addConditionalEdges("router", (state) => state.route, {
        math: "math",
        chat: "chat"
    })
    .addEdge("math", END)
    .addEdge("chat", END)
    .compile();

// 6. 执行测试
console.log("result:", await graph.invoke({ query: "你好" }));
// 输出: { query: "你好", route: "chat", answer: "你说的是: 你好" }

console.log("result:", await graph.invoke({ query: "1+2" }));
// 输出: { query: "1+2", route: "math", answer: "3" }

4.2 执行流程

输入: { query: "1+2" }
  ↓
START
  ↓
router: 检测是否包含 +-* 符号
  ↓
  判断: isMath = true
  设置: route = "math"
  ↓
条件边: 根据 route 选择下一个节点
  ↓
math: eval("1+2") = 3
  设置: answer = "3"
  ↓
END

最终状态: { query: "1+2", route: "math", answer: "3" }

4.3 条件边详解

.addConditionalEdges("router", (state) => state.route, {
    math: "math",    // route === "math" → 执行 math 节点
    chat: "chat"     // route === "chat" → 执行 chat 节点
})

• 第一个参数: 源节点名称
• 第二个参数: 路由函数，返回下一个节点的标识
• 第三个参数: 映射表，将标识映射到实际节点名称

4.4 应用场景

条件路由适用于：

• 意图识别：根据用户输入选择不同的处理逻辑
• 错误处理：根据执行结果决定重试或终止
• 多路径分支：A/B 测试、灰度发布
• 动态编排：根据中间结果动态调整工作流

---

五、状态持久化：MemorySaver

5.1 为什么需要持久化？

问题场景：
1. Agent 执行到一半，需要用户授权
2. 工作流失败，需要从中断点恢复
3. 多轮对话，需要保持上下文

解决方案：
使用 MemorySaver 保存状态，支持：
- 中断和恢复
- 多会话隔离
- 状态回溯

5.2 完整代码示例

// checkpoint-memory.mjs
import {
    Annotation,
    END,
    START,
    MemorySaver,
    StateGraph
} from '@langchain/langgraph';

// 1. 定义状态结构
const StateAnnotation = Annotation.Root({
    visitCount: Annotation({
        reducer: (_prev, next) => next,
        default: () => 0
    }),
    message: Annotation({
        reducer: (_prev, next) => next,
        default: () => ""
    })
});

// 2. 计数节点
function recordVisit(state) {
    const visitCount = state.visitCount + 1;
    const message = visitCount === 1 
        ? "这是第一次访问" 
        : `这是第${visitCount}次访问`;
    return { visitCount, message };
}

// 3. 构建工作流
const graph = new StateGraph(StateAnnotation)
    .addNode("recordVisit", recordVisit)
    .addEdge(START, "recordVisit")
    .addEdge("recordVisit", END);

// 4. 添加 MemorySaver
const checkpoint = new MemorySaver();
const app = graph.compile({ checkpointer: checkpoint });

// 5. 多会话测试
const user1Options = {
    configurable: { thread_id: "用户小张" }
};

const res1 = await app.invoke({}, user1Options);
console.log(res1);
// 输出: { visitCount: 1, message: "这是第一次访问" }

const res1_2 = await app.invoke({}, user1Options);
console.log(res1_2);
// 输出: { visitCount: 2, message: "这是第2次访问" }

const user2Options = {
    configurable: { thread_id: "用户小李" }
};

const res2 = await app.invoke({}, user2Options);
console.log(res2);
// 输出: { visitCount: 1, message: "这是第一次访问" }

5.3 执行流程

用户小张的第一次访问:
  thread_id: "用户小张"
  ↓
START
  ↓
recordVisit: visitCount = 0 + 1 = 1
  ↓
END
  ↓
MemorySaver 保存状态: { visitCount: 1 }

用户小张的第二次访问:
  thread_id: "用户小张"
  ↓
START
  ↓
recordVisit: visitCount = 1 + 1 = 2  ← 基于上次状态
  ↓
END
  ↓
MemorySaver 保存状态: { visitCount: 2 }

用户小李的第一次访问:
  thread_id: "用户小李"  ← 不同的 thread_id
  ↓
START
  ↓
recordVisit: visitCount = 0 + 1 = 1  ← 独立的状态
  ↓
END

5.4 关键概念

thread_id（会话 ID）

const options = {
    configurable: { thread_id: "用户小张" }
};

• 每个 thread_id 对应一个独立的状态空间
• 不同用户、不同对话使用不同的 thread_id
• 相同 thread_id 的多次调用会共享状态

MemorySaver（内存保存器）

const checkpoint = new MemorySaver();
const app = graph.compile({ checkpointer: checkpoint });

• 将状态保存到内存
• 支持多会话隔离
• 进程重启后状态丢失

持久化方案

内存保存: MemorySaver
  - 优点: 简单快速
  - 缺点: 进程重启后丢失

数据库保存: 
  - SQLite: 适合单机应用
  - Redis: 适合分布式系统
  - PostgreSQL: 适合生产环境

---

六、中断与恢复：Human-in-the-Loop

6.1 应用场景

场景 1: 需要用户授权
Agent 准备删除文件 → 暂停 → 询问用户 → 继续执行

场景 2: 需要人工审核
Agent 生成代码 → 暂停 → 人工审核 → 继续执行

场景 3: 需要外部输入
Agent 执行到一半 → 暂停 → 等待用户输入 → 继续执行

6.2 实现原理

工作流执行流程:
1. 执行到某个节点
2. 节点返回 interrupt 信号
3. MemorySaver 保存当前状态
4. 工作流暂停
5. 用户处理中断（授权、输入等）
6. 恢复工作流，从中断点继续

6.3 代码示例

import { interrupt } from '@langchain/langgraph';

// 需要用户授权的节点
const deleteFileNode = async (state) => {
    const filePath = state.filePath;
    
    // 中断，等待用户授权
    const userConfirmed = await interrupt({
        message: `确认删除文件 ${filePath}?`,
        type: "confirm"
    });
    
    if (!userConfirmed) {
        return { status: "cancelled" };
    }
    
    // 执行删除操作
    await deleteFile(filePath);
    return { status: "deleted" };
};

---

七、可视化：Mermaid 流程图

7.1 自动生成流程图

const drawable = await graph.getGraphAsync();
const mermaid = drawable.drawMermaid({ withStyles: true });
console.log(mermaid);

7.2 输出示例

graph TD
    START --> router
    router -->|math| math
    router -->|chat| chat
    math --> END
    chat --> END
    
    style START fill:#f9f,stroke:#333
    style END fill:#bbf,stroke:#333
    style router fill:#dfd,stroke:#333
    style math fill:#fdd,stroke:#333
    style chat fill:#ddf,stroke:#333

7.3 可视化优势

• ✅ 直观展示工作流结构
• ✅ 便于调试和理解
• ✅ 支持文档生成
• ✅ 可用于团队沟通

---

八、实战案例：多 Agent 协作系统

8.1 场景描述

构建一个智能客服系统，包含：

• 主 Agent: 接收用户问题，分发给子 Agent
• 搜索 Agent: 负责知识库检索
• 计算 Agent: 负责数学计算
• 代码 Agent: 负责代码执行

8.2 工作流设计

用户输入
  ↓
主 Agent (意图识别)
  ↓
  ├─→ 搜索 Agent (RAG)
  ├─→ 计算 Agent (数学)
  └─→ 代码 Agent (编程)
  ↓
结果整合
  ↓
返回用户

8.3 核心代码

import { StateGraph, Annotation } from '@langchain/langgraph';

// 状态定义
const StateAnnotation = Annotation.Root({
    query: Annotation({ reducer: (_, n) => n, default: () => "" }),
    intent: Annotation({ reducer: (_, n) => n, default: () => "" }),
    result: Annotation({ reducer: (_, n) => n, default: () => "" })
});

// 主 Agent: 意图识别
const mainAgent = (state) => {
    const query = state.query;
    let intent = "search";
    
    if (/[+-*/]/.test(query)) intent = "math";
    else if (/代码|编程|函数/.test(query)) intent = "code";
    
    return { intent };
};

// 搜索 Agent
const searchAgent = async (state) => {
    // 调用 RAG 检索
    const result = await ragSearch(state.query);
    return { result };
};

// 计算 Agent
const mathAgent = (state) => {
    const result = eval(state.query);
    return { result: String(result) };
};

// 代码 Agent
const codeAgent = async (state) => {
    const result = await executeCode(state.query);
    return { result };
};

// 构建工作流
const graph = new StateGraph(StateAnnotation)
    .addNode("main", mainAgent)
    .addNode("search", searchAgent)
    .addNode("math", mathAgent)
    .addNode("code", codeAgent)
    .addEdge(START, "main")
    .addConditionalEdges("main", (state) => state.intent, {
        search: "search",
        math: "math",
        code: "code"
    })
    .addEdge("search", END)
    .addEdge("math", END)
    .addEdge("code", END)
    .compile();

---

九、最佳实践

9.1 状态设计

// ✅ 推荐：细粒度状态
const StateAnnotation = Annotation.Root({
    query: Annotation({ reducer: (_, n) => n, default: () => "" }),
    intent: Annotation({ reducer: (_, n) => n, default: () => "" }),
    result: Annotation({ reducer: (_, n) => n, default: () => "" })
});

// ❌ 不推荐：粗粒度状态
const StateAnnotation = Annotation.Root({
    data: Annotation({ reducer: (_, n) => n, default: () => ({}) })
});

9.2 错误处理

// ✅ 推荐：在节点内部处理错误
const safeNode = (state) => {
    try {
        const result = riskyOperation();
        return { result, error: null };
    } catch (err) {
        return { result: null, error: err.message };
    }
};

// ❌ 不推荐：依赖外部错误处理
const unsafeNode = (state) => {
    const result = riskyOperation(); // 可能抛出异常
    return { result };
};

9.3 状态更新策略

// 覆盖更新
reducer: (_prev, next) => next

// 追加更新
reducer: (prev, next) => prev + next

// 数组合并
reducer: (prev, next) => [...prev, ...next]

// 对象合并
reducer: (prev, next) => ({ ...prev, ...next })

---

十、总结

10.1 核心概念回顾

概念	说明
State	工作流的数据载体，在节点间传递
Node	工作单元，接收状态，返回新状态
Edge	连接节点，定义执行顺序
Conditional Edge	根据状态动态选择下一个节点
MemorySaver	持久化状态，支持中断和恢复

10.2 LangGraph 的优势

✅ 网状工作流：支持分支、循环、条件路由
✅ 状态管理：持久化状态，支持中断和恢复
✅ 多 Agent 协作：主 Agent 分发任务，子 Agent 并行处理
✅ 可视化：自动生成 Mermaid 流程图
✅ Token 优化：每个 Agent 只保留必要的 Prompt

10.3 适用场景

• 复杂的多步骤任务
• 需要多 Agent 协作的场景
• 需要人工介入的工作流
• 需要状态持久化的应用
• 需要可视化的流程编排

10.4 下一步学习

• 学习 LangGraph 的高级特性（子图、并行执行）
• 学习如何与 LangSmith 集成（监控和调试）
• 学习生产环境部署（分布式、高可用）
• 学习更多实战案例（RAG、代码生成、数据分析）

---

附录：运行示例

# 安装依赖
pnpm install

# 运行基础示例
node src/basic-graph.mjs

# 运行条件路由示例
node src/conditional-routing.mjs

# 运行状态持久化示例
node src/checkpoint-memory.mjs

---

作者：基于实际项目整理
日期：2026-09-10