LangChain 的链式调用（Chain）跑线性流程很顺，可一旦流程需要”记状态、做分支、循环重试”，链就撑不住了。LangGraph 把 LLM 应用建模成图：节点是步骤，边是流向，图可以分支可以成环，状态在节点之间显式传递。本文从零实现一个邮件处理智能体：解析邮件 → 分类 → 自动回简单邮件、复杂邮件调 API 升级，完整覆盖状态图、条件边、循环、Pydantic 结构化输出与 with_structured_output，以及给 Agent 写测试的方法。
一、背景：从 Chain 到 Graph，差的是什么
先看 Chain 的天花板。一条 prompt | model | parser 链：输入进、输出出，中间没有”记忆”，也没有”如果…就…”。想做”先分类，垃圾邮件直接删、简单问题自动回、复杂问题转人工”？链只能写一坨 if-else 把多个链粘起来，状态靠闭包和全局变量传，循环重试靠 while 手写， dritten Schritt 之后代码就没法看了。
LangGraph（LangChain 生态的开源 Python 框架，1.x 长期支持版 API 已稳定）解决的正是编排层问题：状态（State） 让信息在步骤间显式流转；条件边（conditional edges） 让图按数据走不同分支；环（cycles） 让”校验→重做”这类循环成为一等公民；节点里照样可以用 LangChain 的模型与工具。图跑起来之后，顶层再包一层”自主决策用哪个工具”的逻辑，就是 Agent。本教程的邮件智能体就是这条路：先手写状态图，再让 Agent 自己走图。
你需要：Python 中级水平（懂类和方法）、LangChain 基础概念、一个 LLM Key（示例用 OpenAI，可换任何服务商）。
二、原理：状态、节点、边、条件边、Agent
State（状态）：一个 TypedDict，定义图 Steps 之间传递哪些字段。每个节点读状态、写回增量，框架负责合并。状态是显式的——调试时随时打印”跑到这一步状态是什么”，这是相对闭包传参最大的可维护性胜利。
Node（节点）：一个 Python 函数 (state) -> dict，返回要更新的状态片段。可以是调 LLM 分类、调 API 发邮件、写数据库，任何逻辑都能装进去。
Edge（边）：节点之间的固定连线，add_edge("parse", "classify") 表示解析完一定去分类。
Conditional edge（条件边）：按函数返回值动态路由，add_conditional_edges("classify", route, {"simple": "reply", "complex": "escalate", "spam": "delete"})。没有它就没有分支；配合指回前驱节点的边，就形成环（如”草稿质检不过 → 回炉重写”，设最大轮数防死循环）。
Agent：把”下一步走哪”交给 LLM 决定的图。节点里放工具（发邮件、查 API），模型每次看状态决定调哪个工具、还是结束。图是骨架，模型是大脑。
Pydantic + with_structured_output：邮件解析这种”从非结构化文本抽字段”的活，不能靠正则碰运气。定义 Pydantic 模型锁死字段与类型，再用 LangChain 的 with_structured_output 让模型直接返回该模型的实例——解析失败在类型层就拦住，而不是下游空指针。
三、环境准备
python -m venv .venv && source .venv/bin/activate
python -m pip install langgraph langchain-openai "pydantic[email]"
python -c "import langgraph; print('ok')"
export OPENAI_API_KEY="你的key"
教程基于 LangGraph 1.2，构图 API 自 1.0 长期支持版保持稳定。OpenAI 可换成任何 LangChain 支持的模型，只需改模型初始化一行。
四、分步实战：邮件处理智能体
步骤 1：测试集先行—— example_emails.py
AI 应用必须有固定测试集，否则你永远在”感觉变好了”。先写死几封典型邮件：
# example_emails.py
SIMPLE_EMAIL = """\
From: alice@example.com
Subject: 请问发票什么时候开？
你好，我上周下的订单 #8848，发票什么时候能开好？
"""
COMPLEX_EMAIL = """\
From: bob@example.com
Subject: 合同条款需要法务确认
你好，附件合同第 5.2 条的赔偿上限与我们上次谈的不一致，
请法务同学确认后再回复我，涉及金额约 200 万。
"""
SPAM_EMAIL = """\
From: winner@lottery-xxx.biz
Subject: 恭喜您中奖！！！点击领取
您已被选中为幸运用户，点击链接领取百万大奖……
"""
后面每实现一步都拿这三封跑，对错一目了然。
步骤 2：Pydantic 锁死邮件结构 + with_structured_output 解析
# schemas.py
from pydantic import BaseModel, EmailStr, Field
from typing import Literal
class ParsedEmail(BaseModel):
sender: EmailStr
subject: str
body: str
category: Literal["simple", "complex", "spam"] = "simple"
summary: str = Field(default="", description="一句话摘要")
order_id: str | None = None
解析节点用 with_structured_output 直接产出实例：
# nodes_parse.py
from langchain_openai import ChatOpenAI
from schemas import ParsedEmail
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = llm.with_structured_output(ParsedEmail)
PARSE_PROMPT = """\
从下面邮件原文抽取字段：
- sender/subject/body 照录
- summary：一句话中文摘要
- order_id：如有订单号则抽取，否则 null
- category 先填 simple，分类由下游节点复核
邮件原文：
{raw}
"""
def parse_node(state):
raw = state["raw_email"]
parsed: ParsedEmail = parser.invoke(PARSE_PROMPT.format(raw=raw))
return {"parsed": parsed}
with_structured_output 的价值：模型输出不符合 schema 时 LangChain 层直接抛错重试，而不是把脏字符串传给下游。这是从”字符串编程”升级到”类型编程”的关键一步。
步骤 3：第一张状态图——解析 → 分类
# graph_v1.py
from typing import TypedDict
from langgraph.graph import StateGraph, END
from schemas import ParsedEmail
class EmailState(TypedDict):
raw_email: str
parsed: ParsedEmail | None
label: str # simple | complex | spam
def classify_node(state: EmailState):
p = state["parsed"]
text = (p.subject + "\n" + p.body).lower()
if any(k in text for k in ["中奖", "点击领取", "lottery", "百万大奖"]):
return {"label": "spam"}
if any(k in text for k in ["合同", "法务", "赔偿", "200万", "确认"]):
return {"label": "complex"}
return {"label": "simple"}
builder = StateGraph(EmailState)
builder.add_node("parse", parse_node)
builder.add_node("classify", classify_node)
builder.set_entry_point("parse")
builder.add_edge("parse", "classify")
builder.add_edge("classify", END)
graph = builder.compile()
if __name__ == "__main__":
from example_emails import SIMPLE_EMAIL, SPAM_EMAIL, COMPLEX_EMAIL
for name, mail in [("simple", SIMPLE_EMAIL), ("spam", SPAM_EMAIL),
("complex", COMPLEX_EMAIL)]:
out = graph.invoke({"raw_email": mail})
print(name, "->", out["label"])
跑通标准：三封邮件标签全对。StateGraph(EmailState) 声明状态，add_node 注册步骤，add_edge 连线，compile() 生成可 invoke 的图。这张图是线性的——下一步让它分叉。
步骤 4：条件边——三类邮件三条路
# graph_v2.py（在 v1 基础上加）
def route(state: EmailState) -> str:
return state["label"] # 返回值即分支 key
def reply_node(state: EmailState):
p = state["parsed"]
draft = f"你好，订单 {p.order_id or '（查询中）'} 的发票已在处理，3 个工作日内开好。"
return {"draft_reply": draft, "action": "replied"}
def escalate_node(state: EmailState):
return {"action": "escalated_to_legal",
"note": f"转法务：{state['parsed'].summary}"}
def delete_node(state: EmailState):
return {"action": "deleted_as_spam"}
builder = StateGraph(EmailState)
builder.add_node("parse", parse_node)
builder.add_node("classify", classify_node)
builder.add_node("reply", reply_node)
builder.add_node("escalate", escalate_node)
builder.add_node("delete", delete_node)
builder.set_entry_point("parse")
builder.add_edge("parse", "classify")
builder.add_conditional_edges(
"classify", route,
{"simple": "reply", "complex": "escalate", "spam": "delete"},
)
builder.add_edge("reply", END)
builder.add_edge("escalate", END)
builder.add_edge("delete", END)
graph = builder.compile()
记得把 draft_reply/action/note 字段补进 EmailState。add_conditional_edges 三个参数：从哪个节点出发、路由函数、分支映射表。拿三封测试邮件重跑：simple 应产出 draft_reply，complex 应被转法务，spam 应被删除。
步骤 5：环——草稿质检不过回炉重写
自动回信最怕”语气差、答非所问”。加一个质检节点 + 回边：
def review_node(state):
draft = state.get("draft_reply", "")
p = state["parsed"]
ok = ("发票" in draft or "订单" in draft) and len(draft) < 300
tries = state.get("tries", 0) + 1
return {"approved": ok or tries >= 3, "tries": tries}
def route_review(state) -> str:
return "done" if state["approved"] else "rewrite"
def rewrite_node(state):
d = state["draft_reply"]
return {"draft_reply": "你好呀！" + d + "（如有疑问请回复本邮件）"}
# 接线：reply -> review；review --done--> END；review --rewrite--> rewrite --\
# rewrite --> review（成环，tries 上限 3 防死循环）
环是 LangGraph 相对 Chain 的本质差别：线性管道表达不了”不合格就重做”。任何环必须有退出条件（这里是质检通过或 3 次上限），否则一次坏输入就能让图转到天荒地老。
步骤 6：升级成 Agent——让模型自己决定调什么工具
前面路由是手写规则。现在把工具（发邮件、建法务工单）暴露给模型，让它看状态自主决策：
from langchain.tools import tool
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件。to 为收件人，subject/body 为主题正文。"""
print(f"[SEND] to={to} subject={subject}\n{body}")
return "sent"
@tool
def create_legal_ticket(summary: str, amount: str = "") -> str:
"""为复杂邮件建法务工单。"""
print(f"[TICKET] {summary} 金额:{amount}")
return "ticket-001"
agent_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools(
[send_email, create_legal_ticket]
)
def agent_node(state):
p = state["parsed"]
msg = agent_llm.invoke(
f"邮件：{p.subject}\n{p.body}\n摘要：{p.summary}\n"
"简单问题直接用 send_email 回复；涉及合同法务用 create_legal_ticket；"
"垃圾推广不处理直接结束。"
)
return {"agent_msg": msg}
def route_tool(state) -> str:
calls = getattr(state["agent_msg"], "tool_calls", None) or []
if not calls:
return END
return calls[0]["name"] # send_email | create_legal_ticket
把 agent_node 和两个工具节点连进图，条件边按工具名分发，工具执行完指回 agent_node（又一个环：模型看工具结果再决策，直到不再调工具为止）。测试：simple 邮件应触发 send_email，complex 触发 create_legal_ticket，spam 零工具调用直接结束。打印每次的 tool_calls，确认决策链可审计。
步骤 7：给 Agent 写测试——断言行为而不是断言字符串
# test_agent.py
from example_emails import SIMPLE_EMAIL, COMPLEX_EMAIL, SPAM_EMAIL
def run(mail):
return graph.invoke({"raw_email": mail})
def test_simple_replied():
out = run(SIMPLE_EMAIL)
assert out["action"] in ("replied",) or "draft_reply" in out
def test_complex_escalated():
out = run(COMPLEX_EMAIL)
assert out.get("action") == "escalated_to_legal" or "ticket" in str(out)
def test_spam_deleted():
out = run(SPAM_EMAIL)
assert out.get("action") == "deleted_as_spam"
def test_loop_bounded():
out = run(SIMPLE_EMAIL)
assert out.get("tries", 1) <= 3
要点：断言状态字段和工具调用，不逐字断言模型生成的自然语言（那会把测试写成玄学）。循环必须有”最大轮数”断言。LLM 抖动大时，把 temperature 置 0、关键节点用规则兜底，保证测试可重复。
五、常见坑
坑 1：状态字段忘声明。 节点返回了 draft_reply 但 EmailState 没定义，更新静默丢失。从严做法：节点返回前对照 state 定义自查。
坑 2：条件边漏分支。 路由函数返回了映射表里没有的 key，图直接抛错。路由返回值用 Literal 类型锁死，或加默认分支兜底。
坑 3：环没有退出条件。 “重写直到满意”没有轮数上限，一次病态输入烧掉一整晚 token。每个环配计数器 + 硬上限，这是铁律。
坑 4：全图一个 temperature。 解析/分类要 0（稳定），起草回复可以稍高（自然）。不同节点用不同 temperature 的模型实例，别一参到底。
坑 5：字符串解析邮件。 发件人、订单号用正则硬抽，换个格式全崩。Pydantic + with_structured_output 是正道，类型层拦脏数据。
坑 6：断言模型原话。 测试断言”回复必须包含某句话”，模型换个措辞就红。断言状态、标签、工具调用这些结构化事实。
坑 7：LangChain 和 LangGraph 职责混淆。 模型、提示词、工具是 LangChain 的活；状态、路由、循环是 LangGraph 的活。混着写也能跑，但出了问题分不清该查哪边。
附：完整工程文件清单与运行顺序
本教程所有代码拼成可跑工程：
mail-agent/
├── schemas.py # ParsedEmail（Pydantic）
├── nodes_parse.py # parse_node（with_structured_output）
├── example_emails.py # 三封测试邮件
├── graph_v1.py # 解析→分类（线性图，验证标签全对）
├── graph_v2.py # +条件边三分支（验证分流正确）
├── graph_v3.py # +质检环（验证回炉+轮数上限）
├── graph_agent.py # +工具与模型决策（验证 tool_calls）
└── test_agent.py # 四个行为断言
运行顺序就是 v1→v2→v3→agent，每一步拿三封邮件验收后再往下走。调试技巧：任何一步输出不对，先打印该节点的输入状态——状态显式是 LangGraph 相对闭包传参最大的红利，90% 的 bug 都是”上游没写进来”而不是”本节点算错了”。图的执行轨迹可以用 graph.get_graph().draw_mermaid() 画成 Mermaid 图，分支和环一目了然，贴进文档给同事讲架构非常好用。
常见问答
问：StateGraph 和普通函数调用链最大的体感差别是什么？ 可观测性。链跑崩了你只能逐段 print；图跑崩了你直接看”哪个节点写的状态不对”，每个节点的输入输出都是显式快照。状态即日志，这是用图的第一红利。
问：条件边和代码里写 if-else 有什么区别？ if-else 把路由逻辑埋进节点函数，图结构看不出来；条件边让分支成为图的一等结构，Mermaid 一画全团队看懂。分支超过 3 条或带环时，条件边的可维护性碾压 if-else。
问：Agent 不听话乱调工具怎么办？ 三招：系统提示把”何时调哪个工具”写死；工具 description 写清前置条件；关键动作（如真发邮件）先走”生成草稿→人工确认”的半自动模式。自主性是逐步放开的，不是一步到位的。
问：流式输出怎么做？ 图支持按节点粒度 stream，graph.stream() 逐节点吐增量，前端逐块渲染。token 级流式在 LLM 节点内部处理。邮件场景按节点 stream 完全够用：解析完先显示分类，起草完再显示草稿。
问：多智能体协作怎么建模？ 一个子图就是一个智能体，主图用条件边在子图间调度，状态里加消息走廊字段。先把单智能体玩熟，多智能体只是”图里套图”，概念不增加，复杂度线性涨。
问：LangGraph 一定要配 LangChain 用吗？ 不一定。节点里可以调任何 LLM SDK、任何 API，LangGraph 只管状态与路由。但用了 LangChain 的模型、工具、解析器，节点代码会短很多，两者同源，配合属于顺水推舟。
问：生产部署注意什么？ 图 compile 一次多处复用；长循环图配 checkpointer 做断点续跑；工具节点加超时和重试；线上采样真实邮件定期回放测试集，标签漂移（如新型垃圾邮件话术）及时补进 example_emails。图即代码，享受 CI 待遇。
问：学完这只邮件智能体，下一步学什么？ 三个方向：持久化（checkpointer + 中断续跑，做可暂停的人机协同流程）、子图复用（把分类器、起草器封成子图，主图只做调度）、可观测（LangSmith 追踪每次调用的输入输出与 token 消耗）。概念都在本文的地基上，逐层加高即可。
问：节点之间传大对象（如整封邮件附件）怎么办？ 状态里只传引用（路径、ID），大对象放外部存储，节点按需读写。状态是”账本”不是”仓库”，塞大对象既慢又让调试快照没法看，这是初学者最常见的性能坑。
问：图越画越大怎么拆？ 按”一个子图只干一件事”拆：解析、分类、起草、质检各成子图，主图只剩调度线。子图独立测试、独立复用，主图保持十个节点以内可读。图的代码组织和普通代码一样，闻到味道就拆。
六、总结
LangGraph 的学习路径就是这只邮件智能体的生长史：TypedDict 状态让数据流转显式化 → 节点与边搭出线性图 → 条件边长出分支 → 回边长出循环（配退出条件）→ Pydantic + with_structured_output 把非结构化输入变成类型化事实 → 工具 + 模型决策长成 Agent → 状态断言守住测试。记住两句话：凡是”如果…就…”就用条件边，凡是”不行就重来”就用环。掌握这两样，任何复杂度的 LLM 工作流你都能画成图、测得住、跑得稳。 点击阅读原文
参考资料：https://realpython.com/langgraph-python/ 点击阅读原文
相关文章
