LangGraph 快速使用说明

一、LangGraph 是什么

LangGraph 是由 LangChain 团队开发的开源框架（MIT 协议），专门用于构建有状态的、基于图结构的 LLM 应用。它的核心思想很直接：把你的 AI 应用逻辑建模为一张有向图——节点（Node）是具体的操作步骤，边（Edge）是步骤之间的流转规则，状态（State）是在节点之间传递的共享数据。

与传统的链式调用（Chain）不同，LangGraph 天然支持循环和条件分支，这让它特别适合构建需要"思考-行动-观察"反复迭代的 Agent 系统。

LangGraph 可以独立于 LangChain 使用，但两者配合使用效果更好。LangChain 提供模型接口和工具集成，LangGraph 负责编排和流程控制。

LangGraph 的核心优势：

• 有状态执行：自动管理和传递状态，节点之间无需手动传参
• 支持循环：Agent 可以反复调用工具直到任务完成，不像简单 Chain 只能走一遍
• 条件路由：根据运行时的结果动态决定下一步走哪个节点
• 持久化：内置 checkpoint 机制，支持中断后恢复、回溯历史状态
• 人工介入（Human-in-the-loop）：可在关键节点暂停，等待人工审核后继续
• 流式输出：原生支持 token 级别的流式响应

适用场景：

• 需要调用外部工具的对话 Agent
• 多步骤的 RAG 流水线（检索、评估、重写、生成）
• 多 Agent 协作系统
• 需要人工审批的自动化工作流
• 任何需要循环决策的 LLM 应用

---

二、安装与环境准备

2.1 基础安装

pip install langgraph

如果要配合 LangChain 的模型接口使用（推荐），还需要安装对应的集成包：

# 使用 OpenAI
pip install langchain-openai

# 使用 Anthropic
pip install langchain-anthropic

# 使用 Google
pip install langchain-google-genai

2.2 环境变量

# 根据你使用的模型设置对应的 API Key
export OPENAI_API_KEY="sk-..."
export ANTHROPIC_API_KEY="sk-ant-..."

2.3 版本要求

LangGraph 需要 Python 3.9 及以上版本。建议使用 Python 3.11+，以获得更好的类型提示支持。截至 2025 年，LangGraph 的最新稳定版本在 PyPI 上持续更新，建议通过 pip install -U langgraph 获取最新版。

---

三、核心概念详解

理解以下四个概念，就能掌握 LangGraph 的 80% 用法。

3.1 State（状态）

State 是整个图的"记忆"。它是一个 TypedDict，定义了在图中流转的所有数据字段。每个节点都能读取和更新 State。

from typing import TypedDict

class MyState(TypedDict):
    query: str           # 用户输入
    result: str          # 最终结果
    steps: list[str]     # 中间步骤记录

关键机制——Reducer： 默认情况下，节点返回的字段值会直接覆盖 State 中的对应字段。但对于消息列表这种需要"追加"而非"覆盖"的场景，LangGraph 提供了 Reducer 机制：

from typing import Annotated
from langgraph.graph.message import add_messages

class ChatState(TypedDict):
    # add_messages 会把新消息追加到列表末尾，而不是替换整个列表
    messages: Annotated[list, add_messages]

add_messages 是 LangGraph 内置的 Reducer，它不仅追加消息，还能根据消息 ID 进行去重和更新。

3.2 Node（节点）

节点就是普通的 Python 函数，接收 State 作为输入，返回一个字典表示对 State 的更新。

def analyze(state: MyState) -> dict:
    query = state["query"]
    # 执行某些操作...
    return {"result": f"分析结果：{query}", "steps": ["完成分析"]}

节点函数的返回值不需要包含 State 的所有字段，只需返回本节点要更新的字段即可。

3.3 Edge（边）

边定义了节点之间的连接关系，分为两种：

普通边（Direct Edge）：无条件地从一个节点流向下一个节点。

graph.add_edge("node_a", "node_b")  # 执行完 node_a 后，一定执行 node_b

条件边（Conditional Edge）：根据一个函数的返回值，动态决定走哪条路。

def route(state: MyState) -> str:
    if "错误" in state["result"]:
        return "retry"     # 走 retry 节点
    return "finish"        # 走 finish 节点

graph.add_conditional_edges("check", route, ["retry", "finish"])

3.4 StateGraph（状态图）

StateGraph 是把上述概念组合在一起的构建器。完整的构建流程是：创建 StateGraph → 添加节点 → 添加边 → 编译 → 调用。

from langgraph.graph import StateGraph, START, END

graph = StateGraph(MyState)
graph.add_node("analyze", analyze)
graph.add_edge(START, "analyze")     # START 是内置起点
graph.add_edge("analyze", END)       # END 是内置终点

app = graph.compile()
result = app.invoke({"query": "你好", "result": "", "steps": []})

START 和 END 是 LangGraph 的特殊节点，分别标记图的入口和出口。

---

四、从零构建第一个 Graph

下面通过一个完整的最小示例，展示 LangGraph 的基本用法。

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

# 1. 定义 State
class State(TypedDict):
    text: str
    word_count: int

# 2. 定义节点函数
def process_text(state: State) -> dict:
    text = state["text"]
    cleaned = text.strip().lower()
    return {"text": cleaned}

def count_words(state: State) -> dict:
    words = state["text"].split()
    return {"word_count": len(words)}

# 3. 构建图
graph = StateGraph(State)
graph.add_node("process", process_text)
graph.add_node("count", count_words)

# 4. 定义执行流程
graph.add_edge(START, "process")
graph.add_edge("process", "count")
graph.add_edge("count", END)

# 5. 编译并执行
app = graph.compile()
result = app.invoke({"text": "  Hello World LangGraph  ", "word_count": 0})

print(result)
# {'text': 'hello world langgraph', 'word_count': 3}

执行过程非常清晰：START → process（清洗文本）→ count（统计字数）→ END。State 在节点之间自动传递和更新。

---

五、条件分支——让图会"思考"

条件边是 LangGraph 区别于简单 Chain 的核心特性。通过条件边，图可以在运行时根据实际情况选择不同的执行路径。

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

class State(TypedDict):
    query: str
    category: str
    response: str

def classify(state: State) -> dict:
    query = state["query"].lower()
    if "价格" in query or "多少钱" in query:
        return {"category": "pricing"}
    elif "故障" in query or "报错" in query:
        return {"category": "technical"}
    else:
        return {"category": "general"}

def handle_pricing(state: State) -> dict:
    return {"response": "关于价格问题，请访问我们的定价页面或联系销售团队。"}

def handle_technical(state: State) -> dict:
    return {"response": "关于技术问题，请提供错误日志，我们的工程师会帮您排查。"}

def handle_general(state: State) -> dict:
    return {"response": "感谢您的咨询，请问还有什么可以帮助您的？"}

# 路由函数：根据 category 决定走哪个节点
def route_query(state: State) -> Literal["pricing", "technical", "general"]:
    return state["category"]

# 构建图
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("pricing", handle_pricing)
graph.add_node("technical", handle_technical)
graph.add_node("general", handle_general)

graph.add_edge(START, "classify")
graph.add_conditional_edges("classify", route_query, ["pricing", "technical", "general"])
graph.add_edge("pricing", END)
graph.add_edge("technical", END)
graph.add_edge("general", END)

app = graph.compile()

result = app.invoke({"query": "这个产品多少钱？", "category": "", "response": ""})
print(result["response"])
# 关于价格问题，请访问我们的定价页面或联系销售团队。

路由函数的返回值必须与条件边中定义的目标节点名称一一对应。LangGraph 会根据返回值找到对应的节点并执行。

---

六、构建 Tool-Calling Agent（实战重点）

这是 LangGraph 最常见也最重要的用法：构建一个能调用外部工具的 Agent。Agent 会自主判断是否需要调用工具，调用后根据结果决定是否继续调用或直接回复用户。

6.1 定义工具

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    # 实际项目中这里会调用搜索 API（如 Tavily、SerpAPI 等）
    return f"搜索结果：关于'{query}'的最新信息..."

@tool
def calculate(expression: str) -> str:
    """计算数学表达式"""
    try:
        result = eval(expression)  # 生产环境请使用安全的表达式解析器
        return str(result)
    except Exception as e:
        return f"计算错误：{e}"

tools = [search_web, calculate]

@tool 装饰器来自 LangChain，它会自动将函数签名和 docstring 转换为 LLM 能理解的工具描述格式。

6.2 初始化模型并绑定工具

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)
llm_with_tools = llm.bind_tools(tools)

bind_tools 会把工具列表注册到模型中，使模型在回复时可以选择调用这些工具。

6.3 定义 Agent 节点

from typing import Annotated
from langgraph.graph import MessagesState

# MessagesState 是 LangGraph 内置的快捷状态，等价于：
# class MessagesState(TypedDict):
#     messages: Annotated[list, add_messages]

def agent_node(state: MessagesState) -> dict:
    """调用 LLM，让它决定是直接回复还是调用工具"""
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

6.4 定义工具执行节点

from langchain_core.messages import ToolMessage

# 构建工具名称到工具对象的映射
tools_by_name = {t.name: t for t in tools}

def tool_node(state: MessagesState) -> dict:
    """执行 LLM 决定调用的工具"""
    results = []
    # 获取最后一条消息中的工具调用请求
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        results.append(
            ToolMessage(content=str(observation), tool_call_id=tool_call["id"])
        )
    return {"messages": results}

6.5 定义路由逻辑

from typing import Literal
from langgraph.graph import END

def should_continue(state: MessagesState) -> Literal["tools", END]:
    """判断是否需要继续调用工具"""
    last_message = state["messages"][-1]
    # 如果 LLM 的回复中包含工具调用请求，就走 tools 节点
    if last_message.tool_calls:
        return "tools"
    # 否则结束
    return END

6.6 组装完整的 Agent Graph

from langgraph.graph import StateGraph, START, END, MessagesState

graph = StateGraph(MessagesState)

# 添加节点
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# 定义流程
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, ["tools", END])
graph.add_edge("tools", "agent")  # 工具执行完后回到 agent，形成循环

agent = graph.compile()

注意 graph.add_edge("tools", "agent") 这行——它形成了一个循环。这正是 Agent 的核心模式：LLM 调用工具 → 拿到结果 → 再次让 LLM 判断 → 可能继续调用工具，直到 LLM 认为可以直接回复。

6.7 运行 Agent

from langchain_core.messages import HumanMessage

result = agent.invoke({
    "messages": [HumanMessage(content="帮我算一下 (15 + 27) * 3，然后搜索一下今天的天气")]
})

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

典型的执行流程是：

1. 用户发送消息
2. Agent 节点：LLM 分析后决定先调用 calculate 工具
3. Tools 节点：执行计算，返回结果
4. Agent 节点：LLM 拿到计算结果，决定再调用 search_web
5. Tools 节点：执行搜索，返回结果
6. Agent 节点：LLM 拿到所有结果，组织最终回复
7. 路由判断没有更多工具调用，流向 END

6.8 使用预构建组件简化代码

LangGraph 提供了预构建的 ToolNode 和 tools_condition，可以大幅简化上面的代码：

from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.graph import StateGraph, START, END, MessagesState

graph = StateGraph(MessagesState)
graph.add_node("agent", agent_node)
graph.add_node("tools", ToolNode(tools))  # 替代手写的 tool_node

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", tools_condition)  # 替代手写的 should_continue
graph.add_edge("tools", "agent")

agent = graph.compile()

ToolNode 和 tools_condition 封装了上面手写的逻辑，是生产环境的推荐做法。

---

七、持久化与会话记忆

在实际应用中，Agent 通常需要记住之前的对话内容。LangGraph 通过 Checkpointer 机制实现持久化。

7.1 使用内存 Checkpointer

from langgraph.checkpoint.memory import InMemorySaver

memory = InMemorySaver()
agent = graph.compile(checkpointer=memory)

# 使用 thread_id 来区分不同的会话
config = {"configurable": {"thread_id": "user-123"}}

# 第一轮对话
result1 = agent.invoke(
    {"messages": [HumanMessage(content="我叫小明")]},
    config=config
)

# 第二轮对话——同一个 thread_id，Agent 能记住之前的内容
result2 = agent.invoke(
    {"messages": [HumanMessage(content="你还记得我叫什么吗？")]},
    config=config
)

InMemorySaver 适合开发和测试。生产环境中，LangGraph 支持 SQLite、PostgreSQL 等持久化后端：

from langgraph.checkpoint.sqlite import SqliteSaver

# SQLite 持久化
memory = SqliteSaver.from_conn_string("checkpoints.db")
agent = graph.compile(checkpointer=memory)

7.2 查看和操作历史状态

# 获取当前状态快照
snapshot = agent.get_state(config)
print(snapshot.values["messages"])

# 查看所有历史状态
for state in agent.get_state_history(config):
    print(state.config, len(state.values["messages"]))

---

八、Human-in-the-Loop（人工介入）

在很多业务场景中，AI 的某些决策需要人工确认后才能执行。LangGraph 通过 interrupt_before 或 interrupt_after 参数实现这一点。

# 在执行 tools 节点之前暂停，等待人工确认
agent = graph.compile(
    checkpointer=memory,
    interrupt_before=["tools"]  # 在工具执行前中断
)

config = {"configurable": {"thread_id": "review-session"}}

# 第一次调用——会在 tools 节点前暂停
result = agent.invoke(
    {"messages": [HumanMessage(content="帮我删除所有过期数据")]},
    config=config
)

# 此时可以查看 Agent 打算做什么
snapshot = agent.get_state(config)
pending_tool_calls = snapshot.values["messages"][-1].tool_calls
print(f"Agent 打算调用：{pending_tool_calls}")

# 人工确认后，继续执行（传入 None 表示继续）
result = agent.invoke(None, config=config)

如果人工审核后不同意执行，可以修改状态再继续：

from langchain_core.messages import AIMessage

# 用一个不包含工具调用的消息替换，让 Agent 直接回复
agent.update_state(
    config,
    {"messages": [AIMessage(content="抱歉，该操作需要管理员权限，我无法执行。")]},
    as_node="agent"
)

result = agent.invoke(None, config=config)

---

九、流式输出

LangGraph 支持多种粒度的流式输出，方便前端实时展示 Agent 的执行过程。

9.1 节点级别流式

# stream 返回每个节点执行后的状态更新
for event in agent.stream(
    {"messages": [HumanMessage(content="搜索最新的 AI 新闻")]},
    config=config
):
    for node_name, output in event.items():
        print(f"--- {node_name} ---")
        if "messages" in output:
            for msg in output["messages"]:
                print(msg.content[:100])

9.2 Token 级别流式

# stream_mode="messages" 提供 token 级别的流式输出
for msg, metadata in agent.stream(
    {"messages": [HumanMessage(content="给我写一首诗")]},
    config=config,
    stream_mode="messages"
):
    if msg.content:
        print(msg.content, end="", flush=True)

---

十、子图（Subgraph）——复杂系统的模块化

当系统变得复杂时，可以把部分逻辑封装为子图，然后在主图中作为一个节点使用。

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

# 定义一个子图：负责数据验证
class ValidationState(TypedDict):
    data: str
    is_valid: bool

def check_format(state: ValidationState) -> dict:
    is_valid = len(state["data"]) > 0 and "@" in state["data"]
    return {"is_valid": is_valid}

validation_graph = StateGraph(ValidationState)
validation_graph.add_node("check", check_format)
validation_graph.add_edge(START, "check")
validation_graph.add_edge("check", END)
validation_subgraph = validation_graph.compile()

# 在主图中使用子图
class MainState(TypedDict):
    data: str
    is_valid: bool
    result: str

def process_result(state: MainState) -> dict:
    if state["is_valid"]:
        return {"result": "数据有效，处理成功"}
    return {"result": "数据无效，请检查格式"}

main_graph = StateGraph(MainState)
main_graph.add_node("validate", validation_subgraph)  # 子图作为节点
main_graph.add_node("process", process_result)

main_graph.add_edge(START, "validate")
main_graph.add_edge("validate", "process")
main_graph.add_edge("process", END)

app = main_graph.compile()
result = app.invoke({"data": "user@example.com", "is_valid": False, "result": ""})
print(result["result"])  # 数据有效，处理成功

子图需要和主图共享相同的 State 字段名（或者部分重叠的字段），LangGraph 会自动处理 State 的映射。

---

十一、实战案例：RAG 质量控制 Agent

下面是一个更贴近生产环境的例子——一个带质量评估的 RAG Agent。它会检索文档、评估检索质量，如果质量不达标，会重写查询后重新检索。

from typing import TypedDict, Literal, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o", temperature=0)

class RAGState(TypedDict):
    question: str
    context: str
    answer: str
    quality_score: float
    retry_count: int

def retrieve(state: RAGState) -> dict:
    """模拟文档检索"""
    question = state["question"]
    # 实际项目中这里调用向量数据库
    context = f"检索到的与'{question}'相关的文档内容..."
    return {"context": context}

def evaluate_quality(state: RAGState) -> dict:
    """评估检索质量"""
    prompt = f"请评估以下检索结果与问题的相关性，返回0-1之间的分数。\n问题：{state['question']}\n检索结果：{state['context']}"
    # 简化示例，实际应调用 LLM 并解析分数
    score = 0.8  # 假设得分
    return {"quality_score": score}

def generate_answer(state: RAGState) -> dict:
    """基于检索结果生成回答"""
    prompt = f"根据以下上下文回答问题。\n上下文：{state['context']}\n问题：{state['question']}"
    response = llm.invoke(prompt)
    return {"answer": response.content}

def rewrite_query(state: RAGState) -> dict:
    """重写查询以改善检索质量"""
    prompt = f"原始问题检索效果不佳，请重写以下问题使其更适合检索：\n{state['question']}"
    response = llm.invoke(prompt)
    return {
        "question": response.content,
        "retry_count": state["retry_count"] + 1
    }

def quality_check(state: RAGState) -> Literal["generate", "rewrite", "force_generate"]:
    """根据质量分数决定下一步"""
    if state["quality_score"] >= 0.7:
        return "generate"
    if state["retry_count"] >= 2:
        return "force_generate"  # 重试次数超限，强制生成
    return "rewrite"

# 构建图
graph = StateGraph(RAGState)

graph.add_node("retrieve", retrieve)
graph.add_node("evaluate", evaluate_quality)
graph.add_node("generate", generate_answer)
graph.add_node("rewrite", rewrite_query)

graph.add_edge(START, "retrieve")
graph.add_edge("retrieve", "evaluate")
graph.add_conditional_edges(
    "evaluate",
    quality_check,
    {
        "generate": "generate",
        "rewrite": "rewrite",
        "force_generate": "generate"
    }
)
graph.add_edge("rewrite", "retrieve")  # 重写后重新检索，形成循环
graph.add_edge("generate", END)

rag_agent = graph.compile()

result = rag_agent.invoke({
    "question": "LangGraph 的核心特性是什么？",
    "context": "",
    "answer": "",
    "quality_score": 0.0,
    "retry_count": 0
})

print(result["answer"])

这个图的执行流程是：检索 → 评估 → 如果质量高就生成答案，如果低就重写查询后重新检索。retry_count 作为安全阀，防止无限循环。

---

十二、调试技巧与最佳实践

12.1 可视化图结构

# 生成 Mermaid 图（在 Jupyter 中直接显示）
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))

# 或者输出 Mermaid 文本格式
print(app.get_graph().draw_mermaid())

可视化是调试复杂图的最有效手段，强烈建议在开发阶段始终打开。

12.2 日志和追踪

LangGraph 可以与 LangSmith 集成，实现完整的运行追踪：

export LANGSMITH_API_KEY="ls__..."
export LANGSMITH_TRACING=true

设置环境变量后，每次 invoke 或 stream 的调用都会自动上报到 LangSmith 平台，可以看到每个节点的输入输出、耗时、token 消耗等。

12.3 错误处理

节点函数中的异常不会被 LangGraph 自动捕获，建议在关键节点中加入 try-except：

def safe_tool_node(state: MessagesState) -> dict:
    try:
        # 执行工具调用...
        return {"messages": results}
    except Exception as e:
        error_msg = ToolMessage(
            content=f"工具执行失败：{str(e)}",
            tool_call_id=state["messages"][-1].tool_calls[0]["id"]
        )
        return {"messages": [error_msg]}

12.4 实践建议

State 设计原则：

• 保持 State 精简，只放必要的字段
• 对于需要追加的列表字段，使用 Annotated + Reducer
• 考虑加入控制字段如 retry_count、error_message 用于流程控制

图结构设计原则：

• 每个节点只做一件事，职责单一
• 避免在节点中做复杂的条件判断，把路由逻辑放到条件边中
• 循环一定要有退出条件，防止无限循环
• 使用子图封装可复用的逻辑模块

生产环境清单：

• 使用持久化 Checkpointer（PostgreSQL / SQLite），不要用 InMemorySaver
• 关键操作加入 Human-in-the-loop 审核
• 配置合理的超时和重试机制
• 接入 LangSmith 进行线上监控
• 对 Agent 的工具调用做权限控制和输入校验

---

十三、LangGraph 与其他框架的对比

LangGraph vs LangChain Agents： LangChain Agents 是更高层的抽象，底层已经基于 LangGraph 实现。如果你只需要标准的 ReAct 模式（LLM + 工具循环），直接用 LangChain 的 create_react_agent 更方便。如果需要自定义流程控制、多 Agent 协作、复杂的条件路由，则用 LangGraph。

LangGraph vs CrewAI / AutoGen： CrewAI 和 AutoGen 侧重于多 Agent 角色扮演和对话，抽象层级较高，上手快但灵活性有限。LangGraph 是底层编排框架，可以实现上述框架的所有模式，但需要自己编写更多代码。适合需要精细控制的生产场景。

一句话总结： 如果你需要的是"积木"级别的灵活组装能力，选 LangGraph；如果你需要的是开箱即用的特定场景方案，可以先看高层框架是否满足需求。

---

十四、总结

LangGraph 的核心就三件事：用 State 管理数据，用 Node 封装逻辑，用 Edge 编排流程。理解了这些，剩下的就是根据业务需要组合这些积木。

入门路径建议：

1. 先跑通上面"第一个 Graph"的例子，理解 State 的流转
2. 加入条件边，理解路由机制
3. 构建 Tool-Calling Agent，这是最核心的实战模式
4. 加入 Checkpointer 实现会话记忆
5. 根据需要引入 Human-in-the-loop、子图、流式输出等高级特性

官方资源：

• 文档：https://docs.langchain.com/oss/python/langgraph/overview
• GitHub：https://github.com/langchain-ai/langgraph
• 免费课程：https://academy.langchain.com/courses/intro-to-langgraph
• 示例代码：GitHub 仓库的 examples 目录

LangGraph 仍在快速迭代中，API 可能随版本更新有所变化，建议开发时参考官方文档确认最新用法。