但当你真正把 AutoGen 做的客服系统上线后老板可能会找你谈话老板“哥昨天有个用户说要退款你的 AI 也没查订单状态直接就答应了而且那个订单金额是 10 万块AI 居然没问我直接批了”你“呃…那个 Agent 比较热心肠…”痛点 AutoGen 这种“多 Agent 自由讨论”的模式适合创意型任务写研报、写代码。 但对于业务型任务退款、审批、报销企业需要的是严谨的 SOP (标准作业程序)绝不允许 AI 自由发挥。解决方案 我们需要把 AI 从“自由讨论室”拉到“工厂流水线”上。 这就是LangGraph的使命 ——构建有状态、可控的图应用。— 1. 为什么你需要 LangGraph(Why) 生动类比AutoGen头脑风暴会议️。大家七嘴八舌最后也能讨论出结果但过程不可控适合创意。LangGraph肯德基后厨流水线。工位 1 (意图识别)服务员看单子。是做“香辣鸡腿堡”还是“牛肉堡”决定去哪个分支工位 2 (规则检查)⚖️称重员检查。肉饼有没有少于 100g金额 1000工位 3 (人工审批)质检员发现肉饼颜色不对按停传送带喊店长人类过来确认。店长说“行”才能继续包装店长说“不行”直接扔掉。Human-in-the-loop工位 4 (执行)打包员装袋递给顾客。State (托盘)一个托盘在流水线上流动上面放着汉堡胚用户需求。Node (工位)核心价值 LangGraph 让你能用代码画出这张“流程图”并且强制 AI 沿着线走。最重要的是它支持Human-in-the-loop (人在回路)—— 关键时刻必须人来拍板AI 才能继续。— 2. 核心概念提炼 (Skeleton) LangGraph 的世界里主要有这几个概念State (状态)流水线上的“包裹”。所有节点都在读写这个包裹里的数据比如order_id,status。Node (节点)流水线上的“工位”。可以是 LLM可以是 Python 函数也可以是 API 调用。Edge (边)连接工位的“传送带”。Conditional Edge (条件边)智能分拣机。如果amount 1000传送到“经理室”否则传送到“财务室”。START END图的入口和出口。— 3. 实战项目智能客服“工单自动处理流” ️我们要实现一个严谨的退款系统绝不让 AI 乱批款。 安装依赖pip install langgraph langchain-openai langchain-corelanggraph本次课程的主角。它负责画图定义状态、节点、边并管理整个流水线的运行包括暂停、恢复。langchain-coreLangChain 家族的基础组件。它定义了BaseMessage消息格式、PromptTemplate提示词模版等通用标准相当于乐高的“基础积木块”。langchain-openai连接 OpenAI 模型的桥梁。如果我们的节点需要调用 GPT-4 进行意图识别或生成回复就得靠它。 流程设计图 核心代码拆解第一步定义状态 (包裹里有什么) from typing import TypedDict, Annotated, List, Optionalimport operatorfrom langchain_core.messages import BaseMessageclassAgentState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 聊天记录 (使用 operator.add 追加新消息) order_id: Optional[str] # 订单号 amount: Optional[float] # 订单金额# 状态机核心字段# pending: 初始状态# waiting_approval: 等待经理审批 (金额 1000)# approved: 已批准 (金额 1000 或 经理同意)# rejected: 已拒绝 (经理拒绝) refund_status: str第二步定义节点 (工人在干什么) defnode_process_refund(state: AgentState): 退款规则检查节点 输入: state[order_id] 输出: 更新 state[amount] 和 state[refund_status] # 模拟查询数据库defget_order_amount(order_id):# 假设所有订单都是 1200 元 (大于 1000触发审批)return1200.0 amount get_order_amount(state[order_id]) # 规则引擎逻辑if amount 1000: print(⚠️ 金额 1000, 需要经理审批)# 返回的部分数据会自动 merge 到 State 中return {amount: amount, refund_status: waiting_approval}else: print(✅ 金额 1000, 自动批准)return {amount: amount, refund_status: approved}defnode_human_approval(state: AgentState): 人工审批节点 注意这里其实是个空函数因为我们会在进入这个节点前使用 interrupt_before 暂停。 当人工 update_state 后会直接跳过这个节点的执行进入下一个节点。 pass第三步构建图与人工介入 (关键) 这里最神奇的是interrupt_before。它告诉 LangGraph**在进入human_approval节点之前把程序暂停把状态存到数据库里MemorySaver然后等待指令**注意这里的“暂停”不是进程退出而是图的执行挂起。如果你用的是 Web 服务这意味着 API 请求结束等待下一个 API 请求来唤醒它。from langgraph.graph import StateGraph, ENDfrom langgraph.checkpoint.memory import MemorySaver# 构建图workflow StateGraph(AgentState)# 1. 添加节点 (Add Nodes)# 注册我们的“工人”和“工位”workflow.add_node(process, node_process_refund)workflow.add_node(human_approval, node_human_approval) # 2. 设置入口 (Set Entry Point)# 告诉图从哪里开始跑 (这里假设直接进入 process)workflow.set_entry_point(process)# 3. 添加条件边 (Conditional Edges)# 相当于流水线上的“分拣机”defroute_process(state):# 如果状态是 waiting_approval就传送到 human_approval 工位if state[refund_status] waiting_approval:returnhuman_approval# 否则直接结束 (END)return END# 将 process 节点连接到分拣逻辑 (route_process)# 意味着process 跑完后下一步去哪儿问问 route_processworkflow.add_conditional_edges(process, route_process)# 4. 添加普通边 (Edges)# 经理审批完直接结束workflow.add_edge(human_approval, END)# 5. 编译图 (Compile)# memory: 用于保存图的运行状态 (Checkpointer)memory MemorySaver()app workflow.compile( checkpointermemory, # interrupt_before: 在进入这些节点之前强制暂停# 这是实现 Human-in-the-loop 的关键 interrupt_before[human_approval] )第四步运行与交互 (Human-in-the-loop) 当程序暂停时你可以去喝杯咖啡。回来后经理输入yes程序会从断点处恢复执行。# 1. 配置 (Config)# thread_id: 每个用户的对话都有唯一的 ID类似微信号config {configurable: {thread_id: thread-1}}# 2. 初始运行 (Run)# 我们只给一个空的输入因为这里假设从 process 节点开始它会自己去查 order_idprint( 系统启动...)for event in app.stream({order_id: BIG888}, configconfig):# stream 会逐步输出每个节点的执行结果pass# 3. 检查断点 (Check Interrupt)# 运行结束后检查一下图的状态state app.get_state(config)if state.next andhuman_approvalin state.next:# 发现下个节点是 human_approval说明被暂停了 print(f 触发风控等待经理审批... (金额: {state.values[amount]})) user_input input(经理是否批准(yes/no): )if user_input yes:# 4. 人工干预 (Human Intervention)# 经理说 OK我们直接修改图里的状态 print(✅ 经理已批准更新状态为 approved) app.update_state(config, {refund_status: approved})else: print(❌ 经理已拒绝更新状态为 rejected) app.update_state(config, {refund_status: rejected})# 5. 恢复执行 (Resume)# 再次调用 stream这次传入 NoneLangGraph 会自动从断点处继续跑 print(▶️ 继续执行后续流程...)for event in app.stream(None, configconfig):pass 运行结果展示当你运行这段代码时你会看到类似下面的交互过程 系统启动...⚠️ 金额 1000, 需要经理审批 触发风控等待经理审批... (金额: 1200.0)经理是否批准(yes/no): yes✅ 经理已批准更新状态为 approved▶️ 继续执行后续流程...✅ 金额 1000, 自动批准 (注意这里可能是后续节点的日志)看这就是 Human-in-the-loop 的魅力。AI 跑得再快关键时刻也得停下来听你的。这种“暂停-修改-继续”的能力正是传统自动化脚本做不到的也是构建负责任 AI (Responsible AI)的基石。学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】