聊《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 resultDemo 跑通了效果还行。然后接入真实环境问题来了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 stateDemo 阶段你可能会把所有逻辑塞进一个函数但生产环境这样写调试起来会非常痛苦。---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(checkpointermemory) # 执行时传入 thread_id可以暂停和恢复 config {configurable: {thread_id: ticket-12345}} result app.invoke(initial_state, configconfig)去年帮一个金融客户做 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大模型里的哪类内容。
聊《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 resultDemo 跑通了效果还行。然后接入真实环境问题来了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 stateDemo 阶段你可能会把所有逻辑塞进一个函数但生产环境这样写调试起来会非常痛苦。---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(checkpointermemory) # 执行时传入 thread_id可以暂停和恢复 config {configurable: {thread_id: ticket-12345}} result app.invoke(initial_state, configconfig)去年帮一个金融客户做 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大模型里的哪类内容。
