这篇不先堆名词。我们把《一个LangGraph项目上线后最先暴露的并不是代码问题》拆成几级台阶看完至少知道下一步该学什么、该练什么。摘要先把这篇文章的目标说清楚看完之后你应该能判断这件事值不值得做以及从哪里动手。上周把那个基于 LangGraph 构建的企业内部知识问答 Agent 正式推上了生产环境。按照常规思维这时候我该庆祝 Demo 到 Production 的跨越或者写篇“保姆级教程”教大家怎么装依赖、配 API Key。但现实给了我一记响亮的耳光。上线第一天监控报警没响因为模型没挂业务方投诉没来因为接口通了。直到第三天客服主管拿着日志截图找我问“为什么这个 Agent 帮用户查了三次密码重置第四次却直接报错说‘权限不足’而且没有记录这次失败的尝试”那一刻我才意识到我们在写代码时满脑子都是State怎么流转、Node 怎么执行、Prompt 怎么写得更聪明。我们忽略了 Agent 作为一个“系统”最本质的属性它不是一个黑盒函数而是一个有状态、有分支、甚至有人介入的工作流。如果你也正准备把 Agent 从 Jupyter Notebook 搬进 Kubernetes请先停下手中的 Prompt 优化看看这篇关于“可控性”的复盘。目录为什么脚本式 Agent 无法支撑生产环境State 与 Node给 Agent 建立“记忆中枢”Edge 与条件分支让流程“听话”人工审批节点生产环境的“刹车片”工程化落地日志、监控与回滚总结为什么脚本式 Agent 无法支撑生产环境很多开发者刚接触 LLM 应用时喜欢写这种“脚本式”代码def run_agent(user_input): # 1. 调模型 response llm.invoke(user_input) # 2. 调工具 result tool.execute(response.content) # 3. 返回 return result这种写法在本地测试时丝般顺滑。但在生产环境中它有三大致命缺陷1. 不可回溯一旦中间步骤出错你无法知道是哪一环的问题。是 Prompt 理解错了还是工具调用超时还是数据库连接断了2. 状态丢失LLM 本身是无状态的。如果需要多轮对话中的上下文记忆你需要自己维护一个列表。一旦列表过长Token 费用飙升且上下文窗口溢出整个逻辑就会崩溃。3. 缺乏确定性分支简单的if-else很难处理复杂的业务逻辑。比如“如果用户问价格走 A 流程如果问技术细节走 B 流程如果涉及敏感操作插入人工审核”。LangGraph 的出现本质上是为了解决这个问题把 Agent 的控制权从“概率性的模型输出”收回到“确定性的图结构”中。State 与 Node给 Agent 建立“记忆中枢”在 LangGraph 中State不仅仅是变量它是整个工作流的单一事实来源Single Source of Truth。我之前的一个踩坑经历是试图在每个 Node 里通过全局变量传递数据。结果在多并发下状态完全混乱A 用户的查询结果混进了 B 用户的响应里。正确的做法是定义一个 TypedDict 作为 State 基类明确每个节点读写的数据字段。from typing import TypedDict, Annotated, List import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): # messages: 显式的消息历史支持追加操作 messages: Annotated[List, add_messages] # tools_used: 记录已使用的工具用于防重复调用或审计 tools_used: List[str] # is_escalated: 是否触发人工升级 is_escalated: bool这里的add_messages是一个关键技巧。它保证了无论哪个 Node 往messages里塞了什么系统都能自动合并历史消息而不是简单覆盖。这就是“记忆”的工程化实现。Node 则是纯函数。输入 State输出 State 的部分更新。切记Node 不要有副作用如直接打印日志、修改外部 DB副作用应该放在 Edge 或 Graph 编译后的 Hook 中处理。 这样做的目的是保证 Node 的可测试性和可替换性——你可以随时把“调用 GPT-4”换成“调用本地 Llama3”而不需要重写业务流程。Edge 与条件分支让流程“听话”如果说 Node 是原子操作那 Edge 就是交通规则。LangGraph 提供了两种边1. 普通边Normal Edges固定顺序A - B - C。适用于线性流程。2. 条件边Conditional Edges根据当前 State 决定下一步去哪个 Node。这是 Agent 具备“思考”能力的关键。以刚才提到的“权限升级”为例我们可以定义一个路由函数def route_after_check(state: AgentState): last_message state[messages][-1] # 假设大模型在最后一步输出了结构化 JSON包含 needs_human 字段 try: decision json.loads(last_message.content) if decision.get(needs_human): return human_approval_node else: return finalize_response_node except: return error_handler_node workflow.add_conditional_edges( tool_execution_node, route_after_check, { human_approval_node: human_approval_node, finalize_response_node: END, error_handler_node: error_handler_node } )这里有一个容易被忽视的细节循环Loops。如果你的业务逻辑需要“模型生成草稿 - 检查员修改 - 模型再次润色”你可以创建一个自环。LangGraph 天然支持max_iterations设置防止模型陷入无限循环。在生产环境中永远不要信任模型的自我纠错能力无限次成功设置上限是最低成本的兜底策略。人工审批节点生产环境的“刹车片”这是本期文章最想强调的差异化点。在 Demo 阶段我们追求的是“全自动”。但在生产环境尤其是涉及资金、隐私、敏感内容时“人”是最高级的容错机制。LangGraph 支持interrupt_before和interrupt_after。这意味着你可以在某个 Node 执行前后暂停工作流等待人类介入。# 在编译时配置中断点 app workflow.compile(interrupt_before[sensitive_action_node]) # 运行时 state app.invoke({messages: [{role: user, content: 帮我删除所有用户数据}]}) # 此时程序会暂停等待外部回调恢复 # state app.update_state(state.graph_id, {approved: True}, as_nodehuman_approval)这种做法的价值在于1. 审计合规谁批准了这次敏感操作时间戳是多少这些都在 State 里留痕。2. 容灾当模型出现幻觉输出了危险的指令时人工节点可以拦截它而不是让它执行下去。3. 灵活性你可以动态调整哪些节点需要人工审核无需修改代码只需配置图结构。很多团队上线崩盘不是因为模型笨而是因为没有给模型装上刹车。工程化落地日志、监控与回滚回到开篇那个“权限不足”的 Bug。问题出在哪里出在tools_used状态没有被正确持久化到日志系统中。当我们使用 LangGraph 时State 的变化是内部的。为了让监控生效我们需要结合 OpenTelemetry 或自定义 Callbacks。一个简单的实践建议在每个 Node 执行前后记录 State 的快照。import time def logging_decorator(node_func): def wrapper(state): start_time time.time() result node_func(state) duration time.time() - start_time # 这里将 state 和 duration 发送到你的日志服务 (ELK/Splunk) log_to_monitoring(node_namenode_func.__name__, state_snapshotresult, cost_msduration * 1000) return result return wrapper此外版本控制至关重要。LangGraph 的图结构是可以序列化的。建议将图的定义存储在 Git 中每次发布带上一个版本号。一旦线上出现严重逻辑错误可以通过回滚图结构版本快速恢复而不是去改代码重新部署。这比传统的“热修复”要安全得多。总结LangGraph 不仅仅是一个库它是一种工程化思维的转变。从脚本式调用到图工作流我们获得的不是更聪明的 AI而是更可控的系统。State 提供了可追溯的记忆。Nodes 提供了模块化能力。Edges 提供了确定的逻辑分支。Interrupts 提供了人机协同的安全阀。当你在下一个项目中考虑“要不要用 Agent”时先问自己三个问题1. 这个流程是否有多步决策2. 是否需要保存中间状态以便调试3. 最关键的操作是否有“后悔药”如人工审核如果答案是肯定的那么 LangGraph 就是你的首选。否则一个简单的 Chain 或 Script 可能更轻量、更高效。记住在 AI 应用工程中确定性比智能更珍贵。资料展示下面是我整理的AI大模型学习资料和工具包预览适合收藏后按主题逐步学习。如果你想看完整资料目录可以在评论区留言「资料」也欢迎告诉我你更关注AI大模型里的哪类内容。
LangGraph 工作流：用一次交付过程做复盘
这篇不先堆名词。我们把《一个LangGraph项目上线后最先暴露的并不是代码问题》拆成几级台阶看完至少知道下一步该学什么、该练什么。摘要先把这篇文章的目标说清楚看完之后你应该能判断这件事值不值得做以及从哪里动手。上周把那个基于 LangGraph 构建的企业内部知识问答 Agent 正式推上了生产环境。按照常规思维这时候我该庆祝 Demo 到 Production 的跨越或者写篇“保姆级教程”教大家怎么装依赖、配 API Key。但现实给了我一记响亮的耳光。上线第一天监控报警没响因为模型没挂业务方投诉没来因为接口通了。直到第三天客服主管拿着日志截图找我问“为什么这个 Agent 帮用户查了三次密码重置第四次却直接报错说‘权限不足’而且没有记录这次失败的尝试”那一刻我才意识到我们在写代码时满脑子都是State怎么流转、Node 怎么执行、Prompt 怎么写得更聪明。我们忽略了 Agent 作为一个“系统”最本质的属性它不是一个黑盒函数而是一个有状态、有分支、甚至有人介入的工作流。如果你也正准备把 Agent 从 Jupyter Notebook 搬进 Kubernetes请先停下手中的 Prompt 优化看看这篇关于“可控性”的复盘。目录为什么脚本式 Agent 无法支撑生产环境State 与 Node给 Agent 建立“记忆中枢”Edge 与条件分支让流程“听话”人工审批节点生产环境的“刹车片”工程化落地日志、监控与回滚总结为什么脚本式 Agent 无法支撑生产环境很多开发者刚接触 LLM 应用时喜欢写这种“脚本式”代码def run_agent(user_input): # 1. 调模型 response llm.invoke(user_input) # 2. 调工具 result tool.execute(response.content) # 3. 返回 return result这种写法在本地测试时丝般顺滑。但在生产环境中它有三大致命缺陷1. 不可回溯一旦中间步骤出错你无法知道是哪一环的问题。是 Prompt 理解错了还是工具调用超时还是数据库连接断了2. 状态丢失LLM 本身是无状态的。如果需要多轮对话中的上下文记忆你需要自己维护一个列表。一旦列表过长Token 费用飙升且上下文窗口溢出整个逻辑就会崩溃。3. 缺乏确定性分支简单的if-else很难处理复杂的业务逻辑。比如“如果用户问价格走 A 流程如果问技术细节走 B 流程如果涉及敏感操作插入人工审核”。LangGraph 的出现本质上是为了解决这个问题把 Agent 的控制权从“概率性的模型输出”收回到“确定性的图结构”中。State 与 Node给 Agent 建立“记忆中枢”在 LangGraph 中State不仅仅是变量它是整个工作流的单一事实来源Single Source of Truth。我之前的一个踩坑经历是试图在每个 Node 里通过全局变量传递数据。结果在多并发下状态完全混乱A 用户的查询结果混进了 B 用户的响应里。正确的做法是定义一个 TypedDict 作为 State 基类明确每个节点读写的数据字段。from typing import TypedDict, Annotated, List import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): # messages: 显式的消息历史支持追加操作 messages: Annotated[List, add_messages] # tools_used: 记录已使用的工具用于防重复调用或审计 tools_used: List[str] # is_escalated: 是否触发人工升级 is_escalated: bool这里的add_messages是一个关键技巧。它保证了无论哪个 Node 往messages里塞了什么系统都能自动合并历史消息而不是简单覆盖。这就是“记忆”的工程化实现。Node 则是纯函数。输入 State输出 State 的部分更新。切记Node 不要有副作用如直接打印日志、修改外部 DB副作用应该放在 Edge 或 Graph 编译后的 Hook 中处理。 这样做的目的是保证 Node 的可测试性和可替换性——你可以随时把“调用 GPT-4”换成“调用本地 Llama3”而不需要重写业务流程。Edge 与条件分支让流程“听话”如果说 Node 是原子操作那 Edge 就是交通规则。LangGraph 提供了两种边1. 普通边Normal Edges固定顺序A - B - C。适用于线性流程。2. 条件边Conditional Edges根据当前 State 决定下一步去哪个 Node。这是 Agent 具备“思考”能力的关键。以刚才提到的“权限升级”为例我们可以定义一个路由函数def route_after_check(state: AgentState): last_message state[messages][-1] # 假设大模型在最后一步输出了结构化 JSON包含 needs_human 字段 try: decision json.loads(last_message.content) if decision.get(needs_human): return human_approval_node else: return finalize_response_node except: return error_handler_node workflow.add_conditional_edges( tool_execution_node, route_after_check, { human_approval_node: human_approval_node, finalize_response_node: END, error_handler_node: error_handler_node } )这里有一个容易被忽视的细节循环Loops。如果你的业务逻辑需要“模型生成草稿 - 检查员修改 - 模型再次润色”你可以创建一个自环。LangGraph 天然支持max_iterations设置防止模型陷入无限循环。在生产环境中永远不要信任模型的自我纠错能力无限次成功设置上限是最低成本的兜底策略。人工审批节点生产环境的“刹车片”这是本期文章最想强调的差异化点。在 Demo 阶段我们追求的是“全自动”。但在生产环境尤其是涉及资金、隐私、敏感内容时“人”是最高级的容错机制。LangGraph 支持interrupt_before和interrupt_after。这意味着你可以在某个 Node 执行前后暂停工作流等待人类介入。# 在编译时配置中断点 app workflow.compile(interrupt_before[sensitive_action_node]) # 运行时 state app.invoke({messages: [{role: user, content: 帮我删除所有用户数据}]}) # 此时程序会暂停等待外部回调恢复 # state app.update_state(state.graph_id, {approved: True}, as_nodehuman_approval)这种做法的价值在于1. 审计合规谁批准了这次敏感操作时间戳是多少这些都在 State 里留痕。2. 容灾当模型出现幻觉输出了危险的指令时人工节点可以拦截它而不是让它执行下去。3. 灵活性你可以动态调整哪些节点需要人工审核无需修改代码只需配置图结构。很多团队上线崩盘不是因为模型笨而是因为没有给模型装上刹车。工程化落地日志、监控与回滚回到开篇那个“权限不足”的 Bug。问题出在哪里出在tools_used状态没有被正确持久化到日志系统中。当我们使用 LangGraph 时State 的变化是内部的。为了让监控生效我们需要结合 OpenTelemetry 或自定义 Callbacks。一个简单的实践建议在每个 Node 执行前后记录 State 的快照。import time def logging_decorator(node_func): def wrapper(state): start_time time.time() result node_func(state) duration time.time() - start_time # 这里将 state 和 duration 发送到你的日志服务 (ELK/Splunk) log_to_monitoring(node_namenode_func.__name__, state_snapshotresult, cost_msduration * 1000) return result return wrapper此外版本控制至关重要。LangGraph 的图结构是可以序列化的。建议将图的定义存储在 Git 中每次发布带上一个版本号。一旦线上出现严重逻辑错误可以通过回滚图结构版本快速恢复而不是去改代码重新部署。这比传统的“热修复”要安全得多。总结LangGraph 不仅仅是一个库它是一种工程化思维的转变。从脚本式调用到图工作流我们获得的不是更聪明的 AI而是更可控的系统。State 提供了可追溯的记忆。Nodes 提供了模块化能力。Edges 提供了确定的逻辑分支。Interrupts 提供了人机协同的安全阀。当你在下一个项目中考虑“要不要用 Agent”时先问自己三个问题1. 这个流程是否有多步决策2. 是否需要保存中间状态以便调试3. 最关键的操作是否有“后悔药”如人工审核如果答案是肯定的那么 LangGraph 就是你的首选。否则一个简单的 Chain 或 Script 可能更轻量、更高效。记住在 AI 应用工程中确定性比智能更珍贵。资料展示下面是我整理的AI大模型学习资料和工具包预览适合收藏后按主题逐步学习。如果你想看完整资料目录可以在评论区留言「资料」也欢迎告诉我你更关注AI大模型里的哪类内容。
