如果你刚开始学 AI Agent，很容易把它理解成“让大模型回答问题”。但真正的 Agent，关键不只是“回答”，而是它会先判断，再决定要不要调用工具，拿到工具结果后继续往下走，最后再输出答案。

这类流程如果只靠普通链式调用来写，通常会越来越乱：
输入进来，先调用模型；如果模型说要查天气，就再调一次工具；工具结果回来，再调一次模型；如果工具失败，可能还要重试；如果用户没说城市，还得补问。
流程一复杂，代码就会开始到处塞 if/else。

这正是 LangGraph 的价值所在。

根据 LangGraph 官方文档，它本质上是一个“图式工作流框架”：

• State：共享状态，保存当前运行快照

• Node：处理逻辑节点，读取状态并返回状态更新

• Edge：决定接下来执行哪个节点

• StateGraph：把这些节点和边编译成可执行图

• 图编译后可以 invoke()、stream()、异步执行，并且天然适合有循环、分支、工具调用、持久化和人工介入的 Agent 流程

官方文档还特别强调：LangGraph 是面向有状态、可循环、可恢复执行的 Agent 编排运行时，而不只是把多个步骤简单串起来。这也是它比普通链更适合 Agent 的核心原因。

这篇文章我们就从零开始，搭一个最小可运行的天气查询 Agent，让你真正理解：

• 什么是 State

• 什么是 Node

• 什么是 Router

• 为什么 Agent 天然适合用 LangGraph

• 一个“用户问天气 → Agent 决策 → 调用工具 → 返回答案”的完整执行流程是怎么跑起来的

---

一、先理解：LangGraph 到底在解决什么问题？

先看一个最简单的用户请求：

“

“北京今天天气怎么样？”

对于这个问题，一个真正的 Agent 不应该一上来就瞎答，而应该经历下面的过程：

1. 接收用户消息

2. 判断这是不是一个需要工具的问题

3. 如果需要，提取城市名，比如“北京”

4. 调用天气工具

5. 拿到天气结果

6. 再组织成自然语言回复用户

   

LangGraph 天气查询 Agent 的完整执行流程图

注意，这里面不是单次“输入→输出”，而是一个带状态的多步决策过程。

如果以后再扩展一下需求，比如：

• 如果用户没说城市，要追问

• 如果天气接口报错，要重试

• 如果用户还问“顺便告诉我穿衣建议”，还要继续推理

• 如果要支持多轮对话，还得保留消息历史

这已经明显不是简单 Prompt 能优雅解决的问题了。

LangGraph 的思路是：
把 Agent 看成一个状态机（state machine）或图（graph）。

每一个步骤是一个节点，每一次跳转由边来控制，而所有节点共享同一个状态。

---

二、新手最容易混淆的 3 个概念：State、Node、Router

这是初学 LangGraph 最容易绕晕的地方，我们先讲透。

1. State：当前运行过程中的“共享记忆”

你可以把 State 理解成 Agent 在这一轮执行中随身带着的一份工作记录。

例如我们的天气 Agent，State 里可以放这些信息：

• messages：对话消息列表

• city：解析出的城市

• tool_result：天气工具返回的数据

• final_answer：最终输出给用户的答案

在 LangGraph 官方 Graph API 中，State 的核心特点是：

• 所有节点都能读它

• 节点执行后返回的是 Partial State，也就是“我只更新其中一部分字段”

• 图运行时会把这些更新合并回共享状态中

也就是说，Node 不是直接“打印结果”，而是“修改状态”。

---

2. Node：一个处理步骤

Node 本质就是一个函数。
它接收当前状态，做一些逻辑处理，然后返回要更新的字段。

例如：

• agent_node：让大模型判断是否需要调用工具

• tool_node：真正执行天气查询

• respond_node：根据工具结果生成最终答案

所以你可以把 Node 理解成：

“

Agent 工作流中的一个“处理站”

---

3. Router：决定下一步去哪

Router 不是一个独立的官方类名，而是开发中常见的说法。
通常指的是：根据当前状态，决定图接下来走哪条边的函数。

例如：

• 如果模型已经发起了工具调用 → 去 tool_node

• 如果模型已经可以直接回答 → 去 respond_node

• 如果执行结束了 → 去 END

在 LangGraph 里，这类逻辑一般通过 add_conditional_edges() 来实现。

所以最直白的理解是：

• State：现在手里有哪些信息

• Node：当前这一步要做什么

• Router：下一步该去哪里

  

State、Node、Router 三个概念的对比关系图

---

三、为什么 LangGraph 适合 Agent，而不只是普通链式调用？

很多人第一次接触 LangGraph 时会问：

“

“我直接写一个链，不也能先 LLM、再工具、再 LLM 吗？”

可以，但两者适合的场景不同。

普通链式调用更像流水线

链式调用通常适合：

• 步骤固定

• 基本没有分支

• 不需要循环

• 不需要状态长期保存

• 不需要失败恢复

例如：

用户输入 → 提取关键词 → 总结结果 → 输出

这种很适合链。

---

Agent 更像“动态决策流程”

Agent 的特点是：

• 是否调用工具，运行时才知道

• 调用哪个工具，运行时才知道

• 可能会循环多轮：思考 → 调工具 → 再思考

• 可能需要中断、恢复、回放

• 可能需要保留复杂状态

而 LangGraph 官方文档强调的优势正是这些：

• Durable execution：长流程可以持久化与恢复

• Human-in-the-loop：中途人工介入

• Stateful：天然管理状态

• Streaming / Debugging / Observability：适合生产级 Agent 追踪和调试

一句话总结：

“

普通链擅长“固定步骤”，LangGraph 擅长“会分支、会循环、会调用工具的有状态 Agent”。

LangGraph 与普通链式调用的对比图

---

四、先搭一个最小版天气 Agent

下面我们用 Python + LangGraph 搭一个完整示例。

为了聚焦 LangGraph 核心概念，这里天气工具先用模拟数据，不接真实天气 API。等你跑通流程后，再把工具替换成真实接口即可。

---

五、安装依赖

pip install -U langgraph langchain langchain-openai

如果你用 OpenAI 模型，还需要配置环境变量：

export OPENAI_API_KEY=你的_key

Windows PowerShell:

setx OPENAI_API_KEY "你的_key"

---

六、定义天气工具

先定义一个最简单的工具。LangChain 官方 Tools 文档中推荐使用 @tool 装饰器定义工具，这也是 LangGraph 里最常见的工具接入方式。

from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。输入必须是城市名。"""
    mock_weather_data = {
        "北京": "晴，22°C，东北风 2 级",
        "上海": "多云，26°C，东南风 3 级",
        "广州": "小雨，29°C，南风 2 级",
        "深圳": "阴，28°C，微风",
    }
    return mock_weather_data.get(city, f"暂时没有 {city} 的天气数据")

这里先不追求真实，而是先把“工具调用”这个动作建立起来。

---

七、定义 State

LangGraph 官方 Graph API 文档推荐使用 TypedDict 或 Pydantic 来定义状态。
如果你的工作流围绕消息展开，还可以使用 MessagesState。

我们的例子正好是聊天场景，所以直接继承 MessagesState 最省事。

from typing import Annotated, TypedDict, Optional
from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from langchain.messages import AnyMessage

class WeatherAgentState(MessagesState):
    city: Optional[str]
    tool_result: Optional[str]
    final_answer: Optional[str]

这里要注意：

• MessagesState 已经帮你准备好了 messages

• messages 会以适合消息场景的方式合并

• 我们额外补充了 city、tool_result、final_answer

你也可以不继承 MessagesState，手写：

class WeatherAgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    city: Optional[str]
    tool_result: Optional[str]
    final_answer: Optional[str]

这两种写法都可以。对新手来说，继承 MessagesState 更直观。

---

八、定义模型

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools([get_weather])

这里的关键是 bind_tools([get_weather])。
绑定后，模型就具备了“按工具 schema 发起工具调用”的能力。

---

九、定义第一个节点：Agent 决策节点

这个节点的职责是：

• 读取当前消息

• 让模型判断要不要调用 get_weather

• 把模型输出追加回 messages

from langchain.messages import SystemMessage

def agent_node(state: WeatherAgentState):
    system_prompt = SystemMessage(
        content=(
            "你是一个天气查询助手。"
            "如果用户在询问天气，并且提供了城市，优先调用 get_weather 工具。"
            "如果用户没有提供城市，不要调用工具，直接礼貌追问用户想查询哪个城市。"
            "当已经拿到工具结果后，请基于结果生成最终回答。"
        )
    )

    response = llm_with_tools.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

LangGraph 天气 Agent 最小代码结构示意图

这里很多新手会疑惑：

“

为什么这里只返回 {"messages": [response]}，没有直接返回最终答案？

因为这个节点还只是“决策节点”。
它的工作是让模型先决定：

• 要不要调工具

• 还是直接回答

真正执行工具，是下一个节点的事情。

这就是 LangGraph 的好处：把“思考”和“行动”拆开。

---

十、定义工具节点

LangGraph 里处理工具调用，最方便的方式是使用预置的 ToolNode。

from langgraph.prebuilt import ToolNode

tool_node = ToolNode([get_weather])

它会自动读取上一条 AIMessage 里的 tool calls，执行对应工具，并把结果写回消息流。

不过为了让新手更容易理解“发生了什么”，我们再额外写一个节点，把工具结果从消息里提取出来，存进显式状态字段 tool_result。

---

十一、定义结果整理节点

from langchain.messages import ToolMessage

def collect_tool_result_node(state: WeatherAgentState):
    last_message = state["messages"][-1]

    if isinstance(last_message, ToolMessage):
        return {
            "tool_result": last_message.content
        }
    return {}

这个节点的作用是：

• 从消息列表里取出最后一条消息

• 如果它是 ToolMessage，说明刚刚执行过工具

• 把工具返回值同步到 tool_result

这一步不是必须的，但对新手理解 State 很有帮助：
消息流是一种状态，显式字段也是一种状态。

---

十二、定义最终回答节点

工具结果拿到了，接下来就让模型把它组织成对用户友好的回答。

def respond_node(state: WeatherAgentState):
    tool_result = state.get("tool_result", "")
    user_question = state["messages"][0].content if state["messages"] else ""

    prompt = [
        SystemMessage(content="你是一个简洁、友好的天气助手，请基于工具结果回答用户，不要编造。"),
        *state["messages"]
    ]

    response = llm.invoke(prompt)
    return {
        "messages": [response],
        "final_answer": response.content
    }

这个节点会生成最终用户看到的内容，并写入：

• messages

• final_answer

---

十三、定义 Router：决定下一步去哪

这是 LangGraph 最关键的部分之一。

我们希望流程是：

• Agent 决策后，如果模型发起工具调用 → 去工具节点

• 如果模型没有工具调用 → 直接结束

• 工具执行后 → 去收集结果节点

• 收集完结果 → 再回到回答节点

• 回答完成 → 结束

先写判断函数：

def route_after_agent(state: WeatherAgentState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

这个 router 的逻辑非常简单：

• 如果 AIMessage 里有 tool_calls

• 说明模型决定调用工具

• 那么下一步就走到 tools

• 否则直接结束

注意：这里的 "tools"、"end" 不是随便写的，它们会映射到图里配置的节点或终点。

---

十四、把整个图组装起来

现在开始真正使用 StateGraph。

from langgraph.graph import StateGraph, START, END

builder = StateGraph(WeatherAgentState)

builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_node("collect_tool_result", collect_tool_result_node)
builder.add_node("respond", respond_node)

builder.add_edge(START, "agent")

builder.add_conditional_edges(
    "agent",
    route_after_agent,
    {
        "tools": "tools",
        "end": END
    }
)

builder.add_edge("tools", "collect_tool_result")
builder.add_edge("collect_tool_result", "respond")
builder.add_edge("respond", END)

graph = builder.compile()

到这里，一个最小可运行的天气 Agent 图就搭完了。

---

十五、完整流程图解

我们用文字流程图把它画出来：

START
  ↓
agent（让模型判断：直接回答 or 调用工具）
  ├── 如果需要工具 → tools（执行 get_weather）
  │                      ↓
  │               collect_tool_result（提取工具结果写入 state）
  │                      ↓
  │               respond（基于工具结果生成最终回答）
  │                      ↓
  │                     END
  │
  └── 如果不需要工具 → END

如果你想把这个流程理解得更“运行时”一些，可以看成这样：

1. 用户发来：“北京今天天气怎么样？”

2. agent 节点读到消息，模型判断：这是天气问题，需要调用 get_weather(city="北京")

3. Router 检查到存在 tool_calls

4. 图跳转到 tools

5. tools 执行 get_weather

6. 工具返回：“晴，22°C，东北风 2 级”

7. collect_tool_result 把结果写入 state["tool_result"]

8. respond 基于工具结果生成自然语言答案

9. 图到达 END

这就是一个标准的“先判断，再行动，再生成答案”的 Agent 闭环。

---

十六、运行示例

from langchain.messages import HumanMessage

result = graph.invoke({
    "messages": [HumanMessage(content="北京今天天气怎么样？")],
    "city": None,
    "tool_result": None,
    "final_answer": None,
})

print(result["final_answer"])
print(result["messages"][-1].content)

你大概率会得到类似结果：

北京今天天气晴，气温 22°C，东北风 2 级。整体天气不错，适合出行。

---

十七、给你一份可以直接运行的完整代码

下面把前面的代码拼成一份完整版本。

from typing import Optional
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain.messages import HumanMessage, SystemMessage, ToolMessage
from langgraph.graph import StateGraph, START, END, MessagesState
from langgraph.prebuilt import ToolNode

# 1. 定义工具
@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。输入必须是城市名。"""
    mock_weather_data = {
        "北京": "晴，22°C，东北风 2 级",
        "上海": "多云，26°C，东南风 3 级",
        "广州": "小雨，29°C，南风 2 级",
        "深圳": "阴，28°C，微风",
    }
    return mock_weather_data.get(city, f"暂时没有 {city} 的天气数据")

# 2. 定义状态
class WeatherAgentState(MessagesState):
    city: Optional[str]
    tool_result: Optional[str]
    final_answer: Optional[str]

# 3. 初始化模型
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
llm_with_tools = llm.bind_tools([get_weather])

# 4. Agent 决策节点
def agent_node(state: WeatherAgentState):
    system_prompt = SystemMessage(
        content=(
            "你是一个天气查询助手。"
            "如果用户在询问天气，并且提供了城市，优先调用 get_weather 工具。"
            "如果用户没有提供城市，不要调用工具，直接礼貌追问用户想查询哪个城市。"
            "当已经拿到工具结果后，请基于结果生成最终回答。"
        )
    )
    response = llm_with_tools.invoke([system_prompt] + state["messages"])
    return {"messages": [response]}

# 5. 工具节点
tool_node = ToolNode([get_weather])

# 6. 提取工具结果节点
def collect_tool_result_node(state: WeatherAgentState):
    last_message = state["messages"][-1]
    if isinstance(last_message, ToolMessage):
        return {"tool_result": last_message.content}
    return {}

# 7. 最终回答节点
def respond_node(state: WeatherAgentState):
    prompt = [
        SystemMessage(content="你是一个简洁、友好的天气助手，请基于工具结果回答用户，不要编造。"),
        *state["messages"]
    ]
    response = llm.invoke(prompt)
    return {
        "messages": [response],
        "final_answer": response.content
    }

# 8. Router
def route_after_agent(state: WeatherAgentState):
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return"tools"
    return"end"

# 9. 构建图
builder = StateGraph(WeatherAgentState)

builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
builder.add_node("collect_tool_result", collect_tool_result_node)
builder.add_node("respond", respond_node)

builder.add_edge(START, "agent")
builder.add_conditional_edges(
    "agent",
    route_after_agent,
    {
        "tools": "tools",
        "end": END
    }
)
builder.add_edge("tools", "collect_tool_result")
builder.add_edge("collect_tool_result", "respond")
builder.add_edge("respond", END)

graph = builder.compile()

# 10. 运行
result = graph.invoke({
    "messages": [HumanMessage(content="北京今天天气怎么样？")],
    "city": None,
    "tool_result": None,
    "final_answer": None,
})

print("最终答案：", result["final_answer"])

---

十八、这段代码里，LangGraph 概念分别对应哪里？

很多人代码能跑，但脑子里还是不清楚图是怎么组成的。我们把代码和概念一一对上。

1. State 对应哪里？

class WeatherAgentState(MessagesState):
    city: Optional[str]
    tool_result: Optional[str]
    final_answer: Optional[str]

这就是共享状态定义。
所有节点都围绕它读写。

---

2. Node 对应哪里？

def agent_node(...):
def collect_tool_result_node(...):
def respond_node(...):
tool_node = ToolNode([get_weather])

这些都是节点。
其中 ToolNode 是 LangGraph 提供的预制节点。

---

3. Edge 对应哪里？

builder.add_edge(START, "agent")
builder.add_edge("tools", "collect_tool_result")
builder.add_edge("collect_tool_result", "respond")
builder.add_edge("respond", END)

这些是固定边，表示确定的跳转关系。

---

4. Router 对应哪里？

def route_after_agent(state):
    ...

加上：

builder.add_conditional_edges(...)

这就是条件路由。
Agent 不是每次都去工具节点，Router 会动态决定。

---

5. 编译对应哪里？

graph = builder.compile()

官方文档明确提到：StateGraph 只是构建器，必须 compile() 后才能 invoke()。
这点新手经常忘。

---

十九、再进一步：如果用户没给城市怎么办？

例如用户只问：

“

“今天天气怎么样？”

这时一个合格的天气 Agent 不应该盲调工具，而应该先追问：

“

“请问你想查询哪个城市的天气？”

而我们的 agent_node 提示词里已经写了这个规则，因此模型通常不会发起工具调用。

执行流程会变成：

1. 用户问：“今天天气怎么样？”

2. agent 判断：缺少城市

3. 不调用工具

4. Router 返回 end

5. 图结束

这个例子特别适合帮助你理解：

“

Agent 的本质不是“必须调用工具”，而是“有能力根据状态决定是否调用工具”。

---

二十、想做成真正的循环 Agent，应该怎么改？

上面的示例是一个“单次工具调用”的简化图。
但 LangGraph 真正强大的地方在于：它非常适合做循环。

比如你可以改成下面这种结构：

START
  ↓
agent
  ├── 调工具 → tools → agent
  └── 直接回答 → END

这才是经典 Agent Loop：

1. 模型先思考

2. 需要工具就调工具

3. 工具结果回来后再交给模型

4. 模型继续判断还要不要下一步

5. 不需要了才结束

也就是说，工具执行后不是必须进 respond，而是可以回到 agent 再做一次决策。

这样做的好处是：

• 支持多个工具串联

• 支持多轮工具调用

• 更接近真实 Agent

例如代码结构会变成：

builder.add_edge("tools", "agent")

而 Router 则继续判断：

• 有工具调用 → 继续工具节点

• 没工具调用 → END

这就是 LangGraph 为什么非常适合 Agent：
它天然支持循环，而循环正是 Agent 的典型运行形态。

---

二十一、新手最常见的 5 个坑

坑 1：把 State 当成普通入参

很多人会写出这样的思路：

• 节点 A 返回一个字符串

• 节点 B 接收这个字符串

这更像普通函数调用，不是 LangGraph 的核心用法。

LangGraph 里更推荐的思路是：

• 所有节点围绕共享 State 工作

• 节点返回 Partial State 更新

---

坑 2：忘了 compile()

StateGraph 只是图构建器。
真正可执行的是编译后的 graph。

错误思路：

builder.invoke(...)

正确思路：

graph = builder.compile()
graph.invoke(...)

---

坑 3：搞不清消息和状态字段的关系

很多人问：

“

“既然 messages 里已经有工具结果了，为什么还要 tool_result？”

答案是：
不一定要。

但显式字段有两个好处：

1. 更利于调试

2. 更利于后续节点直接读取结构化信息

所以：

• 小 Demo 里你可以只用 messages

• 真正项目里常常会额外维护结构化 State 字段

---

坑 4：把 Router 理解成模型

Router 不是模型。
Router 是你写的一个普通 Python 函数，用来根据状态决定下一跳。

模型负责“思考”。
Router 负责“跳线”。

---

坑 5：以为 ToolNode 会自动帮你完成所有业务状态更新

ToolNode 很方便，但它主要处理的是：

• 读取工具调用

• 执行工具

• 把结果写回消息

如果你想把工具结果同步到自己的业务字段，比如：

• weather_json

• city

• risk_level

那通常还需要额外节点或让工具返回更结构化的数据。

---

二十二、怎么把模拟工具换成真实天气 API？

当你已经跑通这个示例后，升级非常简单。
只要改 get_weather() 就行，图结构几乎不用动。

例如：

import requests
from langchain.tools import tool

@tool
def get_weather(city: str) -> str:
    """查询指定城市的天气。输入必须是城市名。"""
    # 伪代码：替换为你自己的天气接口
    resp = requests.get("https://your-weather-api.com/weather", params={"city": city})
    data = resp.json()
    return f"{city}：{data['weather']}，{data['temp']}°C，{data['wind']}"

这正是 LangGraph 的另一个优点：

“

工作流编排和具体工具实现是解耦的。

你可以先把图跑通，再逐步把模拟工具替换成真实工具。

---

二十三、如果你想继续升级，这 4 个方向最值得学

当你学会这个天气 Agent 后，下一步建议你按这个顺序进阶。

1. 做成循环 Agent

让 tools -> agent，而不是直接去 respond。
这样你会真正理解 Agent Loop。

---

2. 给 State 增加更多业务字段

例如：

• intent

• location

• need_forecast

• retry_count

这样你会开始体会“有状态工作流”的价值。

---

3. 接入持久化与记忆

LangGraph 官方文档把持久化、长执行恢复、记忆都作为核心能力。
当你的 Agent 需要多轮对话或中断恢复时，这会非常重要。

---

4. 接入 LangSmith 做可观测性

一旦图变复杂，仅靠 print 很难调试。
可视化节点执行路径、状态变化和耗时，是 LangGraph 进入生产实践的重要一环。

---

二十四、最后，用一句话把 LangGraph 讲明白

如果你读到这里，还有点模糊，我给你一个最适合新手记住的定义：

“

LangGraph 就是把 Agent 的思考、工具调用、分支决策和状态管理，组织成一个可执行图。

在这个图里：

• State 负责记住当前进展

• Node 负责执行某一步逻辑

• Router/Conditional Edge 负责决定下一步去哪

• Loop 负责让 Agent 可以反复思考和行动，直到任务完成

而天气查询 Agent，正是理解这一切的最佳入门案例。

因为它足够简单，却完整包含了 Agent 的核心结构：

• 用户提问

• 模型判断

• 工具调用

• 状态更新

• 最终回复

当你真正把这个例子跑通，你对 LangGraph 的理解就不再停留在“看过概念”，而是已经建立起了最关键的直觉。

---

二十五、本文小结

我们今天完成了 3 件事：

1. 理解了 LangGraph 的核心概念

   • State 是共享状态

   • Node 是处理步骤

   • Router 是下一跳决策函数

2. 搭建了一个完整的天气查询 Agent

   • 用户问天气

   • Agent 判断是否调用工具

   • 工具返回结果

   • 最终生成回答

3. 明白了为什么 LangGraph 比普通链更适合 Agent

   • 它不是只做顺序调用

   • 它擅长有状态、可分支、可循环的工作流

   • 这正是 Agent 的天然形态

如果你接下来想真正从“会跑 Demo”进阶到“会设计 Agent”，建议你马上做两个练习：

• 把天气工具改成真实 API

• 把流程改成 agent -> tools -> agent 的循环结构

只要你亲手做完这两步，LangGraph 的门就算真正入了。