目录

一、LangGraph 核心介绍

1.1 基本概述

1.2 核心概念

1.3 主要特点

1.4 使用场景

1.5 与LangChain的关系

二、LangGraph 快速开始（实操落地）

2.1 环境准备

2.2 创建ReAct Agent（快速上手）

2.3 自定义工作流（核心实操）

2.3.1 基础聊天机器人（无工具）

2.3.2 添加工具调用（增强能力）

2.3.3 添加记忆（实现多轮对话连贯）

三、核心进阶：状态持久化与记忆管理

3.1 记忆的两种类型

3.2 InMemoryStore 与 MemorySaver 的区别

3.3 结合两种存储的实操示例

四、实战案例：LangGraph 实现RAG与人工介入

4.1 实现RAG（检索增强生成）

4.2 加入分支：检索不到答案转人工处理

五、总结与拓展

---

在大模型应用开发中，简单的线性任务链已无法满足复杂场景需求，而LangGraph作为LangChain生态下的图结构工作流框架，凭借其状态管理、循环逻辑和多主体协作能力，成为构建智能代理（Agent）的核心工具。本文将从基础介绍到实操落地，带你快速掌握LangGraph的核心用法，避开冗余细节，聚焦实用知识点。

一、LangGraph 核心介绍

1.1 基本概述

LangGraph是LangChain团队开源的一款底层编排框架，专为构建基于大语言模型（LLM）的复杂、有状态、多主体应用而设计。它通过图结构（Graph）建模工作流，相比传统线性工作流，具备更强的灵活性和控制能力，尤其适合需要循环推理、状态管理和多主体协作的场景（如智能代理、多步骤任务处理）。

核心定位：以图结构为核心，简化复杂AI工作流的构建、部署与管理，支持LLM与外部工具的无缝集成。

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

1.2 核心概念

LangGraph的核心能力源于其图结构设计，关键概念可总结为3点：

• 图结构（Graph Structure）：工作流被抽象为有向图，节点（Nodes）代表具体操作（调用LLM、执行工具等），边（Edges）定义节点执行顺序，支持普通边和条件边（动态选择下一步）。

• 状态管理（State Management）：共享状态贯穿整个图执行过程，记录上下文信息，每个节点可读取状态、执行任务并更新状态，确保多步骤交互的一致性。

• 循环能力（Cyclical Workflows）：区别于LangChain的线性任务链（LCEL），支持循环逻辑，可实现反复推理、动态交互（如Agent循环调用工具直到达成目标）。

1.3 主要特点

• 灵活性：精细控制工作流逻辑与状态更新，适配复杂业务需求；

• 持久性：内置状态保存与恢复机制，支持错误恢复和长时间任务；

• 多主体协作：支持多个Agent分工协作，通过图结构协调交互；

• 工具集成：轻松对接外部工具（如搜索API）和自定义函数；

• 人机交互：支持“human-in-the-loop”，关键步骤可由人类介入决策。

1.4 使用场景

LangGraph尤其适合以下3类场景，覆盖大部分复杂AI应用需求：

• 对话代理：构建具备上下文记忆、动态调整策略的智能聊天机器人；

• 多步骤任务：处理需分解为多个阶段的复杂任务（如研究、写作、数据分析）；

• 多代理系统：协调多个Agent分工（如一个负责搜索、一个负责总结）。

1.5 与LangChain的关系

两者同属LangChain生态，但定位不同，可互补使用：

• LangGraph是独立模块，可单独使用，也可结合LangChain组件（提示模板、工具接口）；

• LangChain擅长线性任务链（DAG），LangGraph专注于复杂循环和多主体场景。

二、LangGraph 快速开始（实操落地）

以下实操基于通义千问模型（qwen-max/qwen-plus），简化冗余运行日志，保留核心代码与步骤，可直接复用。

2.1 环境准备

先安装所需依赖（集中整理，避免重复）：

!pip install -U langgraph langchain langchain-community dashscope langchain-tavily pymupdf faiss-cpu

2.2 创建ReAct Agent（快速上手）

LangGraph内置ReAct架构Agent，可快速配置LLM、工具和记忆，实现简单的工具调用能力：

from langchain_community.chat_models.tongyi import ChatTongyi
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.memory import InMemorySaver

# 自定义工具（示例：获取天气）
def get_weather(city: str) -> str:  
    """Get weather for a given city."""
    return f"It's always sunny in {city}!"

# 配置：内存存储（短期记忆）、LLM、工具
checkpointer = InMemorySaver()
model = ChatTongyi(model="qwen-max", temperature=0)

# 创建ReAct Agent
agent = create_react_agent(
    model=model,
    tools=[get_weather],
    checkpointer=checkpointer,
    prompt="You are a helpful assistant"
)

# 运行Agent（简化调用，无需冗余输出）
config = {"configurable": {"thread_id": "1"}}
agent.invoke(
    {"messages": [{"role": "user", "content": "上海的天气怎样"}]},
    config
)

2.3 自定义工作流（核心实操）

自定义工作流是LangGraph的核心能力，分为3个关键步骤：构建状态、定义节点、配置边，以下分场景实现。

2.3.1 基础聊天机器人（无工具）

构建最简单的图工作流，仅包含一个聊天节点，实现基础对话：

from typing import Annotated
from langchain_community.chat_models.tongyi import ChatTongyi
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

# 1. 定义状态（存储对话消息）
class State(TypedDict):
    messages: Annotated[list, add_messages]  # add_messages自动追加消息

# 2. 创建图构建器，定义节点
graph_builder = StateGraph(State)
llm = ChatTongyi(model="qwen-plus", temperature=0)

# 聊天节点：读取状态中的消息，调用LLM并返回新消息
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 3. 添加节点和边（START→chatbot→END）
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

# 编译图（核心步骤）
graph = graph_builder.compile()

# 运行（简化交互逻辑）
def stream_graph_updates(user_input: str):
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        for value in event.values():
            print("Assistant:", value["messages"][-1].content)

2.3.2 添加工具调用（增强能力）

给聊天机器人添加Tavily搜索引擎，实现“提问→工具调用→回答”的闭环，核心是添加工具节点和条件边：

from langchain_tavily import TavilySearch
from langgraph.prebuilt import ToolNode

# 1. 初始化工具
tavily_search = TavilySearch(max_results=5)
tools = [tavily_search]

# 2. 给LLM绑定工具（告知可调用的工具）
llm_with_tools = llm.bind_tools(tools)

# 3. 重写聊天节点（使用绑定工具的LLM）
def chatbot(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# 4. 添加工具节点，配置条件边（判断是否需要调用工具）
graph_builder = StateGraph(State)
graph_builder.add_node("chatbot", chatbot)
tool_node = ToolNode([tavily_search])
graph_builder.add_node("tools", tool_node)

# 条件边：根据LLM输出判断是否调用工具
def route_tools(state: State):
    messages = state.get("messages", [])
    if not messages:
        raise ValueError("No messages found in state")
    ai_message = messages[-1]
    # 有工具调用则跳转到工具节点，否则结束
    if hasattr(ai_message, "tool_calls") and len(ai_message.tool_calls) > 0:
        return "tools"
    return END

# 配置边：START→chatbot，chatbot→工具/END，工具→chatbot（形成闭环）
graph_builder.add_conditional_edges("chatbot", route_tools, {"tools": "tools", END: END})
graph_builder.add_edge("tools", "chatbot")
graph_builder.add_edge(START, "chatbot")

# 编译并运行（逻辑同前，可复用stream_graph_updates函数）
graph = graph_builder.compile()

2.3.3 添加记忆（实现多轮对话连贯）

通过检查点（checkpointer）实现状态持久化，让机器人记住对话上下文，核心是编译图时添加记忆存储：

from langgraph.checkpoint.memory import InMemorySaver

# 初始化内存记忆（生产环境建议用Redis/Sqlite/Postgres存储）
memory = InMemorySaver()

# 编译图时添加记忆（关键步骤）
graph = graph_builder.compile(checkpointer=memory)

# 运行时指定thread_id，确保状态复用
config = {"configurable": {"thread_id": "1"}}
# 第一次交互
graph.stream({"messages": [{"role": "user", "content": "Hi there! My name is Lutianyu."}]}, config)
# 第二次交互（可记住姓名）
graph.stream({"messages": [{"role": "user", "content": "Remember my name?"}]}, config)

三、核心进阶：状态持久化与记忆管理

3.1 记忆的两种类型

LangGraph的记忆功能分为短期和长期，适配不同场景：

• 短期记忆：线程范围内的记忆，用于单个对话线程，通过checkpointer保存，可随时恢复，确保单轮对话连贯；

• 长期记忆：跨对话线程共享，可自定义命名空间（如按用户ID划分），用于保存用户偏好、历史信息等，实现个性化交互。

3.2 InMemoryStore 与 MemorySaver 的区别

两者均用于存储数据，但定位不同，可结合使用：

• InMemoryStore：基于内存的临时存储，侧重快速访问，适合短期会话数据；

• MemorySaver：用于状态持久化，可将数据保存到数据库，适合长期存储、跨会话复用。

3.3 结合两种存储的实操示例

实现“记忆用户信息→跨线程复用”，核心是结合InMemoryStore（存储长期记忆）和MemorySaver（存储短期状态）：

from langgraph.store.memory import InMemoryStore
from langchain_community.embeddings import DashScopeEmbeddings
import uuid
from langgraph.graph import MessagesState

# 1. 初始化长期记忆存储（结合嵌入模型）
in_memory_store = InMemoryStore(
    index={
        "embed": DashScopeEmbeddings(model="text-embedding-v1"),
        "dims": 1536,
    }
)

# 2. 定义节点：调用LLM，检索/存储长期记忆
def call_model(state: MessagesState, config, *, store):
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)  # 按用户ID划分记忆空间
    # 检索用户相关记忆
    memories = store.search(namespace, query=str(state["messages"][-1].content))
    info = "\n".join([d.value["data"] for d in memories])
    
    # 存储新记忆（当用户要求"remember"时）
    last_message = state["messages"][-1]
    if "remember" in last_message.content.lower():
        store.put(namespace, str(uuid.uuid4()), {"data": "User name is Lutianyu"})
    
    # 调用LLM，传入记忆上下文
    response = model.invoke(
        [{"role": "system", "content": f"User info: {info}"}] + state["messages"]
    )
    return {"messages": response}

# 3. 构建图并编译（结合短期记忆和长期记忆）
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_edge(START, "call_model")
graph = builder.compile(checkpointer=MemorySaver(), store=in_memory_store)

四、实战案例：LangGraph 实现RAG与人工介入

4.1 实现RAG（检索增强生成）

结合FAISS向量库和LangGraph，构建“文档检索→LLM回答”的工作流，核心是添加检索节点：

from langchain_community.embeddings import DashScopeEmbeddings
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import PyMuPDFLoader
from langchain.prompts import ChatPromptTemplate, HumanMessagePromptTemplate

# 1. 文档加载、切分、构建向量库
loader = PyMuPDFLoader("./data/deepseek-v3-1-4.pdf")  # 替换为你的文档路径
pages = loader.load_and_split()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=512, chunk_overlap=200)
texts = text_splitter.create_documents([page.page_content for page in pages[:2]])
embeddings = DashScopeEmbeddings(model="text-embedding-v1")
db = FAISS.from_documents(texts, embeddings)
retriever = db.as_retriever(search_kwargs={"k": 5})

# 2. 定义检索节点：根据用户问题检索文档
def retrieval(state: MessagesState):
    user_query = state["messages"][-1] if state["messages"] else ""
    docs = retriever.invoke(str(user_query))
    # 构建提示模板，传入检索结果
    prompt = ChatPromptTemplate.from_messages([
        HumanMessagePromptTemplate.from_template("请根据对话历史和下面提供的信息回答用户问题:\n{query}")
    ])
    messages = prompt.invoke("\n".join([doc.page_content for doc in docs])).messages
    return {"messages": messages}

# 3. 构建RAG工作流：START→检索→聊天→END
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("retrieval", retrieval)
graph_builder.add_node("chatbot", chatbot)  # 复用之前的chatbot节点
graph_builder.add_edge(START, "retrieval")
graph_builder.add_edge("retrieval", "chatbot")
graph_builder.add_edge("chatbot", END)
graph = graph_builder.compile()

4.2 加入分支：检索不到答案转人工处理

通过条件边实现“检索验证→正常回答/人工介入”的分支逻辑，提升应用健壮性：

from langchain.schema import HumanMessage, AIMessage
from typing import Literal
from langgraph.types import interrupt, Command

# 1. 验证节点：判断检索信息是否能回答问题
def verify(state: MessagesState) -> Literal["chatbot", "ask_human"]:
    verify_msg = HumanMessage("请判断已知信息是否能回答用户问题，直接输出Y或N")
    ret = llm.invoke(state["messages"] + [verify_msg])
    return "chatbot" if "Y" in ret.content else "ask_human"

# 2. 人工介入节点：中断流程，等待人工输入
def ask_human(state: MessagesState):
    user_query = state["messages"][-2].content
    human_response = interrupt({"question": user_query})  # 中断并获取人工输入
    return {"messages": [AIMessage(human_response)]}

# 3. 重构工作流，添加分支
graph_builder = StateGraph(MessagesState)
graph_builder.add_node("retrieval", retrieval)
graph_builder.add_node("chatbot", chatbot)
graph_builder.add_node("ask_human", ask_human)

# 配置边：START→检索，检索→验证→chatbot/ask_human，ask_human/chatbot→END
graph_builder.add_edge(START, "retrieval")
graph_builder.add_conditional_edges("retrieval", verify)
graph_builder.add_edge("ask_human", END)
graph_builder.add_edge("chatbot", END)

# 编译（需添加记忆，支持中断后恢复）
graph = graph_builder.compile(checkpointer=MemorySaver())

五、总结与拓展

LangGraph的核心价值的是“用图结构解耦复杂工作流”，通过节点、边和状态管理，实现循环逻辑、多主体协作和状态持久化，相比传统线性框架，更适合构建复杂AI代理。

除本文介绍的功能外，LangGraph还支持并行处理、子图管理、历史动作回放（调试）、多智能体协作等高级特性，更多细节可参考官方文档。

拓展建议：生产环境中，将InMemorySaver替换为RedisSaver/PostgresSaver，提升状态存储的稳定性；结合LangChain的其他组件（如提示模板、工具集），进一步简化开发流程。