聊《LangGraph实战：真正难的不是调用，而是稳定交付》之前，先说一句实在的：别急着背概念，先看它在真实项目里到底解决什么问题。

摘要

摘要：最近帮一个团队做 Agent 工单处理系统的 Code Review，发现一个问题：Demo 跑得好好的，接入权限和日志之后直接崩盘。回头一看，他们从头到尾没搞清楚"工作流"和"脚本"的区别。这篇文章复盘那次踩坑，顺便把 LangGraph 里 State、Edge、人工审批这些关键概念讲清楚——不聊概念，只聊实际工程中怎么用。

目录

• 为什么需要图工作流
• State 与 Node：别把状态散落在各处
• Edge 与条件分支：动态路由才是 Agent 的灵魂
• 人工审批节点：Demo 里不需要，生产里必须有
• 工程化落地：权限、日志、可观测性
• 总结

---

目录

• 为什么需要图工作流
• State 与 Node：别把状态散落在各处
• Edge 与条件分支：动态路由才是 Agent 的灵魂
• 人工审批节点：Demo 里不需要，生产里必须有
• 工程化落地：权限、日志、可观测性
• 总结

为什么需要图工作流

去年我接了一个内部项目：工单智能处理 Agent。需求很简单——用户提交工单，Agent 自动分析、分类、路由给对应团队。

一开始我写了个很"干净"的脚本：

def process_ticket(ticket): analysis = llm.analyze(ticket) # 分析工单 category = llm.classify(analysis) # 分类 if category == "技术": result = tech_team_handler(ticket, analysis) elif category == "售后": result = after_sales_handler(ticket, analysis) else: result = escalate_to_human(ticket) return result

Demo 跑通了，效果还行。然后接入真实环境，问题来了：

1. 权限控制：不同用户提交工单，能访问的数据不同，但脚本里没有权限上下文
2. 日志缺失：出了问题不知道是哪一步出错，LLM 的输出是什么
3. 可重试性差：某个节点失败了，整个流程得从头再来
4. 人工介入难：售后场景经常需要人工确认，但脚本写死了自动流程

这些问题单个看都不难解决，但组合起来，脚本架构就撑不住了。

这时候我才真正理解 LangGraph 的设计哲学：Agent 不是线性脚本，而是有状态、有条件分支、可中断的图结构。

---

State 与 Node：别把状态散落在各处

LangGraph 的关键概念是 State。State 不是简单的变量传递，而是整个工作流的"记忆"。

回到工单系统的例子，我们的 State 应该包含什么？

from typing import TypedDict, Annotated import operator class TicketState(TypedDict): # 工单基本信息 ticket_id: str user_id: str content: str # 处理结果（带合并操作符，支持多路径写入） analysis: Annotated[str, operator.add] category: str # 权限上下文 user_permissions: list[str] # 人工审批状态 needs_human_review: bool human_approval: bool # 日志追踪 step_log: Annotated[list[str], operator.add]

这里有两个关键点：

第一，State 是全局共享的。每个 Node 都能读取和写入 State，不需要像脚本那样层层传递参数。

第二，使用 Annotated 和 operator 控制写入行为。operator.add表示追加，适合日志这种累积数据；普通字段则是覆盖写入。

Node 的设计原则也很简单：每个 Node 只负责一件事。

def analyze_ticket(state: TicketState) -> TicketState: """分析工单内容""" analysis = call_llm(f"请分析以下工单：{state['content']}") state['analysis'] = analysis state['step_log'].append(f"[{datetime.now()}] 分析完成") return state def classify_ticket(state: TicketState) -> TicketState: """分类工单""" category = call_llm(f"根据分析结果分类：{state['analysis']}") state['category'] = category state['step_log'].append(f"[{datetime.now()}] 分类完成: {category}") return state

Demo 阶段你可能会把所有逻辑塞进一个函数，但生产环境这样写，调试起来会非常痛苦。

---

Edge 与条件分支：动态路由才是 Agent 的灵魂

脚本的分支是静态的：if category == "技术": ...

LangGraph 的 Edge 是动态的：根据 State 的值，决定下一步走哪个 Node。

from langgraph.graph import StateGraph, END # 定义图 graph = StateGraph(TicketState) # 添加节点 graph.add_node("analyze", analyze_ticket) graph.add_node("classify", classify_ticket) graph.add_node("route", route_ticket) graph.add_node("tech_handler", tech_team_handler) graph.add_node("after_sales", after_sales_handler) graph.add_node("escalate", escalate_to_human) # 添加边 graph.add_edge("analyze", "classify") # 条件边：根据分类结果路由 def route_by_category(state: TicketState) -> str: if state["category"] == "技术": return "tech_handler" elif state["category"] == "售后": return "after_sales" else: return "escalate" graph.add_conditional_edges( "classify", route_by_category, { "tech_handler": "tech_handler", "after_sales": "after_sales", "escalate": "escalate" } ) # 所有处理节点完成后结束 graph.add_edge("tech_handler", END) graph.add_edge("after_sales", END) graph.add_edge("escalate", END) # 编译图 app = graph.compile()

这里有个实战经验：条件函数应该尽量简单，逻辑放在 Node 里。如果route_by_category需要调用 LLM 或者做复杂判断，应该拆成一个独立的 Node。

去年踩过的一个坑：我在条件边里直接调了 LLM 做路由决策，结果每次调用都耗时 3-5 秒，而且无法缓存。后来改成预分类 + 条件路由，性能提升了 10 倍。

---

人工审批节点：Demo 里不需要，生产里必须有

这是我最想强调的部分。

Demo 阶段的 Agent 都是全自动的，但真实业务中，关键决策需要人工确认。比如工单金额超过 1000 元，或者涉及敏感操作。

LangGraph 支持在图执行过程中暂停，等待外部输入：

def check_approval_needed(state: TicketState) -> str: """检查是否需要人工审批""" amount = extract_amount(state["content"]) if amount > 1000: state["needs_human_review"] = True return "await_approval" return "continue" def human_approval_node(state: TicketState) -> TicketState: """等待人工审批""" # 这里可以集成消息队列、Webhook 等 approval = wait_for_human_decision(state["ticket_id"]) state["human_approval"] = approval state["step_log"].append(f"[{datetime.now()}] 人工审批结果: {approval}") return state # 添加审批相关节点和边 graph.add_node("check_approval", check_approval_needed) graph.add_node("await_approval", human_approval_node) graph.add_conditional_edges( "classify", check_approval_needed, { "await_approval": "await_approval", "continue": "route" } ) # 审批后的路由 graph.add_conditional_edges( "await_approval", lambda state: "route" if state["human_approval"] else "escalate", {"route": "route", "escalate": "escalate"} )

这里的关键是：审批节点不应该阻塞整个线程。实际生产中，我们会把wait_for_human_decision换成异步消息队列，Node 执行完后图进入"等待"状态，审批通过后从检查点恢复执行。

LangGraph 的检查点机制（Checkpointer）解决了这个问题：

from langgraph.checkpoint.memory import MemorySaver # 使用检查点 memory = MemorySaver() app = graph.compile(checkpointer=memory) # 执行时传入 thread_id，可以暂停和恢复 config = {"configurable": {"thread_id": "ticket-12345"}} result = app.invoke(initial_state, config=config)

去年帮一个金融客户做 Agent 时，他们要求所有超过 5000 元的操作必须人工审批。如果没有 LangGraph 的检查点机制，我们得自己实现状态持久化、消息队列、恢复逻辑——至少多一周的工作量。

---

工程化落地：权限、日志、可观测性

回到文章开头的问题：为什么 Demo 能跑，团队接手就崩？

根本原因是：Demo 阶段忽略了工程化约束。

权限控制

权限不应该在 LLM 调用时才考虑，而应该在 State 初始化时就注入：

def inject_permissions(state: TicketState) -> TicketState: """根据用户 ID 注入权限""" user_id = state["user_id"] permissions = get_user_permissions(user_id) # 从权限服务获取 # 权限校验：用户只能处理自己权限范围内的工单 if "tech" not in permissions and state.get("category") == "技术": state["step_log"].append("[权限拒绝] 用户无权处理技术类工单") raise PermissionError(f"用户 {user_id} 无权处理技术类工单") state["user_permissions"] = permissions return state

日志追踪

State 里的step_log字段就是天然的日志。每次 Node 执行后记录关键信息，生产环境可以直接查询：

def get_execution_log(thread_id: str) -> list[str]: """获取某次执行的完整日志""" config = {"configurable": {"thread_id": thread_id}} # 从检查点获取历史状态 snapshots = list(app.get_history(config)) logs = [] for snapshot in snapshots: if "step_log" in snapshot["channel_values"]: logs.extend(snapshot["channel_values"]["step_log"]) return logs

可观测性

接入现有的日志系统（比如 ELK、Loki），把step_log推送出去。同时记录每次 LLM 调用的输入输出，方便后续调试：

import logging logger = logging.getLogger(__name__) def analyze_ticket(state: TicketState) -> TicketState: logger.info(f"分析工单: {state['ticket_id']}") logger.debug(f"工单内容: {state['content']}") analysis = call_llm(f"请分析以下工单：{state['content']}") logger.info(f"分析结果: {analysis}") state['analysis'] = analysis return state

---

总结

从脚本到 LangGraph 工作流，本质上是从线性思维到图思维的转变。

Demo 阶段的 Agent 追求的是"能跑通"，生产阶段的 Agent 追求的是"可控、可观测、可维护"。这三者的区别，决定了你的项目是停留在 POC 阶段，还是真正能上线。

几个实战建议：

1. State 设计先于 Node 实现。State 设计不好，后面改起来非常痛苦。
2. 每个 Node 保持单一职责。不要为了省事把多个逻辑塞进一个 Node。
3. 条件分支逻辑尽量简单。复杂的判断拆成独立 Node。
4. 尽早接入人工审批节点。不要等到上线前才加，架构设计阶段就要考虑。
5. 日志和权限从第一天就设计。不要等出问题再补，那时候改架构成本很高。

LangGraph 不是银弹，但它提供了一个清晰的框架，让 Agent 从"脚本"变成"系统"。这个转变，就是 Demo 和生产环境之间那道最难的坎。

资料展示

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

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