1. 项目缘起从单打独斗到团队协作的AI Agent进化最近在折腾LangChain生态发现一个挺有意思的现象大家用LangChain搭出来的应用大多还是“单线程”的。一个Agent接到任务吭哧吭哧从头干到尾遇到复杂点的问题比如既要分析数据又要写报告还要检查逻辑就显得有点力不从心。这就像让一个研究员同时干研究、挑刺和写总结的活儿质量难免打折扣。于是多Agent协同的框架开始冒头其中LangGraph是LangChain官方力推的一个。但官方文档一上来就是StateGraph、Nodes、Edges这些概念配合着动辄几百行的示例代码新手很容易被劝退。我就在想能不能用最少的代码把多Agent协同的核心流程跑通让大家先看到“森林”再去看“树木”所以就有了这个项目用大约100行Python代码实现一个经典的“研究员Researcher vs 批评家Critic 总结员Summarizer”的多Agent工作流。这个模式非常实用研究员负责生成初稿批评家负责挑毛病总结员负责提炼最终成果形成了一个完整的“创作-评审-定稿”闭环。通过这个最小可运行示例你能快速理解LangGraph是如何编排多个AI智能体Agent像团队一样协同工作的。2. LangGraph核心三要素图、状态与边在动手写代码之前得先搞明白LangGraph里最关键的三个概念图Graph、状态State和边Edges。你可以把整个多Agent系统想象成一个公司的项目流程图。图Graph就是这张流程图本身它定义了整个工作流的骨架。在LangGraph里我们创建一个StateGraph对象它就是我们的画布。状态State是这个工作流中流转的“项目文件夹”。所有Agent需要共享和传递的信息都放在这里面。在LangGraph中状态通常用一个TypedDict来定义明确每个字段的类型。在我们的例子里这个“文件夹”里可能会装这些东西input: 用户最初提出的问题。research_report: 研究员生成的初步报告。critique: 批评家写的评审意见。final_summary: 总结员产出的最终摘要。可能还有一个next字段用来决定下一步该去哪个部门节点。节点Nodes就是流程图里的各个“部门”每个部门负责一项具体任务。在我们的系统里三个部门分别是研究员、批评家、总结员。每个部门节点都是一个Python函数它接收当前的“项目文件夹”状态在里面加工或添加内容然后把更新后的文件夹返回。边Edges是连接各部门的“箭头”决定了工作流的走向。边分为两种条件边Conditional Edges: 像是一个路口的分支根据“项目文件夹”里的某个条件比如state[‘next’]的值来决定下一步去哪。这实现了if-else的逻辑。普通边Normal Edges: 就是固定的单行道从一个节点无条件地指向下一个节点。搞懂了这三样整个LangGraph的编排逻辑就清晰了我们定义好状态结构创建几个节点函数然后用边把它们按照业务逻辑连接起来最后编译成一个可执行的工作流。3. 环境搭建与模型选择轻量起步的务实之选多Agent听起来高大上但起步环境很简单。为了确保这个示例足够“最小”我们选择最轻量的依赖和最具性价比的模型方案。首先安装核心库。我们只需要LangChain和LangGraph以及调用大模型所需的包。打开终端执行以下命令pip install langchain langgraph langchain-openai这里没有安装庞大的langchain-community因为我们的示例很纯粹只用OpenAI的模型。langchain-openai这个专用包更轻量。接下来是模型选择。多Agent系统意味着多次调用LLM成本是需要考虑的因素。因此我强烈建议使用OpenAI的GPT-3.5-Turbo作为起点。它的响应速度、成本约为GPT-4的1/10到1/20和性能对于构建和调试一个原型系统来说是完全够用的。等到工作流逻辑完全跑通再考虑换用GPT-4来提升最终输出质量也不迟。你需要准备一个OpenAI的API Key。如果你还没有可以去OpenAI官网注册获取。然后在代码中通过环境变量设置它export OPENAI_API_KEY你的sk-...密钥或者在Python脚本中直接设置import os os.environ[“OPENAI_API_KEY”] “你的sk-...密钥”模型初始化代码如下from langchain_openai import ChatOpenAI # 初始化LLM温度设为0.7让输出有一定创造性但不过于天马行空 llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.7)这个llm对象将作为我们三个Agent共用的“大脑”。虽然它们角色不同但底层能力来自同一个模型我们通过设计不同的系统提示词System Prompt来赋予它们不同的角色和职责。这样做的好处是节省资源也简化了代码结构。4. 定义工作流状态与智能体节点现在我们来具体搭建这个三Agent系统。首先需要明确我们的“项目文件夹”——状态里要放哪些东西。我们使用TypedDict来定义状态结构这能让代码的类型提示更清晰方便后续开发和调试。from typing import TypedDict, Literal class AgentState(TypedDict): “”“多Agent工作流的共享状态。”“” # 原始输入的问题 input: str # 研究员生成的初步报告 research_report: str # 批评家给出的评审意见 critique: str # 总结员生成的最终摘要 final_summary: str # 决定下一个执行节点的键这是一个字面量类型 next: Literal[“researcher”, “critic”, “summarizer”, “end”]这里next字段使用了Literal类型限定它只能是这几个字符串之一这相当于为我们工作流的路线图规定了几个固定的站点名。接下来创建三个节点函数也就是我们的三个“部门员工”。研究员节点它的任务是基于用户输入生成一份详细的初步报告。def research_node(state: AgentState) - AgentState: “”“研究员Agent根据输入问题生成初步研究报告。”“” from langchain_core.prompts import ChatPromptTemplate # 研究员的系统提示词定义其角色和任务 research_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位严谨的研究员。请针对用户提出的问题提供一份详细、客观、信息丰富的初步报告。报告应结构清晰包含背景、分析和关键点。”), (“human”, “{input}”) ]) # 构建调用链提示词 LLM research_chain research_prompt | llm # 调用模型获取报告文本 response research_chain.invoke({“input”: state[“input”]}) report_content response.content # 更新状态存入报告并指示下一步去找批评家 return { “research_report”: report_content, “next”: “critic” }注意这个函数返回了一个字典它会被LangGraph自动合并到全局的AgentState中。我们设置了“next”: “critic”这意味着完成研究后工作流会自动前往批评家节点。批评家节点它的任务是严格评审研究员的报告找出不足。def critique_node(state: AgentState) - AgentState: “”“批评家Agent评审研究报告指出其不足、偏见或遗漏。”“” from langchain_core.prompts import ChatPromptTemplate critique_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位苛刻的批评家。你的任务是严格审阅一份研究报告找出其中的逻辑漏洞、事实不清、论证不充分、潜在偏见或信息遗漏之处。请提供具体、有建设性的批评意见。”), (“human”, “请审阅以下研究报告\n\n{report}\n\n请给出你的批评意见。”) ]) critique_chain critique_prompt | llm response critique_chain.invoke({“report”: state[“research_report”]}) critique_content response.content # 更新状态存入批评意见并指示下一步去找总结员 return { “critique”: critique_content, “next”: “summarizer” }批评家只关注报告内容它不需要再看原始问题。它的输出是针对报告的“挑刺”意见。总结员节点它的任务是综合原始问题、研究报告和批评意见生成一份最终的、高质量的摘要。def summarize_node(state: AgentState) - AgentState: “”“总结员Agent综合所有信息生成最终摘要。”“” from langchain_core.prompts import ChatPromptTemplate summarize_prompt ChatPromptTemplate.from_messages([ (“system”, “你是一位资深的总结员。请基于原始问题、初步研究报告以及批评家的意见合成一份最终的、高质量的摘要。这份摘要应该吸收批评中的合理建议修正报告的不足做到精炼、准确、全面。直接输出最终摘要即可。”), (“human”, “原始问题{question}\n\n初步研究报告{report}\n\n批评意见{critique}\n\n请生成最终摘要”) ]) summarize_chain summarize_prompt | llm response summarize_chain.invoke({ “question”: state[“input”], “report”: state[“research_report”], “critique”: state[“critique”] }) summary_content response.content # 更新状态存入最终摘要并指示工作流结束 return { “final_summary”: summary_content, “next”: “end” }总结员是工作流的最后一环它拥有最全面的信息视图因此其产出的质量理论上应该高于研究员的初版报告。“next”: “end”标志着工作流正常终点。5. 构建与编译LangGraph工作流部门和员工都准备好了现在需要一位“项目经理”来绘制流程图并把大家组织起来。这个项目经理就是StateGraph。from langgraph.graph import StateGraph, END # 1. 初始化一个状态图并指定我们定义的状态类型 workflow StateGraph(AgentState) # 2. 添加三个节点将我们之前定义的函数注册到图中 workflow.add_node(“researcher”, research_node) workflow.add_node(“critic”, critique_node) workflow.add_node(“summarizer”, summarize_node)这里END是LangGraph提供的一个特殊节点代表工作流的终点。接下来设置工作流的起点和边连接线。# 3. 设置入口点工作流从“研究员”节点开始 workflow.set_entry_point(“researcher”) # 4. 添加边定义节点之间的流转关系 workflow.add_edge(“researcher”, “critic”) # 研究员完成后无条件前往批评家 workflow.add_edge(“critic”, “summarizer”) # 批评家完成后无条件前往总结员 workflow.add_edge(“summarizer”, END) # 总结员完成后工作流结束我们的流程是线性的研究员 - 批评家 - 总结员 - END。这是一个最简单的链式结构非常适合入门理解。在更复杂的场景中你可以使用add_conditional_edges来创建分支比如让批评家判断报告是否合格不合格则打回给研究员重写。最后将这张流程图“编译”成一个可执行的应用。# 5. 编译图生成可执行的应用 app workflow.compile()这个app对象就是我们最终得到的多Agent协同系统。你可以把它想象成一个封装好的函数只要输入初始状态它就会按照我们设计好的流程自动运行下去。6. 运行示例与结果分析看AI团队如何工作让我们用一个具体的问题来测试这个系统。比如我们想了解“可再生能源如太阳能和风能当前面临的主要挑战是什么”# 定义初始状态 initial_state: AgentState { “input”: “可再生能源如太阳能和风能当前面临的主要挑战是什么”, “research_report”: “”, # 初始为空 “critique”: “”, “final_summary”: “”, “next”: “researcher” # 设置起点 } # 运行工作流 final_state app.invoke(initial_state)app.invoke()方法会启动工作流并返回执行完毕后的最终状态。我们可以打印出每个环节的产出。研究员输出示例研究报告指出可再生能源主要面临四大挑战1间歇性与不稳定性太阳能和风能依赖天气发电量波动大对电网稳定性构成压力。2储能技术瓶颈大规模、低成本、长寿命的储能方案如电池尚不成熟难以平抑波动。3基础设施与电网升级现有电网是为集中式、可调度的化石能源设计的需要巨额投资改造以适应分布式可再生能源。4经济与政策因素尽管成本下降但初始投资高且受政策补贴波动影响大。研究员给出了一个结构清晰、要点明确的报告符合其角色设定。批评家输出示例批评意见报告框架清晰但存在以下不足1深度不够仅列举了挑战未深入分析每个挑战背后的技术或经济根源例如储能瓶颈具体是能量密度、成本还是材料问题。2缺乏数据支撑没有引用任何具体数据如平准化度电成本LCOE对比、电网消纳比例等来量化挑战的严重性。3遗漏关键点未提及土地资源争夺、制造业供应链如光伏硅料、风机叶片的瓶颈以及环境影响如风机对鸟类、光伏板回收问题。4解决方案提及过少挑战应对措施一带而过显得悲观。批评家非常“尽职”它没有简单夸奖而是尖锐地指出了报告在深度、数据、全面性和解决方案上的缺失。这正是我们想要的“挑刺”效果。总结员输出示例最终摘要可再生能源太阳能、风能的发展主要受制于四大核心挑战。首先其固有的间歇性与不稳定性对电网实时平衡提出高技术要求这源于自然资源的不可控性。其次储能技术成为关键瓶颈当前电池技术在成本、规模及寿命上尚无法完全满足电网级长时间尺度调峰需求。第三传统电网基础设施是基于集中式、可调度电源设计的向适应高比例分布式、波动性电源的智能电网转型需要巨额投资与时间。最后经济性与政策环境仍存不确定性初始投资成本高且严重依赖政策补贴的连续性。 此外还需关注土地资源竞争、全球供应链脆弱性如硅料、稀土以及全生命周期环境影响设备制造与回收等衍生问题。克服这些挑战需多管齐下包括持续技术创新以降低储能成本、推动电网智能化改造、制定长期稳定的政策框架以及发展循环经济以应对环境压力。总结员的输出明显提升了。它吸收了批评家的意见结构上更严谨将挑战归纳为“核心”与“衍生”内容上更深入点出了“电网级长时间尺度调峰”、“智能电网转型”、“全生命周期环境影响”等具体概念表述上也更加专业和精准。最终摘要的质量显著优于最初的研报。通过这个运行示例你可以清晰地看到信息如何在三个Agent间流动和迭代原始问题 - 初步答案 - 批判性反馈 - 融合优化的最终答案。这正是多Agent协同的核心价值通过角色分工和流程编排实现单一个体难以达到的思考深度和输出质量。7. 关键配置解析与实战避坑指南虽然代码不长但里面有几个关键配置点和容易踩坑的地方值得单独拿出来说说。1. 温度参数Temperature的权衡 我们在初始化LLM时设置了temperature0.7。这个值对于创造性任务如研究、总结比较友好能产生更多样的表达。但对于批评家角色你可能希望它的输出更稳定、更聚焦于逻辑本身。一个更精细的做法是为不同节点使用不同的LLM配置。例如creative_llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.8) # 用于研究员和总结员 strict_llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0.1) # 用于批评家然后在各自的节点函数里使用对应的LLM。这能更好地匹配不同角色的特性。2. 状态更新的合并策略 注意我们节点函数返回的是字典如{“research_report”: content, “next”: “critic”}。LangGraph默认会使用update_state函数将这个字典合并到全局状态中。这意味着你只需要返回发生变化的字段。这是一种非常简洁的设计。但务必确保返回字典中的键名与AgentState中定义的字段名完全一致否则会导致运行时错误。3. 提示词工程是灵魂 多Agent系统的效果90%取决于提示词写得好不好。几个原则角色定义要清晰像“严谨的研究员”、“苛刻的批评家”、“资深的总结员”这些角色锚定词非常重要。任务指令要具体避免“写得好一点”这种模糊要求。要用“提供一份详细、客观、信息丰富的初步报告”、“找出逻辑漏洞、事实不清…”、“合成一份最终的、高质量的摘要”这样明确的指令。提供上下文给总结员的提示词中明确列出了它可用的所有材料原始问题、报告、批评这是它能进行综合提升的基础。迭代优化实际运行后根据输出结果反复调整提示词。比如如果批评家总是说得太笼统就在提示词里加上“请提供具体、有建设性的批评意见并举例说明”。4. 错误处理与超时控制 生产环境中网络调用可能失败LLM可能超时。目前的简单示例没有包含错误处理。一个健壮的系统应该考虑在节点函数内使用try…except包裹LLM调用。利用LangGraph的检查点Checkpointing和中断Interruption机制实现工作流的暂停、恢复和错误状态处理。这对于长时运行的任务至关重要。为app.invoke()设置超时参数。5. 调试与可视化 LangGraph提供了强大的调试和可视化工具。你可以通过app.get_graph().draw_mermaid()生成Mermaid图代码粘贴到支持Mermaid的编辑器如Typora、GitHub Markdown中就能看到工作流的可视化图形这对于理解复杂流程非常有帮助。此外在开发时可以在节点函数内打印日志跟踪状态的演变过程。8. 从最小示例到实际应用扩展思路与进阶玩法这个“研究员-批评家-总结员”的三角结构是一个强大的模式但LangGraph的能力远不止于此。理解了基础我们可以探索更多可能性。1. 引入循环与条件分支 这是让工作流“智能”起来的关键。比如我们可以修改流程让批评家来决定报告是否通过from langgraph.graph import END def should_loop_back(state: AgentState) - Literal[“researcher”, “summarizer”]: “”“根据批评意见的严厉程度决定是打回重写还是继续总结。”“” critique state[“critique”] # 一个简单的启发式规则如果批评意见中包含“严重不足”、“重写”等词则打回 if any(word in critique.lower() for word in [“严重不足”, “重写”, “完全不行”]): return “researcher” # 返回研究员节点重新研究 else: return “summarizer” # 前往总结员节点 # 在构建图时使用条件边 workflow.add_conditional_edges( “critic”, # 从批评家节点出发 should_loop_back, # 条件判断函数 { “researcher”: “researcher”, # 如果返回“researcher”则跳转到研究员节点 “summarizer”: “summarizer” # 如果返回“summarizer”则跳转到总结员节点 } ) # 注意此时需要删除原来的 workflow.add_edge(“critic”, “summarizer”)这样就实现了一个简单的质量检查循环直到批评家基本满意才会进入总结阶段。2. 并行执行与聚合 有些任务可以并行处理以提高效率。例如研究员可以同时生成报告的“技术部分”和“市场部分”然后由另一个节点聚合。这需要用到LangGraph的多线程map和reduce节点或者通过创建子图来实现。3. 集成外部工具与知识 真正的Agent不能只靠LLM空想。我们可以让研究员节点具备联网搜索能力通过TavilySearchResults等工具让总结员能够查询内部知识库。只需在对应的节点函数中将工具调用链集成到LangChain的Chain中即可。例如在research_node中可以构建一个Search - LLM的序列链。4. 构建更复杂的多轮对话Agent 这个示例是单向流水线。你可以利用LangGraph的状态管理构建一个能与用户进行多轮交互的对话Agent。将用户的每次输入追加到状态的历史记录中并设计一个“路由节点”来决定是调用工具、查询知识库还是直接由LLM回复。5. 与LangChain的生态结合 LangGraph完美兼容LangChain的其它组件。你可以轻松地将Runnable、Tools、Retrievers检索器、Memory记忆等融入你的图中。例如为工作流添加一个向量数据库检索节点让研究员在动笔前先查找相关资料。这个100行代码的示例就像给你展示了一辆汽车的发动机、底盘和车轮是如何组装起来并跑起来的。基于这个最小可运行框架你可以根据自己的业务需求更换更强大的“发动机”模型添加“变速箱”条件逻辑、“座椅”用户界面和“导航系统”外部工具造出一辆能应对复杂路况的专属智能车。多Agent协同的世界大门就从这里打开了。