基于 LangChain 1.0 的 LangGraph 高级应用文章目录基于 LangChain 1.0 的 LangGraph 高级应用1. 深度对比Workflow vs Agent1.1 Workflow 实现示例内容审核1.2 Agent 实现示例内容审核2. 高级状态管理Annotated 与 Reducer3. 人机协作循环 (Human-in-the-Loop) 新规范3.1 核心代码剖析3.2 外部客户端如何挂起与唤醒4. 多智能体协作 (Multi-Agent) 架构模式 查看完整专栏LangChain 1.0 与 LangGraph 学习笔记特别说明本文为个人学习笔记内容仅供学习与交流使用禁止转载或用于商业用途。笔记为个人理解与总结可能存在疏漏或偏差欢迎读者参考并自行甄别。在掌握了 LangGraph 的基本图结构Node、Edge、Conditional Edge之后本篇笔记将深入探讨 LangGraph 的高级应用。我们将对比 Workflow 和 Agent 的不同流转实现深挖底层的状态聚合机制Reducer解析最新版本的精准中断与人工干预Human-in-the-loop。1. 深度对比Workflow vs Agent在业务落地中我们常常面临一个选择是写死规则Workflow还是让 AI 自己做主AgentWorkflow工作流基于硬编码规则驱动。执行顺序固定状态确定极度稳定可控但缺乏灵活性。Agent智能体基于大模型 (LLM)驱动。模型根据上下文自行决定下一步去哪灵活智能但有一定的不可控性和耗时。1.1 Workflow 实现示例内容审核以一个内容审核系统为例。如果是 Workflow我们会把它拆分为多个硬编码的规则节点按顺序流水线执行importrefromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDclassContentModerationState(TypedDict):content:strhas_spam_pattern:booldecision:strreason:strdefcheck_spam(state:ContentModerationState)-dict:[Process] 基于正则规则检查垃圾信息contentstate[content]# 规则重复字符超过5次has_spambool(re.search(r(.)\1{5,},content))return{has_spam_pattern:has_spam}defmake_decision(state:ContentModerationState)-dict:[Process] 综合判定ifstate.get(has_spam_pattern,False):return{decision:rejected,reason:触发垃圾信息正则}return{decision:approved,reason:规则校验通过}workflow_graphStateGraph(ContentModerationState)workflow_graph.add_node(check_spam,check_spam)workflow_graph.add_node(decide,make_decision)# 固定的线性执行顺序workflow_graph.add_edge(START,check_spam)workflow_graph.add_edge(check_spam,decide)workflow_graph.add_edge(decide,END)workflow_appworkflow_graph.compile()1.2 Agent 实现示例内容审核同样的任务如果交给 Agent节点会变得非常精简所有的判断逻辑都交由 LLM 的invoke结果来动态路由importjsonfromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportSystemMessage,HumanMessageclassAgentModerationState(TypedDict):content:stranalysis:strdecision:strmodelinit_chat_model(modelyour-model-name,model_provideropenai,...)defanalyze_content(state:AgentModerationState)-dict:[Process] 使用 LLM 进行语义级审核分析system_prompt你是一个内容审核员。请分析内容并返回 JSON {decision: approved 或 rejected 或 needs_review, reason: ...}responsemodel.invoke([SystemMessage(contentsystem_prompt),HumanMessage(contentf待审核内容{state[content]})])# 实际应用中建议使用 with_structured_output此处为简化演示resultjson.loads(response.content)return{decision:result.get(decision),analysis:result.get(reason)}defshould_auto_decide(state:AgentModerationState)-Literal[end,human_review]:[Route] 根据 LLM 的判定动态路由ifstate[decision]needs_review:returnhuman_reviewreturnenddefhuman_review_node(state:AgentModerationState)-dict:[System] 人工审核节点占位returnstate agent_graphStateGraph(AgentModerationState)agent_graph.add_node(analyze,analyze_content)agent_graph.add_node(human_review,human_review_node)agent_graph.add_edge(START,analyze)agent_graph.add_conditional_edges(analyze,should_auto_decide,{end:END,human_review:human_review})agent_graph.add_edge(human_review,END)agent_appagent_graph.compile()特性WorkflowAgent决策方式基于硬编码规则 (if-else/正则)基于 LLM 对上下文的语义理解灵活性固定流程极难处理边界情况动态路由灵活判断边缘 Case性能/成本毫秒级0 Token 成本较慢需网络请求有消耗适用场景规则明确的结构化数据处理规则模糊、需要理解语义的复杂场景2. 高级状态管理Annotated 与 ReducerState是 LangGraph 的灵魂。如果多个节点同时向状态中写入数据旧数据会被覆盖吗为了解决这个问题我们需要引入 Python 的Annotated机制和自定义的Reducer聚合器。Annotated允许我们为类型标注附加元数据LangGraph 正是通过这些元数据来判断状态合并逻辑的。fromtypingimportTypedDict,Annotatedfromoperatorimportaddfromlanggraph.graph.messageimportadd_messages# 1. 自定义一个去重的 Reducer 函数defmerge_unique(old:list,new:list)-list:合并两个列表并自动去重returnlist(set(oldnew))defmerge_dicts_deep(old:dict,new:dict)-dict:递归深度合并字典resultold.copy()forkey,valueinnew.items():ifisinstance(result.get(key),dict)andisinstance(value,dict):result[key]merge_dicts_deep(result[key],value)else:result[key]valuereturnresult# 2. 定义高级复合状态classAdvancedState(TypedDict):# 普通字段新值会直接【覆盖】旧值user_name:str# 累加字段利用 operator.add新值会与旧值【相加】total_tokens:Annotated[int,add]# 消息列表官方提供的专门针对大模型消息的合并去重器messages:Annotated[list,add_messages]# 标签列表使用我们自己手写的 merge_unique 去重合并tags:Annotated[list[str],merge_unique]# 元数据使用我们手写的字典深层合并器metadata:Annotated[dict,merge_dicts_deep]通过这种极其细粒度的状态合并控制我们能在非常复杂的并发工作流中确保每一个节点产出的数据都能安全、正确地汇总到全局 State 中。3. 人机协作循环 (Human-in-the-Loop) 新规范在实际业务中当 AI 准备执行敏感操作如付款、发送全员邮件时系统必须暂停并等待人类操作。在 LangGraph 最新版本中推荐使用interrupt()函数触发中断并通过Command(resume...)恢复流转。这一切的前提是必须挂载持久化存储Checkpointer。3.1 核心代码剖析importuuidfromtypingimportOptionalfromtyping_extensionsimportTypedDictfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.constantsimportSTART,ENDfromlanggraph.graphimportStateGraphfromlanggraph.typesimportinterrupt,CommandclassState(TypedDict):messages:listlast_answer:Optional[str]human_feedback:Optional[str]# --- 1. Agent 思考节点 ---defagent_node(state:State):print([Process] Agent 正在生成回答...)# ... 省略大模型调用代码 ...mock_answer我已经起草好了邮件准备发送。return{last_answer:mock_answer,human_feedback:None}# --- 2. 核心人工审核节点 ---defreview_node(state:State):answerstate[last_answer]# 重点调用 interrupt 会立刻阻断程序的继续执行并将参数抛给外部。# 它的返回值 feedback 就是未来外部唤醒它时传入的值。feedbackinterrupt(f[Interrupt] AI准备执行\n{answer}\n请输入修改意见输入 ok 批准)feedbackfeedback.strip()iffeedback.lower()ok:print([Success] 人类审核通过流程结束。)returnCommand(gotoEND)print([Action] 人类要求修改打回 Agent 重做。)# 重点使用 Command 动态决定下一个节点并携带更新后的状态returnCommand(gotoagent,update{human_feedback:feedback})# --- 3. 编译图必须挂载 Checkpoint ---builderStateGraph(State)builder.add_node(agent,agent_node)builder.add_node(review,review_node)builder.add_edge(START,agent)builder.add_edge(agent,review)# 此处使用 MemorySaver 用于本地演示生产环境应使用 SqliteSaver/PostgresSavergraphbuilder.compile(checkpointerMemorySaver())3.2 外部客户端如何挂起与唤醒由于加入了interrupt()当我们调用stream()或invoke()时程序会在中途停住。外部驱动代码必须这么写if__name____main__:# 配置唯一的会话 ID保证中断恢复时能找到上下文config{configurable:{thread_id:str(uuid.uuid4())}}# 1. 初始运行current_input{messages:[帮我写一封感谢信]}whileTrue:interrupt_promptNoneinterruptedFalse# 迭代获取图的执行流forchunkingraph.stream(current_input,config):if__interrupt__inchunk:interruptedTrue# 获取在 interrupt(...) 函数里传入的提示信息interrupt_promptchunk[__interrupt__][0].value# 如果没有触发中断说明图已经跑到 END 了跳出大循环ifnotinterrupted:break# 2. 模拟前端弹窗获取人类真实反馈user_feedbackinput(interrupt_prompt\n)# 3. 核心机制唤醒图# 用 Command(resume...) 重新发起请求LangGraph 会自动从刚才中断的节点继续往下跑current_inputCommand(resumeuser_feedback)这种设计非常优雅地剥离了“业务流”和“前端交互流”是真正生产可用的 HIL (Human-in-the-Loop) 方案。4. 多智能体协作 (Multi-Agent) 架构模式单体 Agent 能力再强也无法搞定错综复杂的巨型项目。多智能体协作是通往更高智能的必经之路。常见的多智能体组织架构有以下几种顺序协作 (Sequential)Agent A 查资料 → Agent B 写初稿 → Agent C 润色。线性传递。并行协作 (Parallel)面对一个大问题拆分为多个子问题交由不同领域的 Agent 并发处理最后由一个统一节点汇总。主控专家模式 (Supervisor-Expert)也就是我们常说的“包工头与打工人”。主控 Agent 负责任务拆解和路由将特定任务分发给特定的专家 Agent 子图。协商模式 (Consensus)多个智能体通过相互对话、辩论达成一致后再输出。