青戈AI小栈
会员限时8折
LangGraph 生产级精讲:从状态机到上线

LangGraph 是什么:从关键词分流到第一张图

8,743 字约 22 分钟读完305 次阅读更新于 2026/10/22

登录后收藏

阅读设置

字号
行距
代码

用户提交一条工单:买错了型号要退款,软件登录不上要找客服。程序先判断类型,再走对应的回复流程。这个系列围绕客服工单 Agent「工单管家」,逐篇加上模型、工具、人工审批和持久化。

本篇先用关键词分流,写出第一张 LangGraph 图。会 Python 的函数、字典和类型标注就能跟着做;模型和工具从下一篇开始接。

环境准备

要求 Python 3.12+。在终端新建项目目录,进入目录后创建虚拟环境并安装 LangGraph:

powershell
# Windows(macOS / Linux:建 venv 用 python3,激活用 source .venv/bin/activate)
mkdir ticket-agent
cd ticket-agent
py -m venv .venv
.venv\Scripts\Activate.ps1
python -m pip install langgraph

本篇的代码在 langgraph 1.2.x(langchain-core 1.6.x)、Python 3.14.4 下实跑,下面的解释以这组版本为准。

先看一个不用框架的写法

不借助任何框架,分流逻辑大概长这样:

python
def dispatch(ticket: str) -> str:
    if "退款" in ticket or "退钱" in ticket:
        return "已收到退款申请……"
    elif "发票" in ticket:
        return "发票信息已登记……"
    else:
        return "工单已记录……"

现在这么写完全没问题。麻烦在往后:工单类型会越来越多,还要接模型、查订单、退款得人工确认,这个函数会越来越长,判断、流程、状态混在一起。下面的任务就是把「下一步做什么」从函数里拆出来,交给图来管。图版本先只保留退款和其他两条分支,发票那条放到练习里补——分支少一条,主干更容易看清。

LangGraph 的三个基本件

LangGraph 把一次任务执行拆成三件事:

  • 状态(State):一个字典,存任务当前的全部信息。
  • 节点(Node):一个函数,读状态、做一件事,返回「要更新的字段」。
  • 边(Edge):决定节点之间怎么走。固定顺序直接连;要根据状态选路的用条件边(Conditional Edge),由一个路由函数返回分支名。

三件事定义完,compile 成图,每次 invoke 从 START 跑到 END。

LangGraph 管执行流程;模型可以用 LangChain 的组件,也可以直接用模型服务的 SDK。本篇只写流程,装一个 langgraph 就够了。

写第一个图

在刚才的 ticket-agent 目录创建 main.py,把下面的完整代码写进去并保存。代码分四块:State、节点函数、建图、执行。先跑一次,再拆开看。

python
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class TicketState(TypedDict):
    ticket: str    # 原始工单内容
    category: str  # 分类结果:refund / other
    reply: str     # 给用户的回复


def classify_ticket(state: TicketState) -> dict:
    text = state["ticket"]
    if "退款" in text or "退钱" in text:
        return {"category": "refund"}
    return {"category": "other"}


def route(state: TicketState) -> str:
    return state["category"]


def refund_reply(state: TicketState) -> dict:
    return {"reply": "已收到退款申请,客服将在 1 个工作日内与你联系处理。"}


def other_reply(state: TicketState) -> dict:
    return {"reply": "工单已记录,客服会尽快回复。"}


builder = StateGraph(TicketState)
builder.add_node("classify", classify_ticket)
builder.add_node("refund_reply", refund_reply)
builder.add_node("other_reply", other_reply)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"refund": "refund_reply", "other": "other_reply"})
builder.add_edge("refund_reply", END)
builder.add_edge("other_reply", END)

graph = builder.compile()

result = graph.invoke({"ticket": "买错了型号,申请退款"})
print(result["category"])
print(result["reply"])

result = graph.invoke({"ticket": "软件一直登录不上"})
print(result["category"])
print(result["reply"])

先跑一次

powershell
python main.py
输出
refund
已收到退款申请,客服将在 1 个工作日内与你联系处理。
other
工单已记录,客服会尽快回复。

再看两个细节:

  • 第二次 invoke 的结果里没有第一次的影子。图不存数据,每次调用独立;跨调用的会话记忆靠 checkpoint,后面会讲。
  • classify_ticket 只动了 category,reply 是后面的节点补上的。这就是「节点只更新返回字段」。

代码逐段说明

State:工单在图里的样子

TicketState 是一个 TypedDict,三个字段:ticket 存原始文本,category 存分类结果,reply 存回复文案。它在节点之间流动,每个节点只改自己负责的字段。TypedDict 是字典形状的说明,不会自动创建三个字段,也不做运行时校验。调用时只传 ticket,后两个字段由节点逐步补上:

输出
输入:       {"ticket": "买错了型号,申请退款"}
分类之后:   {"ticket": "买错了型号,申请退款", "category": "refund"}
回复之后:   {"ticket": "买错了型号,申请退款", "category": "refund", "reply": "已收到退款申请,客服将在 1 个工作日内与你联系处理。"}

所以在分类节点之前读 state["category"],或者在回复节点之前读 state["reply"],都会因字段还不存在而报 KeyError。

节点:一个函数干一件事

classify_ticket 读 ticket,返回 {"category": ...};两个 *_reply 各写一句 reply。本篇使用的节点返回字典,字典的键得是 State 里声明过的字段——返回什么,图就更新什么;没返回的保持原样。

route 是个例外:它不是节点,只是给条件边用的路由函数,返回 "refund" 或 "other" 这种分支名,用来告诉图下一步去哪。

建图:九行逐行看

上面那些函数只是零件,真正把流程固定下来的是这九行:

python
builder = StateGraph(TicketState)                  # ① 建一个空图
builder.add_node("classify", classify_ticket)      # ② 注册节点:名字 → 函数
builder.add_node("refund_reply", refund_reply)     # ③
builder.add_node("other_reply", other_reply)       # ④
builder.add_edge(START, "classify")                # ⑤ 声明入口
builder.add_conditional_edges(                     # ⑥ 声明出口
    "classify", route, {"refund": "refund_reply", "other": "other_reply"}
)
builder.add_edge("refund_reply", END)              # ⑦ 走到终点
builder.add_edge("other_reply", END)               # ⑧
graph = builder.compile()                          # ⑨ 变成可执行对象

① StateGraph(TicketState):建一个空图,并声明这张图用哪个 State。参数是类型本身,不是数据;此刻图里既没有节点也没有边,只有「State 长什么样」这一条信息——它决定节点函数的注解怎么写(state: TicketState)。

②③④ add_node(名字, 函数):把节点注册进图。两个参数是节点名(字符串)和节点函数。名字不是给函数用的,是给边用的:⑤⑥⑦⑧ 里写的全是名字。名字和函数名一样只是习惯,写成 add_node("step1", classify_ticket) 照样能跑。少注册一个节点,指向它的边会在 compile() 时被拒:ValueError: Found edge ending at unknown node ...。

⑤ add_edge(START, "classify"):声明入口。START 是 LangGraph 提供的常量,代表「图开始执行」,它不是节点。没有从 START 出发的边,compile() 直接报 ValueError: Graph must have an entrypoint: add at least one edge from START to another node。

⑥ add_conditional_edges(源节点, 路由函数, 映射表):唯一一条「不写死下一步」的边。三个参数分别是:从哪个节点出发("classify");用哪个函数决定去向(route,它读 State、返回一个字符串);怎么把返回值翻译成节点名({"refund": "refund_reply", ...})。

为什么不让路由函数直接返回节点名,而要一张映射表?因为路由函数判断的是业务(这单是不是退款),节点名是实现(谁来回这句话)。分开之后,改节点名不用动业务判断,看映射表就知道每种结果去哪;映射表本身也把「一共有几种走法」摊开了。代价是映射表必须覆盖路由函数所有可能的返回值——漏一个,运行时就是 KeyError(练习第 2 题会让你亲手撞一次)。

⑦⑧ add_edge(节点, END):END 和 START 一样是常量,表示「执行结束」。这两行是声明意图:走到这里就收工。技术上不写也能跑(没有出边的节点会被当成终点),但写了之后,读代码的人一眼能看出流程有几个出口、分别在哪——以后加了环或分支,也不容易漏。

⑨ compile():把上面这套定义校验一遍,变成能 invoke 的对象。校验都发生在这里:入口有没有、边指向的节点存不存在、条件边的目标存不存在。它检查的是图的结构;路由返回了映射表没有的值、节点返回了错误的数据,仍要到 invoke() 才会发现。

自己写一个图,按这五步

把这九行还原成「怎么写」的过程,就是五步:

  1. 先把流程画出来:本例是「分类 → 退款回复 / 其他回复 → 结束」。先有流程,才有图。
  2. 给每一步起名字:classify、refund_reply、other_reply——这些就是节点名,连边全靠它们。
  3. 定 State:每一步要读什么、要写什么,列成 TypedDict 的字段(这里三个:ticket、category、reply)。
  4. 写节点函数:一个函数一步,第一个参数是 state,返回要更新的字段组成的字典。
  5. 建图:StateGraph(State) → 逐个 add_node(名字, 函数) → 按流程连边(入口用 START、出口用 END、要选路的用 add_conditional_edges)→ compile()。

以后改流程,也照这五步对号入座:

想做的事 改哪里
加一步 写节点函数 + add_node + 前后各连一条边;若它由分支决定,还要在映射表里加一项
改判断条件 只改节点函数或路由函数,图结构不动
改节点名 改 add_node 的第一个参数,以及所有指向它的边和映射表
换一套流程 先动 State 的字段,再按这五步重排节点与边

编译与执行

builder.compile() 把定义变成可执行的 graph。graph.invoke(...) 触发一次执行:START → classify → 路由 → 某个 reply 节点 → END。invoke 返回的是执行结束后的完整 State。

常见报错

PowerShell 提示无法加载 Activate.ps1

虚拟环境已经建好,但激活脚本被限制了。可以直接用虚拟环境里的解释器,不必修改系统策略。在 ticket-agent 目录执行 .venv\Scripts\python.exe -m pip install langgraph,运行时执行 .venv\Scripts\python.exe main.py。

ModuleNotFoundError: No module named 'langgraph'

装的 Python 和跑的 Python 不是同一个(venv 没激活,或者 IDE 的解释器选错了)。在同一条命令行里执行 python -m pip install langgraph,装和跑用同一个解释器。

langgraph.errors.GraphRecursionError: Recursion limit of 10007 reached without hitting a stop condition.

图里有环,无限循环走不到头,通常是条件边的路由结果指回了上游节点(比如分类完又跳回分类)。检查路由函数有没有可能返回它自己所在的节点;调试时给 invoke 传 config={"recursion_limit": 20},让它早点报错。顺带说清楚:漏连 add_edge(..., END) 不会报这个错,节点没有出边时图会把它当终点。

改了节点里的代码,输出没变

分两种情况。节点返回的不是字典(比如直接返回了字符串),会报 langgraph.errors.InvalidUpdateError: Expected dict, got ...;返回的是字典但键名拼错了,图不报错,但 State 里那个字段不会被更新,invoke 的返回里就找不到它。

KeyError: 'other'(同一段报错里还跟着一行 During task with name ...)

路由返回了映射表里没有的分支名。给 add_conditional_edges 的映射表补上对应的一项。

下一步

到这里,你已经有一个能跑的 3 节点状态机,结构是「分类 → 路由 → 回复」。下一篇把 classify_ticket 换成真正调模型的节点——结构不用动,只换分类的判断来源。

练习

改代码之前先预测。下面四题都在本篇的 main.py 上改:动手改之前,先把结果写在注释里,再跑起来对照。第 1–3 题各自从正文原版开始,做完一题就还原,尤其不要把第 1 题固定退款的 route 带进第 2 题。第 4 题也从原版补分支。

  1. 让 route 永远走退款分支:

    python
    def route(state: TicketState) -> str:
        return "refund"

    跑 graph.invoke({"ticket": "软件一直登录不上"})。result["category"] 和 result["reply"] 分别是什么?

  2. 把条件边的映射表漏掉 "other":

    python
    builder.add_conditional_edges("classify", route, {"refund": "refund_reply"})

    再跑同一条工单,会发生什么?

  3. 删掉 builder.add_edge("refund_reply", END),跑一条退款工单。会不会报错?

  4. 把发票接回图里。前面不用框架的那版有「发票」这条分支,图版本只做了退款和其他,现在补齐:工单里带「发票」时分类返回 "invoice",新增节点 invoice_reply,回复「发票信息已登记,我们会尽快为你开具。」。

    跑 graph.invoke({"ticket": "发票开错了,需要重开"}),要拿到 invoice 和这句回复。

验收: 四题都能说出原因;第 1 题能讲清 category 和 reply 为什么对不上;第 3 题能解释为什么少一条边也没事;第 4 题能说清一条分支要动哪几处,以及漏掉其中哪一处会报什么错。

查看提示

前三题都只改一两行,第 4 题是把一条分支从头接到尾。卡住的话按这个顺序想:谁写了 category、谁写了 reply;路由函数的返回值去哪张表里查;节点没有出边时图会拿它怎么办;第 4 题再问自己一句——一条分支要想跑通,得同时具备哪几样东西。

查看参考答案
  1. category 是 other,reply 是「已收到退款申请,客服将在 1 个工作日内与你联系处理。」

    两个字段看起来不一致,因为它们出自不同节点:classify_ticket 已经把 category 写成了 other,固定路由只是让 refund_reply 补上了 reply,没有回头改 category。路由函数只决定「走哪个节点」,不改写 State 里已有的字段。

  2. 报 KeyError: 'other',同一段报错里还跟着一行 During task with name 'classify' and id ...——LangGraph 用它指出是哪个节点出的错。

    工单被判成 other,路由返回 "other",映射表里没有这一项,图找不到要去的节点。映射表必须覆盖路由函数所有可能的返回值。

  3. 不报错,正常跑完,输出 refund 和退款话术。

    节点没有出边时,图把它当终点处理。所以漏连 add_edge(..., END) 不会触发 GraphRecursionError——那个错来自图里真的有环,比如路由结果指回了上游节点。

  4. 一条分支要五处同时到位:classify_ticket 里加 if "发票" in text: return {"category": "invoice"}、写 invoice_reply 节点、add_node("invoice_reply", invoice_reply)、映射表补 "invoice": "invoice_reply"、补 add_edge("invoice_reply", END)。

    跑「发票开错了,需要重开」拿到 invoice 和发票话术。漏掉哪一处,后果不一样:分类判断漏了,工单落到 other,走的是其他话术,图不报错;节点函数或 add_node 漏了,compile() 时报 ValueError: At 'classify' node, 'route' branch found unknown target 'invoice_reply';映射表漏了这一项,报 KeyError: 'invoice'(和第 2 题同一个机制);出边漏了不报错,节点没有出边时图把它当终点(和第 3 题同一个机制)。

第 4 题的完整 main.py 如下。第 1–3 题的改动都已还原;在 ticket-agent 目录执行 python main.py:

python
from typing import TypedDict

from langgraph.graph import END, START, StateGraph


class TicketState(TypedDict):
    ticket: str    # 原始工单内容
    category: str  # 分类结果:refund / invoice / other
    reply: str     # 给用户的回复


def classify_ticket(state: TicketState) -> dict:
    text = state["ticket"]
    if "退款" in text or "退钱" in text:
        return {"category": "refund"}
    if "发票" in text:
        return {"category": "invoice"}
    return {"category": "other"}


def route(state: TicketState) -> str:
    return state["category"]


def refund_reply(state: TicketState) -> dict:
    return {"reply": "已收到退款申请,客服将在 1 个工作日内与你联系处理。"}


def invoice_reply(state: TicketState) -> dict:
    return {"reply": "发票信息已登记,我们会尽快为你开具。"}


def other_reply(state: TicketState) -> dict:
    return {"reply": "工单已记录,客服会尽快回复。"}


builder = StateGraph(TicketState)
builder.add_node("classify", classify_ticket)
builder.add_node("refund_reply", refund_reply)
builder.add_node("invoice_reply", invoice_reply)
builder.add_node("other_reply", other_reply)
builder.add_edge(START, "classify")
builder.add_conditional_edges("classify", route, {"refund": "refund_reply", "invoice": "invoice_reply", "other": "other_reply"})
builder.add_edge("refund_reply", END)
builder.add_edge("invoice_reply", END)
builder.add_edge("other_reply", END)

graph = builder.compile()

for ticket in ["买错了型号,申请退款", "发票开错了,需要重开", "软件一直登录不上"]:
    result = graph.invoke({"ticket": ticket})
    print(result["category"], "|", result["reply"])
输出
refund | 已收到退款申请,客服将在 1 个工作日内与你联系处理。
invoice | 发票信息已登记,我们会尽快为你开具。
other | 工单已记录,客服会尽快回复。
有帮助
登录后提问 / 纠错

提问与反馈

仅你和管理员可见。我的反馈

设置你的昵称

在英雄榜和讨论中,使用你自己的名字。

2–12 个字符0 / 12