关键词LangGraph, LangChain, 状态机, SOP, Human-in-the-loop大家好我是飞哥前两周我们学习了LlamaIndex(最强大脑) 和AutoGen(最强团队)。按理说有了这两个神器什么应用做不出来但当你真正把 AutoGen 做的客服系统上线后老板可能会找你谈话老板“飞哥昨天有个用户说要退款你的 AI 也没查订单状态直接就答应了而且那个订单金额是 10 万块AI 居然没问我直接批了”你“呃…那个 Agent 比较热心肠…”痛点AutoGen 这种“多 Agent 自由讨论”的模式适合创意型任务写研报、写代码。但对于业务型任务退款、审批、报销企业需要的是严谨的 SOP (标准作业程序)绝不允许 AI 自由发挥。解决方案我们需要把 AI 从“自由讨论室”拉到“工厂流水线”上。这就是LangGraph的使命 ——构建有状态、可控的图应用。1. 为什么你需要 LangGraph(Why) 生动类比AutoGen头脑风暴会议️。大家七嘴八舌最后也能讨论出结果但过程不可控适合创意。LangGraph肯德基后厨流水线。State (托盘)一个托盘在流水线上流动上面放着汉堡胚用户需求。Node (工位)工位 1 (意图识别)服务员看单子。是做“香辣鸡腿堡”还是“牛肉堡”决定去哪个分支工位 2 (规则检查)⚖️称重员检查。肉饼有没有少于 100g金额 1000工位 3 (人工审批)质检员发现肉饼颜色不对按停传送带喊店长人类过来确认。店长说“行”才能继续包装店长说“不行”直接扔掉。Human-in-the-loop工位 4 (执行)打包员装袋递给顾客。核心价值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 乱批款。 安装依赖pipinstalllanggraph langchain-openai langchain-corelanggraph本次课程的主角。它负责画图定义状态、节点、边并管理整个流水线的运行包括暂停、恢复。langchain-coreLangChain 家族的基础组件。它定义了BaseMessage消息格式、PromptTemplate提示词模版等通用标准相当于乐高的“基础积木块”。langchain-openai连接 OpenAI 模型的桥梁。如果我们的节点需要调用 GPT-4 进行意图识别或生成回复就得靠它。 流程设计图是退款闲聊金额 1000金额 1000经理同意经理拒绝用户输入意图识别节点退款规则检查节点直接回复自动批准经理审批节点拒绝退款 核心代码拆解第一步定义状态 (包裹里有什么) fromtypingimportTypedDict,Annotated,List,Optionalimportoperatorfromlangchain_core.messagesimportBaseMessageclassAgentState(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.0amountget_order_amount(state[order_id])# 规则引擎逻辑ifamount1000: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 请求来唤醒它。fromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpoint.memoryimportMemorySaver# 构建图workflowStateGraph(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 工位ifstate[refund_status]waiting_approval:returnhuman_approval# 否则直接结束 (END)returnEND# 将 process 节点连接到分拣逻辑 (route_process)# 意味着process 跑完后下一步去哪儿问问 route_processworkflow.add_conditional_edges(process,route_process)# 4. 添加普通边 (Edges)# 经理审批完直接结束workflow.add_edge(human_approval,END)# 5. 编译图 (Compile)# memory: 用于保存图的运行状态 (Checkpointer)memoryMemorySaver()appworkflow.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( 系统启动...)foreventinapp.stream({order_id:BIG888},configconfig):# stream 会逐步输出每个节点的执行结果pass# 3. 检查断点 (Check Interrupt)# 运行结束后检查一下图的状态stateapp.get_state(config)ifstate.nextandhuman_approvalinstate.next:# 发现下个节点是 human_approval说明被暂停了print(f 触发风控等待经理审批... (金额:{state.values[amount]}))user_inputinput(经理是否批准(yes/no): )ifuser_inputyes:# 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(▶️ 继续执行后续流程...)foreventinapp.stream(None,configconfig):pass 运行结果展示当你运行这段代码时你会看到类似下面的交互过程 系统启动... ⚠️ 金额 1000, 需要经理审批 触发风控等待经理审批... (金额: 1200.0) 经理是否批准(yes/no): yes ✅ 经理已批准更新状态为 approved ▶️ 继续执行后续流程... ✅ 金额 1000, 自动批准 (注意这里可能是后续节点的日志)看这就是 Human-in-the-loop 的魅力。AI 跑得再快关键时刻也得停下来听你的。这种“暂停-修改-继续”的能力正是传统自动化脚本做不到的也是构建负责任 AI (Responsible AI)的基石。4. 飞哥总结 AutoGen是文科生擅长创意、协作、发散思维。LangGraph是理科生擅长逻辑、规则、流程控制。在真正的企业级 AI 应用如 ERP、CRM、OA 系统中LangGraph才是那个能让老板放心的“定海神针”。下期预告做好了功能怎么知道 AI 到底稳不稳定会不会哪天突然变笨了下周我们将进入LLMOps (运维与监控)的世界教你用LangSmith监控 AI 的一举一动创作不易记得关注飞哥 点赞、收藏哦~~下篇见