​
当AI技术从单一能力输出迈向复杂任务处理时，具备流程调度能力的智能体（AI Agent）成为开发核心。在众多构建工具中，LangGraph凭借其透明化、高可控的特性脱颖而出，成为衔接大语言模型（LLM）与实际业务场景的关键框架。无论是自动化客户服务、智能内容生成，还是精准的数据分析助手，LangGraph都能为开发者提供底层支撑。但对于刚接触AI Agent的开发者来说，技术概念的繁杂往往让人无从下手。本文将以实战为导向，带你吃透LangGraph的核心逻辑，快速实现从理论到代码的落地。

1、LangGraph是什么

LangGraph是一款专注于构建各类工作流程（Workflow）与智能体（Agent）的基础设施框架。和众多力求简化操作的无代码（No-code）或低代码（Low-code）开发框架不同，LangGraph走的是一条更注重透明性与可控性的道路——它不会对提示词或架构进行过多的抽象封装，而是直接为开发者提供三大核心优势：

控制（Control）：能轻松定义或组合各种工作流程与智能体，每个环节都尽在掌控。

持久化（Persistence）：框架可对图状态进行持久化存储，这让开发记忆功能和人类干预功能成为可能。

测试、调试、部署（Testing, Debugging, Deployment）：具备完善的测试、调试和部署支持，让开发过程更顺畅。

这样的设计理念意味着，使用LangGraph时，你不会被框架的“黑盒”所限制，而是能清晰了解每一步的执行逻辑，并根据需求进行精准调整。

2、控制能力：搭建你的首个图（Graph）

2.1、核心概念解析

在LangGraph里，LLM应用被定义为一种图结构，该图包含三个基本要素：

• 图状态（Graph State）：图状态是所有基于LangGraph的LLM应用的核心，任务执行过程中所需的全部信息都会存储在图状态中。
• 节点（Nodes）：节点是更新状态的执行单元，需要明确在哪些节点更新状态信息。
• 边（Edges）：边是节点流转的路径，描述了节点之间的连接关系。

简单来讲，图状态是LangGraph的核心，节点是更新状态的执行单元，而边则是节点流转的路径。

2.2、定义状态模式（State schema）

在LangGraph中，用StateGraph类来表示一个图。在初始化StateGraph之前，需要先定义一个状态模式，它可以是Python中任何带有getattr()方法的对象，例如字典、类或者Pydantic对象。不同的选择各有特点：

• TypedDict：速度快，但不支持默认值。
• dataclass：速度也很快，支持state.foo的访问方式，且支持默认值。
• Pydantic：速度较慢，但支持数据验证。

让我们从一个简单的例子开始。首先，我们使用TypedDict来定义一个StateGraph对象：

import sys
sys.path.append("..")
from typing import TypedDict
from langgraph.graph import StateGraph, START, END

from src.agents_intro import model_with_tools, write_email # 导入之前的代码
from src.utils import show_graph

class StateSchema(TypedDict):
    request: str
    email: str

workflow = StateGraph(StateSchema)

这里我们定义了一个包含request和email两个字段的 State schema，然后用它初始化了一个StateGraph对象，名称为 workflow。

2.3、添加节点（Nodes）和边（Edges）

实际应用的图中，会有很多边和节点，边是不需要定义的，但是节点需要提前定义好。节点的作用是更新 graph state，更新的方式需要关注一下，默认情况下，如果节点中更新了某个key的值，会直接覆盖旧的值，返回值中没有涉及的字段不会被更新。

def write_email_node(state: StateSchema) -> StateSchema:
    # 处理请求的具体代码
    output = model_with_tools.invoke(state['request'])
    args = output.tool_calls[0]['args']
    email = write_email.invoke(args)
    return {'email': email}  # 只更新email字段，request字段保持不变

在这个函数中，email字段被覆盖更新，而request字段并未更新，因此保持不变。

LangGraph也支持自定义更新的逻辑。

定义好节点后，我们需要将节点和边添加到图中：

workflow = StateGraph(StateSchema)
workflow.add_node("write_email_node", write_email_node)
workflow.add_edge(START, "write_email_node")
workflow.add_edge("write_email_node", END)

app = workflow.compile()

上述代码需要注意几个要点：

• START和END是LangGraph默认的开始和结束节点
• 边不需要单独定义，只需要指定连接关系
• 图需要compile之后才能运行

2.4、执行简单流程

现在我们可以执行这个简单的工作流程：

result = app.invoke({
    "request": "写一封邮件给张三，主题是关于项目进展，内容是：我们正在按照计划推进项目，预计下周完成。"
})

执行后，图状态被更新，request字段保持不变，email字段更新为write_email的输出。这个流程是一个非常简单的直线流转：Start -> write_email_node -> End。

2.5、构建条件分支流程

在实际应用中，我们经常需要根据不同条件执行不同的逻辑。LangGraph支持条件边，让节点间的流转可以按条件执行。

首先，我们定义两个新的节点：

• call_llm 用于根据用户的请求调用LLM
• run_tool 用于执行LLM输出的工具调用

from langgraph.graph import MessagesState

defcall_llm(state: MessagesState) -> MessagesState:
    """运行LLM"""
    output = model_with_tools.invoke(state['messages'])
    return {"messages": [output]}

defrun_tool(state: MessagesState) -> MessagesState:
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        observation = write_email.invoke(tool_call['args'])
        result.append({
            "role": "tool",
            "content": observation,
            "tool_call_id": tool_call["id"]
        })
    return {"messages": result}

这里有几个重要概念需要说明：

• MessagesState是一个预定义的StateGraph子类，包含messages字段
• 当节点更新messages字段时，会自动扩展（extend）初始数据，而不是覆盖
• run_tool中的observation很好地体现了工具的作用：与环境交互，并返回交互结果作为LLM对外部环境的观察

接下来，我们定义条件判断函数：

from typing importLiteral

defshould_continue(state: MessagesState) -> Literal["run_tool", "__end__"]:
    """判断是否需要转到工具处理器，或者结束"""
    messages = state["messages"]
    last_message = messages[-1]
    
    # 如果最后一条消息是工具调用
    if last_message.tool_calls:
        return"run_tool"
    # 否则，停止（回复用户）
    return END

NODE_NAME_MAPPER = {
    "run_tool": "run_tool",
    END: END
}

这个函数判断是否需要使用工具，并将返回值映射到对应的节点名称。注意这里的END可以理解成和__end__是同一个内容。

最后，我们通过add_conditional_edges函数将条件判断加入到图中：

workflow = StateGraph(MessagesState)
workflow.add_node("call_llm", call_llm)
workflow.add_node("run_tool", run_tool)
workflow.add_edge(START, "call_llm")
workflow.add_conditional_edges("call_llm", should_continue, NODE_NAME_MAPPER)
workflow.add_edge("run_tool", END)

app = workflow.compile()

这样，我们就构建了一个具有条件分支的智能体，它能够根据LLM的输出决定是否需要调用工具。结构如下图所示：

条件分支图

2.6、使用预构建的智能体

为了方便开发者快速构建智能体，LangGraph提供了一个抽象方法。使用这个方法你可以快速的构建agent，你只需要提供：

• LLM作为大脑
• 一组工具
• Prompt指令

from langgraph.prebuilt import create_react_agent

agent = create_react_agent(
    model=llm,
    tools=[write_email],
    prompt="使用提供的工具完成用户的请求",
)

上述代码创建了一个 ReAct agent，这个预构建的智能体本质上就是LLM借助工具与环境的循环交互，它会自动处理工具调用的逻辑。它的结构图如下：

预构建的智能体

3、持久化能力：赋予智能体记忆

LangGraph的持久化功能是其一大亮点，它可以持久化存储图状态，使开发记忆功能和人类干预功能成为可能。

3.1、理解持久化概念

在深入使用持久化功能之前，我们需要理解几个核心概念：

Thread（线程）：表示围绕一个特定主题展开的一系列相关讨论或消息的集合，即"主题线索"或"讨论串"。在LangGraph中，thread可以理解为一次任务执行过程中不同时间点的所有状态的集合。

Checkpoint（检查点）：表示在某个时间点thread的快照，里面存储着当前时间点的状态信息。

Super-step（超级步骤）：任务执行过程中，每个顺序执行的节点被称为一个super-step。

从概念上讲，快照（StateSnapshot）、Checkpoint、Super-step是同一时间点的不同层次的概念表达。正如官方文档所说：

Checkpoint is a snapshot of the graph state saved at each super-step and is represented by StateSnapshot object.

checkpointer

有了持久化能力后，任何时间任何步骤的状态信息都可以被存储下来，这为我们后续在智能体中实现中断、回溯、恢复等一系列功能提供了基础。

3.2、选择合适的存储方式

Checkpoint是LangGraph中针对持久化存储所设计的一个抽象类，它定义了如何将状态信息存储到不同的介质中。LangGraph提供了多种Checkpoint的实现：

• InMemorySaver：将状态信息存储到内存中
• SqliteSaver：将状态信息存储到SQLite数据库中
• PostgresSaver：将状态信息存储到PostgreSQL数据库中

让我们使用InMemorySaver来演示如何使用持久化功能：

from langgraph.checkpoint.memory import InMemorySaver

agent = create_react_agent(
    model=llm,
    tools=[write_email],
    prompt="使用提供的工具完成用户的请求",
    checkpointer=InMemorySaver(),
)

对于自定义构建的智能体，你可以使用下面的方式来指定持久化存储的实现。

graph = workflow.compile(checkpointer=InMemorySaver())

3.3、实现对话记忆

现在让我们看看持久化如何在实际应用中发挥作用。首先，我们需要为每个对话线程指定一个唯一的 thread_id，它完全是自定义的，一个新的 thread_id表示一个新的任务的开始：

config = {"configurable": {"thread_id": "1"}}  # 指定当前thread的id

result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "给出一些写邮件的建议",
    }]
}, config)

通过get_state方法，我们可以通过thread_id获取最新的状态。最新的state中保留着所有的历史对话信息：

config = {"configurable": {"thread_id": "1"}}
state = agent.get_state(config)

for message in state.values["messages"]:
    message.pretty_print()

当我们继续对话时，新的对话内容会被添加到最新的state中，实现了真正的对话记忆功能：

# 继续对话
result = agent.invoke({
    "messages": [{
        "role": "user", 
        "content": "很好，请使用第3条建议，给我的同事写一封信，告诉他项目进度一切正常。",
    }]
}, config)

智能体能够记住之前提到的"第3条建议"，并据此生成相应的邮件内容。这展示了持久化功能如何让智能体具备了上下文记忆能力。

3.4、查看历史状态

LangGraph还提供了查看历史状态的功能。通过get_state_history方法，你可以获取某个thread的所有历史状态：

history = agent.get_state_history(config)
for state in history:
    print(f"Step: {state.step}, Next: {state.next}")

这个功能对于调试和理解智能体的执行过程非常有用，你可以清楚地看到每一步的状态变化。

4、总结

通过这篇文章，我们深入了解了LangGraph的两大核心能力：控制和持久化。

在控制能力方面，LangGraph为我们提供了一个清晰而灵活的图构建方式。通过State、Nodes和Edges的组合，我们可以构建从简单的线性流程到复杂的条件分支流程。无论是自定义构建还是使用预构建的智能体，LangGraph都给了开发者足够的控制权，让你能够精确地定义智能体的行为逻辑。

在持久化能力方面，LangGraph通过Thread、Checkpoint和Super-step的概念，为智能体提供了强大的状态管理和记忆功能。这不仅让智能体能够记住历史对话，还为实现更复杂的功能（如中断恢复、人类干预等）奠定了基础。

LangGraph的设计哲学是透明和可控，它不会将复杂性隐藏在抽象层后面，而是让开发者能够清楚地了解每一个环节的工作原理。这种设计让LangGraph成为了构建生产级智能体应用的理想选择。

当你开始使用LangGraph时，建议从简单的线性流程开始，逐步掌握State、Nodes和Edges的使用方法，然后再探索条件分支和持久化功能。随着对框架理解的深入，你会发现LangGraph为构建复杂智能体系统提供了无限的可能性。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线科技企业深耕十二载，见证过太多因技术卡位而跃迁的案例。那些率先拥抱 AI 的同事，早已在效率与薪资上形成代际优势，我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在大模型的学习中的很多困惑。我们整理出这套 AI 大模型突围资料包：

• ✅ 从零到一的 AI 学习路径图
• ✅ 大模型调优实战手册（附医疗/金融等大厂真实案例）
• ✅ 百度/阿里专家闭门录播课
• ✅ 大模型当下最新行业报告
• ✅ 真实大厂面试真题
• ✅ 2025 最新岗位需求图谱

所有资料 ⚡️ ，朋友们如果有需要 《AI大模型入门+进阶学习资源包》，下方扫码获取~

① 全套AI大模型应用开发视频教程

（包含提示工程、RAG、LangChain、Agent、模型微调与部署、DeepSeek等技术点）

② 大模型系统化学习路线

作为学习AI大模型技术的新手，方向至关重要。 正确的学习路线可以为你节省时间，少走弯路；方向不对，努力白费。这里我给大家准备了一份最科学最系统的学习成长路线图和学习规划，带你从零基础入门到精通！

③ 大模型学习书籍&文档

学习AI大模型离不开书籍文档，我精选了一系列大模型技术的书籍和学习文档（电子版），它们由领域内的顶尖专家撰写，内容全面、深入、详尽，为你学习大模型提供坚实的理论基础。

④ AI大模型最新行业报告

2025最新行业报告，针对不同行业的现状、趋势、问题、机会等进行系统地调研和评估，以了解哪些行业更适合引入大模型的技术和应用，以及在哪些方面可以发挥大模型的优势。

⑤ 大模型项目实战&配套源码

学以致用，在项目实战中检验和巩固你所学到的知识，同时为你找工作就业和职业发展打下坚实的基础。

⑥ 大模型大厂面试真题

面试不仅是技术的较量，更需要充分的准备。在你已经掌握了大模型技术之后，就需要开始准备面试，我精心整理了一份大模型面试题库，涵盖当前面试中可能遇到的各种技术问题，让你在面试中游刃有余。

以上资料如何领取？

为什么大家都在学大模型？

最近科技巨头英特尔宣布裁员2万人，传统岗位不断缩减，但AI相关技术岗疯狂扩招，有3-5年经验，大厂薪资就能给到50K*20薪！

不出1年，“有AI项目经验”将成为投递简历的门槛。

风口之下，与其像“温水煮青蛙”一样坐等被行业淘汰，不如先人一步，掌握AI大模型原理+应用技术+项目实操经验，“顺风”翻盘！

这些资料真的有用吗？

这份资料由我和鲁为民博士(北京清华大学学士和美国加州理工学院博士)共同整理，现任上海殷泊信息科技CEO，其创立的MoPaaS云平台获Forrester全球’强劲表现者’认证，服务航天科工、国家电网等1000+企业，以第一作者在IEEE Transactions发表论文50+篇，获NASA JPL火星探测系统强化学习专利等35项中美专利。本套AI大模型课程由清华大学-加州理工双料博士、吴文俊人工智能奖得主鲁为民教授领衔研发。

资料内容涵盖了从入门到进阶的各类视频教程和实战项目，无论你是小白还是有些技术基础的技术人员，这份资料都绝对能帮助你提升薪资待遇，转行大模型岗位。

以上全套大模型资料如何领取？