1.安装与入门

LangGraph 是 LangChain 生态中一个低级别的编排框架，专门用于构建、管理和部署长时间运行、有状态的智能体系统。与 LangChain 提供的高级抽象不同，LangGraph 给予开发者对智能体行为更细粒度的控制，同时内置了持久化执行、人工介入、记忆管理等生产级特性。Klarna、Replit、Elastic 等公司已经在使用 LangGraph 构建下一代智能体应用。

LangGraph 环境安装配置

安装 LangGraph 非常简单，直接使用 pip 即可。建议创建独立的虚拟环境来管理依赖。

pip install -U langgraph

这个命令会安装 LangGraph 核心库以及必要的依赖。如果计划使用 LangChain 的模型集成，可以同时安装 langchain：

pip install -U langgraph langchain

对于 JavaScript/TypeScript 开发者，可以使用 npm 或 yarn：

npm install @langchain/langgraph @langchain/core

安装完成后，可以通过简单的导入测试来验证环境是否正常：

from langgraph.graph import StateGraph, START
from typing_extensions import TypedDict

class State(TypedDict):
    message: str

# 如果能正常导入并创建图对象，说明安装成功
graph_builder = StateGraph(State)

环境配置的关键在于理解 LangGraph 的依赖关系。LangGraph 可以独立使用，但通常会配合 LangChain 的模型包装器、工具定义和提示词模板。如果项目需要这些功能，建议一并安装相关包。

对于生产环境，还需要考虑检查点存储（checkpointer）的后端配置。LangGraph 支持内存、SQLite、PostgreSQL 等多种存储后端，这部分会在后续章节详细讨论。

创建首个 StateGraph 图

StateGraph 是 LangGraph 的核心抽象，它将工作流建模为有向图，其中节点（Node）执行具体操作，边（Edge）控制执行流程。每个图都有一个状态（State），在节点间传递和演化。

让我们从一个最简单的例子开始：一个两节点的顺序执行图。

from langgraph.graph import START, StateGraph
from typing_extensions import TypedDict

class State(TypedDict):
    text: str

def node_a(state: State) -> dict:
    return {"text": state["text"] + "a"}

def node_b(state: State) -> dict:
    return {"text": state["text"] + "b"}

# 创建图并定义状态模式
graph = StateGraph(State)

# 添加节点
graph.add_node("node_a", node_a)
graph.add_node("node_b", node_b)

# 定义边：从 START 到 node_a，从 node_a 到 node_b
graph.add_edge(START, "node_a")
graph.add_edge("node_a", "node_b")

# 编译图
compiled_graph = graph.compile()

# 执行图
result = compiled_graph.invoke({"text": ""})
print(result)  # 输出: {'text': 'ab'}

这段代码展示了 LangGraph 的基本模式。首先定义状态模式，这里使用 TypedDict 指定状态包含一个 text 字段。然后创建两个节点函数，每个函数接收当前状态并返回更新。

关键点在于节点函数的返回值。node_a 接收 {"text": ""}，返回 {"text": "a"}，这个更新会被应用到状态上，使状态变为 {"text": "a"}。接着 node_b 接收这个状态，返回 {"text": "ab"}，最终结果就是 {"text": "ab"}。

图的编译步骤很重要。.compile() 会检查图结构的合法性，比如是否有孤立的节点，同时准备运行时所需的通道和检查点机制。编译后的图才能被调用执行。

START 是一个特殊的节点，表示图的入口。LangGraph 会自动将用户输入路由到 START 节点连接的节点。在这个例子中，START -> node_a -> node_b 形成了一条简单的执行链。

定义聊天机器人处理节点

现在让我们构建一个更实用的例子：一个能调用 LLM 的聊天机器人节点。这需要引入 LangChain 的模型包装器。

from typing import Annotated
from typing_extensions import TypedDict
from langchain_core.messages import HumanMessage, AIMessage
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    # messages 字段使用 add_messages reducer，自动追加而非覆盖
    messages: Annotated[list, add_messages]

def chatbot_node(state: State) -> dict:
    # 这里简化处理，实际会调用 LLM
    last_message = state["messages"][-1]
    response = f"收到: {last_message.content}"
    return {"messages": [AIMessage(content=response)]}

# 构建图
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot_node)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

graph = graph_builder.compile()

# 测试
result = graph.invoke({
    "messages": [HumanMessage(content="你好")]
})
print(result["messages"][-1].content)  # 输出: 收到: 你好

这个例子引入了几个重要概念。首先是 add_messages reducer，它定义了状态更新的方式。当节点返回新的消息时，reducer 会将这些消息追加到现有列表，而不是替换整个列表。这对于对话场景至关重要，因为需要保留历史上下文。

节点函数的设计模式很清晰：接收状态，执行逻辑，返回状态更新。更新必须是字典形式，键对应状态模式中的字段。LangGraph 会自动应用这些更新。

对于真实的 LLM 调用，代码会是这样：

from langchain.chat_models import init_chat_model

# 初始化模型，支持 OpenAI、Anthropic、Google 等多种提供商
llm = init_chat_model("anthropic:claude-3-5-sonnet-latest")

def chatbot_node(state: State) -> dict:
    # 调用 LLM 生成回复
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

模型通过 bind_tools 方法可以绑定工具，让 LLM 能够决定何时调用外部函数。这在构建智能体时非常关键。

# 定义工具
def get_weather(city: str) -> str:
    """获取指定城市的天气"""
    return f"{city}的天气总是晴朗的！"

# 绑定工具到模型
llm_with_tools = llm.bind_tools([get_weather])

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

当 LLM 决定使用工具时，返回的 AIMessage 会包含 tool_calls 字段。LangGraph 提供了预建的 ToolNode 来处理工具调用：

from langgraph.prebuilt import ToolNode, tools_condition

# 创建工具节点
tool_node = ToolNode(tools=[get_weather])

# 添加到图
graph_builder.add_node("tools", tool_node)

# 条件边：根据 LLM 输出决定下一步
graph_builder.add_conditional_edges(
    "chatbot",
    tools_condition,  # 如果有工具调用则路由到 tools，否则结束
    ["tools", END]
)

# 工具执行后返回 chatbot
graph_builder.add_edge("tools", "chatbot")

tools_condition 是一个预建的路由函数，检查最后一条消息是否有工具调用。如果有，路由到 tools 节点；否则，路由到 END。这种条件路由是构建循环智能体的基础。

编译并执行图运行

编译是图构建的最后一步，也是执行的前提。编译过程不仅验证图结构，还配置运行时行为，如检查点、断点、线程管理等。

from langgraph.checkpoint.memory import InMemorySaver

# 创建内存检查点保存器
memory = InMemorySaver()

# 编译时传入检查点
graph = graph_builder.compile(checkpointer=memory)

# 现在可以执行带记忆的对话
config = {"configurable": {"thread_id": "conversation_1"}}

# 第一次调用
result1 = graph.invoke(
    {"messages": [HumanMessage(content="记住我的名字是 Alice")]},
    config=config
)

# 第二次调用，能记住上下文
result2 = graph.invoke(
    {"messages": [HumanMessage(content="我叫什么名字？")]},
    config=config
)

检查点机制是 LangGraph 的核心优势。每次节点执行后，状态会被自动保存。通过指定 thread_id，可以在不同调用间恢复状态，实现对话记忆。

执行图时有几种模式：

# 1. invoke：同步阻塞调用，返回最终结果
result = graph.invoke(input, config)

# 2. stream：流式输出中间状态
for event in graph.stream(input, config, stream_mode="values"):
    print(event)  # 每次状态更新都会输出

# 3. astream：异步流式
async for event in graph.astream(input, config):
    print(event)

stream_mode 参数控制输出内容：

• "values"：输出完整状态
• "updates"：只输出被更新的字段
• "debug"：包含详细的调试信息

对于长时间运行的任务，流式输出能让应用实时反馈进度，提升用户体验。

配置对象 config 除了 thread_id，还可以传递其他运行时参数：

config = {
    "configurable": {
        "thread_id": "conversation_1",
        "user_id": "user_123",
        "model_name": "claude-3-5-sonnet"
    },
    "tags": ["production", "chatbot"],
    "metadata": {"session_type": "support"}
}

这些信息会被传递给节点函数，用于日志记录、权限控制、模型选择等场景。

图结构可视化输出

理解复杂图结构的最佳方式是可视化。LangGraph 提供了多种可视化方法，帮助开发者直观地理解决策流程。

最常用的是 Mermaid 图表生成：

from IPython.display import Image, display

# 生成 PNG 图片
try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    # 可能需要安装额外依赖
    pass

对于非 Jupyter 环境，可以保存到文件：

import io
from PIL import Image

# 获取图片数据
png_data = graph.get_graph().draw_mermaid_png()

# 保存到文件
with open("graph.png", "wb") as f:
    f.write(png_data)

生成的图表清晰展示了节点和边的关系。对于条件边，会标注出分支条件。例如，一个带有工具调用的智能体图会显示：

• START -> chatbot
• chatbot -> tools（条件：有工具调用）
• chatbot -> END（条件：无工具调用）
• tools -> chatbot

这种可视化在调试复杂路由逻辑时特别有用。可以一眼看出执行路径，发现潜在的死循环或孤立节点。

除了 PNG，还可以生成 Mermaid 语法文本：

mermaid_code = graph.get_graph().draw_mermaid()
print(mermaid_code)

输出类似：

graph TD;
    __start__([<p>__start__</p>]):::first
    chatbot(chatbot)
    tools(tools)
    __end__([<p>__end__</p>]):::last
    __start__ --> chatbot;
    chatbot -.-> tools;
    chatbot -.-> __end__;
    tools --> chatbot;
    classDef default fill:#f2f0ff,line-height:1.2

这段代码可以直接粘贴到支持 Mermaid 的编辑器（如 Mermaid Live Editor）中渲染，方便在文档或代码仓库中分享。

对于大型图，LangGraph Studio 提供了交互式可视化。可以实时查看状态变化、节点执行顺序、消息传递等细节。这是生产环境调试的首选工具。

在代码中，也可以通过 get_graph() 方法获取图的元数据：

graph_info = graph.get_graph()
print("节点:", list(graph_info.nodes.keys()))
print("边:", graph_info.edges)

这有助于动态检查图结构，特别是在构建可配置的工作流时。

可视化不仅是调试工具，也是设计工具。在编写代码前，先用纸笔画出图结构，明确节点职责和路由逻辑，能避免很多常见错误。LangGraph 的声明式 API 让这种设计驱动开发变得自然。

至此，我们已经掌握了 LangGraph 的核心工作流程：安装、定义状态、创建节点、连接边、编译、执行和可视化。这些基础概念将贯穿后续所有高级特性。下一章将深入探讨状态管理，包括更复杂的状态模式、reducer 函数和消息状态管理。