LangGraph 工作流:从排查路径看工程价值

发布时间:2026/8/5 12:21:55
LangGraph 工作流:从排查路径看工程价值
这篇不先堆名词。我们把《LangGraph并不难难的是知道什么时候不该用》拆成几级台阶看完至少知道下一步该学什么、该练什么。摘要最近线上出了个问题一个 Agent 把用户数据批量导出了日志里只有任务完成四个字没人知道它调了哪些接口、走了哪些分支、审批人是谁。团队吵了一架最后共识是Demo 能跑的项目根本不值钱。我复盘了一下之前踩过的坑发现 LangGraph 其实不难难的是你知道什么时候不该用它以及怎么把 Demo 扩成能上线的系统。目录为什么需要图工作流State 与 NodeEdge 与条件分支人工审批节点工程化落地总结为什么需要图工作流先说结论如果你的 Agent 只有一条直线流程别用 LangGraph。我见过太多人把 LangChain 的 LCEL 链或者简单的顺序调用硬套进图结构里结果代码反而更乱了。图工作流的真正价值在于解决分支、循环、状态共享、人工介入这四个问题。实际场景是什么比如一个报销审批 Agent用户提交申请系统校验金额是否超限如果超限转人工审批否则自动通过审批通过后调用财务接口失败重试最多三次这个流程里有判断、有循环、有状态用图来表达是最自然的。如果你硬要用 if-else 堆出来代码会迅速失控。我的判断标准很简单流程里有没有如果……就……否则……的分支有没有需要记忆的状态需不需要停下来等人操作需要重试吗 满足两个以上才值得考虑 LangGraph。State 与 NodeState 是图的核心。很多人把 State 理解成全局变量这是错的。State 应该是纯数据描述当前任务的完整上下文。Node 是纯函数接收 State返回更新后的 State。看一段代码from typing import TypedDict, Annotated import operator class ApprovalState(TypedDict): # 使用 operator.add 表示列表追加 messages: Annotated[list, operator.add] # 普通字段直接覆盖 approval_status: str # 金额字段 amount: float # 重试次数 retry_count: int这里有个细节messages用了Annotated[list, operator.add]意思是每次 Node 执行时新消息会追加到列表末尾而不是覆盖。这是 LangGraph 的状态合并机制理解这一点很重要。Node 的实现也很简单def validate_amount(state: ApprovalState) - dict: if state[amount] 10000: return {approval_status: pending_human} return {approval_status: auto_approved} def call_finance_api(state: ApprovalState) - dict: # 实际调用财务接口 success simulate_api_call(state) if not success: return {retry_count: state[retry_count] 1} return {approval_status: completed}注意 Node 不直接修改 State而是返回一个 dictLangGraph 负责合并。这种设计让调试变得可能——你可以在任意节点打印 State看到完整上下文。Edge 与条件分支条件路由是图工作流的精髓。LangGraph 支持两种 Edge普通 Edge 和 Conditional Edge。普通 Edge 就是固定的跳转从 Node A 到 Node B。Conditional Edge 根据 State 的值决定下一步去哪里。from langgraph.graph import StateGraph, END workflow StateGraph(ApprovalState) # 添加节点 workflow.add_node(validate, validate_amount) workflow.add_node(human_approval, human_approval_node) workflow.add_node(finance, call_finance_api) # 设置入口 workflow.set_entry_point(validate) # 条件路由 workflow.add_conditional_edges( validate, route_by_status, { pending_human: human_approval, auto_approved: finance } ) # 普通边 workflow.add_edge(human_approval, finance) workflow.add_edge(finance, END)def route_by_status(state: ApprovalState) - str: return state[approval_status]这里有个坑条件函数返回的值必须和路由字典的 key 完全匹配否则 LangGraph 会报错。我见过有人在这里栽跟头调试了半天才发现是字符串大小写问题。另一个常见错误是忘记处理所有分支。如果你的条件函数可能返回多个值路由字典必须覆盖所有情况否则图会卡住。人工审批节点这是 Demo 和上线最大的区别之一。Demo 里的人工审批通常是打印一条日志等开发者手动改代码。上线后的人工审批需要真正的暂停-恢复机制。LangGraph 提供了interrupt()函数来实现这个功能def human_approval_node(state: ApprovalState) - dict: # 暂停执行等待人工输入 result interrupt({ message: f金额 {state[amount]} 元需要人工审批, state_snapshot: state }) # 恢复执行result 是人工输入的值 approval result.get(approved, False) if approval: return {approval_status: approved} else: return {approval_status: rejected}调用interrupt()后图的执行会暂停State 会被序列化保存。你可以通过 API 查询当前状态等待人工操作后传入新的输入恢复执行。这段代码在生产环境里需要配合持久化存储比如 Redis 或数据库否则进程重启后状态就丢了。这是很多 Demo 忽略的问题。工程化落地回到开头那个问题为什么 Demo 能跑上线就崩我总结了三个关键点第一权限控制。 Agent 调用的每个工具都应该有明确的权限边界。财务接口只能由审批通过的流程调用不能因为模型感觉应该调用就调用。我在代码里加了白名单机制ALLOWED_TOOLS {query_balance, submit_reimbursement, send_notification} def tool_call_node(state: ApprovalState) - dict: tool_name state.get(next_tool) if tool_name not in ALLOWED_TOOLS: raise PermissionError(fTool {tool_name} not allowed) # 执行工具调用第二日志可观测。 每个节点的输入输出都要记录而且要有唯一的事务 ID。我通常用 trace_id 贯穿整个流程import uuid def make_traced_node(node_func): def wrapped(state: ApprovalState) - dict: trace_id str(uuid.uuid4()) print(f[TRACE {trace_id}] Entering {node_func.__name__}) print(f[TRACE {trace_id}] Input: {state}) result node_func(state) print(f[TRACE {trace_id}] Output: {result}) return result return wrapped第三错误处理。 重试策略要写在图里而不是靠模型重试一下。对于关键操作还要有降级方案。def call_finance_api(state: ApprovalState) - dict: if state[retry_count] 3: return {approval_status: failed, error: max_retries_exceeded} try: success simulate_api_call(state) if not success: return {retry_count: state[retry_count] 1} return {approval_status: completed} except Exception as e: return {retry_count: state[retry_count] 1, error: str(e)}总结LangGraph 不是银弹。如果你的场景很简单LCEL 链就够了。如果你的 Agent 需要分支判断、状态记忆、人工介入、重试机制图工作流才是正解。Demo 和上线之间隔的不是模型能力是权限、日志、错误处理这些 boring engineering。我见过太多人把时间花在选择模型上却忽略了工作流的可维护性。最后给个建议先画流程图再写代码。 把节点、边、条件、状态全部画出来你会发现很多问题在设计阶段就能解决而不是上线后崩了才来补救。Agent 开发的核心竞争力正在从能不能跑转向能不能可控。LangGraph 提供了工具但怎么用取决于你对业务的理解。资料展示下面是我整理的AI大模型学习资料和工具包预览适合收藏后按主题逐步学习。如果你想看完整资料目录可以在评论区留言「资料」也欢迎告诉我你更关注AI大模型里的哪类内容。