LangGraph 入门教程：构建 AI Agent 工作流

发表于 | 分类于 ， | 评论数： 0 | 阅读次数：

LangGraph 是 LangChain 推出的用于构建有状态、多 Agent 应用的库。本文将带你从零开始学习 LangGraph，理解核心概念，并通过实际代码示例掌握如何构建 AI Agent 工作流。

目录

  1. 什么是 LangGraph？
  2. 核心概念详解
  3. 环境准备与安装
  4. [第一个 LangGraph 应用](#四第一个-langgraph 应用)
  5. 状态管理与记忆
  6. 条件分支与循环
  7. [多 Agent 协作](#七多-agent 协作)
  8. 实战：客服对话系统
  9. 调试与可视化
  10. 最佳实践与总结

一、什么是 LangGraph？

1.1 LangGraph 简介

LangGraph 是 LangChain 团队于 2024 年推出的开源库，专门用于构建有状态的、多 Agent 的工作流应用。它基于图（Graph）结构，让开发者可以直观地设计和控制 AI Agent 的执行流程。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
┌─────────────────────────────────────────────────────────────┐
│                    LangGraph 定位                            │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  LangChain → 基础 LLM 应用开发框架                            │
│      ↓                                                      │
│  LangGraph → 有状态、多 Agent 工作流编排                       │
│                                                             │
│  核心优势：                                                  │
│  • 可视化工作流设计                                          │
│  • 内置状态管理                                              │
│  • 支持循环和分支                                            │
│  • 可持久化 checkpoint                                       │
│  • 易于调试和测试                                            │
│                                                             │
└─────────────────────────────────────────────────────────────┘

1.2 为什么需要 LangGraph？

传统的 LLM 应用开发面临以下挑战：

挑战 传统方案 LangGraph 方案
状态管理 手动维护对话历史 内置 State 管理
复杂流程 大量 if-else 嵌套 图结构清晰表达
多 Agent 协作 难以协调 天然支持多节点
循环执行 容易死循环 可控的循环机制
调试困难 黑盒执行 可视化 + 断点

1.3 应用场景

  • 🤖 智能客服系统：多轮对话、问题分类、工单流转
  • 📊 数据分析助手：数据查询、分析、报告生成
  • 📝 内容创作工作流：大纲生成、内容撰写、审核修改
  • 🔍 研究助手：信息检索、总结、引用管理
  • 💼 企业自动化：审批流程、任务分配、状态跟踪

二、核心概念详解

2.1 图（Graph）

LangGraph 的核心是有向图，由节点（Node）和边（Edge）组成：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
┌─────────────────────────────────────────────────────────────┐
│                      LangGraph 图结构                        │
│                                                             │
│         ┌─────────┐                                         │
│         │  START  │                                         │
│         └────┬────┘                                         │
│              │                                              │
│              ▼                                              │
│         ┌─────────┐     ┌─────────┐                        │
│         │ Node A  │────▶│ Node B  │                        │
│         └────┬────┘     └────┬────┘                        │
│              │              │                              │
│              ▼              ▼                              │
│         ┌─────────┐     ┌─────────┐                        │
│         │ Node C  │◀────│ Node D  │                        │
│         └────┬────┘     └─────────┘                        │
│              │                                              │
│              ▼                                              │
│         ┌─────────┐                                         │
│         │  END    │                                         │
│         └─────────┘                                         │
│                                                             │
│  Node = 执行单元（函数/Agent/Tool）                          │
│  Edge = 执行流向（条件/无条件）                               │
│                                                             │
└─────────────────────────────────────────────────────────────┘

2.2 节点（Node）

节点是图中的执行单元，可以是：

  • Python 函数：普通函数或异步函数
  • Agent：LangChain Agent
  • Tool：工具调用
  • LLM 调用：直接调用大模型
1
2
3
4
5
6
# 节点示例：一个简单的函数节点
def node_a(state: dict) -> dict:
    """处理用户输入"""
    user_input = state.get("input", "")
    processed = user_input.lower().strip()
    return {"processed_input": processed}

2.3 边（Edge）

边定义节点之间的执行流向，分为：

  • 普通边：无条件执行下一个节点
  • 条件边：根据条件决定执行路径
1
2
3
4
5
6
7
# 条件边示例
def should_continue(state: dict) -> str:
    """根据状态决定下一步"""
    if state.get("confidence", 0) > 0.8:
        return "high_confidence"  # 高置信度路径
    else:
        return "low_confidence"   # 低置信度路径

2.4 状态（State）

状态是贯穿整个工作流的数据容器，在所有节点间共享：

1
2
3
4
5
6
7
8
9
10
11
12
13
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
import operator

# 定义状态结构
class AgentState(TypedDict):
    """工作流状态"""
    input: str                    # 用户输入
    messages: List[str]           # 消息历史
    processed_input: str          # 处理后的输入
    confidence: float             # 置信度
    output: str                   # 最终输出
    step_count: int               # 执行步数

2.5 Checkpoint（检查点）

Checkpoint 用于持久化状态，支持：

  • 断点续传：从中断处继续执行
  • 时间旅行：回滚到历史状态
  • 并行执行：多个分支独立状态
1
2
3
4
5
6
7
8
from langgraph.checkpoint import MemorySaver

# 内存 Checkpoint（开发测试用）
memory = MemorySaver()

# SQLite Checkpoint（生产环境）
# from langgraph.checkpoint.sqlite import SqliteSaver
# memory = SqliteSaver.from_conn_string(":memory:")

三、环境准备与安装

3.1 基础环境

1
2
3
4
5
6
# 创建虚拟环境
python -m venv langgraph_env
source langgraph_env/bin/activate  # Windows: langgraph_env\Scripts\activate

# Python 版本要求：3.9+
python --version  # 建议 3.10+

3.2 安装依赖

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 核心依赖
pip install langgraph langchain langchain-core

# LLM 提供商（选择你需要的）
pip install langchain-openai        # OpenAI
pip install langchain-anthropic     # Anthropic/Claude
pip install langchain-ollama        # Ollama 本地模型

# 工具和功能
pip install langchain-community     # 社区工具
pip install langchain-tools         # 官方工具

# 可选：可视化
pip install langgraph-cli

# 可选：持久化
pip install langgraph-checkpoint-sqlite

3.3 配置 API Key

1
2
3
4
5
6
7
# 设置环境变量（推荐）
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

# 或在代码中配置
import os
os.environ["OPENAI_API_KEY"] = "sk-..."

3.4 验证安装

1
2
3
4
5
6
# test_install.py
import langgraph
from langgraph.graph import StateGraph, START, END

print(f"LangGraph 版本：{langgraph.__version__}")
print("✅ 安装成功！")

四、第一个 LangGraph 应用

4.1 最简单的示例：问候机器人

让我们从最简单的例子开始，构建一个问候机器人：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
# 01_hello_bot.py
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义状态
class State(TypedDict):
    """状态定义"""
    name: str
    greeting: str

# 2. 定义节点函数
def greet_node(state: State) -> State:
    """问候节点"""
    name = state.get("name", "朋友")
    return {"greeting": f"你好，{name}！欢迎使用 LangGraph！"}

# 3. 创建图
builder = StateGraph(State)

# 4. 添加节点
builder.add_node("greet", greet_node)

# 5. 添加边
builder.add_edge(START, "greet")  # 从 START 到 greet 节点
builder.add_edge("greet", END)    # 从 greet 节点到 END

# 6. 编译图
graph = builder.compile()

# 7. 执行
if __name__ == "__main__":
    # 初始状态
    initial_state = {"name": "张三"}
    
    # 运行图
    result = graph.invoke(initial_state)
    
    print("执行结果：")
    print(result)
    # 输出：{'name': '张三', 'greeting': '你好，张三！欢迎使用 LangGraph！'}

4.2 运行示例

1
python 01_hello_bot.py

4.3 可视化执行流程

1
2
3
4
5
6
7
┌─────────┐      ┌─────────┐      ┌─────────┐
│  START  │─────▶│  greet  │─────▶│   END   │
└─────────┘      └─────────┘      └─────────┘
                     │
                     ▼
              执行 greet_node
              返回 greeting

五、状态管理与记忆

5.1 状态累加器

在某些场景下，我们需要累加状态而不是覆盖，比如消息历史：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
# 02_state_accumulator.py
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
import operator

# 使用 Annotated 定义累加器
class ChatState(TypedDict):
    """聊天状态"""
    # 普通字段：新值覆盖旧值
    current_input: str
    
    # 累加字段：新值追加到列表
    messages: Annotated[List[str], operator.add]
    
    # 计数累加
    turn_count: Annotated[int, operator.add]

# 定义节点
def user_input_node(state: ChatState) -> ChatState:
    """用户输入节点"""
    return {
        "current_input": "今天天气不错",
        "messages": ["用户：今天天气不错"],
        "turn_count": 1
    }

def assistant_node(state: ChatState) -> ChatState:
    """助手回复节点"""
    return {
        "messages": ["助手：是的，适合出门活动！"],
        "turn_count": 1
    }

# 创建图
builder = StateGraph(ChatState)
builder.add_node("user", user_input_node)
builder.add_node("assistant", assistant_node)

builder.add_edge(START, "user")
builder.add_edge("user", "assistant")
builder.add_edge("assistant", END)

graph = builder.compile()

# 执行
if __name__ == "__main__":
    result = graph.invoke({})
    
    print("聊天历史：")
    for msg in result["messages"]:
        print(f"  {msg}")
    
    print(f"\n总轮数：{result['turn_count']}")
    
    # 输出：
    # 聊天历史：
    #   用户：今天天气不错
    #   助手：是的，适合出门活动！
    # 
    # 总轮数：2

5.2 多轮对话示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
# 03_multi_turn_chat.py
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
import operator

class ConversationState(TypedDict):
    """对话状态"""
    messages: Annotated[List[dict], operator.add]
    summary: str

def llm_node(state: ConversationState) -> ConversationState:
    """LLM 回复节点（模拟）"""
    last_message = state["messages"][-1]["content"] if state["messages"] else ""
    
    # 模拟 LLM 回复
    responses = {
        "你好": "你好！有什么我可以帮助你的吗？",
        "再见": "再见！祝你有美好的一天！",
        "谢谢": "不客气！随时为你服务。"
    }
    
    response = responses.get(last_message, "我明白了，请继续。")
    
    return {
        "messages": [{"role": "assistant", "content": response}]
    }

# 创建图
builder = StateGraph(ConversationState)
builder.add_node("llm", llm_node)
builder.add_edge(START, "llm")
builder.add_edge("llm", END)

graph = builder.compile()

# 多轮对话
if __name__ == "__main__":
    state = {"messages": [], "summary": ""}
    
    # 第一轮
    state["messages"] = [{"role": "user", "content": "你好"}]
    result = graph.invoke(state)
    print("第一轮：", result["messages"][-1]["content"])
    
    # 第二轮
    state = result
    state["messages"].append({"role": "user", "content": "谢谢"})
    result = graph.invoke(state)
    print("第二轮：", result["messages"][-1]["content"])
    
    # 第三轮
    state = result
    state["messages"].append({"role": "user", "content": "再见"})
    result = graph.invoke(state)
    print("第三轮：", result["messages"][-1]["content"])

六、条件分支与循环

6.1 条件分支

根据状态决定执行路径：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
# 04_conditional_branch.py
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class RouterState(TypedDict):
    """路由状态"""
    query: str
    category: str
    result: str

def classify_node(state: RouterState) -> RouterState:
    """分类节点"""
    query = state["query"].lower()
    
    # 简单关键词分类
    if any(word in query for word in ["价格", "费用", "多少钱"]):
        category = "pricing"
    elif any(word in query for word in ["技术", "如何使用", "教程"]):
        category = "technical"
    else:
        category = "general"
    
    return {"category": category}

def pricing_node(state: RouterState) -> RouterState:
    """价格查询节点"""
    return {"result": "我们的产品价格如下：基础版¥99/月，专业版¥199/月。"}

def technical_node(state: RouterState) -> RouterState:
    """技术支持节点"""
    return {"result": "技术文档请访问 docs.example.com，或联系技术支持。"}

def general_node(state: RouterState) -> RouterState:
    """通用咨询节点"""
    return {"result": "感谢咨询，我们的客服会尽快回复您。"}

# 条件路由函数
def route(state: RouterState) -> Literal["pricing", "technical", "general"]:
    """根据分类决定路由"""
    category = state.get("category", "general")
    return category

# 创建图
builder = StateGraph(RouterState)

# 添加节点
builder.add_node("classify", classify_node)
builder.add_node("pricing", pricing_node)
builder.add_node("technical", technical_node)
builder.add_node("general", general_node)

# 添加边
builder.add_edge(START, "classify")

# 添加条件边
builder.add_conditional_edges(
    "classify",      # 从哪个节点出来
    route,           # 路由函数
    {
        "pricing": "pricing",
        "technical": "technical",
        "general": "general"
    }
)

# 所有分支最终都到 END
builder.add_edge("pricing", END)
builder.add_edge("technical", END)
builder.add_edge("general", END)

graph = builder.compile()

# 测试
if __name__ == "__main__":
    queries = [
        "这个产品多少钱？",
        "如何使用这个功能？",
        "你们公司地址在哪里？"
    ]
    
    for query in queries:
        print(f"\n用户：{query}")
        result = graph.invoke({"query": query})
        print(f"客服：{result['result']}")
        print(f"分类：{result['category']}")

6.2 循环执行

LangGraph 支持可控的循环，通过条件边实现：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
# 05_loop_execution.py
from typing import TypedDict, List, Annotated
from langgraph.graph import StateGraph, START, END
import operator

class ResearchState(TypedDict):
    """研究状态"""
    topic: str
    findings: Annotated[List[str], operator.add]
    step: int
    max_steps: int

def search_node(state: ResearchState) -> ResearchState:
    """搜索节点（模拟）"""
    step = state["step"]
    findings = [
        f"找到关于'{state['topic']}'的第{step}条资料",
        f"相关论文 {step} 篇",
        f"相关专利 {step * 2} 项"
    ]
    
    return {
        "findings": findings,
        "step": 1
    }

def analyze_node(state: ResearchState) -> ResearchState:
    """分析节点"""
    return {
        "findings": [f"分析完成，发现 {len(state['findings'])} 条有效信息"],
        "step": 1
    }

def should_continue(state: ResearchState) -> str:
    """判断是否继续"""
    current_step = state.get("step", 0)
    max_steps = state.get("max_steps", 3)
    
    if current_step < max_steps:
        return "continue"
    else:
        return "end"

def increment_step(state: ResearchState) -> ResearchState:
    """增加步数"""
    return {"step": state["step"] + 1}

# 创建图
builder = StateGraph(ResearchState)

# 添加节点
builder.add_node("search", search_node)
builder.add_node("analyze", analyze_node)
builder.add_node("increment", increment_step)

# 添加边
builder.add_edge(START, "search")
builder.add_edge("search", "analyze")
builder.add_edge("analyze", "increment")

# 条件循环
builder.add_conditional_edges(
    "increment",
    should_continue,
    {
        "continue": "search",  # 继续循环
        "end": END             # 结束
    }
)

graph = builder.compile()

# 执行
if __name__ == "__main__":
    initial_state = {
        "topic": "AI Agent",
        "findings": [],
        "step": 0,
        "max_steps": 3
    }
    
    result = graph.invoke(initial_state)
    
    print(f"研究主题：{result['topic']}")
    print(f"执行步数：{result['step']}")
    print(f"发现条数：{len(result['findings'])}")
    print("\n发现内容：")
    for finding in result["findings"]:
        print(f"  - {finding}")

6.3 循环控制最佳实践

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
# 防止无限循环的模式
class SafeLoopState(TypedDict):
    """安全循环状态"""
    data: any
    iteration: int
    max_iterations: int
    should_stop: bool

def loop_node(state: SafeLoopState) -> SafeLoopState:
    """循环节点"""
    # 必须增加迭代计数
    return {
        "iteration": state["iteration"] + 1,
        "data": process(state["data"])
    }

def check_stop(state: SafeLoopState) -> str:
    """检查停止条件"""
    # 双重保护：业务条件 + 最大迭代次数
    if state["should_stop"] or state["iteration"] >= state["max_iterations"]:
        return "stop"
    return "continue"

# 始终设置合理的 max_iterations
initial_state = {
    "data": initial_data,
    "iteration": 0,
    "max_iterations": 10,  # 最多循环 10 次
    "should_stop": False
}

七、多 Agent 协作

7.1 多 Agent 架构

LangGraph 的核心优势是支持多 Agent 协作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
┌─────────────────────────────────────────────────────────────┐
│                    多 Agent 协作架构                          │
│                                                             │
│                      ┌───────────┐                          │
│                      │  用户输入  │                          │
│                      └─────┬─────┘                          │
│                            │                                │
│                            ▼                                │
│                      ┌───────────┐                          │
│                      │ 路由 Agent │                          │
│                      └─────┬─────┘                          │
│                            │                                │
│           ┌────────────────┼────────────────┐               │
│           │                │                │               │
│           ▼                ▼                ▼               │
│    ┌───────────┐   ┌───────────┐   ┌───────────┐           │
│    │ 技术 Agent │   │ 销售 Agent │   │ 客服 Agent │           │
│    └─────┬─────┘   └─────┬─────┘   └─────┬─────┘           │
│           │                │                │               │
│           └────────────────┼────────────────┘               │
│                            │                                │
│                            ▼                                │
│                      ┌───────────┐                          │
│                      │ 汇总 Agent │                          │
│                      └─────┬─────┘                          │
│                            │                                │
│                            ▼                                │
│                      ┌───────────┐                          │
│                      │  最终输出  │                          │
│                      └───────────┘                          │
│                                                             │
└─────────────────────────────────────────────────────────────┘

7.2 多 Agent 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
# 06_multi_agent.py
from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, START, END
import operator

class MultiAgentState(TypedDict):
    """多 Agent 状态"""
    input: str
    messages: Annotated[List[dict], operator.add]
    assigned_agent: str
    result: str

def router_agent(state: MultiAgentState) -> MultiAgentState:
    """路由 Agent：判断问题类型"""
    input_text = state["input"].lower()
    
    # 简单路由逻辑
    if any(word in input_text for word in ["bug", "错误", "问题", "故障"]):
        assigned = "technical"
    elif any(word in input_text for word in ["价格", "购买", "订单", "支付"]):
        assigned = "sales"
    else:
        assigned = "support"
    
    return {
        "assigned_agent": assigned,
        "messages": [{"role": "router", "content": f"分配给 {assigned} Agent"}]
    }

def technical_agent(state: MultiAgentState) -> MultiAgentState:
    """技术 Agent"""
    return {
        "messages": [{"role": "technical", "content": "技术团队已收到您的问题，将在 24 小时内回复。"}],
        "result": "technical_ticket_001"
    }

def sales_agent(state: MultiAgentState) -> MultiAgentState:
    """销售 Agent"""
    return {
        "messages": [{"role": "sales", "content": "我们有多种套餐可选，基础版¥99/月，专业版¥199/月。"}],
        "result": "sales_quote_001"
    }

def support_agent(state: MultiAgentState) -> MultiAgentState:
    """客服 Agent"""
    return {
        "messages": [{"role": "support", "content": "感谢咨询，请问有什么可以帮助您的？"}],
        "result": "support_case_001"
    }

def summarize_agent(state: MultiAgentState) -> MultiAgentState:
    """汇总 Agent"""
    return {
        "messages": [{"role": "summary", "content": f"处理完成，结果：{state['result']}"}]
    }

def route_to_agent(state: MultiAgentState) -> Literal["technical", "sales", "support"]:
    """路由到具体 Agent"""
    return state.get("assigned_agent", "support")

# 创建图
builder = StateGraph(MultiAgentState)

# 添加节点
builder.add_node("router", router_agent)
builder.add_node("technical", technical_agent)
builder.add_node("sales", sales_agent)
builder.add_node("support", support_agent)
builder.add_node("summarize", summarize_agent)

# 添加边
builder.add_edge(START, "router")

# 条件路由
builder.add_conditional_edges(
    "router",
    route_to_agent,
    {
        "technical": "technical",
        "sales": "sales",
        "support": "support"
    }
)

# 所有 Agent 完成后到汇总
builder.add_edge("technical", "summarize")
builder.add_edge("sales", "summarize")
builder.add_edge("support", "summarize")
builder.add_edge("summarize", END)

graph = builder.compile()

# 测试
if __name__ == "__main__":
    test_cases = [
        {"input": "系统出现 bug，无法登录"},
        {"input": "我想购买专业版，有什么优惠？"},
        {"input": "你们公司地址在哪里？"}
    ]
    
    for case in test_cases:
        print(f"\n{'='*50}")
        print(f"用户：{case['input']}")
        print(f"{'='*50}")
        
        result = graph.invoke(case)
        
        print("\n处理过程：")
        for msg in result["messages"]:
            print(f"  [{msg['role']}] {msg['content']}")
        
        print(f"\n最终结果：{result['result']}")

7.3 Agent 间通信

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
# Agent 间传递信息的模式
class HandoffState(TypedDict):
    """交接状态"""
    context: dict          # 共享上下文
    history: List[dict]    # 对话历史
    current_agent: str     # 当前 Agent
    handoff_count: int     # 交接次数

def agent_a(state: HandoffState) -> HandoffState:
    """Agent A"""
    # 处理一部分
    result = process_part_a(state["context"])
    
    # 更新上下文，传递给下一个 Agent
    return {
        "context": {**state["context"], "part_a_result": result},
        "history": [{"agent": "A", "action": "processed"}],
        "current_agent": "B",
        "handoff_count": 1
    }

def agent_b(state: HandoffState) -> HandoffState:
    """Agent B"""
    # 使用 Agent A 的结果
    part_a_result = state["context"].get("part_a_result")
    result = process_part_b(part_a_result)
    
    return {
        "context": {**state["context"], "part_b_result": result},
        "history": [{"agent": "B", "action": "processed"}],
        "current_agent": "C",
        "handoff_count": 1
    }

八、实战：客服对话系统

8.1 系统架构

让我们构建一个完整的客服对话系统：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
# 07_customer_service.py
from typing import TypedDict, List, Annotated, Literal
from langgraph.graph import StateGraph, START, END
import operator
from datetime import datetime

# ========== 状态定义 ==========
class CustomerServiceState(TypedDict):
    """客服系统状态"""
    # 输入输出
    user_input: str
    response: str
    
    # 对话历史
    messages: Annotated[List[dict], operator.add]
    
    # 工单信息
    ticket_id: str
    ticket_status: str
    priority: str
    
    # 路由信息
    intent: str
    confidence: float
    assigned_department: str
    
    # 执行控制
    step_count: Annotated[int, operator.add]
    max_steps: int

# ========== 节点定义 ==========
def intent_recognition_node(state: CustomerServiceState) -> CustomerServiceState:
    """意图识别节点"""
    input_text = state["user_input"].lower()
    
    # 模拟意图识别（实际应使用 LLM）
    intents = {
        "technical": ["bug", "错误", "故障", "无法", "报错"],
        "billing": ["账单", "发票", "付款", "退款", "扣费"],
        "account": ["账号", "密码", "登录", "注册", "修改"],
        "product": ["产品", "功能", "价格", "套餐", "购买"]
    }
    
    intent_scores = {}
    for intent, keywords in intents.items():
        score = sum(1 for kw in keywords if kw in input_text)
        intent_scores[intent] = score
    
    # 选择最高分意图
    best_intent = max(intent_scores, key=intent_scores.get)
    confidence = intent_scores[best_intent] / 5.0  # 归一化
    
    return {
        "intent": best_intent,
        "confidence": min(confidence, 1.0),
        "messages": [{"role": "system", "content": f"识别意图：{best_intent}"}]
    }

def triage_node(state: CustomerServiceState) -> CustomerServiceState:
    """分诊节点：决定是否需要人工"""
    confidence = state.get("confidence", 0)
    input_text = state["user_input"].lower()
    
    # 低置信度或紧急情况转人工
    if confidence < 0.4:
        assigned = "human"
    elif any(word in input_text for word in ["投诉", "举报", "紧急", "立刻"]):
        assigned = "human"
        priority = "high"
    else:
        assigned = "auto"
        priority = "normal"
    
    return {
        "assigned_department": assigned,
        "priority": priority,
        "messages": [{"role": "system", "content": f"分诊结果：{assigned}"}]
    }

def auto_response_node(state: CustomerServiceState) -> CustomerServiceState:
    """自动回复节点"""
    intent = state.get("intent", "general")
    
    # 预设回复模板
    responses = {
        "technical": {
            "response": "您好，我们已收到您的技术问题。请提供以下信息：1. 错误截图 2. 操作步骤 3. 浏览器版本。我们的技术团队会在 24 小时内回复。",
            "ticket_status": "open",
            "ticket_id": f"TECH{datetime.now().strftime('%Y%m%d%H%M%S')}"
        },
        "billing": {
            "response": "您好，关于账单问题，您可以登录账户在'账单中心'查看详细记录。如需人工协助，请提供订单号。",
            "ticket_status": "resolved",
            "ticket_id": f"BILL{datetime.now().strftime('%Y%m%d%H%M%S')}"
        },
        "account": {
            "response": "您好，账号相关问题可以通过'账户设置'页面自助处理。如需重置密码，请点击登录页的'忘记密码'。",
            "ticket_status": "resolved",
            "ticket_id": f"ACCT{datetime.now().strftime('%Y%m%d%H%M%S')}"
        },
        "product": {
            "response": "您好，我们提供三种套餐：基础版¥99/月、专业版¥199/月、企业版定制。您可以在官网查看详细介绍。",
            "ticket_status": "resolved",
            "ticket_id": f"PROD{datetime.now().strftime('%Y%m%d%H%M%S')}"
        }
    }
    
    response_data = responses.get(intent, responses["product"])
    
    return {
        "response": response_data["response"],
        "ticket_status": response_data["ticket_status"],
        "ticket_id": response_data["ticket_id"],
        "messages": [{"role": "assistant", "content": response_data["response"]}],
        "step_count": 1
    }

def human_handoff_node(state: CustomerServiceState) -> CustomerServiceState:
    """人工交接节点"""
    ticket_id = f"HUMAN{datetime.now().strftime('%Y%m%d%H%M%S')}"
    
    return {
        "response": f"您好，您的问题较为复杂，我们将安排专业客服为您处理。工单号：{ticket_id}，预计 2 小时内回复。",
        "ticket_id": ticket_id,
        "ticket_status": "pending_human",
        "messages": [{"role": "assistant", "content": f"转接人工客服，工单号：{ticket_id}"}],
        "step_count": 1
    }

def summarize_node(state: CustomerServiceState) -> CustomerServiceState:
    """汇总节点"""
    return {
        "messages": [{
            "role": "system",
            "content": f"处理完成 | 工单：{state.get('ticket_id', 'N/A')} | 状态：{state.get('ticket_status', 'unknown')}"
        }],
        "step_count": 1
    }

# ========== 条件函数 ==========
def route_by_department(state: CustomerServiceState) -> Literal["auto", "human"]:
    """根据分诊结果路由"""
    return state.get("assigned_department", "auto")

def should_summarize(state: CustomerServiceState) -> str:
    """判断是否需要汇总"""
    return "summarize"

# ========== 创建图 ==========
builder = StateGraph(CustomerServiceState)

# 添加节点
builder.add_node("intent", intent_recognition_node)
builder.add_node("triage", triage_node)
builder.add_node("auto", auto_response_node)
builder.add_node("human", human_handoff_node)
builder.add_node("summarize", summarize_node)

# 添加边
builder.add_edge(START, "intent")
builder.add_edge("intent", "triage")

# 条件路由
builder.add_conditional_edges(
    "triage",
    route_by_department,
    {
        "auto": "auto",
        "human": "human"
    }
)

# 到汇总
builder.add_edge("auto", "summarize")
builder.add_edge("human", "summarize")
builder.add_edge("summarize", END)

# 编译
graph = builder.compile()

# ========== 测试 ==========
if __name__ == "__main__":
    test_cases = [
        {"user_input": "系统一直报错，无法登录", "max_steps": 10},
        {"user_input": "我想了解一下产品价格", "max_steps": 10},
        {"user_input": "我要投诉！你们的服务太差了！", "max_steps": 10},
        {"user_input": "asdfghjkl 随机输入", "max_steps": 10}
    ]
    
    print("="*70)
    print("客服对话系统测试")
    print("="*70)
    
    for i, case in enumerate(test_cases, 1):
        print(f"\n【测试用例 {i}】")
        print(f"用户：{case['user_input']}")
        print("-"*70)
        
        result = graph.invoke(case)
        
        print(f"客服：{result['response']}")
        print(f"\n处理详情：")
        print(f"  意图：{result.get('intent', 'unknown')}")
        print(f"  置信度：{result.get('confidence', 0):.2f}")
        print(f"  工单号：{result.get('ticket_id', 'N/A')}")
        print(f"  状态：{result.get('ticket_status', 'unknown')}")
        print(f"  优先级：{result.get('priority', 'normal')}")
        print(f"  执行步数：{result.get('step_count', 0)}")
    
    print("\n" + "="*70)

8.2 运行结果示例

1
python 07_customer_service.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
======================================================================
客服对话系统测试
======================================================================

【测试用例 1】
用户：系统一直报错，无法登录
----------------------------------------------------------------------
客服：您好，我们已收到您的技术问题。请提供以下信息：1. 错误截图 2. 操作步骤 3. 浏览器版本。我们的技术团队会在 24 小时内回复。

处理详情：
  意图：technical
  置信度：0.60
  工单号：TECH20260323203000
  状态：open
  优先级：normal
  执行步数：2

【测试用例 2】
用户：我想了解一下产品价格
----------------------------------------------------------------------
客服：您好，我们提供三种套餐：基础版¥99/月、专业版¥199/月、企业版定制。您可以在官网查看详细介绍。

处理详情：
  意图：product
  置信度：0.40
  工单号：PROD20260323203000
  状态：resolved
  优先级：normal
  执行步数：2

【测试用例 3】
用户：我要投诉！你们的服务太差了！
----------------------------------------------------------------------
客服：您好，您的问题较为复杂，我们将安排专业客服为您处理。工单号：HUMAN20260323203000，预计 2 小时内回复。

处理详情：
  意图：general
  置信度：0.00
  工单号：HUMAN20260323203000
  状态：pending_human
  优先级：high
  执行步数：2

【测试用例 4】
用户：asdfghjkl 随机输入
----------------------------------------------------------------------
客服：您好，您的问题较为复杂，我们将安排专业客服为您处理。工单号：HUMAN20260323203000，预计 2 小时内回复。

处理详情：
  意图：general
  置信度：0.00
  工单号：HUMAN20260323203000
  状态：pending_human
  优先级：high
  执行步数：2

======================================================================

九、调试与可视化

9.1 启用调试模式

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
# 08_debug_mode.py
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class DebugState(TypedDict):
    value: int

def node_a(state: DebugState) -> DebugState:
    print(f"[Node A] 输入：{state}")
    result = {"value": state["value"] + 1}
    print(f"[Node A] 输出：{result}")
    return result

def node_b(state: DebugState) -> DebugState:
    print(f"[Node B] 输入：{state}")
    result = {"value": state["value"] * 2}
    print(f"[Node B] 输出：{result}")
    return result

builder = StateGraph(DebugState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

graph = builder.compile()

# 使用 stream 调试
if __name__ == "__main__":
    print("执行流程：")
    for step, output in graph.stream({"value": 1}):
        print(f"步骤：{step}")
        print(f"输出：{output}")
        print("-"*40)

9.2 可视化图结构

1
2
3
4
5
# 使用 LangGraph CLI 可视化
langgraph visualize graph.py

# 或导出为图片
langgraph export graph.py --output graph.png

9.3 使用 Checkpoint 调试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
# 09_checkpoint_debug.py
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint import MemorySaver
from typing import TypedDict, List, Annotated
import operator

class DebugState(TypedDict):
    messages: Annotated[List[str], operator.add]
    step: int

def node_a(state: DebugState) -> DebugState:
    return {
        "messages": ["执行了 Node A"],
        "step": 1
    }

def node_b(state: DebugState) -> DebugState:
    return {
        "messages": ["执行了 Node B"],
        "step": 1
    }

# 启用 Checkpoint
memory = MemorySaver()
builder = StateGraph(DebugState)
builder.add_node("a", node_a)
builder.add_node("b", node_b)
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)

# 编译时传入 checkpointer
graph = builder.compile(checkpointer=memory)

# 执行并保存状态
if __name__ == "__main__":
    config = {"configurable": {"thread_id": "test_thread_1"}}
    
    # 第一次执行
    result = graph.invoke({"messages": [], "step": 0}, config)
    print("第一次执行结果：", result)
    
    # 获取历史状态
    print("\n历史状态：")
    for state in graph.get_state_history(config):
        print(f"  步数：{state.values.get('step')}")
        print(f"  消息：{state.values.get('messages')}")
        print("-"*40)

十、最佳实践与总结

10.1 设计原则

原则 说明 示例
单一职责 每个节点只做一件事 意图识别、回复生成、工单创建分开
状态明确 清晰定义 State 结构 使用 TypedDict 定义所有字段
可控循环 设置最大迭代次数 max_iterations 保护
错误处理 每个节点都要处理异常 try-except 包裹
可测试性 节点函数纯函数化 便于单元测试

10.2 常见陷阱

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
# ❌ 错误：无限循环
def bad_loop(state):
    if not condition:
        return "loop"  # 没有退出机制
    return "end"

# ✅ 正确：带保护的循环
def good_loop(state):
    if state["iteration"] >= state["max_iterations"]:
        return "end"
    if condition:
        return "loop"
    return "end"

# ❌ 错误：状态覆盖
def bad_state(state):
    return {"messages": ["new message"]}  # 覆盖了历史消息

# ✅ 正确：状态累加
def good_state(state):
    return {"messages": ["new message"], "step_count": 1}  # 使用 Annotated[operator.add]

10.3 性能优化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
# 1. 使用异步节点
async def async_node(state):
    result = await async_llm_call(state["input"])
    return {"output": result}

# 2. 批量处理
def batch_node(state):
    items = state["items"]
    results = process_batch(items, batch_size=10)
    return {"results": results}

# 3. 缓存结果
from functools import lru_cache

@lru_cache(maxsize=100)
def cached_process(input_text):
    return llm_process(input_text)

10.4 学习资源

资源 链接
LangGraph 官方文档 https://langchain-ai.github.io/langgraph/
LangChain 文档 https://python.langchain.com/
GitHub 仓库 https://github.com/langchain-ai/langgraph
示例代码 https://github.com/langchain-ai/langgraph/tree/main/examples

总结

本文介绍了 LangGraph 的核心概念和使用方法：

核心要点

  1. 图结构：节点 + 边 = 工作流
  2. 状态管理：TypedDict 定义，支持累加器
  3. 条件路由：根据状态决定执行路径
  4. 循环控制：设置最大迭代次数保护
  5. 多 Agent：天然支持多角色协作
  6. Checkpoint：持久化状态，支持断点续传

下一步

  • 📚 阅读官方文档深入学习
  • 🔧 尝试修改示例代码
  • 🚀 构建自己的 Agent 工作流
  • 💡 分享你的使用经验

本文基于 LangGraph 最新版本编写，如有更新请参考官方文档。

标签：#LangGraph #LangChain #AI Agent #工作流 #大模型

最后更新: 2026 年 3 月 23 日

感谢你的阅读，如果文章对你有帮助，可以请作者喝杯茶!