langchain-ai/langgraph · 上手攻略

• 仓库：langchain-ai/langgraph
• 链接：https://github.com/langchain-ai/langgraph
• 分类：agent
• 作者：Tom
• 更新：2026-07-06

这是什么

LangGraph 是 LangChain 团队出品的状态化 Agent 编排框架（orchestration framework），用于构建、运行和管理"长时间运行、有状态"的多步骤 AI Agent 工作流。底层受 Google Pregel 和 Apache Beam 启发，接口设计参考了 NetworkX；用纯 Python 开发，License 为 MIT。

它解决的核心问题是：当 Agent 需要在多轮对话中维护状态、调用多个工具、加入人工审核节点、或从失败中恢复时，如何工程化地管理这些复杂流程。LangGraph 本身不是 LLM provider，也不绑定具体模型——任何支持 tool-calling 的模型（Claude、GPT-4、Gemini、DeepSeek 等）都可以接入。

解决什么问题

快速安装

pip install -U langgraph

# 可选：配套的 LangChain 模型集成
pip install -U langchain langchain-anthropic langchain-openai

⚠️ 版本注意：langgraph 与旧版 langchain 生态有 API 差异，建议搭配 langchain >= 0.3 使用。具体依赖请参考 PyPI: langgraph。

核心用法

基础概念：StateGraph

LangGraph 的核心是状态图（StateGraph）：你定义节点（Nodes）和边（Edges），每个节点是一个 Python 函数，每个节点接收当前状态（State）并返回状态更新。

最小示例：计算器 Agent（Graph API）

以下代码来自 LangGraph 官方 Quickstart，完整演示了构建一个 Tool-calling Agent 的全流程。

# Step 1: 定义工具和模型
from langchain.tools import tool
from langchain.chat_models import init_chat_model

# 初始化模型（支持 claude-sonnet-4-6 / gpt-4 / gemini 等）
model = init_chat_model("claude-sonnet-4-6", temperature=0)

@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`."""
    return a * b

@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`."""
    return a + b

@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`."""
    return a / b

tools = [add, multiply, divide]
tools_by_name = {tool.name: tool for tool in tools}
model_with_tools = model.bind_tools(tools)

# Step 2: 定义状态（State）
from langchain.messages import AnyMessage
from typing_extensions import TypedDict, Annotated
import operator

class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], operator.add]  # 新消息追加，而非覆盖
    llm_calls: int

# Step 3: 定义 LLM 节点
from langchain.messages import SystemMessage

def llm_call(state: MessagesState):
    """LLM 决定是否调用工具"""
    return {
        "messages": [
            model_with_tools.invoke(
                [SystemMessage(content="You are a helpful arithmetic assistant.")]
                + state["messages"]
            )
        ],
        "llm_calls": state.get('llm_calls', 0) + 1
    }

# Step 4: 定义工具节点
from langchain.messages import ToolMessage

def tool_node(state: MessagesState):
    """执行工具调用"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(
            ToolMessage(content=str(observation), tool_call_id=tool_call["id"])
        )
    return {"messages": result}

# Step 5: 定义条件路由（是否继续调用工具）
from typing import Literal
from langgraph.graph import StateGraph, START, END

def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tool_node"  # 继续调用工具
    return END  # 没有工具调用，结束

# Step 6: 构建并编译图
agent_builder = StateGraph(MessagesState)

agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    {"tool_node": "tool_node", END: END}
)
agent_builder.add_edge("tool_node", "llm_call")

agent = agent_builder.compile()

# 运行 Agent
from langchain.messages import HumanMessage

messages = [HumanMessage(content="Add 3 and 4, then multiply by 2.")]
result = agent.invoke({"messages": messages, "llm_calls": 0})

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

# 可视化图（导出 Mermaid）
from IPython.display import Image, display
display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

核心 API 小结

进阶功能

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

# 在节点处设置 interrupt，让 Agent 暂停等待人工确认
from langgraph.types import interrupt

def review_node(state: MessagesState):
    # 暂停执行，等待人工审批
    interrupt("请人工审核以下结果：")
    return {"messages": state["messages"]}

调用时：

# checkpoint=True 让状态可以恢复
config = {"configurable": {"thread_id": "my-session"}}
for chunk in agent.astream(input, config):
    print(chunk)

Memory（记忆）

LangGraph 支持两种记忆： - Short-term：消息列表，当前会话内追加 - Long-term：通过 Checkpoint（checkpointing）实现跨会话持久化

# 持久化 checkpoint（支持 Redis、PostgreSQL、SQLite 等后端）
from langgraph.checkpoint.memory import MemorySaver

checkpointer = MemorySaver()  # 内存存储，生产环境建议换持久化后端
agent = agent_builder.compile(checkpointer=checkpointer)

子图（Subgraph）

大型工作流可以拆成多个子图，子图有自己的状态类型，通过 enter_node/leave_node 控制调用边界。

典型适用场景

• 多步骤 Tool-calling Agent：如研究 Agent（搜索 → 读取 → 总结 → 写作）、客服 Agent（理解意图 → 查库 → 回复）
• 需要人工审核的关键决策：金融、医疗、法律等需要人机协作的场景
• 长时间运行任务：如批量处理、复杂数据分析管道
• 多 Agent 协作：主 Agent 负责任务分发，子 Agent 各自执行专业子任务
• 需要断点恢复的工作流：用户中断后能从上次状态继续，而非从头开始

坑与注意

与同类对比

⚠️ 注意：AutoGen、CrewAI 等更偏"多 Agent 角色协作"，LangGraph 更偏"工作流编排 + 状态管理"，两者定位有差异但有重叠。

一句话推荐结论

如果你需要构建"能记住上下文、能从失败中恢复、能在执行中接受人工审核"的生产级 Agent 工作流，LangGraph 是目前 Python 生态里最成熟、最完整的解决方案；它的图模型让复杂 Agent 流程变得可可视化、可测试、可部署——但需要接受一定的学习曲线，建议搭配 LangSmith 做调试。