1. 项目概述从单兵作战到团队协作的Agent进化如果你已经玩过LangChain搭建过几个简单的AI应用可能会发现一个瓶颈单个Agent智能体的能力终究是有限的。它就像一个全能的“瑞士军刀”虽然功能多但面对复杂任务时要么处理得不够专业要么流程冗长低效。比如你想让AI帮你分析一份财报并生成投资建议一个Agent可能既要懂财务数据提取又要懂市场分析还要会写报告结果往往是每个环节都做得马马虎虎。这正是“多Agent系统”要解决的问题。它不再是让一个AI“大包大揽”而是组建一个分工明确的AI团队。在这个团队里有的Agent擅长信息检索研究员有的精于数据分析分析师有的专攻文案撰写编辑。它们通过一套精密的协作机制像流水线一样高效完成任务。而LangGraph正是构建和管理这个AI团队的“超级大脑”和“指挥中心”。我最初接触多Agent概念时觉得它很“科幻”直到用LangGraph真正搭建出一个能自动处理用户咨询、分派任务、汇总结果的小系统后才体会到其威力。它不仅仅是技术的堆砌更是一种工程思维的转变——从设计单个“超人”AI转向设计一个高效、稳定、可扩展的“组织架构”。本篇我们就深入LangGraph的核心看看如何让多个Agent学会分工协作构建一个真正智能的多Agent系统。2. 核心设计LangGraph如何为Agent赋予“组织智慧”在深入代码之前我们必须理解LangGraph设计多Agent系统的核心哲学。它不像传统编程那样用if-else硬编码流程而是借鉴了图计算和状态机的思想将协作过程抽象为一张“工作流图”。2.1 状态State团队的共享工作台与记忆中枢你可以把State想象成一个团队项目的共享白板或云端协作文档。所有Agent都能看到它并在上面读写信息。这个State不是一个简单的变量而是一个结构化的字典TypedDict定义了整个系统运行期间需要维护的所有数据。为什么状态管理如此关键因为它解决了多Agent协作中最核心的问题信息同步与上下文传递。没有统一的状态Agent之间就是信息孤岛A处理完的结果B无法获取协作无从谈起。在LangGraph中定义一个状态通常如下所示我们以一个客服工单处理系统为例from typing import TypedDict, Annotated from typing_extensions import TypedDict import operator class AgentState(TypedDict): # 用户原始输入的问题 messages: Annotated[list, operator.add] # 经过意图识别后的分类如“技术咨询”、“账单问题”、“产品反馈” intent: str # 根据意图路由到的下一个Agent名称 next_agent: str # 各个Agent处理后的中间结果key为Agent名 processed_results: dict # 最终要返回给用户的答案 final_answer: str这里有几个设计要点Annotated[list, operator.add]这是LangGraph的一个精妙设计。它声明messages是一个列表并且当多个节点Agent向其中添加消息时使用operator.add即列表的操作来自动合并。这确保了对话历史的完整记录。明确的职责字段intent和next_agent是控制流的关键。processed_results用于收集阶段性成果避免数据丢失。final_answer是聚合输出的目标。可扩展性你可以根据业务需要任意添加字段如user_id用户会话、priority任务优先级、required_tools所需工具列表等。这个State对象会随着工作流的执行在各个Agent节点间流转和更新是维系整个团队的“记忆纽带”。2.2 节点Node团队中的专业成员节点是工作流中实际执行操作的单元通常就是一个Agent。每个节点是一个函数它接收当前的State执行逻辑如调用大模型、使用工具然后返回更新后的State。一个典型的节点函数结构如下def technical_support_agent(state: AgentState): 技术客服Agent专门处理技术问题 # 1. 从状态中获取最新用户消息 human_message state[“messages”][-1] # 2. 构建该Agent专属的提示词System Prompt赋予其专业角色 system_prompt “””你是一名专业的技术支持工程师擅长解决软件安装、API使用、错误调试等问题。 请根据用户提问提供清晰、逐步的解决方案。如果问题需要更高级的工程师介入请明确指出。“”” # 3. 调用大模型这里以LangChain的ChatModel为例 llm ChatOpenAI(model“gpt-4”) messages [ SystemMessage(contentsystem_prompt), HumanMessage(contenthuman_message[“content”]) ] response llm.invoke(messages) # 4. 将处理结果存入状态 new_state { “processed_results”: { **state.get(“processed_results”, {}), “technical_support”: response.content }, “messages”: state[“messages”] [AIMessage(contentresponse.content)] } # 5. 返回更新后的状态只返回需要更新的部分LangGraph会自动合并 return new_state关键技巧在返回new_state时我们通常只返回发生了变化的字段。LangGraph的底层机制会自动将其与旧状态合并。这符合函数式编程的思想让状态变更更可控、可预测。2.3 边Edge团队协作的规则与流程边定义了节点之间的流转条件也就是“接下来谁该干活”。这是实现动态路由和复杂逻辑的核心。LangGraph提供了两种主要方式1. 条件边Conditional Edge这是最常用的路由机制。通过一个路由函数根据当前State的内容决定下一个节点是谁。def router(state: AgentState) - str: “”“核心路由逻辑根据识别出的意图分配任务”“” intent state.get(“intent”) if intent “technical_support”: return “call_technical_agent” # 指向技术客服节点 elif intent “billing”: return “call_billing_agent” # 指向财务客服节点 elif intent “feedback”: return “call_feedback_collector” # 指向反馈收集节点 else: return “call_general_agent” # 指向通用客服节点2. 入口与出口边入口边通过add_edge(start, node_id)设置决定工作流从哪个节点开始。出口边通过add_edge(node_id, END)设置标记某个节点是工作流的终点。当流程到达END时整个图执行完毕。通过节点和边的组合你就能绘制出一张完整的AI团队协作流程图。LangGraph的StateGraph类就是用来构建这张图的画布。3. 构建实战从零搭建一个多Agent客服系统理论说得再多不如动手搭一个。我们以构建一个智能客服系统为例它需要处理三类问题技术咨询、账单查询和产品反馈。系统会自动识别用户意图并路由给对应的专家Agent处理。3.1 第一步定义团队结构与工作流我们的团队由四个成员和一个“调度员”组成意图识别AgentDispatcher负责分析用户问题分类意图。技术客服Agent解决技术问题。财务客服Agent处理账单和支付问题。反馈收集Agent收集并结构化用户反馈。汇总Agent可选如果需要整合多个Agent的结果可以增加此角色。工作流设计如下用户输入 - 意图识别Agent - (路由) - 专业Agent处理 - 返回结果3.2 第二步实现各个Agent节点首先实现我们的“调度员”——意图识别Agent。它的提示词设计至关重要需要能准确区分用户意图。def intent_classifier_agent(state: AgentState): “”“意图识别节点”“” llm ChatOpenAI(model“gpt-3.5-turbo”, temperature0) # 意图识别要求精确温度调低 classifier_prompt “”” 你是一个意图分类器。请将用户的输入严格分类为以下类别之一 - technical_support: 涉及软件错误、安装问题、API使用、技术故障等。 - billing: 涉及费用、账单、扣款、订阅、价格等。 - feedback: 涉及产品建议、使用体验、功能请求、表扬或投诉等。 - general: 不属于以上任何一类的普通问候或咨询。 只输出类别名称不要有任何其他解释。 用户输入{user_input} “””.format(user_inputstate[“messages”][-1][“content”]) response llm.invoke([HumanMessage(contentclassifier_prompt)]) intent response.content.strip().lower() # 将识别出的意图更新到状态中供路由函数使用 return {“intent”: intent}接下来实现三个专业Agent。它们的结构类似但拥有不同的“专业领域知识”通过System Prompt体现。def billing_agent(state: AgentState): “”“财务客服Agent”“” llm ChatOpenAI(model“gpt-4”) # 模拟一个查询用户账单的工具函数 def query_billing(user_query: str) - str: # 这里应连接真实的数据库或API返回模拟数据 return “尊敬的用户您本月的账单总额为99元其中基础套餐89元增值服务10元。到期日为2023-12-01。” tools [StructuredTool.from_function(funcquery_billing)] agent create_openai_tools_agent(llm, tools) # 执行Agent它会自动判断是否需要调用工具 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) result agent_executor.invoke({“input”: state[“messages”][-1][“content”]}) return { “processed_results”: {“billing”: result[“output”]}, “final_answer”: result[“output”], # 对于单一路径结果可直接作为最终答案 “messages”: state[“messages”] [AIMessage(contentresult[“output”])] }注意在实际项目中technical_support_agent和feedback_collector_agent的实现会接入不同的工具和知识库。例如技术Agent可以连接知识库文档反馈Agent可以结构化数据并存入数据库。3.3 第三步组装工作流图这是LangGraph最核心的一步我们将各个节点和边组装起来。from langgraph.graph import StateGraph, END # 1. 初始化一个状态图并指定我们定义的状态类型 workflow StateGraph(AgentState) # 2. 添加节点即我们定义的各个Agent函数 workflow.add_node(“intent_classifier”, intent_classifier_agent) workflow.add_node(“technical_support”, technical_support_agent) workflow.add_node(“billing”, billing_agent) workflow.add_node(“feedback_collector”, feedback_collector_agent) workflow.add_node(“general_agent”, general_agent) # 通用兜底Agent # 3. 设置入口点所有对话都从意图识别开始 workflow.set_entry_point(“intent_classifier”) # 4. 定义路由逻辑函数 def route_by_intent(state: AgentState): intent state.get(“intent”, “general”) if intent “technical_support”: return “technical_support” elif intent “billing”: return “billing” elif intent “feedback”: return “feedback_collector” else: return “general_agent” # 5. 添加条件边从意图识别节点根据路由结果指向不同专业节点 workflow.add_conditional_edges( “intent_classifier”, # 源节点 route_by_intent, # 路由决策函数 { “technical_support”: “technical_support”, “billing”: “billing”, “feedback_collector”: “feedback_collector”, “general_agent”: “general_agent” } ) # 6. 为每个专业Agent添加出口边指向结束点 workflow.add_edge(“technical_support”, END) workflow.add_edge(“billing”, END) workflow.add_edge(“feedback_collector”, END) workflow.add_edge(“general_agent”, END) # 7. 编译图使其可执行 app workflow.compile()至此一个具备基本路由功能的多Agent客服系统骨架就搭建完成了。你可以通过app.invoke()来运行它。3.4 第四步运行与调试运行这个系统非常简单只需传入初始状态。# 初始化状态包含用户的第一条消息 initial_state { “messages”: [{“role”: “user”, “content”: “我的账单好像扣错了能查一下吗”}], “intent”: “”, # 初始为空由意图识别节点填充 “next_agent”: “”, “processed_results”: {}, “final_answer”: “” } # 执行图 final_state app.invoke(initial_state) print(final_state[“final_answer”]) # 输出可能为“尊敬的用户您本月的账单总额为99元...”执行过程是自动的intent_classifier节点会判断用户意图为billing路由函数将其指向billing节点财务客服Agent调用工具查询账单并生成回复流程结束。4. 高级模式与核心技巧打造更智能的协作网络基础的单次路由只能解决简单问题。现实中任务往往需要多个Agent接力或会商。LangGraph通过**子图Subgraph和循环Cycle**来支持更复杂的协作模式。4.1 模式一顺序接力Sequential Chain有些任务像流水线必须按顺序执行。例如“翻译-总结-润色”任务。# 在图中添加顺序边即可 workflow.add_edge(“translator_agent”, “summarizer_agent”) workflow.add_edge(“summarizer_agent”, “polisher_agent”) workflow.add_edge(“polisher_agent”, END)心得在顺序接力中状态的设计要格外注意。每个节点应只修改自己负责的字段如translated_text,summary,polished_text避免覆盖上游数据。使用Annotated[dict, operator.add]或Annotated[list, operator.add]能很好地合并字典或列表。4.2 模式二循环迭代Loop until Done对于需要反复修正、审核的任务可以使用循环。例如一个代码生成Agent需要经过一个代码审查Agent的多次审核直到通过为止。def code_review_router(state: AgentState): “”“根据审查结果决定是继续循环还是结束”“” if state.get(“review_status”) “approved”: return END else: return “code_generator” # 返回代码生成节点重新生成 workflow.add_conditional_edges(“code_reviewer”, code_review_router)这里code_generator和code_reviewer两个节点会形成一个循环直到review_status变为approved。必须设置明确的终止条件否则会造成无限循环。4.3 模式三并行处理与聚合Fork-Join这是多Agent系统威力最大的模式。例如分析一家公司可以同时派出“市场分析Agent”、“财务分析Agent”和“竞品分析Agent”并行工作最后由一个“报告合成Agent”汇总。 LangGraph本身不直接提供“并行”节点但可以通过设计状态和路由来模拟Fork分发一个路由节点根据任务列表将状态复制多份或设置一个tasks队列。并行执行实际上在单线程中仍是顺序执行不同Agent但逻辑上是并行的。如果需要真正的并发可以在每个节点函数内部使用异步或线程池。Join聚合所有并行任务完成后路由到一个聚合节点合并所有processed_results。实现提示可以在状态中设置一个tasks列表记录所有需要并行处理的任务项。每个处理节点完成后将自己标记为完成。聚合节点检查所有任务是否都已完成然后进行汇总。4.4 核心技巧状态设计的艺术状态设计是多Agent系统的基石几个经验分享扁平化 vs 嵌套化尽量保持状态扁平避免过深的嵌套字典否则在更新和访问时容易出错。可以为不同的Agent组设计不同的顶级字段。不可变思想在节点函数中尽量不对传入的state直接修改而是创建包含更新字段的新字典返回。这使数据流更清晰便于调试和回溯。使用Annotated进行自动合并对于列表如消息历史和字典如结果收集强烈推荐使用Annotated声明合并操作符让LangGraph自动处理合并逻辑简化代码。预留调试字段可以添加debug_trace或execution_path列表记录每个节点的执行顺序和耗时对于后期优化性能至关重要。5. 避坑指南与性能优化在实际部署中你会遇到许多文档里没写的坑。以下是我从多个项目中总结出的核心经验。5.1 常见问题与排查问题现象可能原因排查步骤与解决方案图编译失败提示状态字段错误1.State的TypedDict定义与节点返回值不匹配。2. 节点返回了未在State中声明的字段。1. 检查TypedDict的所有字段名和类型。2. 确保每个节点返回的字典键都在State定义中。3. 使用print或日志输出每个节点返回的state片段进行比对。路由失灵总是走到默认或错误分支1. 路由函数route_by_intent逻辑有误。2. 上游节点如意图识别写入状态的字段名或值与路由函数读取的不一致。3. 条件边的映射字典dict的key与路由函数返回值不匹配。1. 在路由函数开头打印state确认输入是否正确。2. 检查意图识别Agent的输出是否准确可能大模型没按指令输出。3. 确保映射字典的key与路由函数所有可能的返回值完全一致。无限循环程序不结束1. 循环条件永远无法满足如review_status永远不是approved。2. 缺少指向END节点的边。1. 在循环体内添加计数器如state[“loop_count”]超过阈值则强制跳出并报错。2. 检查图结构确保至少有一条路径能到达END。使用workflow.get_graph().draw_mermaid()可视化图形检查。多Agent协作时上下文混乱1. 每个Agent的提示词System Prompt角色定义不清导致行为越界。2.messages历史过长包含无关对话干扰当前Agent。1. 为每个Agent设计独特、强约束的System Prompt例如“你只负责财务问题对于技术问题请告知用户将转接专员”。2. 在状态传递时可以只传递最近几轮对话或关键摘要而非全部历史。设计一个summarizer节点来压缩历史。执行速度慢1. 串行调用多个大模型耗时叠加。2. 单个Agent的提示词过于复杂或调用了慢速工具如网络请求。1.对于无依赖的节点考虑并行化。虽然LangGraph图是顺序执行但可以在节点函数内部使用asyncio.gather并发调用多个LLM或工具。2. 优化提示词减少不必要的上下文。对工具调用做超时和缓存。5.2 性能与成本优化策略多Agent系统由于多次调用LLM成本和延迟容易成为瓶颈。分层模型策略并非所有节点都需要GPT-4。意图识别、路由判断等对创造力要求低的任务可以使用更便宜、更快的模型如gpt-3.5-turbo甚至小型开源模型。只有核心的内容生成、复杂推理节点才使用强模型。缓存与记忆对于相同或相似的输入结果应该被缓存。可以利用LangChain的RedisSemanticCache或简单的LRUCache。对于用户会话将历史记忆向量化存储每次只检索相关片段而非传入全部历史。流式输出与异步如果前端支持对于长流程任务可以采用流式输出让用户感知到进度。同时将图中所有节点的调用都改为异步async/await可以大幅提升I/O密集型任务的吞吐量。图的编译与预热app workflow.compile()编译后的图对象可以复用。在服务启动时预编译并保持长连接避免每次请求都重新编译。5.3 可观测性与监控系统复杂了没有监控就是“黑盒”。结构化日志在每个节点的开始和结束记录日志包含node_id,input_state_snapshot,output_state_snapshot,latency等。链路追踪Trace利用LangSmith或自定义工具为每次完整的图执行生成一个trace_id串联所有节点的调用便于复盘和调试。关键指标监控每个节点的调用次数、平均响应时间、错误率。监控每次图执行的总耗时和总Token消耗用于成本分析和性能告警。6. 超越客服多Agent系统的广阔应用场景掌握了基础架构后你会发现这套范式能应用到无数场景。它本质是一套任务分解、动态调度、结果聚合的通用框架。智能研发助手ProductManagerAgent解析模糊的需求文档生成产品特性列表。ArchitectAgent根据特性列表设计系统架构图和技术栈。CoderAgent根据架构生成模块代码。CodeReviewerAgent检查代码提出修改意见循环。TestWriterAgent生成单元测试。DocumentAgent生成API文档。个性化学习教练DiagnosticAgent通过对话评估学员的知识水平。PlannerAgent生成个性化的学习路径。ContentRetrieverAgent从知识库中提取相关资料。QuizGeneratorAgent生成练习题。ExplainerAgent对错题进行针对性讲解。自动化数据分析报告QueryInterpreterAgent理解用户的自然语言分析需求。SQLGeneratorAgent生成SQL查询语句。DataFetcherAgent执行查询获取数据。AnalystAgent进行数据分析和可视化图表生成。NarratorAgent用文字描述分析洞察形成报告。在每个场景中LangGraph扮演的都是那个“看不见的指挥家”它不直接生产内容但确保了专业的人Agent在正确的时间以正确的顺序处理正确的事情。这种将复杂智能任务“流水线化”、“组织化”的能力才是多Agent系统以及LangGraph带给我们的真正价值。