如果你已经会写一个简单 Agent下一步真正难的不是“再多调一个工具”而是让它稳定完成复杂任务前一步的结果怎么保存下一步该谁执行什么时候继续什么时候结束如果中途需要返工流程又该回到哪里LangGraph 解决的就是这个问题。它不是一个让模型变聪明的魔法工具而是一个用来组织 Agent 工作流的框架。它把复杂任务拆成一个个节点用状态保存上下文用边控制执行路线再用检查点让流程可以恢复。读完这篇文章你至少能搞清楚三件事第一Graph、State、Node、Edge、Checkpointer 分别是什么第二为什么条件边和检查点对 Agent 很重要第三多个 Agent 协作时为什么监督者模式更适合流程清晰的任务。前面会用加法器把概念讲简单后面再过渡到多 Agent 协作案例。一、先安装 LangGraph开始之前先安装 LangGraphpip install langgraph如果后面要做联网搜索和大模型调用还会用到 Tavily、LangChain、模型服务等依赖。但理解 LangGraph 的核心概念不需要一开始就把所有工具都接上。我们先从最小例子开始。二、用一个加法器理解 Graph 的基本结构LangGraph 里的 Graph可以先理解为一张“流程图”。这张图里有三件最重要的东西State流程中一直传递的数据。Node真正执行任务的函数或 Agent。Edge节点之间的连线决定执行顺序。先看一个简单加法器。它的目标很朴素传入一个新数字把它加入 numbers 列表再计算 total。这个例子不复杂但非常适合用来理解 LangGraph 的基本结构。from typing import TypedDict from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph #定义状态类 #total 表示所有的字段都是可选的可以不用全部初始化 class AdderState(TypedDict,totalFalse): numbers: list total: int new_number: int def add_number(state: AdderState) - AdderState: new_number state.get(new_number,0) numbers state.get(numbers, []) numbers.append(new_number) total sum(numbers) #更新状态信息 return {numbers: numbers, total: total} #创建一个 状态图 构造器 builder StateGraph(AdderState) #添加节点 builder.add_node(add,add_number) #设置入门点 builder.set_entry_point(add) #添加边执行完 add -》 结束 builder.add_edge(add,END) #编译 状态图 graph graph builder.compile() #初始化 状态 initial_state {numbers: [], total: 0, new_number: 5} result graph.invoke(initial_state) print(f结果{result})运行结果: 结果{numbers: [5], total: 5, new_number: 5}这段代码虽然简单但已经把 LangGraph 的基本工作方式讲清楚了。首先AdderState 定义了整个流程会用到哪些数据。这里有数字列表 numbers、总和 total以及本次新增的数字 new_number。其次add_number 是一个节点函数。它接收 state读取里面的数据完成计算再返回新的字段。LangGraph 会把返回结果合并回原来的状态里。最后builder.add_edge(add, END) 表示执行完 add 节点之后流程结束。所以Graph 并不是一个神秘概念。它就是把“数据怎么传、函数怎么跑、下一步去哪里”这三件事组织起来。三、为什么需要条件边因为真实流程不总是直线上面的加法器只执行一次流程很简单开始 → add → 结束。但真实任务经常不是这样。比如我们希望不断加 1直到 total 达到 10 才停止。这时候流程就需要循环。接下来把例子稍微升级让加法器不断加 1直到 total 达到 10 才停止。这个例子用来说明条件边为什么重要。# 带循环的 Graph def add_one(state: AdderState) - AdderState: numbers state.get(numbers, []) numbers.append(1) total sum(numbers) #更新状态信息 return {numbers: numbers, total: total} builder2 StateGraph(AdderState) builder2.add_node(add,add_one) builder2.set_entry_point(add) #条件边 def should_continue(state:AdderState) -str: if state[total] 10: return add else: return end #设置 条件分支 #add 源节点的名称 #should_continue 条件判断的函数 #end:END 条件判断后 执行的 节点 builder2.add_conditional_edges(add,should_continue,{ add:add, end:END }) graph2 builder2.compile() result2 graph2.invoke({numbers: [], total: 0}) print(f结果{result2})运行结果:结果{numbers: [1, 1, 1, 1, 1, 1, 1, 1, 1, 1], total: 10}这里的关键是 add_conditional_edges。普通边只告诉 Graph“执行完这个节点固定去下一个节点。”条件边则告诉 Graph“执行完这个节点以后先看一下当前 state再决定下一步去哪里。”这个能力非常重要。因为 Agent 工作流里经常会出现这种判断资料够不够文章写完了吗审阅通过了吗如果不通过要不要回到写作节点从这个加法器开始我们已经能看到 LangGraph 的核心优势它不是只能执行一条直线而是可以根据状态做分支和循环。四、Checkpointer让流程有记忆可以接着跑理解了条件边之后再看 Checkpointer也就是检查点。如果说 State 是当前流程的数据那么 Checkpointer 就是把这些数据保存下来的机制。它可以让同一个会话在下一次调用时恢复之前的状态。这里使用 InMemorySaver 做一个最小示例#检查点 builder3 StateGraph(AdderState) builder3.add_node(add,add_number) builder3.set_entry_point(add) builder3.add_edge(add,END) checkpointer InMemorySaver() graph3 builder3.compile(checkpointercheckpointer) config{ configurable:{thread_id:session_123} } result3 graph3.invoke({numbers: [], total: 0, new_number: 3}, configconfig) print(f第一次执行结果{result3}) result4 graph3.invoke({new_number: 5}, configconfig) print(f第二次执行结果{result4})这里的重点是 thread_id。第一次执行结果{numbers: [3], total: 3, new_number: 3}第二次执行结果{numbers: [3, 5], total: 8, new_number: 5}第一次调用时我们传入 3状态里会记录 numbers 和 total。第二次调用时我们只传入 new_number: 4但因为使用的是同一个 thread_idLangGraph 可以恢复之前的状态再继续计算。这就像给工作流加了“记忆”。在简单加法器里它只是记住了之前加过的数字。但放到真实 Agent 系统里它可以记住用户需求、历史消息、搜索结果、文章草稿、审阅意见和最终结果。这也是 LangGraph 很适合复杂 Agent 应用的原因任务可以中断可以继续可以围绕同一个会话逐步推进。五、为什么需要多个 Agent理解了 State、Node、Edge、Checkpointer 之后我们再看多 Agent 就顺很多。一个典型的复杂场景是研究写作类任务。它不是简单生成一段文字而是包含资料获取、信息整理、结构组织、内容生成、质量检查和修改反馈等多个环节。如果把这些环节都交给同一个 Agent它当然可以尝试完成。但任务越复杂单个 Agent 越容易混乱它既要找资料又要组织内容还要做审阅最后还要自己判断下一步。如果这些事情都交给一个 Agent它当然也能尝试完成。但问题是任务越复杂单个 Agent 越容易混乱。它既要搜索又要写作还要审阅还要决定下一步做什么。多个 Agent 的意义就在这里让每个 Agent 专注于一个领域或工具集。研究 Agent 负责资料写作 Agent 负责表达审阅 Agent 负责质量检查协调者负责调度流程。这样做有三个好处。第一专业性更强。每个 Agent 的系统提示词可以更聚焦。第二可维护性更好。哪个环节出问题就改哪个节点。第三更容易扩展。后面要加数据分析 Agent、代码 Agent、翻译 Agent都可以作为新节点接进来。不过多个 Agent 也带来新问题它们应该怎么协作六、三种 Agent 协作模式多 Agent 协作常见有三种模式网络模式、监督者模式和层级模式。理解这三种模式才能知道什么时候该让 Agent 自由沟通什么时候该让一个 Supervisor 统一调度。1. 网络模式Agent 之间自由通信网络模式里Agent 之间可以自由发送消息没有中心协调者。这种模式适合开放式协作比如头脑风暴、多源信息收集、观点碰撞。它的优点是灵活每个 Agent 都可以直接和其他 Agent 交流。但它也有明显缺点协调成本高。因为没有一个统一调度者任务很容易出现重复、冲突或者方向不一致。所以网络模式更适合开放探索不太适合流程明确的任务。2. 监督者模式一个中央 Agent 负责调度监督者模式里有一个中央 Agent也就是 Supervisor。Supervisor 负责接收任务、拆解任务、分配任务并汇总结果。其他 Agent 是执行者按照 Supervisor 的安排完成各自工作。这个模式非常适合有明确流程的任务比如研究写作、数据处理流水线、报告生成。它的优点是流程清晰、可控性强。当然它也有一个潜在问题Supervisor 可能成为瓶颈。因为所有调度都要经过它所以 Supervisor 的判断质量会直接影响整个系统。3. 层级模式多级监督者逐层分发层级模式可以理解为监督者模式的扩展。顶层监督者负责总目标中层监督者负责模块拆解底层 Agent 负责具体执行。它适合超大规模任务比如企业级自动化系统。它的优势是可扩展性强但实现更复杂。对于刚开始搭建多 Agent 系统的人来说通常不建议一上来就做层级模式。综合来看监督者模式最适合从入门走向实战它既能体现多 Agent 分工又不会让系统一开始就变得过于复杂。七、用 LangGraph 实现监督者 Agent下面看一个监督者 两个工作 Agent 的例子一个搜索 Agent一个总结 Agent。这个例子可以帮助我们理解 Supervisor 如何做任务分发。整体结构是这样的用户输入任务后先进入 supervisor 节点。supervisor 根据用户消息判断任务类型。如果用户说“搜索”“查找”“信息”“数据”就走搜索 Agent如果用户说“总结”“概括”“归纳”就走总结 Agent。这就是监督者模式在 LangGraph 里的落地方式。class MultiAgentState(TypedDict): messages: List current_agent: Optional[str] task_type: Optional[str]这个 State 保存了消息列表、当前 Agent 和任务类型。对于多 Agent 系统来说messages 很关键因为它承载了用户输入和各个 Agent 的输出。搜索 Agent 的职责是使用 Tavily 查询信息并把搜索结果追加到 messages 中def search_agent(state: MultiAgentState): messages state[messages] # 最后一条用户的消息 last_message messages[-1] if messages else None search_query if isinstance(last_message,HumanMessage): search_querylast_message.content try: search_results tavily.search(querysearch_query, search_depthbasic) results[] for result in search_results.get(results, [])[:3]: title result.get(title, ) snippet result.get(snippet, ) results.append(f-{title} : {snippet}) if results: response 搜索结果:\n\n.join(results) else: response 搜索结果: 未找到相关的信息 except Exception as e: response f搜索结果搜索失败,错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesearch_agent)], current_agent: search_agent }总结 Agent 的职责则是读取历史消息把用户问题和搜索结果组织起来再调用模型生成总结。真正决定路线的是 supervisor# 监督者 def supervisor(state: MultiAgentState): messages state[messages] last_message messages[-1] if messages else None if isinstance(last_message,HumanMessage): content last_message.content.lower() if any(keyword in content for keyword in [搜索,查找,信息,数据]): next_nodesearch elif any(keyword in content for keyword in [总结,概括,归纳]): next_nodesummary else: next_nodesearch else: next_nodesearch return {next:next_node}这段代码看起来不复杂但它体现了监督者模式的核心工作节点只负责执行Supervisor 负责判断下一步。然后用条件边把 supervisor 的判断接到不同节点builder.add_conditional_edges( supervisor, lambda state: state.get(next,search), { search:search, summary:summary } ) builder.add_edge(search,END) builder.add_edge(summary,END)到这里一个简单的监督者多 Agent 系统就完成了。完整代码import os from typing import TypedDict, List, Optional from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph from tavily import TavilyClient if not os.getenv(DASHSCOPE_API_KEY): raise ValueError(请先配置DASHSCOPE_API_KEY) tavily TavilyClient(api_keytvly-dev-381111111111111111111111111111111111X) model ChatOpenAI( modelqwen3.5-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0.3 ) class MultiAgentState(TypedDict): messages: List current_agent: Optional[str] task_type: Optional[str] def search_agent(state: MultiAgentState): messages state[messages] # 最后一条用户的消息 last_message messages[-1] if messages else None search_query if isinstance(last_message,HumanMessage): search_querylast_message.content try: search_results tavily.search(querysearch_query, search_depthbasic) results[] for result in search_results.get(results, [])[:3]: title result.get(title, ) snippet result.get(snippet, ) results.append(f-{title} : {snippet}) if results: response 搜索结果:\n\n.join(results) else: response 搜索结果: 未找到相关的信息 except Exception as e: response f搜索结果搜索失败,错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesearch_agent)], current_agent: search_agent } # 总结 def summarize_agent(state: MultiAgentState): messages state[messages] content_to_summarize for message in messages: if isinstance(message,HumanMessage): content_to_summarizef用户{message.content}\n elif isinstance(message,AIMessage): content_to_summarizef搜索结果:{message.content}\n try: summary_promptf请总结一下内容\n\n {content_to_summarize} \n\n 总结: summary_response model.invoke(summary_prompt) response f总结{summary_response.content} except Exception as e: response f总结总结失败 错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesummarize_agent)], current_agent: summarize_agent } # 监督者 def supervisor(state: MultiAgentState): messages state[messages] last_message messages[-1] if messages else None if isinstance(last_message,HumanMessage): content last_message.content.lower() if any(keyword in content for keyword in [搜索,查找,信息,数据]): next_nodesearch elif any(keyword in content for keyword in [总结,概括,归纳]): next_nodesummary else: next_nodesearch else: next_nodesearch return {next:next_node} builder StateGraph(MultiAgentState) builder.add_node(supervisor,supervisor) builder.add_node(search,search_agent) builder.add_node(summary,summarize_agent) builder.set_entry_point(supervisor) builder.add_conditional_edges( supervisor, lambda state: state.get(next,search), { search:search, summary:summary } ) builder.add_edge(search,END) builder.add_edge(summary,END) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) def test_multi_agent_system(): config { configurable: {thread_id: session_123} } result graph.invoke({messages: [HumanMessage(content请搜索2024年奥运会举办的时间和中国代表团的表现)]}, configconfig) print(最后一条消息,result[messages][-1].content) result2 graph.invoke({messages: [HumanMessage(content请总结一下2024年奥运会举办的时间和中国代表团的表现)]}, configconfig) print(最后一条消息,result2[messages][-1].content) if __name__ __main__: test_multi_agent_system()八、架构上应该怎么理解监督者、工作节点和状态从架构上看监督者模式可以拆成三块左边是监督者节点中间是工作节点右边是状态定义。左边是监督者节点。它负责理解用户需求决定调用哪个工作节点。它不一定亲自做搜索、写作或分析而是负责调度计划。中间是工作节点比如研究 Agent、写作 Agent、编码 Agent、数据分析 Agent。每个工作节点执行一个具体任务并把结果写回状态。右边是状态定义。它保存用户输入、历史消息、各工作节点的输出以及最终结果。这个结构对应到 LangGraph 里就是State 保存全局数据。Node 执行具体动作。Edge 连接节点并定义流程。Supervisor 通过条件边决定下一步。所以LangGraph 不是简单地“调用多个 Agent”而是把多 Agent 的协作流程变成一张可以执行、可以判断、可以保存状态的图。九、实战搭建一个研究写作团队再往前一步可以搭建一个更完整的研究写作团队研究员、写作者、审阅者和协调者共同完成一个任务。这个例子比前面的搜索 总结更接近真实应用。角色可以这样理解。研究员 Agent使用 Tavily 搜索工具获取资料并提取摘要。写作者 Agent根据研究员提供的资料撰写初稿。审阅者 Agent检查文章语法、逻辑和风格并提出修改建议。协调者也就是监督者控制流程顺序根据中间结果决定下一步。这里的关键不是给 Agent 起不同名字而是让每个角色都有明确职责。角色边界越清楚整个工作流越容易维护。研究员解决“资料从哪里来”。写作者解决“如何把资料写成文章”。审阅者解决“文章是否通顺、准确、有逻辑”。协调者解决“下一步该谁来做”。对应到代码里State 可以这样定义class ResearchState(TypedDict, totalFalse): # 研究主题 topic: str #研究笔记 research_notes: Optional[str] # 草稿 draft: Optional[str] # 反馈建议 review_feedback: Optional[str] # 最终文章 final_article: Optional[str] #下一步操作 next_step: str这里的字段很清楚topic 是研究主题research_notes 是研究笔记draft 是草稿review_feedback 是审阅意见final_article 是最终文章next_step 用来告诉协调者下一步去哪里。研究员节点执行完后把资料写入 research_notes并把下一步设置为 writedef researcher_node(state: ResearchState) - ResearchState:topic state[topic]response researcher_agent.invoke({messages: [(user, topic)]})notes response[messages][-1].contentreturn {research_notes: notes, next_step: write}写作者节点读取研究笔记生成草稿然后把下一步设置为 reviewdef writer_node(state: ResearchState) - ResearchState: notes state[research_notes] response writer_agent.invoke({messages: [(user, f根据以下资料撰写文字: \n{notes})]}) draft response[messages][-1].content return {draft: draft, next_step: review}审阅者节点读取草稿生成反馈和最终文章def reviewer_node(state: ResearchState) - ResearchState: draft state[draft] response reviewer_agent.invoke({messages: [(user, f请审阅以下文章\n{draft})]}) feedback response[messages][-1].content final draft if 严重问题 not in feedback else draft \n\n【修改建议】\n feedback return {review_feedback: feedback, final_article: final, next_step: end}协调者节点最简单但很关键它读取 next_step告诉 Graph 下一步该走哪个节点。def coordinator(state: ResearchState) - ResearchState: step state.get(next_step, research) return {next_step : step}最后把这些节点接起来builder StateGraph(ResearchState) builder.add_node(coordinator,coordinator) builder.add_node(review,reviewer_node) builder.add_node(write,writer_node) builder.add_node(research,researcher_node) builder.set_entry_point(coordinator) builder.add_conditional_edges(coordinator,lambda s:s[next_step],{ research:research, write:write, review:review, end:END }) builder.add_edge(research,coordinator) builder.add_edge(write,coordinator) builder.add_edge(review,coordinator)这段代码体现了一个完整的监督者式工作流先由 coordinator 判断下一步。如果是 research就让研究员工作。研究完成后回到 coordinator。coordinator 再根据 next_step 调用写作者。写作完成后再回到 coordinator。最后进入审阅审阅完成后结束。这就是 LangGraph 在多 Agent 协作里的核心价值它让每个角色专注自己的任务同时让整个流程保持可控。完整代码:import os from typing import TypedDict, Optional from langchain.agents import create_agent from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph from tavily import TavilyClient if not os.getenv(DASHSCOPE_API_KEY): raise Exception(请设置系统环境变量DASHSCOPE_API_KEY) os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] lsv2_pt1111111111111111a9e35a2ac3b5d_eedec3ebfa os.environ[LANGCHAIN_PROJECT] multi_agent_course # 初始化模型 model ChatOpenAI( modelqwen3.5-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0.3 ) class ResearchState(TypedDict, totalFalse): # 研究主题 topic: str #研究笔记 research_notes: Optional[str] # 草稿 draft: Optional[str] # 反馈建议 review_feedback: Optional[str] # 最终文章 final_article: Optional[str] #下一步操作 next_step: str tavily TavilyClient(api_keytvly-dev-3hY111111111111111111111111111se) tool def search_web(query: str) -str: 搜索互联网获取事实的信息参数 query是 搜索关键词当用户 需要最新的 新闻、天气、事件等时 使用此工具 try: #调用 搜索工具查询关键词获取 最多 3条结果 result tavily.search(query, max_results3) summaries [item[content] for item in result.get(results,[])] return \n.join(summaries) if summaries else 未找到相关的信息 except Exception as e: return f搜索失败{str(e)} researcher_agent create_agent(modelmodel, tools[search_web], system_prompt你是一个研究员。根据用户的主题使用搜索工具收集资料并整理成要点列表) writer_agent create_agent(modelmodel, tools[], system_prompt你是一个专业的写作者。根据研究院提供的资料撰写一篇结构清晰、语言流畅的文章) reviewer_agent create_agent(modelmodel, tools[], system_prompt你是一个审阅编辑。检查文章的逻辑、语法、风格提出修改建议) def researcher_node(state: ResearchState) - ResearchState: topic state[topic] response researcher_agent.invoke({messages: [(user, topic)]}) notes response[messages][-1].content return {research_notes: notes, next_step: write} def writer_node(state: ResearchState) - ResearchState: notes state[research_notes] response writer_agent.invoke({messages: [(user, f根据以下资料撰写文字: \n{notes})]}) draft response[messages][-1].content return {draft: draft, next_step: review} def reviewer_node(state: ResearchState) - ResearchState: draft state[draft] response reviewer_agent.invoke({messages: [(user, f请审阅以下文章\n{draft})]}) feedback response[messages][-1].content final draft if 严重问题 not in feedback else draft \n\n【修改建议】\n feedback return {review_feedback: feedback, final_article: final, next_step: end} def coordinator(state: ResearchState) - ResearchState: step state.get(next_step, research) return {next_step : step} builder StateGraph(ResearchState) builder.add_node(coordinator,coordinator) builder.add_node(review,reviewer_node) builder.add_node(write,writer_node) builder.add_node(research,researcher_node) builder.set_entry_point(coordinator) builder.add_conditional_edges(coordinator,lambda s:s[next_step],{ research:research, write:write, review:review, end:END }) builder.add_edge(research,coordinator) builder.add_edge(write,coordinator) builder.add_edge(review,coordinator) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) def test_search_writing_team(): config { configurable:{ thread_id:session_001 } } result graph.invoke({topic: 人工智能在医疗领域的应用, next_step: research}, configconfig) print(f研究笔记 \nresult.get(research_notes, 无)) print(f文章草稿 \n result.get(draft, 无)) print(f审阅反馈 \n result.get(review_feedback, 无)) print(f最终文章 \n result.get(final_article, 无)) if __name__ __main__: test_search_writing_team()运行结果:结尾LangGraph 不是让 Agent 更聪明而是让流程更可靠回顾一下本文其实不是在讲一个单独 API而是在讲一种构建复杂 Agent 应用的方式。当任务很简单时一个 Agent 就够了。比如问一个问题、调用一个工具、返回一个答案。但当任务变成多步骤、多角色、多轮推进时就需要 LangGraph 这样的框架来管理流程。它通过 State 保存上下文通过 Node 拆分任务通过 Edge 控制流向通过 Checkpointer 保存进度。再结合监督者模式就可以把研究、写作、审阅、总结、搜索这些不同 Agent 组织成一个可维护的系统。最后可以用一句话记住 LangGraph它不是更聪明的 Agent而是让多个 Agent 按照清晰流程协作的工作流框架。如果你理解了这条主线再回头看加法器、条件边、检查点和研究写作团队代码就不会觉得它们是零散示例。它们其实是一条逐步升级的路线先理解图再理解状态再理解分支和恢复最后把这些能力用到多 Agent 协作里。
LangGraph 到底有什么用？一文讲透 AI Agent 工作流
如果你已经会写一个简单 Agent下一步真正难的不是“再多调一个工具”而是让它稳定完成复杂任务前一步的结果怎么保存下一步该谁执行什么时候继续什么时候结束如果中途需要返工流程又该回到哪里LangGraph 解决的就是这个问题。它不是一个让模型变聪明的魔法工具而是一个用来组织 Agent 工作流的框架。它把复杂任务拆成一个个节点用状态保存上下文用边控制执行路线再用检查点让流程可以恢复。读完这篇文章你至少能搞清楚三件事第一Graph、State、Node、Edge、Checkpointer 分别是什么第二为什么条件边和检查点对 Agent 很重要第三多个 Agent 协作时为什么监督者模式更适合流程清晰的任务。前面会用加法器把概念讲简单后面再过渡到多 Agent 协作案例。一、先安装 LangGraph开始之前先安装 LangGraphpip install langgraph如果后面要做联网搜索和大模型调用还会用到 Tavily、LangChain、模型服务等依赖。但理解 LangGraph 的核心概念不需要一开始就把所有工具都接上。我们先从最小例子开始。二、用一个加法器理解 Graph 的基本结构LangGraph 里的 Graph可以先理解为一张“流程图”。这张图里有三件最重要的东西State流程中一直传递的数据。Node真正执行任务的函数或 Agent。Edge节点之间的连线决定执行顺序。先看一个简单加法器。它的目标很朴素传入一个新数字把它加入 numbers 列表再计算 total。这个例子不复杂但非常适合用来理解 LangGraph 的基本结构。from typing import TypedDict from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph #定义状态类 #total 表示所有的字段都是可选的可以不用全部初始化 class AdderState(TypedDict,totalFalse): numbers: list total: int new_number: int def add_number(state: AdderState) - AdderState: new_number state.get(new_number,0) numbers state.get(numbers, []) numbers.append(new_number) total sum(numbers) #更新状态信息 return {numbers: numbers, total: total} #创建一个 状态图 构造器 builder StateGraph(AdderState) #添加节点 builder.add_node(add,add_number) #设置入门点 builder.set_entry_point(add) #添加边执行完 add -》 结束 builder.add_edge(add,END) #编译 状态图 graph graph builder.compile() #初始化 状态 initial_state {numbers: [], total: 0, new_number: 5} result graph.invoke(initial_state) print(f结果{result})运行结果: 结果{numbers: [5], total: 5, new_number: 5}这段代码虽然简单但已经把 LangGraph 的基本工作方式讲清楚了。首先AdderState 定义了整个流程会用到哪些数据。这里有数字列表 numbers、总和 total以及本次新增的数字 new_number。其次add_number 是一个节点函数。它接收 state读取里面的数据完成计算再返回新的字段。LangGraph 会把返回结果合并回原来的状态里。最后builder.add_edge(add, END) 表示执行完 add 节点之后流程结束。所以Graph 并不是一个神秘概念。它就是把“数据怎么传、函数怎么跑、下一步去哪里”这三件事组织起来。三、为什么需要条件边因为真实流程不总是直线上面的加法器只执行一次流程很简单开始 → add → 结束。但真实任务经常不是这样。比如我们希望不断加 1直到 total 达到 10 才停止。这时候流程就需要循环。接下来把例子稍微升级让加法器不断加 1直到 total 达到 10 才停止。这个例子用来说明条件边为什么重要。# 带循环的 Graph def add_one(state: AdderState) - AdderState: numbers state.get(numbers, []) numbers.append(1) total sum(numbers) #更新状态信息 return {numbers: numbers, total: total} builder2 StateGraph(AdderState) builder2.add_node(add,add_one) builder2.set_entry_point(add) #条件边 def should_continue(state:AdderState) -str: if state[total] 10: return add else: return end #设置 条件分支 #add 源节点的名称 #should_continue 条件判断的函数 #end:END 条件判断后 执行的 节点 builder2.add_conditional_edges(add,should_continue,{ add:add, end:END }) graph2 builder2.compile() result2 graph2.invoke({numbers: [], total: 0}) print(f结果{result2})运行结果:结果{numbers: [1, 1, 1, 1, 1, 1, 1, 1, 1, 1], total: 10}这里的关键是 add_conditional_edges。普通边只告诉 Graph“执行完这个节点固定去下一个节点。”条件边则告诉 Graph“执行完这个节点以后先看一下当前 state再决定下一步去哪里。”这个能力非常重要。因为 Agent 工作流里经常会出现这种判断资料够不够文章写完了吗审阅通过了吗如果不通过要不要回到写作节点从这个加法器开始我们已经能看到 LangGraph 的核心优势它不是只能执行一条直线而是可以根据状态做分支和循环。四、Checkpointer让流程有记忆可以接着跑理解了条件边之后再看 Checkpointer也就是检查点。如果说 State 是当前流程的数据那么 Checkpointer 就是把这些数据保存下来的机制。它可以让同一个会话在下一次调用时恢复之前的状态。这里使用 InMemorySaver 做一个最小示例#检查点 builder3 StateGraph(AdderState) builder3.add_node(add,add_number) builder3.set_entry_point(add) builder3.add_edge(add,END) checkpointer InMemorySaver() graph3 builder3.compile(checkpointercheckpointer) config{ configurable:{thread_id:session_123} } result3 graph3.invoke({numbers: [], total: 0, new_number: 3}, configconfig) print(f第一次执行结果{result3}) result4 graph3.invoke({new_number: 5}, configconfig) print(f第二次执行结果{result4})这里的重点是 thread_id。第一次执行结果{numbers: [3], total: 3, new_number: 3}第二次执行结果{numbers: [3, 5], total: 8, new_number: 5}第一次调用时我们传入 3状态里会记录 numbers 和 total。第二次调用时我们只传入 new_number: 4但因为使用的是同一个 thread_idLangGraph 可以恢复之前的状态再继续计算。这就像给工作流加了“记忆”。在简单加法器里它只是记住了之前加过的数字。但放到真实 Agent 系统里它可以记住用户需求、历史消息、搜索结果、文章草稿、审阅意见和最终结果。这也是 LangGraph 很适合复杂 Agent 应用的原因任务可以中断可以继续可以围绕同一个会话逐步推进。五、为什么需要多个 Agent理解了 State、Node、Edge、Checkpointer 之后我们再看多 Agent 就顺很多。一个典型的复杂场景是研究写作类任务。它不是简单生成一段文字而是包含资料获取、信息整理、结构组织、内容生成、质量检查和修改反馈等多个环节。如果把这些环节都交给同一个 Agent它当然可以尝试完成。但任务越复杂单个 Agent 越容易混乱它既要找资料又要组织内容还要做审阅最后还要自己判断下一步。如果这些事情都交给一个 Agent它当然也能尝试完成。但问题是任务越复杂单个 Agent 越容易混乱。它既要搜索又要写作还要审阅还要决定下一步做什么。多个 Agent 的意义就在这里让每个 Agent 专注于一个领域或工具集。研究 Agent 负责资料写作 Agent 负责表达审阅 Agent 负责质量检查协调者负责调度流程。这样做有三个好处。第一专业性更强。每个 Agent 的系统提示词可以更聚焦。第二可维护性更好。哪个环节出问题就改哪个节点。第三更容易扩展。后面要加数据分析 Agent、代码 Agent、翻译 Agent都可以作为新节点接进来。不过多个 Agent 也带来新问题它们应该怎么协作六、三种 Agent 协作模式多 Agent 协作常见有三种模式网络模式、监督者模式和层级模式。理解这三种模式才能知道什么时候该让 Agent 自由沟通什么时候该让一个 Supervisor 统一调度。1. 网络模式Agent 之间自由通信网络模式里Agent 之间可以自由发送消息没有中心协调者。这种模式适合开放式协作比如头脑风暴、多源信息收集、观点碰撞。它的优点是灵活每个 Agent 都可以直接和其他 Agent 交流。但它也有明显缺点协调成本高。因为没有一个统一调度者任务很容易出现重复、冲突或者方向不一致。所以网络模式更适合开放探索不太适合流程明确的任务。2. 监督者模式一个中央 Agent 负责调度监督者模式里有一个中央 Agent也就是 Supervisor。Supervisor 负责接收任务、拆解任务、分配任务并汇总结果。其他 Agent 是执行者按照 Supervisor 的安排完成各自工作。这个模式非常适合有明确流程的任务比如研究写作、数据处理流水线、报告生成。它的优点是流程清晰、可控性强。当然它也有一个潜在问题Supervisor 可能成为瓶颈。因为所有调度都要经过它所以 Supervisor 的判断质量会直接影响整个系统。3. 层级模式多级监督者逐层分发层级模式可以理解为监督者模式的扩展。顶层监督者负责总目标中层监督者负责模块拆解底层 Agent 负责具体执行。它适合超大规模任务比如企业级自动化系统。它的优势是可扩展性强但实现更复杂。对于刚开始搭建多 Agent 系统的人来说通常不建议一上来就做层级模式。综合来看监督者模式最适合从入门走向实战它既能体现多 Agent 分工又不会让系统一开始就变得过于复杂。七、用 LangGraph 实现监督者 Agent下面看一个监督者 两个工作 Agent 的例子一个搜索 Agent一个总结 Agent。这个例子可以帮助我们理解 Supervisor 如何做任务分发。整体结构是这样的用户输入任务后先进入 supervisor 节点。supervisor 根据用户消息判断任务类型。如果用户说“搜索”“查找”“信息”“数据”就走搜索 Agent如果用户说“总结”“概括”“归纳”就走总结 Agent。这就是监督者模式在 LangGraph 里的落地方式。class MultiAgentState(TypedDict): messages: List current_agent: Optional[str] task_type: Optional[str]这个 State 保存了消息列表、当前 Agent 和任务类型。对于多 Agent 系统来说messages 很关键因为它承载了用户输入和各个 Agent 的输出。搜索 Agent 的职责是使用 Tavily 查询信息并把搜索结果追加到 messages 中def search_agent(state: MultiAgentState): messages state[messages] # 最后一条用户的消息 last_message messages[-1] if messages else None search_query if isinstance(last_message,HumanMessage): search_querylast_message.content try: search_results tavily.search(querysearch_query, search_depthbasic) results[] for result in search_results.get(results, [])[:3]: title result.get(title, ) snippet result.get(snippet, ) results.append(f-{title} : {snippet}) if results: response 搜索结果:\n\n.join(results) else: response 搜索结果: 未找到相关的信息 except Exception as e: response f搜索结果搜索失败,错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesearch_agent)], current_agent: search_agent }总结 Agent 的职责则是读取历史消息把用户问题和搜索结果组织起来再调用模型生成总结。真正决定路线的是 supervisor# 监督者 def supervisor(state: MultiAgentState): messages state[messages] last_message messages[-1] if messages else None if isinstance(last_message,HumanMessage): content last_message.content.lower() if any(keyword in content for keyword in [搜索,查找,信息,数据]): next_nodesearch elif any(keyword in content for keyword in [总结,概括,归纳]): next_nodesummary else: next_nodesearch else: next_nodesearch return {next:next_node}这段代码看起来不复杂但它体现了监督者模式的核心工作节点只负责执行Supervisor 负责判断下一步。然后用条件边把 supervisor 的判断接到不同节点builder.add_conditional_edges( supervisor, lambda state: state.get(next,search), { search:search, summary:summary } ) builder.add_edge(search,END) builder.add_edge(summary,END)到这里一个简单的监督者多 Agent 系统就完成了。完整代码import os from typing import TypedDict, List, Optional from langchain_core.messages import HumanMessage, AIMessage from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph from tavily import TavilyClient if not os.getenv(DASHSCOPE_API_KEY): raise ValueError(请先配置DASHSCOPE_API_KEY) tavily TavilyClient(api_keytvly-dev-381111111111111111111111111111111111X) model ChatOpenAI( modelqwen3.5-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0.3 ) class MultiAgentState(TypedDict): messages: List current_agent: Optional[str] task_type: Optional[str] def search_agent(state: MultiAgentState): messages state[messages] # 最后一条用户的消息 last_message messages[-1] if messages else None search_query if isinstance(last_message,HumanMessage): search_querylast_message.content try: search_results tavily.search(querysearch_query, search_depthbasic) results[] for result in search_results.get(results, [])[:3]: title result.get(title, ) snippet result.get(snippet, ) results.append(f-{title} : {snippet}) if results: response 搜索结果:\n\n.join(results) else: response 搜索结果: 未找到相关的信息 except Exception as e: response f搜索结果搜索失败,错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesearch_agent)], current_agent: search_agent } # 总结 def summarize_agent(state: MultiAgentState): messages state[messages] content_to_summarize for message in messages: if isinstance(message,HumanMessage): content_to_summarizef用户{message.content}\n elif isinstance(message,AIMessage): content_to_summarizef搜索结果:{message.content}\n try: summary_promptf请总结一下内容\n\n {content_to_summarize} \n\n 总结: summary_response model.invoke(summary_prompt) response f总结{summary_response.content} except Exception as e: response f总结总结失败 错误信息 {str(e)} return { messages: messages[AIMessage(contentresponse,namesummarize_agent)], current_agent: summarize_agent } # 监督者 def supervisor(state: MultiAgentState): messages state[messages] last_message messages[-1] if messages else None if isinstance(last_message,HumanMessage): content last_message.content.lower() if any(keyword in content for keyword in [搜索,查找,信息,数据]): next_nodesearch elif any(keyword in content for keyword in [总结,概括,归纳]): next_nodesummary else: next_nodesearch else: next_nodesearch return {next:next_node} builder StateGraph(MultiAgentState) builder.add_node(supervisor,supervisor) builder.add_node(search,search_agent) builder.add_node(summary,summarize_agent) builder.set_entry_point(supervisor) builder.add_conditional_edges( supervisor, lambda state: state.get(next,search), { search:search, summary:summary } ) builder.add_edge(search,END) builder.add_edge(summary,END) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) def test_multi_agent_system(): config { configurable: {thread_id: session_123} } result graph.invoke({messages: [HumanMessage(content请搜索2024年奥运会举办的时间和中国代表团的表现)]}, configconfig) print(最后一条消息,result[messages][-1].content) result2 graph.invoke({messages: [HumanMessage(content请总结一下2024年奥运会举办的时间和中国代表团的表现)]}, configconfig) print(最后一条消息,result2[messages][-1].content) if __name__ __main__: test_multi_agent_system()八、架构上应该怎么理解监督者、工作节点和状态从架构上看监督者模式可以拆成三块左边是监督者节点中间是工作节点右边是状态定义。左边是监督者节点。它负责理解用户需求决定调用哪个工作节点。它不一定亲自做搜索、写作或分析而是负责调度计划。中间是工作节点比如研究 Agent、写作 Agent、编码 Agent、数据分析 Agent。每个工作节点执行一个具体任务并把结果写回状态。右边是状态定义。它保存用户输入、历史消息、各工作节点的输出以及最终结果。这个结构对应到 LangGraph 里就是State 保存全局数据。Node 执行具体动作。Edge 连接节点并定义流程。Supervisor 通过条件边决定下一步。所以LangGraph 不是简单地“调用多个 Agent”而是把多 Agent 的协作流程变成一张可以执行、可以判断、可以保存状态的图。九、实战搭建一个研究写作团队再往前一步可以搭建一个更完整的研究写作团队研究员、写作者、审阅者和协调者共同完成一个任务。这个例子比前面的搜索 总结更接近真实应用。角色可以这样理解。研究员 Agent使用 Tavily 搜索工具获取资料并提取摘要。写作者 Agent根据研究员提供的资料撰写初稿。审阅者 Agent检查文章语法、逻辑和风格并提出修改建议。协调者也就是监督者控制流程顺序根据中间结果决定下一步。这里的关键不是给 Agent 起不同名字而是让每个角色都有明确职责。角色边界越清楚整个工作流越容易维护。研究员解决“资料从哪里来”。写作者解决“如何把资料写成文章”。审阅者解决“文章是否通顺、准确、有逻辑”。协调者解决“下一步该谁来做”。对应到代码里State 可以这样定义class ResearchState(TypedDict, totalFalse): # 研究主题 topic: str #研究笔记 research_notes: Optional[str] # 草稿 draft: Optional[str] # 反馈建议 review_feedback: Optional[str] # 最终文章 final_article: Optional[str] #下一步操作 next_step: str这里的字段很清楚topic 是研究主题research_notes 是研究笔记draft 是草稿review_feedback 是审阅意见final_article 是最终文章next_step 用来告诉协调者下一步去哪里。研究员节点执行完后把资料写入 research_notes并把下一步设置为 writedef researcher_node(state: ResearchState) - ResearchState:topic state[topic]response researcher_agent.invoke({messages: [(user, topic)]})notes response[messages][-1].contentreturn {research_notes: notes, next_step: write}写作者节点读取研究笔记生成草稿然后把下一步设置为 reviewdef writer_node(state: ResearchState) - ResearchState: notes state[research_notes] response writer_agent.invoke({messages: [(user, f根据以下资料撰写文字: \n{notes})]}) draft response[messages][-1].content return {draft: draft, next_step: review}审阅者节点读取草稿生成反馈和最终文章def reviewer_node(state: ResearchState) - ResearchState: draft state[draft] response reviewer_agent.invoke({messages: [(user, f请审阅以下文章\n{draft})]}) feedback response[messages][-1].content final draft if 严重问题 not in feedback else draft \n\n【修改建议】\n feedback return {review_feedback: feedback, final_article: final, next_step: end}协调者节点最简单但很关键它读取 next_step告诉 Graph 下一步该走哪个节点。def coordinator(state: ResearchState) - ResearchState: step state.get(next_step, research) return {next_step : step}最后把这些节点接起来builder StateGraph(ResearchState) builder.add_node(coordinator,coordinator) builder.add_node(review,reviewer_node) builder.add_node(write,writer_node) builder.add_node(research,researcher_node) builder.set_entry_point(coordinator) builder.add_conditional_edges(coordinator,lambda s:s[next_step],{ research:research, write:write, review:review, end:END }) builder.add_edge(research,coordinator) builder.add_edge(write,coordinator) builder.add_edge(review,coordinator)这段代码体现了一个完整的监督者式工作流先由 coordinator 判断下一步。如果是 research就让研究员工作。研究完成后回到 coordinator。coordinator 再根据 next_step 调用写作者。写作完成后再回到 coordinator。最后进入审阅审阅完成后结束。这就是 LangGraph 在多 Agent 协作里的核心价值它让每个角色专注自己的任务同时让整个流程保持可控。完整代码:import os from typing import TypedDict, Optional from langchain.agents import create_agent from langchain_core.tools import tool from langchain_openai import ChatOpenAI from langgraph.checkpoint.memory import InMemorySaver from langgraph.constants import END from langgraph.graph import StateGraph from tavily import TavilyClient if not os.getenv(DASHSCOPE_API_KEY): raise Exception(请设置系统环境变量DASHSCOPE_API_KEY) os.environ[LANGCHAIN_TRACING_V2] true os.environ[LANGCHAIN_API_KEY] lsv2_pt1111111111111111a9e35a2ac3b5d_eedec3ebfa os.environ[LANGCHAIN_PROJECT] multi_agent_course # 初始化模型 model ChatOpenAI( modelqwen3.5-plus, api_keyos.getenv(DASHSCOPE_API_KEY), base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1, temperature0.3 ) class ResearchState(TypedDict, totalFalse): # 研究主题 topic: str #研究笔记 research_notes: Optional[str] # 草稿 draft: Optional[str] # 反馈建议 review_feedback: Optional[str] # 最终文章 final_article: Optional[str] #下一步操作 next_step: str tavily TavilyClient(api_keytvly-dev-3hY111111111111111111111111111se) tool def search_web(query: str) -str: 搜索互联网获取事实的信息参数 query是 搜索关键词当用户 需要最新的 新闻、天气、事件等时 使用此工具 try: #调用 搜索工具查询关键词获取 最多 3条结果 result tavily.search(query, max_results3) summaries [item[content] for item in result.get(results,[])] return \n.join(summaries) if summaries else 未找到相关的信息 except Exception as e: return f搜索失败{str(e)} researcher_agent create_agent(modelmodel, tools[search_web], system_prompt你是一个研究员。根据用户的主题使用搜索工具收集资料并整理成要点列表) writer_agent create_agent(modelmodel, tools[], system_prompt你是一个专业的写作者。根据研究院提供的资料撰写一篇结构清晰、语言流畅的文章) reviewer_agent create_agent(modelmodel, tools[], system_prompt你是一个审阅编辑。检查文章的逻辑、语法、风格提出修改建议) def researcher_node(state: ResearchState) - ResearchState: topic state[topic] response researcher_agent.invoke({messages: [(user, topic)]}) notes response[messages][-1].content return {research_notes: notes, next_step: write} def writer_node(state: ResearchState) - ResearchState: notes state[research_notes] response writer_agent.invoke({messages: [(user, f根据以下资料撰写文字: \n{notes})]}) draft response[messages][-1].content return {draft: draft, next_step: review} def reviewer_node(state: ResearchState) - ResearchState: draft state[draft] response reviewer_agent.invoke({messages: [(user, f请审阅以下文章\n{draft})]}) feedback response[messages][-1].content final draft if 严重问题 not in feedback else draft \n\n【修改建议】\n feedback return {review_feedback: feedback, final_article: final, next_step: end} def coordinator(state: ResearchState) - ResearchState: step state.get(next_step, research) return {next_step : step} builder StateGraph(ResearchState) builder.add_node(coordinator,coordinator) builder.add_node(review,reviewer_node) builder.add_node(write,writer_node) builder.add_node(research,researcher_node) builder.set_entry_point(coordinator) builder.add_conditional_edges(coordinator,lambda s:s[next_step],{ research:research, write:write, review:review, end:END }) builder.add_edge(research,coordinator) builder.add_edge(write,coordinator) builder.add_edge(review,coordinator) checkpointer InMemorySaver() graph builder.compile(checkpointercheckpointer) def test_search_writing_team(): config { configurable:{ thread_id:session_001 } } result graph.invoke({topic: 人工智能在医疗领域的应用, next_step: research}, configconfig) print(f研究笔记 \nresult.get(research_notes, 无)) print(f文章草稿 \n result.get(draft, 无)) print(f审阅反馈 \n result.get(review_feedback, 无)) print(f最终文章 \n result.get(final_article, 无)) if __name__ __main__: test_search_writing_team()运行结果:结尾LangGraph 不是让 Agent 更聪明而是让流程更可靠回顾一下本文其实不是在讲一个单独 API而是在讲一种构建复杂 Agent 应用的方式。当任务很简单时一个 Agent 就够了。比如问一个问题、调用一个工具、返回一个答案。但当任务变成多步骤、多角色、多轮推进时就需要 LangGraph 这样的框架来管理流程。它通过 State 保存上下文通过 Node 拆分任务通过 Edge 控制流向通过 Checkpointer 保存进度。再结合监督者模式就可以把研究、写作、审阅、总结、搜索这些不同 Agent 组织成一个可维护的系统。最后可以用一句话记住 LangGraph它不是更聪明的 Agent而是让多个 Agent 按照清晰流程协作的工作流框架。如果你理解了这条主线再回头看加法器、条件边、检查点和研究写作团队代码就不会觉得它们是零散示例。它们其实是一条逐步升级的路线先理解图再理解状态再理解分支和恢复最后把这些能力用到多 Agent 协作里。
