从 Demo 到生产：LangGraph 工作流为什么总翻车？

聊《会用LangGraph只是起点，能解释失败才算真正入门》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

上周做需求评审，团队内部吵了一架。

AI 应用组的同学把 Demo 演示得很漂亮：用户输入问题，Agent 自动调用工具、查询数据、生成报告，整个过程丝滑流畅。业务方很满意，准备上线。

但基础设施组直接泼了冷水："权限怎么控制的？日志能追踪到每一步吗？失败了谁来兜底？"

两拨人各执一词，最后老板拍板：先别急着上线，把工作流重构一下，用 LangGraph 重新设计。

我参与了这次重构，踩了几个坑，也摸到了一些门道。今天把这些经验分享出来，希望能帮到正在纠结 Demo 和生产差距的同学。

目录

• 为什么需要图工作流
• State 与 Node
• Edge 与条件分支
• 人工审批节点
• 工程化落地
• 总结

为什么需要图工作流

先说一个真实场景。

我们有一个智能客服 Agent，最初用简单的 if-else 写出来：

if 用户问价格： 调用价格查询工具 elif 用户问订单： 调用订单查询工具 else: 返回默认回复

Demo 跑通了，用户提问准确率 85%。上线一周后，问题出现了：

• 用户问"我上周买的那个东西多少钱"，Agent 不知道要先查订单再查价格
• 复杂问题需要多轮对话，但每次都是独立处理，没有上下文记忆
• 工具调用失败时，直接返回错误，用户体验很差

这些问题本质上是：你的 Agent 是脚本，不是系统。

脚本是线性的、不可控的；系统是结构化的、可观测的、可干预的。

LangGraph 的核心价值，就是把 Agent 从脚本变成图结构的工作流。图有什么好处？

1. 状态可追踪：每一步的状态都保存在 State 里，你可以随时查看
2. 流程可控制：通过 Edge 定义条件分支，实现复杂逻辑
3. 错误可恢复：失败时可以重试、降级、或转人工
4. 可观测性强：每个节点都是一个独立的执行单元，便于打点

State 与 Node

State 是 LangGraph 的关键概念。

很多初学者把 State 当成普通变量传递，这是错误的。State 是一个共享的状态容器，所有 Node 都可以读写它。

from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: list # 对话历史 user_intent: str # 用户意图 tool_result: dict # 工具调用结果 should_transfer: bool # 是否需要转人工 confidence_score: float # 置信度评分

Node 是执行单元。每个 Node 接收 State，处理后返回更新后的 State。

def intent_recognition(state: AgentState) -> AgentState: """意图识别节点""" messages = state["messages"] last_message = messages[-1]["content"] # 调用大模型识别意图 intent = classify_intent(last_message) return { "user_intent": intent, "confidence_score": intent["confidence"] } def tool_calling(state: AgentState) -> AgentState: """工具调用节点""" intent = state["user_intent"] messages = state["messages"] # 根据意图调用对应工具 result = call_tool(intent, messages) return { "tool_result": result, "messages": messages + [{"role": "assistant", "content": result["response"]}] }

踩坑经验：State 设计要遵循单一职责原则。每个字段都有明确的存在意义，不要为了"可能用到"而添加字段。State 越大，调试越痛苦。

Edge 与条件分支

Edge 是图的"神经系统"，决定流程往哪个方向走。

LangGraph 支持两种 Edge：

1. 普通 Edge：固定跳转，Node A 执行完一定去 Node B
2. 条件 Edge：根据 State 动态选择下一个 Node

def route_by_intent(state: AgentState) -> str: """根据意图路由到不同节点""" intent = state["user_intent"] confidence = state["confidence_score"] if confidence < 0.6: return "ask_for_clarification" elif intent == "price_query": return "price_tool" elif intent == "order_query": return "order_tool" else: return "fallback" # 注册条件 Edge graph.add_conditional_edges( "intent_recognition", route_by_intent, { "ask_for_clarification": "clarify_intent", "price_tool": "price_node", "order_tool": "order_node", "fallback": "fallback_node" } )

关键取舍：条件分支不是越多越好。分支过多会导致图结构复杂，调试困难。我的经验是：超过 5 个分支时，考虑拆分图或使用子图。

人工审批节点

这是生产环境最容易忽略的部分。

Demo 里所有事情都是自动完成的，但生产环境不同。涉及金钱、敏感操作、高置信度判断的场景，必须有人工介入。

def human_review(state: AgentState) -> AgentState: """人工审批节点""" # 发送审批请求 approval_request = { "task_id": generate_task_id(), "state": state, "expires_at": datetime.now() + timedelta(hours=2) } # 等待人工审批（这里用阻塞方式演示，实际应该用异步） approval = wait_for_approval(approval_request) if approval["approved"]: return {"should_transfer": False, "approval_record": approval} else: return {"should_transfer": True, "rejection_reason": approval["reason"]} # 添加人工审批节点 graph.add_node("human_review", human_review) graph.add_edge("high_risk_node", "human_review")

生产建议：人工审批节点必须设计超时机制和降级策略。如果人工长时间不审批，应该有默认处理逻辑，不能无限等待。

工程化落地

从 Demo 到生产，LangGraph 工作流需要解决以下问题：

1. 可观测性

每个节点执行时，记录关键信息：

import time import logging logger = logging.getLogger(__name__) def observability_wrapper(node_func): """节点可观测性装饰器""" def wrapper(state: AgentState) -> AgentState: start_time = time.time() node_name = node_func.__name__ logger.info(f"[{node_name}] 开始执行，输入状态: {state}") try: result = node_func(state) elapsed = time.time() - start_time logger.info(f"[{node_name}] 执行成功，耗时: {elapsed:.2f}s") return result except Exception as e: elapsed = time.time() - start_time logger.error(f"[{node_name}] 执行失败，耗时: {elapsed:.2f}s，错误: {e}") raise return wrapper

2. 错误处理

每个节点都应该有错误处理逻辑：

def robust_tool_calling(state: AgentState) -> AgentState: """带错误处理的工具调用""" try: result = call_tool(state["user_intent"], state["messages"]) return {"tool_result": result} except ToolError as e: logger.warning(f"工具调用失败，降级处理: {e}") return { "tool_result": {"error": str(e), "fallback": True}, "messages": state["messages"] + [ {"role": "assistant", "content": "抱歉，服务暂时不可用，请稍后再试"} ] } except Exception as e: logger.error(f"未知错误: {e}") return {"tool_result": {"error": "系统错误", "fallback": True}}

3. 验收标准

上线前必须验证以下几点：

• 状态完整性：每个节点执行后，State 是否完整、一致
• 边界条件：空输入、异常输入、超时输入是否正确处理
• 可恢复性：失败后能否从断点恢复，而不是从头开始
• 可观测性：每个节点是否都有日志、指标、追踪

总结

LangGraph 不是银弹，它解决的是可控性问题。

Demo 阶段的 Agent，追求的是功能跑通；生产阶段的 Agent，追求的是稳定可控。两者的差距，不在 Prompt 质量，而在工程化程度。

我的建议是：

1.先设计图结构，再写代码。用纸笔画出节点和 Edge，想清楚每个分支的逻辑
2.State 设计要克制。字段越多，调试越难，维护成本越高
3.人工审批节点不能省。涉及金钱、敏感操作，必须有人工介入
4.可观测性要前置。不要等上线后才发现日志缺失、追踪断裂

从 Demo 到生产，最大的挑战不是技术，而是思维转变：从"让功能跑通"到"让系统可控"。

LangGraph 提供了工具，但如何使用，取决于你对边界和取舍的理解。

---

实战建议：如果你正在构建 Agent 工作流，建议从简单的图结构开始，逐步增加复杂度。每增加一个节点或分支，都要问自己：这个复杂性是否必要？能否用更简单的方式实现？

记住：会写 LangGraph 只是入门，能解释失败才算真正入门。

资料展示

下面是我整理的AI大模型学习资料和工具包预览，适合收藏后按主题逐步学习。

如果你想看完整资料目录，可以在评论区留言「资料」；也欢迎告诉我你更关注AI大模型里的哪类内容。