0 · 为什么是 LangGraph(而不是 Chain / AgentExecutor)

很多人学 LangChain 第一步就走错路:学了一堆 LLMChain、AgentExecutor——它们正在被淘汰。现代建 Agent 的正确姿势是 LangGraph。先搞清它解决了什么问题。

传统的"链"(Chain)是一条直线:输入 → A → B → C → 输出。但真实的 Agent 需要:

• 循环:想一步、调个工具、看结果、再想一步……直到任务完成。直线结构天然做不到回头。
• 条件分支:根据模型输出决定"该调工具还是直接回答"。
• 状态:多步之间要记住已经发生了什么(对话历史、中间结果)。
• 可控:老 AgentExecutor 是个黑盒,你很难在中间插手、暂停、人工审批。

LangGraph 把 Agent 建模成一张 有状态的图(Stateful Graph):节点是"做事的步骤",边是"下一步去哪",而且边可以指回去形成循环。这正好对上了 Agent 的本质。

1 · 核心心智模型:Agent 就是一张状态图

这一节是全篇地基。把这四个词刻进脑子,后面所有代码都是它们的组合。

运行时,LangGraph 做的事极其简单:从 START 出发,执行当前节点 → 节点返回的更新合并进 State → 沿着边走到下一个节点 → 直到走到 END。循环就是"边指回了前面的节点"。

上面这张图就是一个最经典的 ReAct Agent:agent 节点思考 → 条件边判断要不要调工具 → 要就去 tools 节点 → 工具结果写回 → 回到 agent 再想……不需要工具了就走 END。整篇教程都在把这张图一点点搭出来。

2 · 环境准备(含第一天就开 trace)

复制# 安装核心包(以 OpenAI 为例,换别的模型就换对应集成包)
pip install langgraph langchain langchain-openai

# 环境变量
export OPENAI_API_KEY="sk-..."

# 强烈建议:第一天就接 LangSmith,能看到图内部每一步在干啥
export LANGCHAIN_TRACING_V2="true"
export LANGCHAIN_API_KEY="ls-..."   # smith.langchain.com 免费注册

3 · 第一个图:从 0 手搭一个最小 Graph

先不碰工具、不碰循环。我们搭一个最简单的图:一个节点,调用模型回答问题。目的是把 StateGraph 的五步骨架跑通。

复制from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# ① 定义 State:这个图里要传递的数据。messages 用特殊 reducer(下一节细讲)
class State(TypedDict):
    messages: Annotated[list, add_messages]

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

# ② 定义节点:就是个普通函数,收 State,返回要更新的字段(dict)
def chatbot(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# ③ 建图,挂节点
builder = StateGraph(State)
builder.add_node("chatbot", chatbot)

# ④ 连边:START → chatbot → END
builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", END)

# ⑤ 编译成可运行对象
graph = builder.compile()

# 跑!
result = graph.invoke({"messages": [{"role": "user", "content": "用一句话解释 LangGraph"}]})
print(result["messages"][-1].content)

4 · State 与 Reducer:整个框架的灵魂

如果只能记一个 LangGraph 的核心概念,就记 Reducer。它决定"节点返回的更新,如何合并进 State"。

看回上面那行:messages: Annotated[list, add_messages]。这里 add_messages 就是一个 reducer。它的含义是:当节点返回新的 messages 时,不要覆盖,而是追加到已有列表后面。

对比:有没有 reducer 的区别

为什么这事关键?因为 Agent 在循环里会反复经过同一个节点,你需要"积累"信息(对话越聊越长),而不是每轮把前面冲掉。Reducer 就是控制"积累 vs 覆盖"的开关。

复制import operator
from typing import Annotated
from typing_extensions import TypedDict

class State(TypedDict):
    messages: Annotated[list, add_messages]   # 追加:对话历史
    step_count: Annotated[int, operator.add]  # 累加:走了几步
    final_answer: str                          # 覆盖:最终答案,只保留最新

5 · 条件边与循环:让图学会"自己决定下一步"

直线图没意思。Agent 的灵魂是"根据当前情况决定去哪"。这靠 add_conditional_edges 实现。

条件边 = 一个路由函数。它读 State,返回一个字符串(下一个节点的名字)。LangGraph 按这个名字决定往哪走。

复制from langgraph.graph import StateGraph, START, END

# 路由函数:看最后一条消息,决定继续还是结束
def route(state: State) -> str:
    last = state["messages"][-1]
    # 如果模型要求调用工具,就去 "tools" 节点;否则结束
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return END

builder.add_conditional_edges(
    "agent",            # 从哪个节点出发判断
    route,               # 路由函数
    {"tools": "tools", END: END}  # 返回值 → 目标节点的映射(可省略)
)

# 关键:tools 跑完,边指回 agent —— 这一条就是"循环"!
builder.add_edge("tools", "agent")

6 · 工具调用:ToolNode 让模型动手

Agent 之所以强,是因为它能调工具(搜索、算数、查数据库)。LangGraph 内置了 ToolNode,帮你省掉手写"解析 tool_calls → 执行 → 把结果包成消息"的样板代码。

复制from langchain_core.tools import tool
from langgraph.prebuilt import ToolNode, tools_condition

# ① 用 @tool 定义工具,docstring 会被模型当成"使用说明"
@tool
def get_weather(city: str) -> str:
    """查询某个城市的天气。"""
    return f"{city} 今天晴,26°C"

tools = [get_weather]

# ② 把工具"绑"给模型,模型才知道有哪些工具可用
llm_with_tools = llm.bind_tools(tools)

def agent(state: State):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# ③ ToolNode 自动执行模型要求的工具调用
tool_node = ToolNode(tools)

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", tool_node)
builder.add_edge(START, "agent")
# tools_condition 是官方预置的路由:有 tool_calls 就去 tools,否则去 END
builder.add_conditional_edges("agent", tools_condition)
builder.add_edge("tools", "agent")
graph = builder.compile()

7 · 完整 ReAct Agent:手搭 vs 一行预置

把前面所有零件拼起来,你就有了一个能思考、会调工具、自动循环的完整 Agent。其实第 6 节的代码已经是了。这里给你官方的一行版,以及"什么时候该用哪个"。

一行版:create_react_agent

复制from langgraph.prebuilt import create_react_agent

# 一行得到一个完整 ReAct Agent(内部就是第6节那张图)
agent = create_react_agent(llm, tools=[get_weather])

result = agent.invoke({"messages": [{"role": "user", "content": "北京天气怎么样?"}]})
print(result["messages"][-1].content)

8 · 记忆与持久化:Checkpointer + thread_id

到这里 Agent 还是"一次性"的——每次 invoke 都从零开始,不记得上次聊了啥。要做多轮对话,需要 Checkpointer(检查点)。

原理:每走完一步,LangGraph 把当前 State 存一份快照(checkpoint),用 thread_id 标识是哪个会话。下次带同样的 thread_id 进来,它自动把历史 State 加载回来接着聊。

复制from langgraph.checkpoint.memory import MemorySaver

# 开发用内存版;生产换 SqliteSaver / PostgresSaver 落盘
memory = MemorySaver()
graph = builder.compile(checkpointer=memory)

# 用 config 里的 thread_id 区分不同会话
config = {"configurable": {"thread_id": "user-42"}}

graph.invoke({"messages": [{"role":"user","content":"我叫小镰刀"}]}, config)
# 第二次,同一个 thread_id,它记得你的名字
r = graph.invoke({"messages": [{"role":"user","content":"我叫什么?"}]}, config)
print(r["messages"][-1].content)   # → 你叫小镰刀

9 · 人工介入(Human-in-the-loop)

真实业务里,Agent 在执行"危险操作"(转账、删数据、发邮件)前,往往需要人点头。LangGraph 让你在图的任意位置暂停,等人工批准再继续。

最简单的方式:编译时声明"在某个节点前先停下"。

复制# 在进入 tools 节点前暂停,把控制权交回给你
graph = builder.compile(
    checkpointer=memory,
    interrupt_before=["tools"]   # 执行工具前先停
)

config = {"configurable": {"thread_id": "t1"}}
graph.invoke({"messages":[{"role":"user","content":"帮我删除日志"}]}, config)

# 此刻图停在 tools 前。你检查它想干啥:
state = graph.get_state(config)
print(state.next)  # → ('tools',) 说明它正等着调工具

# 人工确认 OK → 传 None 让它从断点继续
graph.invoke(None, config)

10 · 流式输出:看 Agent 一步步在想什么

用 invoke 你只拿到最终结果。用 stream,你能实时看到每个节点跑完后的状态——调试和做"打字机效果"都靠它。

复制# stream_mode="values":每步后吐出完整 State
for chunk in graph.stream(
        {"messages":[{"role":"user","content":"北京天气?"}]},
        config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

11 · 观测与评测:对接"自进化"

这一节直接对应你 JD 里的"反思 / 数据飞轮 / 效果调优"。LangGraph + LangSmith 是落地自进化的天然组合。

① Trace:把循环看透

第 2 节配好 LANGCHAIN_TRACING_V2 后,每次运行都会在 LangSmith 里生成一条完整时间线:每个节点的输入输出、模型每轮的 token 消耗、调了哪些工具、为什么。排查"Agent 为什么绕圈/调错工具",看 trace 几秒钟定位。

② Eval:用数字说话

复制# 思路:建一个评估集,跑 Agent,用 LLM-as-judge 打分,量化成功率
from langsmith import Client
from langsmith.evaluation import evaluate

dataset = [
    {"input":"北京天气?", "expect_tool":"get_weather"},
    {"input":"1+1?",      "expect_tool": None},  # 不该乱调工具
]
# 对每条 case 跑 agent → 判断行为是否符合预期 → 算通过率
# 改 prompt / 工具描述前后各跑一次,对比成功率 = A/B 调优

12 · 多智能体:Supervisor 模式

单个 Agent 工具一多就容易"选择困难"。把任务拆给多个专精 Agent、由一个"主管"调度,往往更稳。这就是多智能体。

最常用的是 Supervisor(主管)模式:一个 supervisor 节点根据当前任务,决定把活派给哪个子 Agent(研究员 / 写手 / 审核),子 Agent 干完把结果交回 supervisor,它再决定下一步——本质还是"条件边 + 循环",只是节点变成了一个个 Agent。

13 · 常见坑 & 最佳实践

14 · 学习路径 Checklist

按顺序打勾,每一项都"亲手敲一个能跑的 demo"才算过。

• 跑通最小 Graph(第 3 节五步骨架)
• 给 State 加一个 operator.add 字段,理解 reducer
• 手写一个条件边,做一个会循环的图(第 5 节)
• 用 ToolNode + tools_condition 让模型调一个自定义工具
• 用 create_react_agent 跑一遍,再用 StateGraph 手搭等价版
• 加 MemorySaver,做一个记得住名字的多轮对话
• 加 interrupt_before,做一个"工具执行前要人工批准"的流程
• 接 LangSmith,看一条完整 trace
• 建 10 条评估集,做一次"改前 vs 改后"的成功率对比(自进化起点)
• 读官方 LangChain Academy 的 LangGraph 课,补齐细节

学完这份 checklist,你就具备了 JD 里"懂 Agent 核心架构 + 主流框架 + 自进化落地"的硬实力。下一步:把你某个 launchd 任务用 LangGraph 重写成带评测闭环的真 Agent,作为面试案例。