《2026 LangGraph零基础入门:从简单Agent到复杂多智能体系统的实战指南》第2课:LangGraph 的基础构建块:State、Node、Edge

失业一年了,天天想着怎么翻身。上一课很多人反馈终于看懂LangGraph不是黑盒了,这节课咱们继续用最慢、最细致的节奏,把State、Node、Edge这三个核心彻底讲透。每行代码都配详细注释,零基础也能跟得上。特别增加了大家最容易困惑的“节点间消息传递”和“条件边 vs 普通边优先级”的详细解释。

本课目标

  • 彻底搞懂State的定义和作用
  • 学会编写Node(包括LLM调用和后处理节点)
  • 掌握普通Edge和conditional edges的写法及优先级
  • 理解多个节点之间如何通过State进行消息链接

环境准备

pip install -U langgraph langchain langchain-deepseek langchain-core

设置API Key:

import os
os.environ["DEEPSEEK_API_KEY"] = "sk-你的key"

核心代码实战

代码段1:最基础的图(State + Node + 普通Edge)

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langchain_core.messages import AnyMessage, HumanMessage
from langchain_deepseek import ChatDeepSeek

# 定义图的状态(所有节点共享的数据黑板)
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]

llm = ChatDeepSeek(model="deepseek-chat", temperature=0.7)

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

workflow = StateGraph(State)
workflow.add_node("chat", chat_node)

workflow.add_edge(START, "chat")
workflow.add_edge("chat", END)

graph = workflow.compile()

result = graph.invoke({"messages": [HumanMessage(content="今天心情怎么样?")]})
print("回答:", result["messages"][-1].content)

代码段2:加入工具 + conditional edges

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

@tool
def get_weather(city: str) -> str:
    """查询城市天气(模拟)"""
    if "北京" in city or "beijing" in city.lower():
        return "北京今天多云转阴,最高16℃,最低8℃,有点凉"
    elif "上海" in city or "shanghai" in city.lower():
        return "上海今天晴,22℃,很舒服"
    return f"抱歉,暂时没有{city}的天气数据"

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

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

# 条件路由函数(纯Python判断,不消耗API)
def should_continue(state: State) -> str:
    last_msg = state["messages"][-1]
    if last_msg.tool_calls:
        return "tools"
    return END

tools_node = ToolNode(tools)

workflow = StateGraph(State)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", tools_node)

workflow.add_edge(START, "agent")

workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", END: END}
)

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

graph = workflow.compile()

代码段3:加入reminder后处理节点(重点讲解消息链接与边优先级)

from langchain_core.messages import AIMessage

# ==================== Reminder 后处理节点 ====================
# 重要说明:
# 1. reminder_node 与 agent_node 之间**没有直接消息传递**。
# 2. 它们通过共享的 State(messages列表)实现消息链接:
#    - agent_node 执行完后,把回复追加到 state["messages"]
#    - LangGraph 把最新的 State 传递给下一个节点
#    - reminder_node 可以直接读取 state["messages"][-1](agent刚产生的消息)
def reminder_node(state: State):
    # 读取 agent 刚刚输出的最后一条消息
    last_content = state["messages"][-1].content.lower()
    
    if "凉" in last_content or "冷" in last_content or "低" in last_content:
        reminder = AIMessage(content="天气有点凉,记得穿外套或加件衣服!")
        # 把提醒消息追加回 State
        return {"messages": [reminder]}
    return {}

# 添加 reminder 节点
workflow.add_node("reminder", reminder_node)

# ==================== 边的正确添加顺序与优先级说明 ====================
# 关键知识点:
# - conditional_edges 优先级高于普通边
# - agent_node 执行完后,LangGraph 会先执行 should_continue 判断
# - 只有当 should_continue 返回 END 时,才会走普通边 agent → reminder
workflow.add_edge("agent", "reminder")   # 强制后处理
workflow.add_edge("reminder", END)

# tools 执行完仍回到 agent(形成循环)
workflow.add_edge("tools", "agent")

# 重新编译图
graph = workflow.compile()

# 测试运行
result = graph.invoke({"messages": [HumanMessage(content="今天有点冷,天气温度怎么样?")]})

print("\n最终完整输出:")
for i, msg in enumerate(result["messages"]):
    print(f"{i+1}. {msg.type}: {msg.content}")

运行结果说明(以“今天有点冷,天气温度怎么样?”为例):

  1. Agent 第1次调用 → 决定调用工具
  2. Tools 执行 → 返回天气数据
  3. Agent 第2次调用 → 生成天气回答
  4. should_continue 返回 END → 走普通边进入 Reminder
  5. Reminder 检测到“凉” → 追加提醒消息
  6. 结束

最终用户会看到两条连续的AI消息:天气回答 + 穿衣提醒。

小练习(正好2道)

  • 练习1(基础)
    自己定义一个只包含 query: stranswer: str 的State,构建一个 START → llm_node → END 的简单图。写出State定义和核心三行代码。

  • 练习2(进阶)
    修改 should_continue 函数:如果用户问“现在几点了”,路由到tools节点(需自行定义 @tool def get_time())。写出修改后的 should_continue

本课小结
LangGraph的三大构建块:

  • State:共享数据载体,所有节点通过它传递消息(messages列表是最常见用法)。
  • Node:执行具体逻辑的函数,agent_node 每次执行都会消耗一次LLM调用。
  • Edge:决定流程走向。conditional_edges 优先级高于普通边。只有当 agent_node 执行完且 should_continue 返回 END 时,才算完成一次“思考周期”,此时才会沿着普通边进入后续节点(如 reminder)。

reminder_node 能读取agent的输出并追加提醒,正是因为State在节点之间连续传递。这也是LangGraph可控性的核心。

下节预告
第3课:构建第一个实用单 Agent 图(带工具调用)
我们会做一个真正能帮人写作的Agent(搜索 + 起草 + 润色),代码更完整,但依然带详细注释。下节直接开干!

如果觉得这篇有用,欢迎点赞和关注,一起玩转 LangGraph!

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐