报销审批不是一个简单的线性流程。金额超过一定阈值需要总监加签，某些费用类型要财务预审，发票不完整时还要退回申请人补充材料。传统做法通常用if-else或数据库状态字段硬编码这些规则，结果就是分支越来越多，代码越来越难维护。LangGraph提供了一种基于状态图的实现思路：把审批环节抽象为节点，把环节之间的流动规则抽象为边，把当前流转位置和历史记录放进同一个状态对象里。下面用LangGraph构建一个可运行的报销审批流案例。
一、定义报销状态与审批节点
报销审批的核心状态需要覆盖申请单内容、当前审批层级、审批历史以及最终状态。LangGraph的状态通常用一个TypedDict定义，这样既保留了Python的类型提示，又方便节点函数返回部分更新。状态中的列表字段需要特别处理：如果多个节点都要往历史记录里追加内容，可以使用Annotated类型加上operator.add，LangGraph会自动执行累加而不是覆盖。下面的ExpenseState包含申请单对象、当前阶段、阶段结果、历史记录以及整体状态。
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import operator
from dataclasses import dataclass
@dataclass
class ExpenseRequest:
applicant: str
amount: float
category: str
invoice_attached: bool
class ExpenseState(TypedDict):
request: ExpenseRequest
current_stage: str
stage_result: str
history: Annotated[list, operator.add]
status: str
审批节点函数只负责处理自己这一层的业务规则，并通过返回值更新状态。主管节点检查金额是否超过一万，超过则把当前阶段改为总监审批，否则直接进入财务审批。总监节点处理更高金额的加签逻辑，财务节点则检查发票附件是否齐全。每个节点返回的字典只需要包含发生变化的字段，LangGraph会自动合并到全局状态中。这种局部更新机制避免了在节点之间传递大量无关数据，也让单元测试变得简单。
def supervisor_review(state: ExpenseState):
req = state["request"]
if req.amount > 10000:
return {"history": ["supervisor_pass"], "current_stage": "director_review"}
return {"history": ["supervisor_pass"], "current_stage": "finance_review"}
def director_review(state: ExpenseState):
req = state["request"]
if req.amount > 50000:
return {"history": ["director_pass"], "current_stage": "finance_review", "stage_result": "approve"}
return {"history": ["director_pass"], "current_stage": "finance_review", "stage_result": "approve"}
def finance_review(state: ExpenseState):
req = state["request"]
if not req.invoice_attached:
return {"history": ["finance_reject"], "status": "returned", "current_stage": "END"}
return {"history": ["finance_pass"], "status": "approved", "current_stage": "END"}
把节点职责拆分开之后，业务规则的调整只影响对应节点函数。例如新增预算校验环节，只需要写一个budget_check函数并把条件边接到合适位置，不需要改动其他审批节点。这种低耦合设计对复杂报销流程尤其重要，因为实际业务中审批规则经常随公司政策变化。
二、构建状态图与条件路由
状态图StateGraph是LangGraph的核心，它把节点注册到图里，并通过入口节点和条件边决定执行顺序。创建图实例后，先添加主管、总监、财务三个节点，再设置入口节点为主管。条件边的作用是根据当前状态动态选择下一个节点，例如主管审批完成后，根据current_stage字段决定去总监还是财务。条件边的映射字典键表示节点返回的当前阶段值，值表示要跳转的目标节点名称。
graph = StateGraph(ExpenseState)
graph.add_node("supervisor", supervisor_review)
graph.add_node("director", director_review)
graph.add_node("finance", finance_review)
graph.set_entry_point("supervisor")
graph.add_conditional_edges(
"supervisor",
lambda state: state["current_stage"],
{
"director_review": "director",
"finance_review": "finance",
"END": END
}
)
graph.add_conditional_edges(
"director",
lambda state: state["current_stage"],
{
"finance_review": "finance",
"END": END
}
)
graph.add_conditional_edges(
"finance",
lambda state: state["current_stage"],
{
"END": END
}
)
memory = MemorySaver()
app = graph.compile(checkpointer=memory)
条件边的路由函数非常简单，只需要返回状态中的current_stage字段。这样做的好处是节点内部只负责业务判断并更新状态，图结构负责根据状态决定走向，职责分离清晰。编译图时传入MemorySaver检查点，LangGraph会在每个节点执行后自动保存状态快照，为后续持久化和人工介入打下基础。
request = ExpenseRequest("张三", 12000, "差旅", True)
result = app.invoke({
"request": request,
"current_stage": "supervisor",
"history": [],
"status": "pending"
})
print(result["status"])
运行invoke方法会从入口节点开始依次执行，直到遇到END。上面的例子中金额为12000，超过一万，因此会先经过主管节点，然后进入总监节点，最后到达财务节点并批准。整个流程的路径由状态自动控制，不需要额外编写控制逻辑。
三、持久化状态与人工回退
MemorySaver检查点保证了每次节点执行后的状态都被保存下来。在实际报销系统中，人工审批往往不是瞬间完成的，可能需要等待主管登录系统点击同意或驳回。LangGraph支持断点续批：先把流程运行到需要人工处理的节点，暂停并存储状态。之后前端调用get_state获取当前状态，展示给审批人，审批人操作后再调用update_state更新状态并继续执行。线程ID用来隔离不同的报销单，每个线程ID对应一个独立的审批会话。
def finance_review_with_return(state: ExpenseState):
req = state["request"]
if not req.invoice_attached:
return {
"history": ["finance_return"],
"status": "returned",
"current_stage": "supervisor",
"return_to": "supervisor"
}
return {"history": ["finance_pass"], "status": "approved", "current_stage": "END"}
上面的节点在检测到发票缺失时，不是直接结束流程，而是把当前阶段改回supervisor，并设置一个return_to字段。图结构里可以为supervisor添加条件边，如果return_to为supervisor则重新进入主管节点，形成回退闭环。这种回退机制适用于需要补充材料或重新审核的场景，避免流程中断后无法继续。
config = {"configurable": {"thread_id": "expense-001"}}
current = app.get_state(config)
print(current.values)
app.update_state(config, {
"request": request,
"current_stage": "finance_review",
"history": [],
"status": "pending"
})
app.invoke(None, config)
get_state可以获取当前线程的状态快照，update_state允许外部系统修改状态并从中断点继续执行。在报销审批系统中，前端在审批人提交意见后调用update_state，把审批结果写回状态，然后让图继续往下走。这种设计天然支持人工审批、超时自动提醒和流程扭转，比单纯用数据库状态字段加定时任务灵活得多。
四、异常处理与可视化调试
复杂审批流最大的问题之一就是执行路径不透明。LangGraph提供了draw_mermaid方法，可以直接输出Mermaid格式的图描述代码。把这份代码粘贴到支持Mermaid的编辑器里，就能看到完整的节点和边结构。调试复杂条件分支时，先可视化再定位问题，效率比打印日志高很多。如果流程出现递归异常或节点未处理当前阶段，LangGraph会抛出GraphRecursionError，捕获该异常即可触发人工介入或告警。
from langgraph.errors import GraphRecursionError
try:
final_state = app.invoke({
"request": request,
"current_stage": "supervisor",
"history": [],
"status": "pending"
})
except GraphRecursionError as e:
print("流程执行异常，需要人工介入")
mermaid_code = app.get_graph().draw_mermaid()
print(mermaid_code)
异常处理并不是让流程崩溃，而是把它当作一种业务信号。例如某个节点返回了未在条件边映射中定义的current_stage值，说明业务规则有遗漏，需要在条件边里补充对应映射。借助LangGraph的类型系统和状态检查点，可以快速定位是哪个节点返回了异常值。相比传统工作流引擎的黑盒行为，LangGraph的调试信息更加直观。
LangGraph在复杂报销审批流中的价值不在于消灭所有if-else，而在于把这些条件判断收敛到状态节点和条件边中，让流程结构可以被阅读、测试和修改。当审批规则继续变化，比如增加预算冻结判断、批量审批或并行会签时，只需要在图里增加节点和边，不需要重构整个流程代码。这种可演进的架构对长期维护这类业务系统非常有帮助。