1. LangGraph 是什么

LangGraph 是一个用于构建有状态、多智能体（Multi-Agent）LLM 应用的编排框架，它在 LangChain 生态之上，把「大模型调用、工具调用、路由、循环、人工介入」等能力组织成一张图（Graph）。

你可以把它理解为：传统链式调用（Chain）是一条直线，而 LangGraph 是一张可以循环、分支、状态共享的图，特别适合构建 Agent、聊天机器人、自动化工作流等场景。

LangGraph 具备以下几个特点：

• 状态驱动：所有节点共享一份显式的 State，数据流向清晰、可调试。
• 原生支持循环：Agent 的「思考 → 调用工具 → 再思考」天然就是循环。
• 支持持久化：通过 Checkpointer 可以保存每一步的状态，实现记忆、断点续跑与时间旅行调试。
• 灵活可控：节点、边、条件路由都可以精细定制。

2. 为什么需要 LangGraph

假设你要做一个「能查天气的智能助手」，它的执行流程通常是：

1. 用户提问；
2. 大模型判断是否需要调用天气工具；
3. 如果需要，则调用工具并拿到结果；
4. 把工具结果再次交给大模型生成最终回答；
5. 如果不需要，则直接回答。

这个流程里包含条件判断和循环，用普通的链式调用很难优雅地表达，而 LangGraph 可以用图轻松描述：

3. 核心概念速览

在动手写代码之前，先理解几个核心名词：

概念	作用
State	图中所有节点共享的数据结构，通常用 TypedDict 定义
Node	图中的一个执行单元，对应一个函数
Edge	节点之间的连接，决定执行顺序
Conditional Edge	条件边，根据当前状态动态决定下一个节点
START / END	虚拟节点，表示图的入口和出口
Checkpointer	状态持久化器，用于保存执行历史和实现记忆

4. 环境准备

推荐使用 Python 3.10 及以上版本。安装核心依赖：

pip install langgraph langchain-openai

如果你使用其他模型提供商，也可以安装对应的包，例如：

pip install langchain-anthropic   # Anthropic Claude
pip install langchain-openai       # OpenAI
pip install langchain-ollama       # 本地 Ollama

同时准备好模型 API Key（以 OpenAI 为例）：

import os

os.environ["OPENAI_API_KEY"] = "sk-你的密钥"

5. 第一个 LangGraph 程序：最小可用图

我们从最简单的「单节点图」开始，理解 State、Node 和 Graph 的基本用法。

from typing import TypedDict
from langgraph.graph import StateGraph, START, END


# 1. 定义共享状态
class State(TypedDict):
    messages: list


def chatbot(state: State) -> dict:
    """一个简单的节点：这里暂不调用模型，直接返回一句固定回复。"""
    return {"messages": state["messages"] + ["你好，我是 LangGraph 助手！"]}


# 2. 构建图
graph = StateGraph(State)
graph.add_node("chatbot", chatbot)

graph.add_edge(START, "chatbot")   # 入口连到 chatbot
graph.add_edge("chatbot", END)     # chatbot 连到出口

# 3. 编译并运行
app = graph.compile()
result = app.invoke({"messages": []})
print(result["messages"])

运行后输出类似：

['你好，我是 LangGraph 助手！']

解释一下这段代码：

• State 是所有节点共享的「状态」，节点函数返回的字典会合并回 State；
• add_node 注册节点，节点名即为图中的标识；
• add_edge 连接节点，START 和 END 是内置的入口、出口；
• compile 把图编译为可执行的 app，invoke 传入初始状态并执行。

6. 构建一个带循环的聊天机器人

真实场景中，大模型调用工具后需要把工具结果再送回模型，这就形成了循环。先看一个最小循环例子：

from typing import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    messages: list
    turn: int


llm = ChatOpenAI(model="gpt-4o-mini")


def chatbot(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {
        "messages": state["messages"] + [response],
        "turn": state["turn"] + 1,
    }


def should_continue(state: State) -> str:
    # 只让它跑 3 轮，演示循环后退出
    if state["turn"] < 3:
        return "continue"
    return "end"


graph = StateGraph(State)
graph.add_node("chatbot", chatbot)

graph.add_edge(START, "chatbot")
graph.add_conditional_edges(
    "chatbot",
    should_continue,
    {
        "continue": "chatbot",
        "end": END,
    },
)

app = graph.compile()

result = app.invoke(
    {
        "messages": [HumanMessage(content="请用三句话介绍你自己")],
        "turn": 0,
    }
)

for msg in result["messages"]:
    print(msg.content)

在这个例子中：

• should_continue 是条件路由函数，根据当前 turn 决定回到 chatbot 还是结束；
• add_conditional_edges 把条件函数返回的字符串映射到具体节点；
• 循环让模型能够「多次思考」，这正是 Agent 工作的基础。

7. 接入工具调用（Tool Calling）

有了循环，我们再给模型加一个「查询天气」的工具，体验真正的 Agent 工作流。

from typing import Literal, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.prebuilt import ToolNode


@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气"""
    return f"{city} 今天晴天，气温 26°C。"


tools = [get_weather]
llm = ChatOpenAI(model="gpt-4o-mini").bind_tools(tools)


class State(TypedDict):
    messages: list


def chatbot(state: State) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


# ToolNode 会自动执行模型请求的工具调用，并把结果写成 ToolMessage
tool_node = ToolNode(tools)


def route_after_chatbot(state: State) -> Literal["tools", "__end__"]:
    last_message = state["messages"][-1]
    if getattr(last_message, "tool_calls", None):
        return "tools"
    return "__end__"


graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_node("tools", tool_node)

graph.add_edge(START, "chatbot")
graph.add_conditional_edges(
    "chatbot",
    route_after_chatbot,
    {"tools": "tools", "__end__": END},
)
graph.add_edge("tools", "chatbot")  # 工具执行完，结果送回模型

app = graph.compile()

result = app.invoke(
    {"messages": [HumanMessage(content="北京今天天气怎么样？")]}
)

print(result["messages"][-1].content)

执行流程如下：

要点总结：

• bind_tools 让模型具备生成 tool_calls 的能力；
• ToolNode 负责真正执行工具并返回 ToolMessage；
• 条件路由判断「是否需要调用工具」，从而形成「模型 ↔ 工具」的循环，直到模型认为可以给出最终答案。

8. 持久化：让对话拥有记忆

默认情况下，每次调用 invoke 都是无状态的。要让 Agent 记住多轮对话，需要引入 Checkpointer：

from langgraph.checkpoint.memory import MemorySaver

app = graph.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "user-001"}}

app.invoke(
    {"messages": [HumanMessage(content="我叫小明")]},
    config=config,
)

result = app.invoke(
    {"messages": [HumanMessage(content="我叫什么？")]},
    config=config,
)

print(result["messages"][-1].content)

这里的 thread_id 相当于「会话 ID」：

• 相同 thread_id 共享记忆；
• 不同 thread_id 相互隔离；
• MemorySaver 只保存在内存中，生产环境可替换为数据库（如 SqliteSaver、Postgres 等）。

9. 小结与后续学习

通过本文你已经掌握了 LangGraph 的核心用法：

1. 用 StateGraph 和 State 构建图；
2. 用 add_node、add_edge 连接节点；
3. 用条件边实现循环与路由；
4. 用 ToolNode 接入工具调用，实现 Agent；
5. 用 Checkpointer 实现对话记忆。

后续可以继续深入：

• 子图（Subgraph）：把复杂流程拆分成可复用的子图；
• 多智能体（Multi-Agent）：多个 Agent 节点互相协作；
• 人工介入（Human-in-the-loop）：在关键步骤暂停，等待人工确认；
• 时间旅行调试：借助 Checkpointer 回放到任意历史节点。

LangGraph 官方文档：https://langchain-ai.github.io/langgraph/