LangGraph 详细讲解

LangGraph 是 LangChain 生态中用于构建有状态、多步骤的 AI Agent 应用的图框架。它的核心思想是将 Agent 的推理过程建模为一个状态机，其中节点（Node）执行操作，边（Edge）控制流程，状态（State）在节点间传递。

📦 安装

pip install langgraph langchain langchain-openai

🏗️ 核心概念

🔧 核心 API 详解

1. StateGraph - 创建图

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages
from langchain_core.messages import AnyMessage

# 定义状态结构
class MessagesState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    user_name: str
    
# 创建图
graph = StateGraph(MessagesState)

2. add_node - 添加节点

# 添加节点函数
def my_node(state: MessagesState) -> dict:
    # 处理逻辑...
    return {"messages": [AIMessage(content="回复内容")]}

graph.add_node("node_name", my_node)

3. add_edge / add_conditional_edges - 添加边

# 普通边：顺序执行
graph.add_edge("node_a", "node_b")

# 条件边：根据函数返回动态选择
def router(state: MessagesState) -> str:
    if "需要工具" in state["messages"][-1].content:
        return "tool_node"
    return "end"

graph.add_conditional_edges("assistant", router)

4. set_entry_point / set_finish_point - 设置入口和出口

graph.set_entry_point("assistant")      # 起始节点
graph.set_finish_point("assistant")     # 终止节点

5. compile - 编译图

app = graph.compile()

6. invoke / stream - 执行图

# 同步执行，返回最终状态
result = app.invoke({"messages": [HumanMessage(content="你好")]})

# 流式执行，返回每一步的更新
for chunk in app.stream(
    {"messages": [HumanMessage(content="你好")]},
    stream_mode="updates"  # 'values', 'updates', 'debug'
):
    print(chunk)

📋 MessageState 与 add_messages

from typing import Annotated, TypedDict
from langgraph.graph import add_messages
from langchain_core.messages import AnyMessage

class MessagesState(TypedDict):
    # add_messages 会自动追加新消息到列表末尾
    messages: Annotated[list[AnyMessage], add_messages]
    # 没有 reducer 的字段会被直接覆盖
    step_count: int

🛠️ 内置工具节点 - ToolNode

from langgraph.prebuilt import ToolNode
from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """获取城市天气"""
    return f"{city} 天气晴朗"

tools = [get_weather]
tool_node = ToolNode(tools)

graph.add_node("tools", tool_node)

🤖 预置 Agent - create_react_agent / create_agent

from langchain.agents import create_agent  # 推荐使用
from langchain_openai import ChatOpenAI

agent = create_agent(
    model=ChatOpenAI(model="gpt-3.5-turbo"),
    tools=[get_weather],
    system_prompt="你是一个有用的助手"
)

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气如何？"}]})

💡 完整示例：计算器 Agent

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, add_messages
from langgraph.prebuilt import ToolNode
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage

# 1. 定义工具
@tool
def add(a: float, b: float) -> float:
    """计算两个数的和"""
    return a + b

@tool
def multiply(a: float, b: float) -> float:
    """计算两个数的乘积"""
    return a * b

# 2. 定义状态
class CalculatorState(TypedDict):
    messages: Annotated[list, add_messages]
    result: float

# 3. 初始化模型和工具
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
tools = [add, multiply]
llm_with_tools = llm.bind_tools(tools)

# 4. 定义节点函数
def assistant(state: CalculatorState):
    sys_msg = SystemMessage(content="你是一个计算器助手，根据用户请求计算数学表达式")
    response = llm_with_tools.invoke([sys_msg] + state["messages"])
    return {"messages": [response]}

def should_continue(state: CalculatorState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

# 5. 构建图
graph = StateGraph(CalculatorState)
graph.add_node("assistant", assistant)
graph.add_node("tools", ToolNode(tools))

graph.set_entry_point("assistant")
graph.add_conditional_edges("assistant", should_continue, {
    "tools": "tools",
    "end": "__end__"
})
graph.add_edge("tools", "assistant")

# 6. 编译
app = graph.compile()

# 7. 执行
result = app.invoke({"messages": [HumanMessage(content="(3+4)*2等于多少？")]})
for msg in result["messages"]:
    msg.pretty_print()

🔄 stream_mode 详解

for chunk in app.stream(input, stream_mode="updates"):
    print(chunk)  # 例如: {'assistant': {'messages': [...]}}

⚙️ 高级配置

设置递归限制

config = {"recursion_limit": 50}
result = app.invoke(input, config=config)

Checkpoint 持久化（状态保存）

from langgraph.checkpoint import MemorySaver

memory = MemorySaver()
app = graph.compile(checkpointer=memory)

# 多轮对话保持状态
config = {"configurable": {"thread_id": "user123"}}
app.invoke(input, config=config)

📊 图结构对比

⚠️ 常见错误与解决

💎 总结

• StateGraph 是最核心的构建块，定义状态和节点
• add_messages 自动管理消息历史
• ToolNode 简化工具调用执行
• 条件边 实现智能路由（如是否需要调用工具）
• checkpointer 支持多轮对话持久化

掌握这些后，你可以构建从简单对话到复杂多步推理的各种 Agent 应用。