前面6篇我们一直在用create_agent做单Agent应用——一个Agent搞定所有事。但真实业务往往不是这样的你想让一个Agent负责收集需求另一个Agent写代码第三个Agent做Code Review还要一个统筹全局的主管来协调。这时候create_agent就不够用了因为它是单Agent的封装。你需要的是LangGraph——LangChain 1.0的底层运行时引擎专门做复杂工作流编排和多Agent协作。别被图结构吓到它其实就是个状态机每个节点做一件事边决定接下来走哪条路。你写if-else的时候已经在用状态机了LangGraph只是把它可视化、工程化了。为什么要学LangGraph你可能会想create_agent不是已经跑在LangGraph上了吗我直接用create_agent不就行了区别在于控制粒度场景用create_agent用LangGraph单个Agent工具够用杀鸡用牛刀固定流程多步骤勉强能做但很别扭自然条件分支根据结果走不同路做不到核心多Agent协作做不到核心人工审批后继续需要中间件原生支持长任务暂停/恢复需要中间件原生支持简单说create_agent是快速搭建单Agent的工具LangGraph是编排复杂工作流的引擎。两者不是替代关系是互补关系。LangGraph核心概念三个就够1. State状态整个工作流共享的数据结构。所有节点都能读写State节点之间通过State传递数据。from typing import TypedDict, Annotatedfrom langgraph.graph import add_messagesclass AgentState(TypedDict): messages: Annotated[list, add_messages] # 对话消息 current_task: str # 当前任务 review_result: str # 审核结果 final_output: str # 最终输出Annotated[list, add_messages]的意思是每次节点往messages里写数据是追加而不是覆盖。这比你自己管理列表方便多了。2. Node节点每个节点是一个函数接收State返回State的更新部分。def writer(state: AgentState) - dict: 写作节点根据任务写初稿 task state[current_task] # 调用大模型写初稿 draft model.invoke(f请根据以下任务写一篇初稿{task}) return {messages: [(assistant, f初稿完成{draft.content})]}def reviewer(state: AgentState) - dict: 审核节点审核初稿并给出修改意见 last_msg state[messages][-1].content review model.invoke(f请审核以下内容并给出修改意见{last_msg}) return {review_result: review.content}3. Edge边边决定流程怎么走。分两种普通边A执行完一定走B条件边根据State决定走哪条路from langgraph.graph import StateGraph, ENDdef route_after_review(state: AgentState) - str: 审核后路由通过就走publisher不通过回writer重写 if 通过 in state[review_result]: return publisher return writer实战内容生产流水线我们构建一个三Agent协作的内容生产系统Writer写手→ Reviewer审核→ Publisher发布。审核不通过就打回重写。from langchain_openai import ChatOpenAIfrom langgraph.graph import StateGraph, END, STARTfrom typing import TypedDict, Annotatedfrom langgraph.graph import add_messages# 模型model ChatOpenAI(modelgpt-4o-mini)# 定义状态class WorkflowState(TypedDict): messages: Annotated[list, add_messages] topic: str draft: str review_comment: str revision_count: int final_article: str# 节点1写手def writer(state: WorkflowState) - dict: topic state[topic] revision_count state.get(revision_count, 0) if revision_count 0: # 第一次写 prompt f请根据以下主题写一篇800字的文章{topic} else: # 根据审核意见修改 prompt f请根据审核意见修改文章。原文章{state[draft]}审核意见{state[review_comment]}请修改后输出完整文章。 result model.invoke(prompt) return { draft: result.content, revision_count: revision_count 1, }# 节点2审核def reviewer(state: WorkflowState) - dict: prompt f请审核以下文章从内容准确性、逻辑清晰度、语言表达三个维度给出意见。如果文章质量合格回复通过。如果需要修改给出具体修改建议。文章{state[draft]} result model.invoke(prompt) return {review_comment: result.content}# 节点3发布def publisher(state: WorkflowState) - dict: return {final_article: f✅ 已发布文章\n\n{state[draft]}}# 条件边审核后决定走哪条路def should_revise(state: WorkflowState) - str: if通过in state[review_comment]: returnpublisher if state[revision_count] 3: returnpublisher# 最多改3轮强制发布 returnwriter# 构建图graph StateGraph(WorkflowState)# 添加节点graph.add_node(writer, writer)graph.add_node(reviewer, reviewer)graph.add_node(publisher, publisher)# 添加边graph.add_edge(START, writer) # 开始 → 写手graph.add_edge(writer, reviewer) # 写手 → 审核graph.add_conditional_edges( # 审核 → 条件路由 reviewer, should_revise, {writer: writer, publisher: publisher},)graph.add_edge(publisher, END) # 发布 → 结束# 编译app graph.compile()运行result app.invoke({ topic: LangChain 1.0核心变化解析, messages: [], revision_count: 0,})print(result[final_article])这个流程的执行过程Writer根据topic写初稿Reviewer审核初稿如果审核通过→Publisher发布不通过→打回Writer修改最多修改3轮后强制发布防止死循环这就是create_agent做不到的事——条件分支 循环 多Agent协作在LangGraph里几行代码就实现了。状态持久化长任务暂停和恢复实际项目中有些工作流不是一口气跑完的。比如审核节点可能需要真人看今天看完明天才给反馈。LangGraph用Checkpointer实现状态持久化from langgraph.checkpoint.memory import MemorySaver# 用内存做Checkpointer生产环境换成数据库checkpointer MemorySaver()# 编译时传入checkpointerapp graph.compile(checkpointercheckpointer)# 运行时指定thread_idconfig {configurable: {thread_id: article-001}}# 第一次运行result app.invoke({topic: AI开发趋势, ...}, configconfig)# 暂停后恢复用同一个thread_idresult app.invoke(None, configconfig) # 传None表示从上次断点继续生产环境建议用数据库做CheckpointerPostgreSQL、SQLite等这样即使服务重启了也能恢复。人机协同Human-in-the-Loop有些节点需要人工参与。比如上面的Reviewer节点你想让真人来审核而不是用模型from langgraph.checkpoint.memory import MemorySaver# 在需要人工的节点前打断点graph StateGraph(WorkflowState)# ... 添加节点和边 ...# 编译时指定interrupt_beforeapp graph.compile( checkpointerMemorySaver(), interrupt_before[reviewer], # 在reviewer节点前暂停)# 第一次运行会在reviewer前暂停result app.invoke({topic: ..., ...}, config{configurable: {thread_id: 1}})# 人工审核后修改state并恢复state app.get_state(config)app.update_state(config, {review_comment: 通过写得不错}, as_nodereviewer)# 继续执行result app.invoke(None, configconfig)这种暂停→人工处理→恢复的能力是构建真实业务流程的基础。Supervisor模式多Agent协调者当Agent超过3个时你需要一个主管来协调它们。Supervisor模式的思路是一个中心Agent根据任务分配给不同的子Agent。def supervisor(state: WorkflowState) - dict: 主管节点决定下一步该谁干活 task state[current_task] prompt f你是一个任务协调者。根据以下任务决定下一步由谁执行- writer写作任务- coder编程任务- researcher研究调研任务当前任务{task}只需回复节点名称。 result model.invoke(prompt) return {next_agent: result.content.strip().lower()}# Supervisor路由def supervisor_route(state: WorkflowState) - str: return state.get(next_agent, end)# 构建Supervisor图graph StateGraph(WorkflowState)graph.add_node(supervisor, supervisor)graph.add_node(writer, writer_node)graph.add_node(coder, coder_node)graph.add_node(researcher, researcher_node)graph.add_edge(START, supervisor)graph.add_conditional_edges( supervisor, supervisor_route, {writer: writer, coder: coder, researcher: researcher, end: END},)# 每个子Agent完成后回到supervisorgraph.add_edge(writer, supervisor)graph.add_edge(coder, supervisor)graph.add_edge(researcher, supervisor)Supervisor根据任务内容决定分配给谁子Agent完成后再回到Supervisor决定下一步。这就是多Agent协作的核心模式。学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%免费】