LangGraph 基本概念：State、Node、Edge 入门

上篇你搞清了「为何要 Graph」和选型三句话。这篇是 LangGraph 专题第 02 篇——把三要素写进代码：线性图 → 条件边 → 工具环 → checkpointer。

小明想把「必须拒答」画成图，却不知从哪下笔

选型抄进本子后，小明打开编辑器，准备把客服改成图。盯着空白文件发呆：

State 放哪儿？节点返回整个字典还是只返回改动的字段？条件边和 if 有啥区别？

他跑去找老张：”介绍篇说 Node / Edge / State，能不能一天一台阶拆开？”

老张拉开椅子：”可以。今天核心就三样——节点干活、边指路、状态传数据。 Command、interrupt 今天只点名，深挖留给客服项目篇。”

三要素：一张图记住

老张在白板上写：

State  = 整个工作流里流转的数据（TypedDict）
Node   = 关键工作代码（函数）：读 State → 干活 → 返回「要更新的字段」
Edge   = 连接节点的路径：串行 / 并行 / 条件
Graph  = Node + Edge 组成的工作流

比喻：

State 是共享白板；每个 Node 是来写一笔的同事；Edge 规定下一位谁上场。

构图固定四步：

StateGraph(State)
  → add_node / add_edge / add_conditional_edges
  → compile()
  → invoke(...)

线性图——节点返回

不接 LLM，先练骨架

class GraphState(TypedDict):
  text: str
  steps: list[str]


def normalize(state: GraphState) -> dict:
  text = state["text"].strip().lower()
  steps = list(state.get("steps") or [])
  steps.append(f"{len(steps) + 1}.规范化文本")
  return {"text": text, "steps": steps}  # 只返回要改的字段


def shout(state: GraphState) -> dict:
  steps = list(state.get("steps") or [])
  steps.append(f"{len(steps) + 1}.大写文本")
  return {"text": state["text"].upper() + "!", "steps": steps}


def build_graph():
  g = StateGraph(GraphState)
  g.add_node("normalize", normalize)
  g.add_node("shout", shout)
  g.add_edge(START, "normalize")
  g.add_edge("normalize", "shout")
  g.add_edge("shout", END)
  return g.compile()

老张敲黑板：

1. 输入是完整 State；输出是部分 dict——框架负责合并。
2. 不要在节点里偷偷改全局变量；可测、可恢复都靠「返回更新」。
3. START → normalize → shout → END：串行边，路径写死。

条件边——把 if/else 画成路由表

START → classify ┬─ weather → END
         ├─ policy  → END
         └─ chitchat → END

class GraphState(TypedDict):
    question: str
    route: str
    answer: str
  
def classify(state: GraphState) -> dict:
    """ 分类问题 """
    q = state["question"].lower()
    if any(k in q for k in ("天气", "weather", "温度")): # 判断是否包含天气、weather、温度
        route = "weather"
    elif any(k in q for k in ("请假", "年假", "制度")): # 判断是否包含请假、年假、制度
        route = "policy"
    else: # 否则为闲聊
        route = "chitchat"
    return {"route": route}
  
def weather_node(state: GraphState) -> dict:
    return {"answer": "（天气分支）今日培训城市：晴，22C（模拟）"}

def policy_node(state: GraphState) -> dict:
    return {"answer": "（制度分支）请查阅请假制度：满 1 年年假 5 天（样例）。"}

def chitchat_node(state: GraphState) -> dict:
    return {"answer": "（闲聊分支）你好，我是路由演示助手。"}

def pick_branch(state: GraphState) -> Literal["weather", "policy", "chitchat"]:
    return state["route"]  # type: ignore[return-value]

def build_graph():
    g = StateGraph(GraphState)
    g.add_node("classify", classify)
    g.add_node("weather", weather_node)
    g.add_node("policy", policy_node)
    g.add_node("chitchat", chitchat_node)
    g.add_edge(START, "classify")
    g.add_conditional_edges(
        "classify",
        pick_branch,
        {"weather": "weather", "policy": "policy", "chitchat": "chitchat"},
    )

    g.add_edge("weather", END)
    g.add_edge("policy", END)
    g.add_edge("chitchat", END)

    return g.compile()

def run_demo() -> None:
    app = build_graph()
    cases = [
        "杭州天气怎么样？",
        "年假怎么请？",
        "你好呀",
    ]

    for q in cases:
        out = app.invoke({"question": q, "route": "", "answer": ""})
        print(f"Q: {q}")
        print(f"  route={out['route']} | answer={out['answer']}")

要点：

• classify 写 route 进 State；pick_branch 读 State 决定下一节点名。
• 映射表里的 key 必须和路由函数返回值对得上；建议 Literal + 默认分支。
• 规则路由稳、便宜；换成 LLM 分类更懂自然语言，但贵、要兜底（WorkFlow 篇再比）。

python main.py

Q: 杭州天气怎么样？
  route=weather | answer=（天气分支）今日培训城市：晴，22C（模拟）
Q: 年假怎么请？
  route=policy | answer=（制度分支）请查阅请假制度：…
Q: 你好呀
  route=chitchat | answer=（闲聊分支）你好，我是路由演示助手。

工具环——可见的 ReAct

小明最熟的 while tool_calls，画成图就是：

START → agent ──has tool_calls──► tools ─┐
      ▲                │
      └──────────── loop ────────────┘
      └─ no tool_calls → END

class AgentState(TypedDict):
  messages: Annotated[list[BaseMessage], add_messages]

@tool
def get_weather(city: str) -> str:
    """查询城市天气（模拟）。"""
    data = {"杭州": "小雨，17C", "北京": "晴，15C", "Hangzhou": "rain, 17C"}
    return data.get(city, f"{city}：晴，20C（模拟）")

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

model = ChatOpenAI(
    model=os.environ.get("OPENAI_MODEL", "qwen-plus"),
    api_key=os.environ["OPENAI_API_KEY"],
    base_url=os.environ.get("OPENAI_BASE_URL"),
    temperature=0,
).bind_tools(tools)

def call_model(state: AgentState) -> dict:
    """ 调用模型 """
    ai = model.invoke(state["messages"])
    return {"messages": [ai]}

def should_continue(state: AgentState) -> str:
    """ 判断是否继续 """
    last = state["messages"][-1]
    # 判断是否为AI消息且有工具调用
    if isinstance(last, AIMessage) and last.tool_calls:
        return "tools"
    return "end"

def build_graph():
    g = StateGraph(AgentState)
    g.add_node("agent", call_model)
    g.add_node("tools", tool_node)
    g.add_edge(START, "agent")
    g.add_conditional_edges("agent", should_continue, {"tools": "tools", "end": END})
    g.add_edge("tools", "agent")
    return g.compile()

def run_demo() -> None:
    app = build_graph()
    result = app.invoke(
        {"messages": [HumanMessage(content="杭州天气怎么样？用一句话回答。")]}
    )

    for m in result["messages"]:
        t = getattr(m, "type", "?")
        if getattr(m, "tool_calls", None):
            print(f"[{t}] tool_calls={m.tool_calls}")
        else:
            content = (getattr(m, "content", "") or "")[:200]
            print(f"[{t}] {content}")

老张强调两处：

1. add_messages 是 reducer：多轮消息追加，不是整表覆盖。消息链场景几乎必用。
2. 和 create_agent 比：循环看得见；中间可插审核、检索、限步节点。进阶可在 should_continue 里加「最多 3 步」。

python main.py --demo

典型输出形状（内容随模型略有出入）：

[human] 杭州天气怎么样？用一句话回答。
[ai] tool_calls=[...]
[tool] 小雨，17C
[ai] 杭州今天小雨，约 17°C。
Day3 OK

checkpointer + thread_id——图也有短期记忆

class ChatState(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def chatbot(state: ChatState) -> dict:
    ai = model.invoke(state["messages"])
    return {"messages": [ai]}

def build_graph():
    g = StateGraph(ChatState)
    g.add_node("chatbot", chatbot)
    g.add_edge(START, "chatbot")
    g.add_edge("chatbot", END)
  return g.compile(checkpointer=InMemorySaver())

cfg1 = {"configurable": {"thread_id": "u1"}}
cfg2 = {"configurable": {"thread_id": "u2"}}

app.invoke({"messages": [HumanMessage(content="我叫 Carol。")]}, cfg1)
app.invoke({"messages": [HumanMessage(content="我叫什么？")]}, cfg1)  # 同线程记得
app.invoke({"messages": [HumanMessage(content="我叫什么？")]}, cfg2)  # 异线程不串

没有 checkpointer：每次 invoke 不会自动跨调用恢复，除非你自己传入完整历史。课堂用 InMemorySaver；生产再换 Postgres 等（选修）。

python main.py --demo

=== Day4 demo: graph checkpointer + thread_id ===
thread=u1 | （确认记住 Carol 一类回复）
thread=u1 | Carol（或等价）
thread=u2 | （不知道 / 未在本线程介绍）
Day4 OK

对照表：四天分别练了什么

进阶名词：

三个常见坑

坑 1：节点返回了「完整新 State」却漏字段

症状：steps 或其它键莫名变空。
原因：误以为必须返回全量；漏键若被当成覆盖会出事（视 reducer 而定）。
修法：习惯只返回变更字段；列表字段自己 list(...) 拷贝再改，避免共享引用踩坑。

坑 2：条件边返回值不在映射表里

症状：运行报错或走到意外终点。
原因：pick_branch 返回了 "Weather" 而 map 写的是 "weather"。
修法：Literal 约束；永远留默认分支（如 chitchat）。

坑 3：消息 State 不用 add_messages

症状：第二轮对话把历史盖掉，或类型对不上。
原因：普通 list 默认是覆盖语义。
修法：messages: Annotated[list[BaseMessage], add_messages]（或 MessagesState）。

总结

“三个核心理解：

1. 三要素 —— State 传数据，Node 返回部分更新，Edge 决定下一步（串行/条件/并行）
2. 构图四步 —— StateGraph → 加节点边 → compile → invoke
3. 从管道到 Agent —— 线性边 → 条件路由 → 工具环 → checkpointer 记忆”

LangGraph 专题进度：

01 介绍（为何 Graph）
    ↓
02 基本概念和入门（本篇）
    ↓
03 常见 WorkFlow（七种编排模式）
    ↓
04 客户支持 Email Agent

图不是魔法：它是把你本来要写的 if/while/存档，变成看得见、改得动、测得着的拓扑。