适合前后端/测试等有编程基础的同学手把手带你掌握Agent的底层编排引擎前言通过前面几节课的学习我们掌握了LangChain的Model I/O、Chain、Memory、Retrieval以及上一节课的Tool与create_agent。第11节我们提到create_agent底层基于LangGraph构建。今天我们就要揭开这层面纱直接面对LangGraph本身。一句话定义LangGraph是一个专为构建有状态、多节点执行流程的AI智能体系统设计的Python框架它将状态机State Machine与图结构Graph相结合使得开发者能够直观地用节点边来描述执行逻辑和状态转移。简单说LangGraph是Agent的操作系统内核create_agent是预装好的操作系统而我们要学的是怎么自己写内核。全文约7500字代码均可直接复制运行。一、为什么需要LangGraph——从Chain到Graph的进化1.1 Chain的致命局限只能一条路走到黑传统的LangChain是单向输出的A → B → C。无论用SequentialChain还是LCEL的|管道符本质都是有向无环图DAG——流程只能往前走不能回头。现实场景中的需求需求Chain能否实现为什么翻译质量不合格重新翻译❌需要循环Chain不支持回头根据用户意图走不同分支⚠️ 勉强RouterChain能做简单分支但状态管理困难多轮对话中记住完整上下文❌需要跨节点的持久化状态工具调用后回到模型继续思考❌需要循环Chain是线性的人工审核后继续执行❌需要中断和恢复机制一句话总结Chain适合流程固定的任务Agent适合需要动态决策的任务。1.2 LangGraph的解决方案LangGraph的核心思路是**“状态图”State Graph** ——把整个Agent的运行抽象成一张图图上有节点、有边还有一块在所有节点之间流转的**“公共黑板”——状态State** 。LangGraph带来的能力能力说明循环Cycles节点可以回到之前的节点支持重试和迭代条件分支Conditional Branching根据状态动态选择下一步状态管理State所有节点共享一个状态对象数据流转清晰可控持久化Persistence检查点机制支持断点续传和时间旅行可观测性Observability原生集成LangSmith可视化整个执行过程二、LangGraph的核心三要素State Node Edge构建一个LangGraph应用主要涉及三个概念┌─────────────────────────────────────────────────────────────┐ │ LangGraph 核心三要素 │ │ ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │ │ │ State │ │ Nodes │ │ Edges │ │ │ │ 状态 │ │ 节点 │ │ 边 │ │ │ │ │ │ │ │ │ │ │ │ 公共黑板 │ │ 执行单元 │ │ 流转规则 │ │ │ │ 数据共享 │ │ 处理函数 │ │ 跳转逻辑 │ │ │ └─────────────┘ └─────────────┘ └─────────────┘ │ │ ↓ ↓ ↓ │ │ 所有节点读写 接收State返回更新 决定下一个节点 │ └─────────────────────────────────────────────────────────────┘2.1 State状态—— 节点间的公共黑板State是在所有节点之间流转的共享数据结构。通常用TypedDict或Pydantic模型定义。fromtypingimportTypedDict,Listfromlangchain_core.messagesimportBaseMessageclassAgentState(TypedDict):Agent的全局状态messages:List[BaseMessage]# 对话历史current_step:str# 当前步骤user_id:str# 用户标识tool_results:dict# 工具执行结果缓存State的设计原则所有节点的输入都必须符合State结构所有节点返回的更新也会合并到State中State是增量更新的——节点只需要返回它修改的字段2.2 Node节点—— 执行单元节点是处理State的函数。每个节点接收当前State处理后返回部分状态更新。defmy_node(state:AgentState)-dict:一个典型的节点函数# 1. 从state中读取数据messagesstate[messages]user_inputmessages[-1].content# 2. 执行业务逻辑调用LLM、工具等resultdo_something(user_input)# 3. 返回状态更新只需要返回修改的字段return{messages:messages[AIMessage(contentresult)],current_step:completed}节点的类型节点类型用途LLM节点调用大模型生成内容Tool节点执行外部工具Human节点等待人工输入Router节点决定下一步流向2.3 Edge边—— 流转规则边定义了节点之间的跳转逻辑。LangGraph支持两种边① 普通边Normal Edge无条件跳转graph.add_edge(node_a,node_b)# A执行完一定去Bgraph.add_edge(START,node_a)# 从入口进入Agraph.add_edge(node_b,END)# B执行完结束② 条件边Conditional Edge根据State动态决定# 根据状态值决定走哪条路graph.add_conditional_edges(node_a,# 从哪个节点出发router_function,# 路由函数返回下一个节点的名称{path_b:node_b,path_c:node_c,END:END})三、从零构建第一个LangGraph应用3.1 安装依赖pipinstall-Ulanggraph langchain-openai3.2 基础示例一个简单的处理流水线我们先从一个最简单的例子开始——构建一个输入处理 → 生成回复的两步流程。fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,END# 1. 定义状态 classSimpleState(TypedDict):text:strprocessed_count:int# 2. 定义节点函数 defprocess_input(state:SimpleState)-dict:处理输入将文本转为大写textstate[text]countstate.get(processed_count,0)processed_textfProcessed:{text.upper()}return{text:processed_text,processed_count:count1}defgenerate_response(state:SimpleState)-dict:生成回复textstate[text]countstate[processed_count]responsef{text}(已处理{count}次)return{text:response}# 3. 构建图 # 创建状态图graphStateGraph(SimpleState)# 添加节点graph.add_node(process,process_input)graph.add_node(respond,generate_response)# 添加边定义执行顺序graph.add_edge(START,process)# 入口 → 处理graph.add_edge(process,respond)# 处理 → 回复graph.add_edge(respond,END)# 回复 → 结束# 编译图appgraph.compile()# 4. 执行 resultapp.invoke({text:hello langgraph,processed_count:0})print(result)# 输出: {text: Processed: HELLO LANGGRAPH (已处理 1 次), processed_count: 1}执行流程图START → [process] → [respond] → END3.3 核心API速查API作用StateGraph(StateType)创建状态图指定状态类型graph.add_node(name, function)添加节点graph.add_edge(from, to)添加普通边graph.add_conditional_edges(from, router, mapping)添加条件边graph.compile()编译图为可执行对象app.invoke(input)同步执行app.stream(input)流式执行四、实战带自检功能的翻译官Agent这是LangGraph官方推荐的入门案例。我们要构建一个能自我检查并重试的翻译Agent用户输入 → [翻译] → [检查质量] → 合格? → 结束 ↓ 不合格 → 回到[翻译]带修改意见4.1 完整代码fromtypingimportTypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlangchain_openaiimportChatOpenAIfromlangchain_core.messagesimportHumanMessage,SystemMessage# 1. 定义状态 classTranslationState(TypedDict):source_text:str# 原文target_lang:str# 目标语言translation:str# 翻译结果feedback:str# 检查意见is_acceptable:bool# 是否通过检查retry_count:int# 重试次数# 2. 初始化模型 llmChatOpenAI(modeldeepseek-v4-flash,api_key你的Key,base_urlhttps://api.deepseek.com,temperature0.3)# 3. 翻译节点 deftranslator_node(state:TranslationState)-dict:执行翻译如果有feedback则根据意见修改sourcestate[source_text]targetstate[target_lang]feedbackstate.get(feedback,)retry_countstate.get(retry_count,0)iffeedback:# 有修改意见重新翻译print(f 第{retry_count1}次重译反馈{feedback})promptf 原文{source}目标语言{target}上次翻译的问题{feedback}请根据以上反馈重新翻译确保修正所有问题。 else:# 首次翻译print( 首次翻译...)promptf请将以下文本翻译成{target}\n{source}responsellm.invoke([HumanMessage(contentprompt)])translationresponse.contentreturn{translation:translation,retry_count:retry_count1}# 4. 检查节点 defchecker_node(state:TranslationState)-dict:检查翻译质量translationstate[translation]sourcestate[source_text]retry_countstate.get(retry_count,0)# 模拟检查逻辑实际可用另一个LLM做质量评估# 这里用简单规则演示issues[]# 规则1检查是否太短可能漏译iflen(translation)len(source)*0.3:issues.append(译文过短可能漏译)# 规则2检查是否包含待定等占位词if待定intranslationor??intranslation:issues.append(译文包含占位符)# 规则3最多重试3次ifretry_count3:issues[]# 不再重试ifissues:print(f❌ 检查未通过{, .join(issues)})return{is_acceptable:False,feedback:.join(issues)}else:print(✅ 检查通过)return{is_acceptable:True,feedback:}# 5. 路由函数 defshould_continue(state:TranslationState)-str:决定下一步继续翻译还是结束ifstate[is_acceptable]:returnendelse:returnretry# 6. 构建图 graphStateGraph(TranslationState)# 添加节点graph.add_node(translator,translator_node)graph.add_node(checker,checker_node)# 添加边graph.add_edge(START,translator)graph.add_edge(translator,checker)# 条件边检查通过→结束不通过→回到翻译graph.add_conditional_edges(checker,should_continue,{end:END,retry:translator})# 编译appgraph.compile()# 7. 执行 deftranslate_with_self_check(source_text:str,target_lang:str中文):print(f\n 原文{source_text})print(f 目标语言{target_lang})print(-*40)resultapp.invoke({source_text:source_text,target_lang:target_lang,translation:,feedback:,is_acceptable:False,retry_count:0})print(-*40)print(f✅ 最终译文{result[translation]})print(f 重试次数{result[retry_count]})returnresult# 测试translate_with_self_check(The quick brown fox jumps over the lazy dog. This is a classic pangram used for testing fonts.,中文)4.2 执行流程图┌─────────────────┐ │ START │ └────────┬────────┘ ↓ ┌─────────────────┐ │ translator │ ←──┐ │ (执行翻译) │ │ └────────┬────────┘ │ ↓ │ ┌─────────────────┐ │ │ checker │ │ 重试 │ (检查质量) │ │ └────────┬────────┘ │ ↓ │ ┌─────────────────┐ │ │ should_continue │ │ │ (路由决策) │ │ └────────┬────────┘ │ ↙ ↘ │ is_okTrue is_okFalse │ ↓ ────────┘ ┌─────────────────┐ │ END │ └─────────────────┘这个例子展示了LangGraph最核心的能力——循环。传统的Chain无法做到不通过就重试但LangGraph通过条件边轻松实现了。五、检查点Checkpoint与持久化5.1 什么是检查点LangGraph的检查点Checkpoint机制会在每一步将图状态保存为快照。这带来了四个关键能力能力说明对话记忆跨轮次保留对话历史人机回环在关键节点暂停等待人工审核时间旅行回退到任意历史状态重新执行容错恢复节点失败后从上一个检查点恢复5.2 添加短期记忆Checkpointerfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.graphimportStateGraph,START,END# 1. 创建检查点器checkpointerInMemorySaver()# 开发用生产环境用PostgresSaver# 2. 编译图时传入checkpointergraphStateGraph(AgentState)# ... 添加节点和边 ...appgraph.compile(checkpointercheckpointer)# 3. 执行时指定thread_id会话标识config{configurable:{thread_id:user_123}}# 第一轮对话result1app.invoke({messages:[HumanMessage(content我叫小明)]},configconfig)# 第二轮对话使用相同的thread_idAgent记得之前的内容result2app.invoke({messages:[HumanMessage(content我叫什么名字)]},configconfig)# Agent会回答你叫小明Checkpointer vs Store特性CheckpointerStore存储内容图状态快照应用自定义键值数据作用范围单个线程跨线程记忆类型短期记忆会话级长期记忆用户级典型用途对话连续性、容错用户偏好、共享知识生产环境持久化# 开发环境内存存储fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointerInMemorySaver()# 生产环境PostgreSQLfromlanggraph.checkpoint.postgresimportPostgresSaver checkpointerPostgresSaver.from_conn_string(postgresql://user:passlocalhost/db)# 生产环境SQLitefromlanggraph.checkpoint.sqliteimportSqliteSaver checkpointerSqliteSaver.from_conn_string(checkpoints.db)六、LangGraph vs create_agent如何选择6.1 三者的层级关系LangChain生态中Agent开发有三层抽象┌─────────────────────────────────────────────────────────────┐ │ Deep Agents最高层 │ │ 自带记忆、工具集、子Agent编排 │ ├─────────────────────────────────────────────────────────────┤ │ create_agent中间层 │ │ 预置好的ReAct循环模型工具一接就能跑 │ ├─────────────────────────────────────────────────────────────┤ │ LangGraph最底层 │ │ 所有控制权都在开发者手里完全自定义 │ └─────────────────────────────────────────────────────────────┘6.2 对比表格维度LangGraphcreate_agent控制粒度完全控制每个节点都可定制预置的ReAct循环控制有限学习曲线陡峭平缓开发速度慢需要手动定义图结构快开箱即用灵活性极高支持任意图结构中等只能做ReAct循环适用场景复杂工作流、多Agent协作标准Agent应用6.3 选择建议什么时候用create_agent标准的ReAct Agent思考→工具→思考→输出快速原型验证不需要自定义循环逻辑什么时候用LangGraph需要自定义循环如重试、迭代需要多Agent协作需要人机回环人工审核节点需要复杂的条件分支需要精确控制每一步的执行原则从最高的抽象起步create_agent只有当上层确实满足不了需求时再下沉到LangGraph。七、LangGraph在企业级场景中的价值LangGraph之所以成为企业级Agent的首选框架是因为它解决了传统Agent框架的四个致命缺陷问题LangChain AgentLangGraph黑盒执行无法观察内部状态流转可视化状态图 完整执行日志不可中断一旦启动必须跑完支持任意节点人工介入状态混乱依赖全局变量/记忆显式状态定义与管理调试困难错误堆栈难以追踪精确到节点的错误定位一句话总结“让AI Agent像业务流程一样可审计、可控制、可优化”八、实战小练习作业练习构建一个智能客服路由Agent需求用户输入问题后Agent先分类售后/售前/投诉/其他根据分类路由到对应的处理节点售后 → 查询订单状态售前 → 推荐产品投诉 → 转人工模拟其他 → 通用回答使用条件边实现路由添加检查点支持多轮对话提示代码框架fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDclassCustomerState(TypedDict):user_input:strcategory:Literal[售后,售前,投诉,其他]response:strorder_id:str|None# 1. 分类节点调用LLM判断问题类别defclassify_node(state:CustomerState)-dict:# 调用LLM分类pass# 2. 四个处理节点defafter_sales_node(state:CustomerState)-dict:# 处理售后问题passdefpre_sales_node(state:CustomerState)-dict:# 处理售前问题passdefcomplaint_node(state:CustomerState)-dict:# 处理投诉转人工passdefgeneral_node(state:CustomerState)-dict:# 通用回答pass# 3. 路由函数defroute_by_category(state:CustomerState)-str:returnstate[category]# 4. 构建图graphStateGraph(CustomerState)# ... 添加节点和条件边# 5. 添加checkpointer支持多轮checkpointerInMemorySaver()appgraph.compile(checkpointercheckpointer)结语今天这节课我们深入学习了LangGraph——Agent开发的底层编排引擎知识点核心内容为什么需要LangGraphChain无法支持循环、复杂分支和持久化状态核心三要素State状态 Node节点 Edge边普通边 vs 条件边无条件跳转 vs 基于State动态路由检查点Checkpoint持久化、人机回环、时间旅行、容错恢复LangGraph vs create_agent底层控制 vs 开箱即用根据需求选择企业级价值可审计、可控制、可优化至此模块三框架篇的6节课全部完成课时内容第7节LangChain入门与Model I/O第8节Chain — 构建执行流程第9节Memory — 记忆系统第10节Retrieval — 知识检索与RAG第11节Tool与create_Agent第12节LangGraph — 有状态的Agent工作流 ✅下节课第13节我们将进入模块四进阶篇从ReAct架构深度解析开始真正理解Agent思考→行动→观察循环的底层原理如果觉得有帮助欢迎点赞、收藏、评论三连我们下节课见 本文是《AI Agent开发实战》课程第12节的完整内容系列文章持续更新中关注我不迷路
