LangGraph 是什么？5分钟搞懂核心概念

引言：为什么需要 LangGraph？

想象一下，你正在开发一个智能助手，它不仅能回答简单问题，还能帮你完成复杂的多步骤任务——比如”帮我订一张明天去上海的机票，要上午的航班，价格不超过1000元，如果没有合适的就订高铁”。

这样的任务看似简单，但对AI来说却充满挑战：

• 需要搜索航班信息
• 需要比较价格和时段
• 可能需要备选方案（高铁）
• 整个过程需要记住之前的决策

这就是传统 AI Agent 面临的困境：它们往往是线性的、缺乏状态管理、难以处理复杂的循环和条件分支。

传统 Agent 的三大痛点

痛点一：对话像金鱼一样健忘

传统的 Chain（链式调用）模式就像一条金鱼，只有7秒记忆。每次调用都是独立的，前面的上下文很容易丢失。比如用户说”还是选刚才那个航班吧”，传统 Agent 可能会问：”刚才哪个航班？”

痛点二：流程僵硬，无法回头

想象你在走一条单向道，一旦迈出一步就不能回头。传统 Agent 的执行流程往往是固定的：A → B → C → D。但如果 B 步骤发现信息不足，需要回到 A 重新获取呢？传统模式很难优雅地处理这种”循环”需求。

痛点三：调试困难，黑盒运行

当 Agent 出错时，你往往不知道问题出在哪一步。是整个流程设计有问题？还是某个节点的逻辑有bug？缺乏可视化的调试工具，让开发变成了猜谜游戏。

LangGraph 闪亮登场

LangGraph 是 LangChain 团队推出的一个革命性框架，它用图（Graph）的概念来构建 AI Agent。如果传统 Chain 是一条直线，那 LangGraph 就是一张可以自由连接的网络。

核心优势：

• ✅ 持久化状态：像人类一样有”记忆”，随时知道之前发生了什么
• ✅ 循环支持：流程可以前进、后退、循环，灵活应对复杂场景
• ✅ 可视化调试：配合 LangSmith，像看地图一样理解 Agent 的思考过程
• ✅ 人机协作：支持 Human-in-the-loop，让 AI 和人类无缝配合

接下来，让我们一起走进 LangGraph 的世界，用 5 分钟搞懂它的核心概念。

---

核心概念：用生活比喻理解 State、Node、Edge

LangGraph 的核心其实就三个概念：State（状态）、Node（节点）、Edge（边）。别被这些术语吓到，我们用生活中的例子来理解。

🧳 State（状态）：你的行李箱

想象你正在旅行，State 就是你的行李箱。里面装着你在旅程中需要的一切：衣服、证件、充电器、纪念品…

在 LangGraph 中，State 是一个 Python 类（TypedDict），定义了 Graph 运行过程中需要传递的所有数据：

from typing import TypedDict, List

class TravelState(TypedDict):
    destination: str      # 目的地
    budget: int          # 预算
    itinerary: List[str] # 行程安排
    current_step: str    # 当前步骤

关键特性：

• 每个 Node 都可以读取和修改 State
• State 会随着旅程（Graph 执行）不断更新
• 最后你可以从 State 中拿到所有结果

🏨 Node（节点）：旅程中的每个站点

继续旅行的比喻，Node 就是你旅程中的每个站点：机场、酒店、景点、餐厅…

每个 Node 都是一个函数，它接收当前的 State，做一些处理，然后返回更新后的 State：

def book_hotel(state: TravelState) -> TravelState:
    """预订酒店节点"""
    destination = state["destination"]
    budget = state["budget"]
    
    # 查找符合预算的酒店
    hotel = find_hotel(destination, budget)
    
    # 更新状态
    state["itinerary"].append(f"入住{hotel}")
    state["current_step"] = "hotel_booked"
    
    return state

关键特性：

• Node 是独立的功能单元，可以是一个 LLM 调用、一个工具函数、甚至是一段代码
• 每个 Node 只做一件事，保持简单
• Node 之间通过 State 传递信息

🛤️ Edge（边）：连接站点的道路

Edge 就是连接各个站点的道路。它决定了你从酒店去景点，还是直接去机场。

在 LangGraph 中，Edge 分为两种：

1. 普通 Edge（普通道路）：确定性的连接

# 从酒店预订节点，确定性地前往景点规划节点
graph.add_edge("book_hotel", "plan_attractions")

2. 条件 Edge（分叉路口）：根据条件决定走向

def decide_next_step(state: TravelState) -> str:
    """根据预算决定下一步"""
    if state["budget"] > 5000:
        return "luxury_tour"  # 豪华游
    else:
        return "budget_tour"  # 经济游

# 添加条件边
graph.add_conditional_edges(
    "check_budget",
    decide_next_step,
    {
        "luxury_tour": "luxury_tour",
        "budget_tour": "budget_tour"
    }
)

关键特性：

• Edge 定义了 Graph 的”流程”
• 条件 Edge 让 Graph 有了”智能决策”能力
• 可以形成循环（比如”信息不足，返回重新搜索”）

总结比喻

LangGraph 概念	旅行比喻	作用
State	行李箱	存储旅程中的所有物品和信息
Node	站点（酒店、景点）	执行具体的任务
Edge	道路	决定下一站去哪里

理解了这三个概念，你就掌握了 LangGraph 的 80%！

---

实战：构建一个会”思考”的计算器对话机器人

理论知识讲完了，现在让我们动手写一个真正可运行的代码示例——一个会思考的计算器对话机器人。

这个机器人有什么特别？

• 它能理解自然语言（比如”帮我算一下 25 乘以 4 再加 10”）
• 它会分解任务：先算乘法，再算加法
• 它有记忆：能记住之前的计算结果
• 它会反思：如果计算出错，会尝试修正

完整代码

# calculator_bot.py
import os
import re
import json
from typing import TypedDict, Annotated, Sequence
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# ==================== 1. 定义状态 ====================
class CalculatorState(TypedDict):
    """计算器机器人的状态"""
    messages: Annotated[Sequence[dict], "append"]  # 对话历史
    current_expression: str                        # 当前需要计算的表达式
    calculation_result: float                      # 计算结果
    step_count: int                               # 步骤计数，防止无限循环
    is_complete: bool                             # 是否完成

# ==================== 2. 定义工具函数 ====================
def parse_expression(text: str) -> str:
    """从用户输入中提取数学表达式"""
    # 简单的提取逻辑：找数字和运算符
    patterns = [
        r'(\d+)\s*乘以\s*(\d+)',  # 5 乘以 3
        r'(\d+)\s*\*\s*(\d+)',    # 5 * 3
        r'(\d+)\s*加\s*(\d+)',    # 5 加 3
        r'(\d+)\s*\+\s*(\d+)',    # 5 + 3
        r'(\d+)\s*减\s*(\d+)',    # 5 减 3
        r'(\d+)\s*-\s*(\d+)',     # 5 - 3
    ]
    
    for pattern in patterns:
        match = re.search(pattern, text)
        if match:
            if '乘以' in text or '*' in text:
                return f"{match.group(1)} * {match.group(2)}"
            elif '加' in text or '+' in text:
                return f"{match.group(1)} + {match.group(2)}"
            elif '减' in text or '-' in text:
                return f"{match.group(1)} - {match.group(2)}"
    
    # 如果找不到，直接返回原文中的数字和运算符
    cleaned = re.sub(r'[^\d+\-*/.\s]', '', text)
    return cleaned.strip()

def calculate(expression: str) -> float:
    """安全地计算数学表达式"""
    try:
        # 只允许数字和基本运算符
        allowed_chars = set('0123456789+-*/. ')
        if not all(c in allowed_chars for c in expression):
            raise ValueError("表达式包含非法字符")
        
        result = eval(expression)
        return round(result, 2)
    except Exception as e:
        raise ValueError(f"计算错误: {e}")

# ==================== 3. 定义节点函数 ====================
# 初始化 LLM
llm = ChatOpenAI(
    model="gpt-3.5-turbo",
    temperature=0.7,
    api_key=os.getenv("OPENAI_API_KEY")
)

def understand_input(state: CalculatorState) -> CalculatorState:
    """理解用户输入节点：解析用户的自然语言"""
    messages = state["messages"]
    last_message = messages[-1]["content"] if messages else ""
    
    # 使用 LLM 理解用户意图
    system_prompt = """你是一个计算器助手。请分析用户的输入，提取其中的数学计算需求。
如果用户想进行计算，请输出 JSON 格式：{"intent": "calculate", "expression": "数学表达式"}
如果用户只是闲聊，请输出：{"intent": "chat", "response": "你的回复"}"""
    
    response = llm.invoke([
        SystemMessage(content=system_prompt),
        HumanMessage(content=last_message)
    ])
    
    try:
        result = json.loads(response.content)
        if result.get("intent") == "calculate":
            state["current_expression"] = result.get("expression", parse_expression(last_message))
        else:
            state["current_expression"] = ""
            # 添加 AI 回复到消息列表
            state["messages"] = list(messages) + [{"role": "assistant", "content": result.get("response", "你好！我是计算器机器人，可以帮你做数学计算。")}]
    except:
        # 如果解析失败，使用简单规则
        state["current_expression"] = parse_expression(last_message)
    
    state["step_count"] = state.get("step_count", 0) + 1
    return state

def calculate_node(state: CalculatorState) -> CalculatorState:
    """计算节点：执行数学计算"""
    expression = state.get("current_expression", "")
    
    if not expression:
        return state
    
    try:
        result = calculate(expression)
        state["calculation_result"] = result
        
        # 添加结果到消息
        response_msg = f"计算结果：{expression} = {result}"
        state["messages"] = list(state["messages"]) + [{"role": "assistant", "content": response_msg}]
        state["is_complete"] = True
        
    except Exception as e:
        error_msg = f"计算出错：{str(e)}。让我再试试..."
        state["messages"] = list(state["messages"]) + [{"role": "assistant", "content": error_msg}]
        state["is_complete"] = False
    
    state["step_count"] = state.get("step_count", 0) + 1
    return state

def generate_response(state: CalculatorState) -> CalculatorState:
    """生成回复节点：用自然语言回复用户"""
    if state.get("is_complete"):
        return state
    
    # 如果还没有生成回复，用 LLM 生成
    result = state.get("calculation_result")
    expression = state.get("current_expression", "")
    
    if result is not None:
        prompt = f"用户问了一个计算问题。表达式是：{expression}，结果是：{result}。请用友好自然的语言回复用户。"
        response = llm.invoke([HumanMessage(content=prompt)])
        
        state["messages"] = list(state["messages"]) + [{"role": "assistant", "content": response.content}]
        state["is_complete"] = True
    
    return state

# ==================== 4. 定义条件边 ====================
def should_calculate(state: CalculatorState) -> str:
    """决定下一步走向"""
    # 防止无限循环
    if state.get("step_count", 0) > 10:
        return "end"
    
    # 如果有表达式需要计算，去计算节点
    if state.get("current_expression") and not state.get("calculation_result"):
        return "calculate"
    
    # 如果计算完成，去生成回复
    if state.get("calculation_result") and not state.get("is_complete"):
        return "respond"
    
    # 如果都完成了，结束
    if state.get("is_complete"):
        return "end"
    
    # 默认去计算
    return "calculate"

# ==================== 5. 构建 Graph ====================
def create_calculator_graph():
    """创建计算器机器人的 Graph"""
    # 创建图
    workflow = StateGraph(CalculatorState)
    
    # 添加节点
    workflow.add_node("understand", understand_input)
    workflow.add_node("calculate", calculate_node)
    workflow.add_node("respond", generate_response)
    
    # 设置入口点
    workflow.set_entry_point("understand")
    
    # 添加条件边
    workflow.add_conditional_edges(
        "understand",
        should_calculate,
        {
            "calculate": "calculate",
            "respond": "respond",
            "end": END
        }
    )
    
    workflow.add_conditional_edges(
        "calculate",
        should_calculate,
        {
            "calculate": "calculate",  # 可以循环重试
            "respond": "respond",
            "end": END
        }
    )
    
    workflow.add_edge("respond", END)
    
    # 添加记忆（checkpoint）
    memory = MemorySaver()
    
    return workflow.compile(checkpointer=memory)

# ==================== 6. 运行示例 ====================
def main():
    """主函数：演示如何使用计算器机器人"""
    print("🤖 计算器机器人启动！")
    print("输入 'quit' 退出\n")
    
    # 创建 Graph
    graph = create_calculator_graph()
    
    # 对话循环
    thread_id = "user_session_1"
    config = {"configurable": {"thread_id": thread_id}}
    
    while True:
        user_input = input("你: ")
        
        if user_input.lower() in ['quit', 'exit', '退出']:
            print("👋 再见！")
            break
        
        # 准备输入状态
        initial_state = {
            "messages": [{"role": "user", "content": user_input}],
            "current_expression": "",
            "calculation_result": None,
            "step_count": 0,
            "is_complete": False
        }
        
        # 运行 Graph
        try:
            result = graph.invoke(initial_state, config)
            
            # 获取最后的 AI 回复
            messages = result["messages"]
            for msg in reversed(messages):
                if msg.get("role") == "assistant":
                    print(f"🤖 机器人: {msg['content']}\n")
                    break
                    
        except Exception as e:
            print(f"❌ 出错了: {e}\n")

if __name__ == "__main__":
    main()

代码详解

1. 状态定义（State）

class CalculatorState(TypedDict):
    messages: Annotated[Sequence[dict], "append"]
    current_expression: str
    calculation_result: float
    step_count: int
    is_complete: bool

这里我们定义了机器人需要记住的所有信息：

• messages：对话历史，让机器人知道之前说了什么
• current_expression：当前要计算的表达式
• calculation_result：计算结果
• step_count：步骤计数，防止无限循环
• is_complete：标记是否完成

2. 节点（Nodes）

我们有三个核心节点：

understand 节点：理解用户的自然语言输入，提取数学表达式。

• 使用 LLM 解析用户意图
• 支持自然语言（”5 乘以 3”）和数学符号（”5 * 3”）

calculate 节点：执行实际的数学计算。

• 使用安全的 eval（只允许数字和基本运算符）
• 出错时会捕获异常，让 Graph 有机会重试

respond 节点：生成自然的回复。

• 将计算结果转化为友好的语言
• 让用户感觉在和真人对话

3. 条件边（Conditional Edges）

def should_calculate(state: CalculatorState) -> str:
    if state.get("step_count", 0) > 10:
        return "end"
    if state.get("current_expression"):
        return "calculate"
    # ...

这是 Graph 的”大脑”，决定下一步去哪里：

• 有表达式需要计算 → 去 calculate 节点
• 计算完成但还没回复 → 去 respond 节点
• 全部完成 → 结束
• 步骤太多（防死循环）→ 结束

4. 记忆功能（MemorySaver）

memory = MemorySaver()
return workflow.compile(checkpointer=memory)

这行代码赋予机器人”记忆”：

• 同一个 thread_id 的对话会共享状态
• 用户说”刚才的结果再加 10”，机器人能记得”刚才的结果”

运行效果

当你运行这个程序时，交互可能是这样的：

🤖 计算器机器人启动！
输入 'quit' 退出

你: 帮我算一下 25 乘以 4
🤖 机器人: 计算结果：25 * 4 = 100

你: 刚才的结果再加 50
🤖 机器人: 计算结果：100 + 50 = 150

你: 谢谢！
🤖 机器人: 不客气！有什么计算问题随时找我。

---

可视化调试：LangSmith 集成简介

代码写好了，但如果出现问题怎么调试？这就是 LangSmith 登场的时候了。

什么是 LangSmith？

LangSmith 是 LangChain 提供的调试和监控平台。你可以把它理解为 Graph 的”CT扫描仪”——能看到内部每一步的执行过程。

如何集成

只需要几行代码：

import os

# 设置 LangSmith 环境变量
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_ENDPOINT"] = "https://api.smith.langchain.com"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "calculator-bot"

LangSmith 能看到什么？

当你运行 Graph 后，打开 LangSmith 控制台，你会看到：

1. 执行时间线

• 每个节点什么时候执行
• 执行了多长时间
• 哪个节点最耗时

2. 状态变化追踪

• 进入节点前的 State
• 节点返回后的 State
• 清晰地看到数据是如何流转的

3. LLM 调用详情

• 发送给 LLM 的完整 Prompt
• LLM 返回的原始响应
• Token 使用量和成本

4. 错误分析

• 哪个节点抛出了异常
• 完整的错误堆栈
• 出错时的完整状态快照

调试技巧

技巧一：给节点命名

# 好的命名
workflow.add_node("extract_math_expression", understand_input)

# 避免这种命名
workflow.add_node("node1", understand_input)

技巧二：在关键节点添加日志

def calculate_node(state: CalculatorState) -> CalculatorState:
    print(f"[DEBUG] 计算表达式: {state.get('current_expression')}")
    # ... 计算逻辑
    print(f"[DEBUG] 计算结果: {result}")
    return state

技巧三：使用断点（Breakpoint）

LangGraph 支持在特定节点暂停执行，让人类介入：

# 在 calculate 节点前设置断点
graph = workflow.compile(
    checkpointer=memory,
    interrupt_before=["calculate"]
)

# 运行时会暂停，等待人类确认
result = graph.invoke(state, config)

这样，在机器人执行计算前，你可以先检查它理解的表达式是否正确。

---

总结与下篇预告

今天我们学到了什么？

1. LangGraph 解决了传统 Agent 的痛点：状态管理、循环流程、可视化调试

2. 三个核心概念：

   • State：行李箱，存储所有数据
   • Node：站点，执行具体任务
   • Edge：道路，决定流程走向

3. 实战了一个计算器机器人：

   • 理解自然语言输入
   • 分解并执行计算任务
   • 使用记忆功能保持对话上下文
   • 通过条件边实现智能流程控制

4. LangSmith 调试：让 Graph 的执行过程透明可见

LangGraph 的应用场景

学会了基础，你可以构建：

• 智能客服机器人：支持复杂的多轮对话和工单流转
• 代码审查助手：自动分析代码、提出问题、验证修复
• 研究助理：搜索资料、总结要点、生成报告
• 游戏 NPC：有记忆、能决策的智能角色
• 自动化工作流：审批流程、数据处理、异常处理

下篇预告

在下一篇文章中，我们将深入探讨：

《LangGraph 进阶：多 Agent 协作与持久化存储》

• 如何让多个 Agent 像团队一样协作
• 使用 PostgreSQL 持久化对话历史
• 实现 Human-in-the-loop（人机协作）
• 构建一个完整的多 Agent 客服系统

动手练习

想巩固今天的内容？试试这些练习：

1. 扩展计算器：支持更多运算（除法、括号、函数）
2. 添加验证节点：在计算前验证表达式是否合法
3. 实现撤销功能：让用户可以说”撤销上一步”
4. 集成搜索工具：当计算需要外部数据时（如汇率），自动搜索

参考资源

• LangGraph 官方文档
• LangGraph 概念指南
• 示例代码 GitHub 仓库

---

感谢阅读！ 如果这篇文章对你有帮助，欢迎点赞、收藏、分享给更多朋友。有任何问题，欢迎在评论区留言讨论！

本文代码在 Python 3.9+、LangGraph 0.2+ 环境下测试通过。

---

📚 LangGraph 入门系列

本系列共10篇文章，带你从零掌握 LangGraph：

篇目	标题	链接
01	LangGraph 是什么？	阅读
02	State 状态管理	阅读
03	Node 节点	阅读
04	Tool Calling	阅读
05	Memory 记忆系统	阅读
06	HITL 人机协作	阅读
07	Multi-Agent	阅读
08	Streaming 流式输出	阅读
09	Conditional Edges	阅读
10	Production 生产部署	阅读

当前位置： 第1篇 / 共10篇

• 上一篇：无
• 下一篇：State 状态管理