快速开始 #
环境准备 #
安装 LangGraph #
bash
# 使用 pip 安装
pip install langgraph
# 或使用 uv 安装
uv pip install langgraph
# 安装 LangChain 集成（可选）
pip install langgraph langchain-openai
项目结构 #
text
my-langgraph-project/
├── .env # 环境变量
├── main.py # 主程序
└── requirements.txt # 依赖
环境变量 #
bash
# .env 文件
OPENAI_API_KEY=your-api-key-here
第一个示例：Hello World #
让我们从一个最简单的例子开始，理解 LangGraph 的基本结构。
python
from langgraph.graph import StateGraph, MessagesState, START, END
def mock_llm(state: MessagesState):
return {"messages": [{"role": "ai", "content": "hello world"}]}
graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()
result = graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
print(result)
代码解析 #
text
┌─────────────────────────────────────────────────────────────┐
│ Hello World 示例解析 │
├─────────────────────────────────────────────────────────────┤
│ │
│ 1. 定义状态（MessagesState） │
│ - 预构建的消息状态类型 │
│ - 包含 messages 列表 │
│ │
│ 2. 定义节点（mock_llm） │
│ - 接收状态，返回状态更新 │
│ - 这是 Agent 的核心逻辑 │
│ │
│ 3. 创建图（StateGraph） │
│ - 使用状态类型初始化 │
│ │
│ 4. 添加节点和边 │
│ - START -> mock_llm -> END │
│ │
│ 5. 编译图 │
│ - 验证图结构 │
│ - 生成可执行的应用 │
│ │
│ 6. 执行图 │
│ - invoke() 方法执行 │
│ │
└─────────────────────────────────────────────────────────────┘
构建一个简单的 Agent #
现在让我们构建一个能够使用工具的真实 Agent。
步骤 1：定义工具 #
python
from langchain_core.tools import tool
@tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers together."""
return a * b
tools = [add, multiply]
步骤 2：创建 LLM #
python
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
步骤 3：定义状态 #
python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages
class State(TypedDict):
messages: Annotated[list, add_messages]
步骤 4：定义节点 #
python
def agent_node(state: State):
"""Agent 节点：调用 LLM"""
return {"messages": [llm_with_tools.invoke(state["messages"])]}
步骤 5：定义工具节点 #
python
from langgraph.prebuilt import ToolNode
tool_node = ToolNode(tools)
步骤 6：定义条件边 #
python
from langgraph.graph import END
def should_continue(state: State):
"""判断是否需要调用工具"""
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
步骤 7：构建图 #
python
from langgraph.graph import StateGraph, START
graph = StateGraph(State)
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")
app = graph.compile()
完整代码 #
python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
class State(TypedDict):
messages: Annotated[list, add_messages]
@tool
def add(a: int, b: int) -> int:
"""Add two numbers together."""
return a + b
@tool
def multiply(a: int, b: int) -> int:
"""Multiply two numbers together."""
return a * b
tools = [add, multiply]
llm = ChatOpenAI(model="gpt-4o-mini")
llm_with_tools = llm.bind_tools(tools)
def agent_node(state: State):
return {"messages": [llm_with_tools.invoke(state["messages"])]}
tool_node = ToolNode(tools)
def should_continue(state: State):
messages = state["messages"]
last_message = messages[-1]
if last_message.tool_calls:
return "tools"
return END
graph = StateGraph(State)
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")
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "What is 3 + 4 * 5?"}]})
print(result["messages"][-1].content)
图可视化 #
LangGraph 支持将图结构可视化，帮助理解和调试。
在 Jupyter 中显示 #
python
from IPython.display import Image, display
display(Image(app.get_graph().draw_mermaid_png()))
生成 Mermaid 图 #
python
print(app.get_graph().draw_mermaid())
输出示例：
mermaid
graph
START --> agent
agent --> tools
agent --> END
tools --> agent
流式输出 #
LangGraph 支持流式输出，提供更好的用户体验。
流式 Token #
python
async for event in app.astream_events({"messages": [...]}, version="v2"):
if event["event"] == "on_chat_model_stream":
print(event["data"]["chunk"].content, end="", flush=True)
流式更新 #
python
for chunk in app.stream({"messages": [...]}, stream_mode="updates"):
print(chunk)
流式模式对比 #
text
┌─────────────────────────────────────────────────────────────┐
│ 流式模式对比 │
├─────────────────────────────────────────────────────────────┤
│ │
│ stream_mode="values" │
│ - 输出每个步骤后的完整状态 │
│ - 适合需要完整上下文的场景 │
│ │
│ stream_mode="updates" │
│ - 只输出节点的返回值（增量更新） │
│ - 更高效，适合大多数场景 │
│ │
│ stream_mode="messages" │
│ - 专门用于消息流 │
│ - 适合聊天应用 │
│ │
│ astream_events() │
│ - 最细粒度的流 │
│ - 可以流式输出 LLM token │
│ │
└─────────────────────────────────────────────────────────────┘
添加记忆（Memory） #
让 Agent 能够记住之前的对话。
python
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "conversation-1"}}
result1 = app.invoke(
{"messages": [{"role": "user", "content": "My name is Alice"}]},
config
)
result2 = app.invoke(
{"messages": [{"role": "user", "content": "What's my name?"}]},
config
)
print(result2["messages"][-1].content)
记忆机制说明 #
text
┌─────────────────────────────────────────────────────────────┐
│ 记忆机制 │
├─────────────────────────────────────────────────────────────┤
│ │
│ Checkpointer（检查点）： │
│ - 自动保存每次执行的状态 │
│ - 通过 thread_id 区分不同会话 │
│ │
│ 工作流程： │
│ 1. 首次调用：创建新会话，保存状态 │
│ 2. 后续调用：加载历史状态，继续对话 │
│ │
│ 支持的存储后端： │
│ - MemorySaver：内存存储（开发测试） │
│ - SqliteSaver：SQLite 数据库 │
│ - PostgresSaver：PostgreSQL 数据库 │
│ - RedisSaver：Redis 存储 │
│ │
└─────────────────────────────────────────────────────────────┘
使用预构建 Agent #
LangGraph 提供了预构建的 Agent，可以快速开始。
ReAct Agent #
python
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(llm, tools)
result = agent.invoke({"messages": [{"role": "user", "content": "What is 2 + 2?"}]})
带系统提示的 Agent #
python
agent = create_react_agent(
llm,
tools,
state_modifier="You are a helpful math assistant."
)
调试技巧 #
打印执行步骤 #
python
for step in app.stream({"messages": [...]}):
print(f"Step: {step}")
查看状态历史 #
python
config = {"configurable": {"thread_id": "debug-1"}}
app.invoke({"messages": [...]}, config)
history = app.get_state_history(config)
for state in history:
print(state)
使用 LangSmith #
python
import os
os.environ["LANGSMITH_API_KEY"] = "your-key"
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "my-project"
# 所有执行会自动追踪到 LangSmith
常见问题 #
1. 如何处理错误？ #
python
from langgraph.errors import GraphRecursionError
try:
result = app.invoke({"messages": [...]})
except GraphRecursionError:
print("Reached recursion limit")
2. 如何限制递归深度？ #
python
result = app.invoke(
{"messages": [...]},
config={"recursion_limit": 10}
)
3. 如何获取中间结果？ #
python
for chunk in app.stream({"messages": [...]}, stream_mode="updates"):
for node_name, node_output in chunk.items():
print(f"Node {node_name}: {node_output}")
下一步 #
现在你已经学会了 LangGraph 的基本用法，接下来学习 核心概念，深入理解 LangGraph 的工作原理！
最后更新：2026-03-30