从 LangChain 到 LangGraph：一次有状态工作流的自学实践
LangChain 和 LangGraph 经常一起出现，但解决的问题并不相同。这篇通过一个带分支、循环、审核和状态持久化的客服工单 Demo，记录我从“会调用模型”走向“会组织有状态工作流”的过程。
最近在继续补 Agent 相关的知识，LangChain 和 LangGraph 是两个很难绕开的名字。
刚接触时，我一度把它们理解成两个功能差不多的 Agent 框架：都能调用大模型，也都能连接工具。真正写过一个带分支和循环的流程后，我才慢慢分清它们的关注点：LangChain 更像一套连接模型与外部能力的组件，LangGraph 更像一个负责保存状态和控制执行路线的运行时。
这篇文章记录一下我的学习过程。前半部分先梳理 LangChain、LangGraph 的基本概念以及两者的关系，后半部分实现一个不算玩具、但也没有复杂到难以读懂的客服工单处理 Demo。
这个 Demo 会包含：
- 大模型结构化分类；
- 知识库工具调用；
- 根据工单类型进行条件路由；
- 生成回复后自动审核；
- 审核不通过时带着意见重新生成；
- 多次失败后转人工；
- 使用 Checkpointer 保存每一步状态。
先看一下这次的学习路线：
先从 LangChain 开始
LangChain 是一套用于开发大模型应用的开源框架。它把模型、Prompt、消息、工具、结构化输出、检索等常见能力整理成了相对统一的接口。
如果直接调用某一家模型厂商的 SDK，我们当然也能完成一次对话。但应用稍微复杂一点，马上就会遇到更多问题：Prompt 怎么复用？模型怎样切换？输出怎样变成程序能可靠处理的数据？怎样调用数据库或搜索服务？
LangChain 主要是在解决这些“模型调用之外”的连接问题。
我认为最需要先掌握的几个组件
Model
Model 是整个应用的推理核心。LangChain 为不同供应商提供了相近的调用接口，让上层代码不必和某一家 SDK 完全绑定。
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
response = model.invoke("用一句话解释什么是 LangChain")
print(response.content)这里的价值不只是少写几行请求代码。统一接口还意味着 Prompt、工具和输出解析等上层组件可以围绕同一种调用方式组合。
Prompt
Prompt 不应该只是散落在业务代码里的长字符串。ChatPromptTemplate 可以把角色消息、变量和固定规则放在一个可复用模板中。
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "你是一名技术客服，请用清楚、克制的方式回答。"),
("human", "用户问题：{question}"),
])Runnable 与 LCEL
LangChain Expression Language（LCEL）可以用 | 把多个 Runnable 连接起来。数据从左向右流动，前一个组件的输出成为后一个组件的输入。
from langchain_core.output_parsers import StrOutputParser
chain = prompt | model | StrOutputParser()
answer = chain.invoke({"question": "程序升级后无法启动怎么办？"})这条链很直观，但它基本是一条固定流水线：Prompt 处理完交给模型，模型处理完交给解析器。
Structured Output
自然语言适合给人看，却不适合直接作为程序的路由条件。例如程序希望得到工单类别和紧急程度，如果只让模型随意输出文本，后续还要猜它到底写了“高”“紧急”还是 high。
结构化输出可以用 Pydantic 模型约束结果：
from typing import Literal
from pydantic import BaseModel, Field
class TicketAnalysis(BaseModel):
category: Literal["account", "order", "technical", "other"]
urgency: Literal["low", "medium", "high"]
summary: str = Field(description="一句话概括用户问题")
classifier = model.with_structured_output(TicketAnalysis)
result = classifier.invoke("客户端升级后启动白屏，今天一直无法工作")这样返回的 result.category 就可以直接参与程序判断，而不是再解析一段不稳定的自然语言。
Tool
工具让模型或工作流能够接触外部世界，例如查询订单、检索知识库、读取数据库或调用内部 API。
from langchain.tools import tool
@tool
def query_order(order_id: str) -> str:
"""根据订单号查询订单状态。"""
return f"订单 {order_id} 已发货"@tool 会根据函数签名和文档生成工具描述。不过需要注意，模型只是提出工具调用意图，真正执行函数、检查权限和处理异常的仍然是应用程序。
Chain 和 Agent 不是一回事
学习 LangChain 时，我觉得很容易把 Chain 和 Agent 混在一起。
Chain 的路线通常由程序提前确定。比如“读取文章 → 生成摘要 → 翻译英文”，每次都沿着相同顺序执行。
Agent 则把一部分决策交给模型。模型会根据当前输入判断是否需要工具、调用哪个工具，以及拿到工具结果后是否还要继续行动。
固定流程并不比 Agent 落后。恰恰相反，如果业务规则已经很清楚，就应该把这部分路线明确写进程序；只有真正需要判断的环节，才交给模型。
当一个任务同时包含确定步骤、条件分支、循环、人工审批和长时间状态保存时，只靠一条 Chain 就开始变得别扭了。这也是我继续学习 LangGraph 的原因。
LangGraph 解决什么问题
LangGraph 是一个面向长时间运行、有状态工作流和 Agent 的底层编排框架。
它并不负责替代 LangChain 的模型和工具接口，而是把整个执行过程表示成一张图。官方 Graph API 中最核心的三个概念是：
- State：整个流程共享的状态；
- Node：读取状态、执行工作并返回状态更新的节点；
- Edge：连接节点，决定下一步执行位置的边。
State：工作流的共同记忆
State 可以理解成任务执行过程中的一份共享数据。每个节点读取它需要的字段，再返回自己负责的增量更新。
from typing_extensions import TypedDict
class SupportState(TypedDict, total=False):
ticket: str
category: str
urgency: str
draft: str
approved: bool这里的 State 和聊天记录并不是同一个概念。消息可以是 State 的一部分，但工单类别、审核次数、检索结果这些业务数据同样应该明确保存，而不是全部塞进对话文本。
Node：只完成一个清楚的步骤
节点本质上就是一个 Python 函数。它接收当前 State，并返回需要更新的字段：
def finalize_reply(state: SupportState):
return {"final_answer": state["draft"]}节点最好保持职责单一。如果一个节点里同时完成分类、检索、生成、审核，图虽然看起来简单，内部却重新变成了难以观察的大黑盒。
Edge：固定连接与条件路由
普通 Edge 表示固定的下一步，Conditional Edge 则根据 State 选择不同路线。
workflow.add_edge("retrieve_knowledge", "draft_reply")
workflow.add_conditional_edges("review_reply", route_after_review)图可以产生分支，也可以从后面的节点重新连回前面的节点，因此很适合表达“审核不通过就重写”这样的循环。
Checkpointer：保存每一步快照
LangGraph 可以在编译图时接入 Checkpointer。执行过程会按照 thread_id 保存状态快照，从而支持查看当前状态、从中断处恢复以及实现 Human-in-the-loop。
from langgraph.checkpoint.memory import InMemorySaver
graph = workflow.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "ticket-001"}}InMemorySaver 适合本地学习和测试，进程退出后数据就没有了。生产环境应该换成 SQLite、PostgreSQL 等持久化实现，并同时考虑数据保留期限和敏感信息保护。
LangChain 和 LangGraph 到底是什么关系
现在再把两者放在一起看，就清楚多了：
| 对比项 | LangChain | LangGraph |
|---|---|---|
| 主要关注点 | 模型、Prompt、工具、检索、结构化输出和 Agent 抽象 | 状态、节点、路由、循环、持久化和恢复 |
| 适合的起点 | 快速连接模型和外部能力 | 明确控制复杂任务的执行过程 |
| 流程控制 | 简单 Chain 或预构建 Agent 循环 | 显式图结构，可分支、循环和中断 |
| 状态管理 | 组件或 Agent 上下文 | 工作流的一等公民 |
| 是否必须一起使用 | 否 | 否，LangGraph 也可以独立使用 |
在这次 Demo 中，两者会这样协作：
一句不够严谨但方便记忆的话是：LangChain 准备零件，LangGraph 组织这些零件按什么顺序工作。
Demo：带审核循环的客服工单助手
为什么选择这个案例
如果 Demo 只是 START → 调用模型 → END，确实能运行，却很难体现 LangGraph 的价值。
客服工单刚好包含几种常见的真实需求：
- 先由模型把自然语言工单转成结构化分类；
- 账号安全问题不允许自动回复，必须直接转人工；
- 订单和技术问题需要查询对应知识；
- 普通咨询可以直接生成回复；
- 生成结果还要检查是否准确、完整、没有编造；
- 审核失败时根据意见重写，但不能无限循环；
- 每一步状态都应该能够追踪。
完整流程如下：
这个流程既有确定性规则，也有模型判断。账号安全转人工、最多重写两次属于程序规则；工单分类、草稿生成和质量审核则使用大模型。
1. 安装依赖
本文使用 Python 3.11+，先创建虚拟环境并安装依赖：
python -m venv .venv
pip install -U langchain langchain-openai langgraph pydantic配置模型密钥和可选的模型名称：
# macOS / Linux
export OPENAI_API_KEY="你的 API Key"
export OPENAI_MODEL="gpt-4.1-mini"PowerShell 的写法是：
$env:OPENAI_API_KEY="你的 API Key"
$env:OPENAI_MODEL="gpt-4.1-mini"下面的代码按顺序放进同一个 support_workflow.py 文件即可运行。
2. 定义 State 和结构化结果
import operator
import os
from typing import Annotated, Literal
from langchain.tools import tool
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph
from pydantic import BaseModel, Field
from typing_extensions import TypedDict
class TicketAnalysis(BaseModel):
"""工单分类节点的结构化输出。"""
category: Literal["account", "order", "technical", "other"] = Field(
description="工单类别"
)
urgency: Literal["low", "medium", "high"] = Field(
description="紧急程度"
)
summary: str = Field(description="一句话概括用户的问题")
class ReviewResult(BaseModel):
"""回复审核节点的结构化输出。"""
approved: bool = Field(description="回复是否可以直接发送")
feedback: str = Field(description="审核意见；通过时说明通过原因")
class SupportState(TypedDict, total=False):
ticket: str
category: Literal["account", "order", "technical", "other"]
urgency: Literal["low", "medium", "high"]
summary: str
knowledge: str
draft: str
approved: bool
review_feedback: str
revision_count: int
final_answer: str
trace: Annotated[list[str], operator.add]trace 使用了一个 Reducer。普通字段收到新值时会被覆盖，而 operator.add 会把每个节点返回的列表追加到已有列表中，所以它很适合记录执行轨迹。
3. 准备 LangChain 组件
model = ChatOpenAI(
model=os.getenv("OPENAI_MODEL", "gpt-4.1-mini"),
temperature=0,
)
classification_prompt = ChatPromptTemplate.from_messages([
(
"system",
"""你负责分析客服工单。
类别只能是：
- account：登录、密码、账号安全、身份验证；
- order：付款、发货、退款、订单状态；
- technical：客户端、接口、报错、性能问题；
- other：不属于以上类别的普通咨询。
如果问题导致用户完全无法使用核心功能，或涉及资金和账号安全，
紧急度应为 high。请严格按给定结构返回。""",
),
("human", "工单内容：{ticket}"),
])
classification_chain = (
classification_prompt
| model.with_structured_output(TicketAnalysis)
)
draft_prompt = ChatPromptTemplate.from_messages([
(
"system",
"""你是一名产品技术客服。请基于工单和参考知识生成回复。
要求：
1. 先回应用户当前问题，再给出可执行步骤；
2. 不得编造参考知识中没有的政策、时间和承诺；
3. 信息不足时明确需要用户补充什么；
4. 回复控制在 250 字以内，语气自然，不使用空泛套话。""",
),
(
"human",
"""工单：{ticket}
问题摘要：{summary}
类别：{category}
紧急度：{urgency}
参考知识：{knowledge}""",
),
])
draft_chain = draft_prompt | model | StrOutputParser()
review_prompt = ChatPromptTemplate.from_messages([
(
"system",
"""你是客服回复审核员。请检查回复：
1. 是否真正回答了工单；
2. 是否与参考知识一致；
3. 是否包含清楚、可执行的下一步；
4. 是否出现未经依据的承诺；
5. 表达是否适合直接发给用户。
只有全部满足时才可以 approved=true。""",
),
(
"human",
"""原始工单：{ticket}
参考知识：{knowledge}
待审核回复：{draft}""",
),
])
review_chain = review_prompt | model.with_structured_output(ReviewResult)
revision_prompt = ChatPromptTemplate.from_messages([
(
"system",
"你负责根据审核意见修改客服回复。只输出修改后的完整回复，不解释修改过程。",
),
(
"human",
"""原始工单：{ticket}
参考知识：{knowledge}
当前回复：{draft}
审核意见：{feedback}""",
),
])
revision_chain = revision_prompt | model | StrOutputParser()这里有三类 LangChain 能力：Prompt 模板管理输入，模型负责生成内容，Pydantic 结构负责让分类和审核结果可以直接进入图的路由判断。
再准备一个简化的知识库工具：
KNOWLEDGE_BASE = {
"order": (
"订单付款成功后通常在 24 小时内进入发货流程；"
"用户可在订单详情页查看状态。若超过 24 小时仍未处理，"
"需要提供订单号，由人工客服核查。"
),
"technical": (
"客户端升级后异常时，先退出程序并清理本地缓存，然后重新登录。"
"若问题仍存在，请收集客户端版本、操作系统、错误截图和日志时间。"
"不要直接要求用户删除全部数据。"
),
}
@tool
def search_knowledge_base(category: str) -> str:
"""按客服工单类别查询内部知识库。"""
return KNOWLEDGE_BASE.get(category, "没有找到匹配的知识，请谨慎回答。")实际项目里，这里可以替换成向量数据库、全文检索或内部 API。为了把重点放在图的控制逻辑上，Demo 暂时使用内存字典。
4. 把业务步骤拆成节点
def classify_ticket(state: SupportState) -> dict:
analysis = classification_chain.invoke({"ticket": state["ticket"]})
return {
"category": analysis.category,
"urgency": analysis.urgency,
"summary": analysis.summary,
"trace": [
f"分类完成：category={analysis.category}, urgency={analysis.urgency}"
],
}
def retrieve_knowledge(state: SupportState) -> dict:
knowledge = search_knowledge_base.invoke({
"category": state["category"],
})
return {
"knowledge": knowledge,
"trace": [f"已查询 {state['category']} 类知识库"],
}
def draft_reply(state: SupportState) -> dict:
draft = draft_chain.invoke({
"ticket": state["ticket"],
"summary": state["summary"],
"category": state["category"],
"urgency": state["urgency"],
"knowledge": state.get("knowledge", "没有额外参考知识"),
})
return {
"draft": draft,
"trace": ["已生成第一版回复"],
}
def review_reply(state: SupportState) -> dict:
review = review_chain.invoke({
"ticket": state["ticket"],
"knowledge": state.get("knowledge", "没有额外参考知识"),
"draft": state["draft"],
})
return {
"approved": review.approved,
"review_feedback": review.feedback,
"trace": [
f"审核完成：approved={review.approved}，意见={review.feedback}"
],
}
def revise_reply(state: SupportState) -> dict:
revised = revision_chain.invoke({
"ticket": state["ticket"],
"knowledge": state.get("knowledge", "没有额外参考知识"),
"draft": state["draft"],
"feedback": state["review_feedback"],
})
count = state.get("revision_count", 0) + 1
return {
"draft": revised,
"revision_count": count,
"trace": [f"已根据审核意见完成第 {count} 次重写"],
}
def finalize_reply(state: SupportState) -> dict:
return {
"final_answer": state["draft"],
"trace": ["回复审核通过，流程结束"],
}
def escalate_to_human(state: SupportState) -> dict:
if state.get("category") == "account":
reason = "账号与安全类问题不允许自动处理"
else:
reason = (
"回复连续审核失败："
f"{state.get('review_feedback', '没有审核意见')}"
)
return {
"final_answer": f"该工单已转人工处理。原因：{reason}",
"trace": [f"转人工：{reason}"],
}每个节点只返回自己修改的字段，而不是把整个 State 原样返回。这样可以减少误覆盖，也更容易从执行日志中看出每一步究竟改变了什么。
知识检索节点是确定性调用，没有让模型决定是否搜索。因为图在 order 和 technical 分支上已经明确知道需要知识，继续让模型做一次相同判断只会增加不确定性和调用成本。
5. 编写路由并组装图
def route_after_classification(
state: SupportState,
) -> Literal["retrieve_knowledge", "draft_reply", "escalate_to_human"]:
if state["category"] == "account":
return "escalate_to_human"
if state["category"] in {"order", "technical"}:
return "retrieve_knowledge"
return "draft_reply"
def route_after_review(
state: SupportState,
) -> Literal["finalize_reply", "revise_reply", "escalate_to_human"]:
if state["approved"]:
return "finalize_reply"
if state.get("revision_count", 0) >= 2:
return "escalate_to_human"
return "revise_reply"
workflow = StateGraph(SupportState)
workflow.add_node("classify_ticket", classify_ticket)
workflow.add_node("retrieve_knowledge", retrieve_knowledge)
workflow.add_node("draft_reply", draft_reply)
workflow.add_node("review_reply", review_reply)
workflow.add_node("revise_reply", revise_reply)
workflow.add_node("finalize_reply", finalize_reply)
workflow.add_node("escalate_to_human", escalate_to_human)
workflow.add_edge(START, "classify_ticket")
workflow.add_conditional_edges(
"classify_ticket",
route_after_classification,
)
workflow.add_edge("retrieve_knowledge", "draft_reply")
workflow.add_edge("draft_reply", "review_reply")
workflow.add_conditional_edges(
"review_reply",
route_after_review,
)
workflow.add_edge("revise_reply", "review_reply")
workflow.add_edge("finalize_reply", END)
workflow.add_edge("escalate_to_human", END)
checkpointer = InMemorySaver()
graph = workflow.compile(checkpointer=checkpointer)route_after_review 同时定义了成功出口、重写循环和失败出口。即使模型始终无法生成合格回复，流程也会在两次重写后转人工，不会无限执行。
6. 运行并观察每一步状态
if __name__ == "__main__":
config = {
"configurable": {
"thread_id": "ticket-20260715-001",
},
"recursion_limit": 20,
}
initial_state: SupportState = {
"ticket": (
"升级桌面客户端后启动一直白屏，我已经重装过一次，"
"今天没法正常处理工作，请问应该怎么办？"
),
"revision_count": 0,
"trace": [],
}
for update in graph.stream(
initial_state,
config=config,
stream_mode="updates",
):
node_name, node_update = next(iter(update.items()))
print(f"\n[{node_name}]")
print(node_update)
snapshot = graph.get_state(config)
print("\n=== 执行轨迹 ===")
for item in snapshot.values["trace"]:
print("-", item)
print("\n=== 最终结果 ===")
print(snapshot.values["final_answer"])运行：
python support_workflow.py一次可能的执行路线是：
模型输出具有不确定性，因此实际运行时可能第一版就通过，也可能进入一到两次重写。流程路线可以变化，但每一种路线都受到图中规则的限制。
回头看 Demo 中的关键设计
不是所有判断都交给模型
工单属于哪一类，需要理解自然语言，适合让模型判断；账号安全问题必须转人工，是确定的业务政策，直接写成条件路由更可靠。
我现在更愿意把 Agent 系统理解成“概率能力和确定性软件的组合”，而不是让模型接管所有 if/else。
结构化输出不只是格式好看
TicketAnalysis 和 ReviewResult 直接决定下一条 Edge。如果这里返回自由文本，整个路由就建立在字符串猜测上。结构化输出实际上是在模型与工作流之间建立了一份数据契约。
循环一定要有硬边界
审核不通过就重写，是很自然的需求。但如果只写一条返回边，没有重试次数限制，模型可能一直在同一个问题上消耗 token。
Demo 同时使用 revision_count 和 recursion_limit：前者是业务边界，后者是图运行时的最后保护。
Checkpointer 保存的是过程，不只是结果
拿到最终回复只是最基本的需求。真正排查问题时，我们还想知道分类结果是什么、检索到了什么、为什么被审核拒绝、到底重写了几次。
Checkpointer 让每一步状态都有机会被查看和恢复。不过节点可能在恢复时重新执行，所以涉及付款、发消息、写数据库等副作用时，还要使用幂等键，不能假设节点永远只运行一次。
如果继续往生产方向走
这个 Demo 已经覆盖了 LangGraph 的主要控制方式，但距离生产系统还有一段路。下一步至少可以继续补上：
- 用向量数据库或搜索服务替换内存知识库；
- 使用 PostgreSQL Checkpointer 保存跨进程状态；
- 在高风险回复发送前加入
interrupt()，等待人工确认后恢复； - 为模型调用和外部工具增加超时、重试和降级策略；
- 对工单内容和状态快照做敏感数据脱敏；
- 使用固定测试集评估分类准确率、自动解决率和人工转交率；
- 接入可观测系统，记录每个节点的耗时、token 和失败原因。
还可以进一步把“人工转交”从一句结果文字改成真正的暂停节点：图执行到这里后保存状态，客服人员补充意见，再通过同一个 thread_id 恢复执行。这时 LangGraph 的持久化和 Human-in-the-loop 能力会体现得更明显。
写在最后
这次从 LangChain 学到 LangGraph，最大的收获不是又记住了几个 API，而是对大模型应用的边界有了更清楚的认识。
LangChain 让我可以用统一方式组织模型、Prompt、工具和结构化输出；LangGraph 则把一次复杂任务展开成可以观察和控制的状态图。两者结合后，模型不再只是接收一个 Prompt 然后返回答案，而是成为业务流程中的一个或多个判断节点。
对简单任务，一条 Chain 或一个预构建 Agent 已经够用；当任务开始出现明确状态、条件分支、审核循环、中断恢复时，再引入 LangGraph 会更自然。框架并不是越多越好，关键还是先看问题本身需要什么样的控制能力。
后续我准备继续在这个 Demo 上加入真正的人工审批和持久化存储，看看一个任务在程序重启后如何恢复，以及图结构发生调整后，已有任务状态应该怎样兼容。