前言

随着大型语言模型（LLM）能力的飞速提升，单次调用已经无法满足复杂业务场景的需求。开发者们开始构建多步骤、多角色、具备记忆与反思能力的智能代理（Agent）。在这个领域，LangGraph 是 LangChain 生态中最具代表性的工作流编排框架。

如果说 LangChain 是一个工具箱，LangGraph 就是那张蓝图——它让你用图（Graph）的思维方式，精确描述代理在每个节点（Node）做什么、如何路由（Route）、何时循环（Loop）。

本文将深入解析 LangGraph 的核心概念，并通过完整代码示例演示如何构建一个具备记忆、工具调用与自我反思能力的代理工作流。

---

一、LangGraph 核心概念

1.1 为什么需要 LangGraph？

在 LangChain Expression Language（LCEL）中，链（Chain）是线性的：input → chain → output。但现实中的代理流程往往是有分支、有循环、有状态的：

• 规划代理可能需要多轮自我反思才给出最终方案
• ReAct 代理需要在"思考→行动→观察"之间循环
• 多代理系统需要并行运行并同步结果

LangGraph 的核心价值在于：用有向图（Directed Graph）的方式表达任意复杂的代理逻辑。

1.2 核心组件

组件	作用
State	贯穿整个图执行流程的共享状态对象（通常是一个 dict）
Node	图中的计算节点，一个 Python 函数，接收当前 State，返回更新后的 State
Edge	连接节点的有向边，定义执行流程的走向
Router	条件边（Conditional Edge），根据 State 动态决定下一个节点
Graph	整个工作流的容器，管理节点、边以及入口/出口
checkpointer	持久化检查点，支持暂停、恢复和多轮对话记忆

1.3 与 LangChain 的关系

LangChain Core
    ├── LCEL（Chain）：线性、声明式
    └── LangGraph：图结构、状态机式的代理编排

LangGraph 不是替代 LCEL，而是对复杂代理场景的补充扩展。简单的 RAG 链仍用 LCEL，复杂的代理工作流用 LangGraph。

---

二、状态机与图的执行模型

2.1 StateGraph 入门

LangGraph 的核心是 StateGraph，它的设计哲学接近状态机：

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

# 1. 定义状态结构
class AgentState(TypedDict):
    messages: list[str]
    next_action: str
    loop_count: int

# 2. 定义节点函数
def step_one(state: AgentState) -> AgentState:
    state["messages"].append("Step 1 executed")
    state["loop_count"] += 1
    return state

def step_two(state: AgentState) -> AgentState:
    state["messages"].append("Step 2 executed")
    return state

# 3. 构建图
graph = StateGraph(AgentState)
graph.add_node("step_one", step_one)
graph.add_node("step_two", step_two)

graph.add_edge(START, "step_one")
graph.add_edge("step_one", "step_two")
graph.add_edge("step_two", END)

app = graph.compile()

2.2 条件边（Conditional Edge）

真正的代理需要"判断"能力：

from langgraph.graph import Branch

def should_continue(state: AgentState) -> str:
    if state["loop_count"] < 3:
        return "step_one"  # 继续循环
    return END              # 退出

graph.add_conditional_edges(
    "step_two",
    should_continue,
    {
        "step_one": "step_one",
        END: END
    }
)

---

三、实战：构建一个 ReAct 代理

ReAct（Reasoning + Acting）是目前最流行的代理范式：思考 → 行动 → 观察 → 循环。

3.1 项目结构

langgraph-react-agent/
├── tools.py          # 工具定义
├── agent.py          # 代理核心逻辑
├── graph.py          # 图构建
└── main.py           # 入口

3.2 定义工具

from langchain_core.tools import tool
from langchain_community.utilities import SerpAPIWrapper

@tool
def search_web(query: str) -> str:
    """搜索互联网获取最新信息"""
    serp = SerpAPIWrapper()
    return serp.run(query)

@tool
def calculator(expression: str) -> str:
    """计算数学表达式的值"""
    import math
    try:
        result = eval(expression, {"__builtins__": {}, "math": math})
        return str(result)
    except Exception as e:
        return f"计算错误: {e}"

tools = [search_web, calculator]

3.3 代理节点实现

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
from langgraph.prebuilt import ToolNode

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

SYSTEM_PROMPT = """你是一个智能助手，能够根据用户问题自主选择工具来获取答案。
请按以下格式思考：
Thought: 你需要思考的问题
Action: 工具名称（search_web 或 calculator）
Action Input: 给工具的输入
Observation: 工具返回的结果
...（可重复上述步骤）
Final Answer: 最终答案
"""

def reason_node(state):
    """推理节点：让 LLM 决定下一步行动"""
    messages = [
        SystemMessage(content=SYSTEM_PROMPT),
        *state["messages"]
    ]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

# 工具执行节点
tool_node = ToolNode(tools)

def should_continue(state) -> str:
    """判断是否需要继续工具调用"""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return END

3.4 构建图

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

class AgentState(TypedDict):
    messages: list

graph = StateGraph(AgentState)

graph.add_node("reason", reason_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "reason")
graph.add_conditional_edges(
    "reason",
    should_continue,
    {"tools": "tools", END: END}
)
graph.add_edge("tools", "reason")  # 工具执行完回到推理

app = graph.compile()

# 执行
if __name__ == "__main__":
    result = app.invoke({
        "messages": [HumanMessage(content="北京当前气温是多少度？再计算加上10度等于多少？")]
    })
    print(result["messages"][-1].content)

---

四、带记忆的会话代理

前面的 ReAct 代理没有记忆能力，每次调用都是全新的会话。LangGraph 通过 checkpointer 实现多轮对话记忆：

from langgraph.checkpoint.memory import MemorySaver

# 持久化检查点
memory = MemorySaver()

app = graph.compile(checkpointer=memory)

# 第一轮对话
config = {"configurable": {"thread_id": "user-123"}}
result1 = app.invoke(
    {"messages": [HumanMessage(content="我叫张三")]},
    config=config
)
print(result1["messages"][-1].content)  # AI: 好的，张三

# 第二轮对话（同一 thread_id，AI 记得之前的内容）
result2 = app.invoke(
    {"messages": [HumanMessage(content="我叫什么名字？")]},
    config=config
)
print(result2["messages"][-1].content)  # AI: 你叫张三

thread_id 是会话隔离的键，更换 thread_id 就是新会话，复用则是延续同一个对话。

---

五、多代理协作系统

LangGraph 的图结构天然适合多代理协作场景。下面是一个简化版的"规划 + 执行"双代理系统：

# 规划代理
planner_prompt = """你是一个规划专家，将用户任务分解为步骤。"""
planner_llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def planner_node(state):
    response = planner_llm.invoke([
        SystemMessage(content=planner_prompt),
        *state["messages"]
    ])
    return {"messages": [response], "plan": response.content}

# 执行代理
executor_prompt = """你是执行专家，严格按照计划执行每个步骤。"""
executor_llm = ChatOpenAI(model="gpt-4o").bind_tools(tools)

def executor_node(state):
    response = executor_llm.invoke([
        SystemMessage(content=executor_prompt),
        SystemMessage(content=f"执行计划: {state.get('plan', '')}"),
        *state["messages"]
    ])
    return {"messages": [response]}

# 构建协作图
collab_graph = StateGraph(AgentState)
collab_graph.add_node("planner", planner_node)
collab_graph.add_node("executor", executor_node)
collab_graph.add_edge(START, "planner")
collab_graph.add_edge("planner", "executor")
collab_graph.add_edge("executor", END)

collab_app = collab_graph.compile()

---

六、调试与可视化

LangGraph 内置了美观的可视化功能：

# 输出图结构
app.get_graph().print_ascii()

# 导出 Mermaid 图表
print(app.get_graph().draw_mermaid())

在 Jupyter Notebook 或支持 Mermaid 的 Markdown 渲染器中，可以直接看到工作流的图形化展示。

---

总结

LangGraph 将复杂的代理逻辑抽象为状态、节点、边三大原语，让开发者可以用声明式的方式构建任意复杂的工作流。其核心能力包括：

1. 状态管理：通过 TypedDict + 图执行上下文实现跨节点状态共享
2. 条件路由：支持 if/else 式的动态流程控制
3. 循环控制：天然支持 ReAct 等需要多轮迭代的代理模式
4. 记忆持久化：checkpointer 实现多轮对话与暂停恢复
5. 多代理编排：图的组合能力天然适配多代理协作场景

如果你正在构建需要复杂决策流程、长期记忆、或多个 AI 角色协作的应用，LangGraph 是目前最成熟、最灵活的解决方案。

延伸阅读：

• LangGraph 官方文档
• LangGraph GitHub 仓库
• LangChain 官方文档

---

标签：AI / LangChain / LangGraph / Agent / LLM