适合前后端/测试等有编程基础的同学，手把手带你掌握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适合流程固定的任务；Agent适合需要动态决策的任务。

1.2 LangGraph的解决方案

LangGraph的核心思路是**“状态图”（State Graph）** ——把整个Agent的运行抽象成一张图，图上有节点、有边，还有一块在所有节点之间流转的**“公共黑板”——状态（State）** 。

LangGraph带来的能力：

二、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中读取数据messages=state["messages"]user_input=messages[-1].content# 2. 执行业务逻辑（调用LLM、工具等）result=do_something(user_input)# 3. 返回状态更新（只需要返回修改的字段）return{"messages":messages+[AIMessage(content=result)],"current_step":"completed"}

节点的类型：

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-openai

3.2 基础示例：一个简单的处理流水线

我们先从一个最简单的例子开始——构建一个"输入处理 → 生成回复"的两步流程。

fromtyping_extensionsimportTypedDictfromlanggraph.graphimportStateGraph,START,END# ============ 1. 定义状态 ============classSimpleState(TypedDict):text:strprocessed_count:int# ============ 2. 定义节点函数 ============defprocess_input(state:SimpleState)->dict:"""处理输入：将文本转为大写"""text=state["text"]count=state.get("processed_count",0)processed_text=f"Processed:{text.upper()}"return{"text":processed_text,"processed_count":count+1}defgenerate_response(state:SimpleState)->dict:"""生成回复"""text=state["text"]count=state["processed_count"]response=f"{text}(已处理{count}次)"return{"text":response}# ============ 3. 构建图 ============# 创建状态图graph=StateGraph(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)# 回复 → 结束# 编译图app=graph.compile()# ============ 4. 执行 ============result=app.invoke({"text":"hello langgraph","processed_count":0})print(result)# 输出: {'text': 'Processed: HELLO LANGGRAPH (已处理 1 次)', 'processed_count': 1}

执行流程图：

START → [process] → [respond] → END

3.3 核心API速查

四、实战：带自检功能的"翻译官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. 初始化模型 ============llm=ChatOpenAI(model="deepseek-v4-flash",api_key="你的Key",base_url="https://api.deepseek.com",temperature=0.3)# ============ 3. 翻译节点 ============deftranslator_node(state:TranslationState)->dict:"""执行翻译，如果有feedback则根据意见修改"""source=state["source_text"]target=state["target_lang"]feedback=state.get("feedback","")retry_count=state.get("retry_count",0)iffeedback:# 有修改意见：重新翻译print(f"🔄 第{retry_count+1}次重译，反馈：{feedback}")prompt=f""" 原文：{source}目标语言：{target}上次翻译的问题：{feedback}请根据以上反馈重新翻译，确保修正所有问题。 """else:# 首次翻译print("📝 首次翻译...")prompt=f"请将以下文本翻译成{target}：\n{source}"response=llm.invoke([HumanMessage(content=prompt)])translation=response.contentreturn{"translation":translation,"retry_count":retry_count+1}# ============ 4. 检查节点 ============defchecker_node(state:TranslationState)->dict:"""检查翻译质量"""translation=state["translation"]source=state["source_text"]retry_count=state.get("retry_count",0)# 模拟检查逻辑（实际可用另一个LLM做质量评估）# 这里用简单规则演示issues=[]# 规则1：检查是否太短（可能漏译）iflen(translation)<len(source)*0.3:issues.append("译文过短，可能漏译")# 规则2：检查是否包含"待定"等占位词if"待定"intranslationor"??"intranslation:issues.append("译文包含占位符")# 规则3：最多重试3次ifretry_count>=3: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"]:return"end"else:return"retry"# ============ 6. 构建图 ============graph=StateGraph(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"})# 编译app=graph.compile()# ============ 7. 执行 ============deftranslate_with_self_check(source_text:str,target_lang:str="中文"):print(f"\n📖 原文：{source_text}")print(f"🎯 目标语言：{target_lang}")print("-"*40)result=app.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_ok=True is_ok=False │ ↓ ────────┘ ┌─────────────────┐ │ END │ └─────────────────┘

这个例子展示了LangGraph最核心的能力——循环。传统的Chain无法做到"不通过就重试"，但LangGraph通过条件边轻松实现了。

五、检查点（Checkpoint）与持久化

5.1 什么是检查点？

LangGraph的检查点（Checkpoint）机制会在每一步将图状态保存为快照。这带来了四个关键能力：

5.2 添加短期记忆（Checkpointer）

fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.graphimportStateGraph,START,END# 1. 创建检查点器checkpointer=InMemorySaver()# 开发用，生产环境用PostgresSaver# 2. 编译图时传入checkpointergraph=StateGraph(AgentState)# ... 添加节点和边 ...app=graph.compile(checkpointer=checkpointer)# 3. 执行时指定thread_id（会话标识）config={"configurable":{"thread_id":"user_123"}}# 第一轮对话result1=app.invoke({"messages":[HumanMessage(content="我叫小明")]},config=config)# 第二轮对话（使用相同的thread_id，Agent记得之前的内容）result2=app.invoke({"messages":[HumanMessage(content="我叫什么名字？")]},config=config)# Agent会回答："你叫小明"

Checkpointer vs Store：

生产环境持久化：

# 开发环境：内存存储fromlanggraph.checkpoint.memoryimportInMemorySaver checkpointer=InMemorySaver()# 生产环境：PostgreSQLfromlanggraph.checkpoint.postgresimportPostgresSaver checkpointer=PostgresSaver.from_conn_string("postgresql://user:pass@localhost/db")# 生产环境：SQLitefromlanggraph.checkpoint.sqliteimportSqliteSaver checkpointer=SqliteSaver.from_conn_string("checkpoints.db")

六、LangGraph vs create_agent：如何选择？

6.1 三者的层级关系

LangChain生态中，Agent开发有三层抽象：

┌─────────────────────────────────────────────────────────────┐ │ Deep Agents（最高层） │ │ 自带记忆、工具集、子Agent编排 │ ├─────────────────────────────────────────────────────────────┤ │ create_agent（中间层） │ │ 预置好的ReAct循环，模型+工具一接就能跑 │ ├─────────────────────────────────────────────────────────────┤ │ LangGraph（最底层） │ │ 所有控制权都在开发者手里，完全自定义 │ └─────────────────────────────────────────────────────────────┘

6.2 对比表格

6.3 选择建议

什么时候用create_agent：

• 标准的ReAct Agent（思考→工具→思考→输出）
• 快速原型验证
• 不需要自定义循环逻辑

什么时候用LangGraph：

• 需要自定义循环（如重试、迭代）
• 需要多Agent协作
• 需要人机回环（人工审核节点）
• 需要复杂的条件分支
• 需要精确控制每一步的执行

💡原则：从最高的抽象起步（create_agent），只有当上层确实满足不了需求时，再下沉到LangGraph。

七、LangGraph在企业级场景中的价值

LangGraph之所以成为企业级Agent的首选框架，是因为它解决了传统Agent框架的四个致命缺陷：

一句话总结：“让AI Agent像业务流程一样可审计、可控制、可优化！”

八、实战小练习（作业）

练习：构建一个"智能客服路由Agent"

需求：

1. 用户输入问题后，Agent先分类（售后/售前/投诉/其他）
2. 根据分类路由到对应的处理节点：
   • 售后 → 查询订单状态
   • 售前 → 推荐产品
   • 投诉 → 转人工（模拟）
   • 其他 → 通用回答
3. 使用条件边实现路由
4. 添加检查点支持多轮对话

提示代码框架：

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. 构建图graph=StateGraph(CustomerState)# ... 添加节点和条件边# 5. 添加checkpointer支持多轮checkpointer=InMemorySaver()app=graph.compile(checkpointer=checkpointer)

结语

今天这节课，我们深入学习了LangGraph——Agent开发的底层编排引擎：

至此，模块三（框架篇）的6节课全部完成！

下节课（第13节），我们将进入模块四（进阶篇），从ReAct架构深度解析开始，真正理解Agent"思考→行动→观察"循环的底层原理！

如果觉得有帮助，欢迎点赞、收藏、评论三连！我们下节课见！

📌 本文是《AI Agent开发实战》课程第12节的完整内容，系列文章持续更新中，关注我不迷路！