LangGraph 学习笔记

什么是 LangGraph？

LangGraph 是 LangChain 团队推出的一个用于构建有状态、多步骤 LLM 应用的框架。它的核心思想是把 AI 应用的执行流程建模为一张有向图（DAG），节点是执行逻辑，边是状态流转路径。

相比传统的链式调用（LangChain Chain），LangGraph 最大的特点是支持循环（Cycle），这让构建 Agent 的”思考—行动—观察”循环成为可能。

用户输入 → [Agent节点] → 决策：调用工具 or 返回结果
                ↑              ↓
           [观察结果]  ← [工具节点]

核心概念

1. State（状态）

LangGraph 中所有节点共享一个全局的 State 对象，通常用 TypedDict 定义：

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

class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # 消息历史，自动追加
    tool_result: str

Annotated[list, add_messages] 表示这个字段在更新时是追加而不是覆盖，这对保留对话历史非常关键。

2. Node（节点）

节点就是普通的 Python 函数，接收 State，返回 State 的更新部分：

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")

def agent_node(state: AgentState):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

3. Edge（边）

边定义了节点之间的跳转逻辑：

普通边：graph.add_edge("node_a", "node_b") — 无条件跳转
条件边：根据函数返回值决定下一个节点

def should_continue(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"   # 有工具调用，去执行工具
    return "end"         # 没有，结束

graph.add_conditional_edges("agent", should_continue, {
    "tools": "tool_node",
    "end": END
})

构建一个简单的 ReAct Agent

ReAct（Reasoning + Acting）是目前最经典的 Agent 模式，LangGraph 实现起来非常直观：

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_core.tools import tool

# 1. 定义工具
@tool
def search_web(query: str) -> str:
    """搜索网络获取信息"""
    return f"搜索结果：关于'{query}'的相关内容..."

tools = [search_web]
llm_with_tools = llm.bind_tools(tools)

# 2. 定义节点
def agent(state: AgentState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

tool_node = ToolNode(tools)

# 3. 构建图
builder = StateGraph(AgentState)
builder.add_node("agent", agent)
builder.add_node("tools", tool_node)

builder.set_entry_point("agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")  # 工具执行完回到 agent

graph = builder.compile()

# 4. 运行
result = graph.invoke({"messages": [("user", "今天北京天气怎么样？")]})
print(result["messages"][-1].content)

持久化与断点（Checkpointing）

LangGraph 内置了 Checkpointer 机制，可以保存图的执行状态，支持：

多轮对话的记忆保持
任务中断后恢复
Human-in-the-loop 审批

from langgraph.checkpoint.memory import MemorySaver

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

# 使用 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user_123"}}

# 第一轮
graph.invoke({"messages": [("user", "我叫小明")]}, config=config)

# 第二轮（自动记住上下文）
graph.invoke({"messages": [("user", "你还记得我叫什么吗？")]}, config=config)

多 Agent 协作

LangGraph 支持构建多 Agent 系统，不同 Agent 负责不同职责：

用户 → [Supervisor Agent]
              ├─→ [Research Agent]  （负责搜索信息）
              ├─→ [Coder Agent]     （负责写代码）
              └─→ [Review Agent]    （负责审核输出）

每个子 Agent 本身也是一个 LangGraph 子图，通过 Supervisor 协调调度，形成层次化的 Agent 网络。

与 LangChain 的对比

特性	LangChain Chain	LangGraph
执行结构	线性链	有向图（支持循环）
状态管理	手动传递	全局 State 自动管理
Agent 支持	有限	原生支持复杂 Agent
调试工具	基础	LangSmith 深度集成
适用场景	简单流程	复杂多步骤任务