1. 从单体Agent到协同作战为什么需要嵌套架构如果你最近在折腾AI应用尤其是想搞点复杂的自动化流程大概率已经玩过LangChain或者LangGraph了。一个简单的Agent比如一个能联网查天气、然后帮你规划行程的助手用LangChain的create_react_agent可能半小时就搭出来了。但当你真的想把想法落地比如做一个能自动分析市场报告、生成策略、并执行邮件分发的系统时你会发现单个“全能”Agent很快就变得臃肿不堪逻辑混乱而且一出错整个流程全崩。这就是我当初踩的第一个大坑。我试图让一个Agent既当分析师又当写手还兼职调度员。结果就是提示词Prompt写得像一篇论文思维链Chain-of-Thought长得自己都看不懂最后它经常在分析到一半时突然开始写邮件抬头逻辑彻底错乱。这让我意识到智能体的能力边界和单一职责原则在AI应用开发里同样至关重要。一个Agent不应该什么都懂什么都做而应该像一支训练有素的团队各司其职高效协作。于是嵌套Agent架构就成了解决这个问题的自然选择。它的核心思想很简单“大任务拆小小任务分包层层汇报统一调度”。一个顶层的“经理”Agent我们称之为Orchestrator负责理解用户的终极目标并将其分解成一系列子任务。然后它将这些子任务分派给更专业的“专家”AgentSpecialist去执行。这些专家Agent可能自己也无法独立完成它们会进一步将任务拆解调用最底层的“工具人”AgentWorker或直接使用工具Tools。最终结果层层汇总由顶层Agent整理后交付给用户。这种架构的好处是显而易见的模块化与可维护性每个Agent功能单一逻辑清晰提示词可以精心打磨。修改一个分析逻辑不会影响到邮件发送的模块。容错与鲁棒性一个子任务失败比如网络查询超时通常不会导致整个系统崩溃顶层调度器可以选择重试、跳过或启用备用方案。能力复用一个写好了的“数据清洗专家”Agent可以被多个不同的业务流程调用。清晰的思维过程每一步决策和结果都在明确的层级间传递非常利于调试和追溯你总能知道是哪个环节、哪个Agent出了问题。而LangGraph正是实现这种复杂、有状态、多Agent协作工作流的绝佳框架。它用“图”Graph的模型来定义工作流节点Node可以是Agent、工具或者任何函数边Edge定义了控制流。通过其强大的状态管理State和条件边Conditional Edge我们能轻松构建出这种三层甚至更多层的嵌套决策系统。接下来我就手把手带你从零开始用LangGraph搭建一个实用的三层嵌套Agent架构。2. 实战场景定义一个智能内容运营助手为了不让教程过于抽象我们贯穿全文用一个具体的、可扩展的实战场景来驱动构建一个智能内容运营助手。它的核心任务是用户输入一个核心话题例如“LangGraph的最新特性”助手能够自动完成从信息搜集、内容加工到分发的完整流程。具体来说这个三层架构将这样工作第一层Orchestrator协调器。它接收用户请求理解意图是“内容运营”。它的职责是规划工作流先调研再创作最后分发。它会将任务分解为“调研Agent请搜集关于LangGraph最新特性的资料”“创作Agent请基于调研结果撰写一篇博客草稿”“分发Agent请将博客草稿制作成社交媒体帖子并安排发布”。第二层Specialist专家Agent。这一层包含多个专业Agent。调研专家Research Specialist接收Orchestrator的“调研”指令。它自己可能不直接搜索而是调度更底层的工具。创作专家Writing Specialist接收Orchestrator的“创作”指令。它负责组织内容结构、文风润色。分发专家Distribution Specialist接收Orchestrator的“分发”指令。它需要理解不同平台的特点。第三层Worker/Tools执行层。这是实际动手的一层。工具Tools如DuckDuckGoSearchRun进行网页搜索arxiv工具查询论文requests工具访问特定API获取数据。简单Worker Agent对于一些需要简单逻辑判断后再使用工具的操作可以设立Worker Agent。例如一个“链接提取与筛选Worker”它接收调研专家传来的一堆原始文本从中提取链接并过滤掉广告或不相关链接再将干净的链接列表返回。整个系统的数据流和控制流将通过LangGraph进行编排。下面我们就开始环境搭建和核心Agent的构建。3. 环境搭建与智能体核心构造3.1 初始化项目与安装依赖首先确保你有一个干净的Python环境3.10以上版本推荐。我们使用uv或pip进行包管理这里以pip为例。# 创建项目目录并进入 mkdir nested_agent_demo cd nested_agent_demo # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain langchain-openai langchain-community关键依赖说明langgraph: 本次的核心用于构建工作流图。langchain: 提供了构建Agent所需的大量基础组件如提示模板、记忆、输出解析器等。langchain-openai: 官方维护的OpenAI模型集成我们将使用GPT-4o或GPT-3.5-turbo作为Agent的“大脑”。langchain-community: 包含大量第三方工具和集成如我们即将用到的搜索工具。接下来你需要准备一个OpenAI的API Key。可以将其设置为环境变量export OPENAI_API_KEYyour-api-key-here # 或者在代码中直接设置不推荐用于生产环境3.2 构建基础智能体LangChain Agent as a Node在LangGraph中一切皆节点Node。一个Agent本质上就是一个接收状态、执行动作、并返回新状态的函数。LangChain的create_react_agent函数能快速创建一个基于ReAct范式的Agent我们可以把它包装成LangGraph的节点。我们先从构建最底层的工具调用AgentWorker层开始。假设我们创建一个专门用于“精确搜索”的Worker。import os from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.prompts import PromptTemplate # 1. 初始化大语言模型 llm ChatOpenAI(modelgpt-4o, temperature0) # 使用gpt-4o以获得更好的推理能力temperature设为0保证稳定性 # 2. 定义工具 search_tool DuckDuckGoSearchRun(nameweb_search, description使用此工具在互联网上搜索最新信息。) # 3. 创建Worker Agent的提示词 worker_prompt PromptTemplate.from_template( 你是一个信息检索专家。你的唯一任务是理解用户的信息需求并使用提供的工具进行精准搜索。 用户的需求是{input} 你必须只使用工具来获取信息不要编造答案。 如果一次搜索没有找到满意答案可以分析结果并调整关键词再次搜索。 最终你需要整理并返回清晰、有引用来源的信息摘要。 开始思考 ) # 4. 创建React Agent及其执行器 worker_agent create_react_agent(llmllm, tools[search_tool], promptworker_prompt) worker_agent_executor AgentExecutor(agentworker_agent, tools[search_tool], verboseTrue, handle_parsing_errorsTrue) # 5. 将AgentExecutor包装成LangGraph Node所需的函数 def research_worker_node(state: dict): LangGraph节点函数。接收状态字典调用Worker Agent返回更新后的状态。 状态中应包含input键代表要搜索的查询。 user_query state.get(input, ) if not user_query: return {output: 错误未接收到查询输入。} # 调用Agent执行器 result worker_agent_executor.invoke({input: user_query}) # 将Agent的输出放入状态中假设我们用‘research_result’键来存储 return {research_result: result.get(output, 未找到相关信息。)}这个research_worker_node函数就是一个标准的LangGraph节点。它接收一个状态字典从中取出input调用真正的Agent去执行搜索然后将结果以research_result为键放回状态中。注意这里有一个关键设计点状态State的结构。我们提前规划状态字典里有哪些字段这就像团队协作的共享白板。在上面的Worker层我们定义了input和research_result。在后续的多层架构中状态会成为各Agent间通信的唯一桥梁因此设计一个清晰、一致的状态结构至关重要。我建议在项目开始时用注释或TypedDict明确状态模式State Schema。4. 设计核心三层图架构与状态流转现在我们来构建核心的三层图。我们将创建三个主要的图Graph每个图代表一层然后它们会嵌套调用。4.1 定义全局状态模式首先我们定义整个工作流共享的状态。这通常用一个TypedDict或简单的字典结构来约定。from typing import TypedDict, Annotated, Union from typing_extensions import TypedDict import operator # 定义整个系统的状态类型 class GraphState(TypedDict): 整个嵌套Agent工作流的共享状态。 # 原始输入与最终输出 original_input: str # 用户原始请求如“写一篇关于LangGraph的博客” final_output: Union[str, None] # 最终给用户的结果 # 第一层 Orchestrator 使用的字段 task_plan: Union[str, None] # Orchestrator分解后的任务计划如“1. 调研 2. 写作 3. 分发” current_step: Union[str, None] # 当前正在执行哪个步骤如“researching” # 第二层 Specialist 使用的字段 research_query: Union[str, None] # 传递给调研专家的具体查询 writing_brief: Union[str, None] # 传递给创作专家的内容简报 distribution_platforms: Union[list, None] # 需要分发的平台列表 # 第三层 Worker/Tools 产生的结果 research_result: Union[str, None] # 调研专家/Worker的搜索结果 draft_content: Union[str, None] # 创作专家生成的草稿 distribution_ready_posts: Union[dict, None] # 分发专家准备好的各平台帖子内容 # 错误与重试机制 error: Union[str, None] # 记录运行时错误 retry_count: int # 当前步骤重试次数这个GraphState定义了所有可能用到的字段。在实际运行中并非所有字段都会被同时填充但它提供了一个清晰的蓝图。4.2 构建第三层工具执行图Worker Graph第三层是最具体的执行层。我们构建一个简单的“调研工作流”图它只做一件事调用我们之前定义的research_worker_node。from langgraph.graph import StateGraph, END # 创建第三层图调研工作流 worker_builder StateGraph(GraphState) # 添加节点这个节点就是我们之前定义的函数 worker_builder.add_node(research_worker, research_worker_node) # 设置入口点 worker_builder.set_entry_point(research_worker) # 从research_worker节点执行完后直接结束这个子图。 worker_builder.add_edge(research_worker, END) # 编译第三层图 worker_graph worker_builder.compile()这个worker_graph现在是一个可执行的对象。你可以用它来测试worker_graph.invoke({original_input: 测试, research_query: LangGraph 2024 updates})。它会调用Worker Agent进行搜索并将结果存入research_result。4.3 构建第二层专家调度图Specialist Graph第二层的专家Agent其核心职责是“任务规划与调度”。以Research Specialist为例它不直接搜索而是分析Orchestrator给它的指令research_query判断是否需要搜索、如何搜索然后调用第三层的worker_graph最后对结果进行初步加工。# 1. 首先创建Research Specialist Agent一个更智能的规划者 research_specialist_prompt PromptTemplate.from_template( 你是资深调研分析师。你的目标是为内容创作收集充分、可靠的资料。 上级给你的调研主题是{topic} 请思考并完成以下步骤 1. **理解与拆解**这个主题涉及哪些关键子话题哪些是必须查证的核心概念 2. **制定搜索策略**针对每个子话题设计1-2个精准的搜索关键词。优先考虑官方文档、技术博客、权威社区。 3. **调度执行**你将拥有一个强大的搜索工具由下级Worker执行。请根据你的策略生成最终要执行的搜索查询。 4. **结果整合**拿到搜索结果后你需要过滤掉无关信息将不同来源的信息进行交叉验证和整合形成一份结构化的调研摘要。 请开始你的工作。首先输出你的搜索策略和决定执行的最终查询语句。 最终查询语句必须是清晰、可直接用于搜索的一句话。 ) research_specialist_agent create_react_agent(llmllm, tools[], promptresearch_specialist_prompt) research_specialist_executor AgentExecutor(agentresearch_specialist_agent, tools[], verboseTrue) # 2. 定义第二层图的节点函数 def research_specialist_node(state: GraphState): 调研专家节点。它规划搜索调用Worker图然后整合结果。 topic state.get(research_query) if not topic: return {error: Research specialist did not receive a topic.} # 步骤A专家自己进行规划不调用工具纯LLM推理 plan_result research_specialist_executor.invoke({input: f请为以下主题制定调研计划{topic}}) plan plan_result.get(output, ) # 这里一个简单的实现假设专家输出的最后一行就是“最终查询语句” # 在实际项目中你需要用更稳健的方式如Output Parser来提取这个查询 lines plan.strip().split(\n) final_query lines[-1] if lines else topic # 简化处理实际应解析出查询 print(f[Research Specialist] 分析完成。生成查询{final_query}) # 步骤B调用第三层Worker图来执行实际的搜索 # 注意我们通过更新状态然后调用worker_graph来实现嵌套 # 但更标准的LangGraph方式是使用Send和Subscribe这里为清晰起见采用直接调用子图的方式。 worker_result worker_graph.invoke({ **state, # 传入当前状态 input: final_query, # Worker节点需要input字段 }) raw_data worker_result.get(research_result, ) # 步骤C专家整合加工原始结果 integration_prompt f 以下是你要求的搜索查询结果 查询{final_query} 原始结果{raw_data[:2000]}...可能截断 请完成你的第4步整合信息形成一份给内容创作团队的摘要。 摘要需包含核心观点、数据来源、关键引用、以及尚存疑问或需要进一步核实的地方。 integration_result research_specialist_executor.invoke({input: integration_prompt}) final_research_summary integration_result.get(output, raw_data) # 如果整合失败返回原始数据 # 更新状态 return { research_result: final_research_summary, current_step: research_completed, # 通知上层本步骤完成 } # 3. 构建第二层图这里以调研专家为例写作和分发专家结构类似 specialist_builder StateGraph(GraphState) specialist_builder.add_node(conduct_research, research_specialist_node) specialist_builder.set_entry_point(conduct_research) specialist_builder.add_edge(conduct_research, END) research_specialist_graph specialist_builder.compile()这个research_specialist_graph就代表了一个具备规划能力的专家。它展示了第二层Agent的典型模式接收指令 - 自主规划 - 调度底层资源子图/工具- 加工结果 - 返回。4.4 构建第一层总指挥图Orchestrator Graph第一层Orchestrator是大脑中的大脑。它需要理解用户意图分解任务并决定调用哪个专家子图以及处理专家返回的结果决定下一步是继续还是结束。# 1. 创建Orchestrator Agent orchestrator_prompt PromptTemplate.from_template( 你是智能内容运营助手的总指挥。你的目标是将用户的复杂请求分解为有序的、可执行的任务流。 用户请求{request} 已知可用的专家团队有 1. **调研专家**负责信息搜集与验证。 2. **创作专家**负责撰写高质量文案草稿。 3. **分发专家**负责适配不同平台格式并安排发布。 请按以下步骤思考 1. **请求分析**用户的核心需求是什么最终要交付什么 2. **任务分解**要完成这个需求必须经历哪几个阶段例如调研 - 创作 - 分发。每个阶段的具体输入是什么 3. **流程编排**这些阶段的执行顺序是怎样的是否有并行可能例如调研和初步的框架创作可以同时进行。 4. **输出决策**根据当前状态和阶段结果决定下一步调用哪个专家还是结束流程并交付最终结果。 请输出你的任务分解计划和当前应执行的第一步。 输出格式必须严格遵循 计划[你的整体阶段计划如“1.调研2.创作3.分发”] 下一步[下一步要执行的专家名称如“调研专家”] ) orchestrator_agent create_react_agent(llmllm, tools[], promptorchestrator_prompt) orchestrator_executor AgentExecutor(agentorchestrator_agent, tools[], verboseTrue) # 2. 定义Orchestrator的节点函数 def orchestrator_node(state: GraphState): 总指挥节点。分析请求规划任务并设置下一步该谁执行。 request state.get(original_input, ) if not request: return {error: Orchestrator did not receive input.} # 调用Orchestrator Agent进行分析和规划 analysis orchestrator_executor.invoke({input: request}) analysis_output analysis.get(output, ) # 解析输出获取计划和下一步动作这里需要简单的解析逻辑 plan 未解析到计划 next_step 未知 for line in analysis_output.split(\n): if line.startswith(计划): plan line.replace(计划, ).strip() elif line.startswith(下一步): next_step line.replace(下一步, ).strip() # 根据计划更新状态并决定下一个节点 # 我们将“下一步”动作存入状态由条件边Conditional Edge来判断 return { task_plan: plan, current_step: next_step, # 例如 “调研专家” } # 3. 定义路由函数Conditional Edge def route_after_orchestrator(state: GraphState): 根据Orchestrator设定的current_step决定下一步去哪个节点。 next_step state.get(current_step) if next_step 调研专家: return call_research_specialist elif next_step 创作专家: return call_writing_specialist elif next_step 分发专家: return call_distribution_specialist elif next_step 结束: return compile_final_output else: # 如果无法识别返回错误节点或重新规划 return handle_error # 4. 定义调用各专家子图的节点函数 def call_research_specialist_node(state: GraphState): 调用调研专家子图 # 准备查询这里简化处理将原始输入作为查询。实际中Orchestrator应生成更精确的查询。 query state.get(original_input, ) # 调用第二层图 result_state research_specialist_graph.invoke({**state, research_query: query}) # 合并结果到当前状态 new_state {**state, **result_state} # 执行完后将控制权交回给Orchestrator让它决定下一步 new_state[current_step] orchestrator_review return new_state # 5. 构建完整的第一层图主图 main_builder StateGraph(GraphState) # 添加节点 main_builder.add_node(orchestrator, orchestrator_node) main_builder.add_node(call_research_specialist, call_research_specialist_node) # 这里省略了 call_writing_specialist_node 和 call_distribution_specialist_node 的定义和添加其结构与call_research_specialist_node类似。 main_builder.add_node(compile_final_output, lambda state: {final_output: 整合所有结果形成最终回复。此处为示意}) main_builder.add_node(handle_error, lambda state: {error: f路由失败未知步骤{state.get(current_step)}}) # 设置入口点 main_builder.set_entry_point(orchestrator) # 添加边包括条件边 main_builder.add_edge(handle_error, END) # 错误处理节点直接结束 main_builder.add_edge(compile_final_output, END) # 最终编译节点直接结束 # 从orchestrator节点出来后根据条件路由 main_builder.add_conditional_edges( orchestrator, route_after_orchestrator, { call_research_specialist: call_research_specialist, call_writing_specialist: call_writing_specialist, # 假设已定义 call_distribution_specialist: call_distribution_specialist, # 假设已定义 compile_final_output: compile_final_output, handle_error: handle_error, } ) # 专家节点执行完后都回到orchestrator进行下一轮决策 main_builder.add_edge(call_research_specialist, orchestrator) # ... 其他专家节点同理 # 编译主图 main_graph main_builder.compile()至此一个三层嵌套Agent架构的核心骨架就搭建完成了。main_graph是我们的总入口。它启动后由orchestrator节点分析任务然后通过条件边route_after_orchestrator动态决定调用哪个专家子图。专家子图在执行过程中又会调用更底层的Worker图或工具。执行完毕后流程返回orchestrator节点由其根据当前结果例如research_result是否已就绪和初始计划决定下一步是调用“创作专家”还是进入收尾阶段。5. 运行、调试与关键优化策略5.1 运行与可视化你可以像调用任何函数一样调用这个图# 定义初始状态 initial_state GraphState( original_input请帮我写一篇关于LangGraph三层Agent架构的博客并生成Twitter和LinkedIn的推广帖子。, final_outputNone, task_planNone, current_stepNone, # ... 其他字段初始化为None或默认值 ) # 运行工作流 final_state main_graph.invoke(initial_state) print(最终输出, final_state.get(final_output)) print(任务计划, final_state.get(task_plan)) print(调研结果, final_state.get(research_result)[:500]) # 查看部分调研结果LangGraph提供了一个强大的可视化工具让你能看清整个工作流的执行路径对于调试复杂流程不可或缺。from IPython.display import Image, display try: display(Image(main_graph.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本表示 print(main_graph.get_graph().draw_ascii())5.2 调试技巧与常见问题在开发这类嵌套架构时你一定会遇到各种问题。以下是我总结的几个关键调试点和解决方案状态管理混乱问题Agent A修改了状态字段XAgent B却读到了旧值或覆盖了它。解决严格定义状态模式Schema并使用Annotated类型提示来明确每个字段的合并策略是覆盖、追加还是其他。例如from typing import Sequence class GraphState(TypedDict): research_result: Annotated[Sequence[str], operator.add] # 表示追加到列表 current_step: str # 表示直接覆盖在StateGraph中正确配置state_schemaLangGraph会自动处理合并逻辑。条件路由Conditional Edge失灵问题route_after_orchestrator函数没有返回预期的节点名导致图停滞或报错。解决在路由函数中加入详细的日志打印确保state[“current_step”]的值完全符合你条件判断中的字符串。LLM的输出不稳定最好在Orchestrator的提示词中严格要求输出格式并在解析后做一次标准化如.strip().lower()。子图调用与状态隔离问题直接调用sub_graph.invoke(state)可能会无意间修改主图不希望被修改的状态字段。解决在调用子图时显式地传入一个新的、仅包含子图所需字段的状态字典而不是整个主状态。子图返回后再有选择地合并需要的结果。这增加了控制力避免了副作用。Agent陷入循环或“思考”过长问题某个Agent反复调用工具而不输出最终答案或生成过于冗长的中间链式思考。解决在创建AgentExecutor时设置max_iterations最大迭代次数和early_stopping_method提前停止方法。对于ReAct Agentmax_iterations5通常是个安全的起点。同时优化提示词明确要求“在最多X步内得出结论”。5.3 性能与成本优化嵌套调用意味着多次LLM调用成本和延迟会叠加。以下优化策略很有效缓存Caching对LLM调用和工具调用特别是搜索、API请求实施缓存。LangChain支持多种缓存后端内存、Redis、SQLite。对于不变的查询如“什么是LangGraph”缓存能极大节省成本和时间。异步执行Async如果子任务间没有严格的先后依赖例如调研和创作大纲可以并行使用LangGraph的异步接口ainvoke或利用asyncio.gather来并发执行多个节点能显著降低整体延迟。模型分级使用并非每个Agent都需要最强的GPT-4o。Orchestrator需要最强的规划和理解能力可以用GPT-4o。而Worker层执行简单、格式固定的任务如按模板提取信息完全可以使用更便宜、更快的模型如GPT-3.5-turbo甚至Claude Haiku。在ChatOpenAI初始化时通过model_name参数指定即可。精简提示词与少样本Few-Shot学习仔细打磨每个Agent的提示词移除冗余描述加入清晰的示例Few-Shot能提高一次成功率减少LLM的“胡思乱想”和反复纠正所需的交互轮次。6. 超越三层复杂工作流与高级模式三层架构是一个清晰的起点但真实世界的需求可能更复杂。LangGraph能让你轻松扩展动态子图调用上面的例子是“静态”嵌套即图结构在编译时就固定了。LangGraph支持动态节点允许Orchestrator在运行时根据情况决定创建并调用一个全新的、临时组装的子图实现真正的动态工作流。人工介入Human-in-the-Loop你可以在图中加入一个“人工审核”节点。例如在创作专家生成草稿后流程暂停将草稿通过接口发送给用户审批只有收到批准信号后才继续执行分发任务。这通过中断Interruption和外部事件机制实现。多专家协作与竞争一个任务可以同时分发给多个同类型的专家如三个不同的调研专家使用不同策略然后由一个“评审”节点对它们的结果进行综合比较或投票选出最佳答案。这类似于一个AI委员会。循环与迭代优化可以轻松设置循环边。例如如果Orchestrator对创作专家生成的初稿不满意它可以带着修改意见让流程跳回“创作专家”节点进行重写直到满足某个质量阈值通过条件边判断为止。搭建这样一个系统最深刻的体会是设计比编码更重要。在写第一行代码之前花时间在白板上画出Agent之间的数据流、状态字段的演变、以及异常分支的处理会节省你大量的调试时间。LangGraph提供的不仅是一个工具更是一种描述和实现复杂、有状态AI协作思维的范式。它迫使你以清晰、模块化的方式思考问题而这正是构建可靠、可维护AI应用的关键。 LangGraph三层嵌套Agent架构：从单体智能体到协同工作流实战
1. 从单体Agent到协同作战为什么需要嵌套架构如果你最近在折腾AI应用尤其是想搞点复杂的自动化流程大概率已经玩过LangChain或者LangGraph了。一个简单的Agent比如一个能联网查天气、然后帮你规划行程的助手用LangChain的create_react_agent可能半小时就搭出来了。但当你真的想把想法落地比如做一个能自动分析市场报告、生成策略、并执行邮件分发的系统时你会发现单个“全能”Agent很快就变得臃肿不堪逻辑混乱而且一出错整个流程全崩。这就是我当初踩的第一个大坑。我试图让一个Agent既当分析师又当写手还兼职调度员。结果就是提示词Prompt写得像一篇论文思维链Chain-of-Thought长得自己都看不懂最后它经常在分析到一半时突然开始写邮件抬头逻辑彻底错乱。这让我意识到智能体的能力边界和单一职责原则在AI应用开发里同样至关重要。一个Agent不应该什么都懂什么都做而应该像一支训练有素的团队各司其职高效协作。于是嵌套Agent架构就成了解决这个问题的自然选择。它的核心思想很简单“大任务拆小小任务分包层层汇报统一调度”。一个顶层的“经理”Agent我们称之为Orchestrator负责理解用户的终极目标并将其分解成一系列子任务。然后它将这些子任务分派给更专业的“专家”AgentSpecialist去执行。这些专家Agent可能自己也无法独立完成它们会进一步将任务拆解调用最底层的“工具人”AgentWorker或直接使用工具Tools。最终结果层层汇总由顶层Agent整理后交付给用户。这种架构的好处是显而易见的模块化与可维护性每个Agent功能单一逻辑清晰提示词可以精心打磨。修改一个分析逻辑不会影响到邮件发送的模块。容错与鲁棒性一个子任务失败比如网络查询超时通常不会导致整个系统崩溃顶层调度器可以选择重试、跳过或启用备用方案。能力复用一个写好了的“数据清洗专家”Agent可以被多个不同的业务流程调用。清晰的思维过程每一步决策和结果都在明确的层级间传递非常利于调试和追溯你总能知道是哪个环节、哪个Agent出了问题。而LangGraph正是实现这种复杂、有状态、多Agent协作工作流的绝佳框架。它用“图”Graph的模型来定义工作流节点Node可以是Agent、工具或者任何函数边Edge定义了控制流。通过其强大的状态管理State和条件边Conditional Edge我们能轻松构建出这种三层甚至更多层的嵌套决策系统。接下来我就手把手带你从零开始用LangGraph搭建一个实用的三层嵌套Agent架构。2. 实战场景定义一个智能内容运营助手为了不让教程过于抽象我们贯穿全文用一个具体的、可扩展的实战场景来驱动构建一个智能内容运营助手。它的核心任务是用户输入一个核心话题例如“LangGraph的最新特性”助手能够自动完成从信息搜集、内容加工到分发的完整流程。具体来说这个三层架构将这样工作第一层Orchestrator协调器。它接收用户请求理解意图是“内容运营”。它的职责是规划工作流先调研再创作最后分发。它会将任务分解为“调研Agent请搜集关于LangGraph最新特性的资料”“创作Agent请基于调研结果撰写一篇博客草稿”“分发Agent请将博客草稿制作成社交媒体帖子并安排发布”。第二层Specialist专家Agent。这一层包含多个专业Agent。调研专家Research Specialist接收Orchestrator的“调研”指令。它自己可能不直接搜索而是调度更底层的工具。创作专家Writing Specialist接收Orchestrator的“创作”指令。它负责组织内容结构、文风润色。分发专家Distribution Specialist接收Orchestrator的“分发”指令。它需要理解不同平台的特点。第三层Worker/Tools执行层。这是实际动手的一层。工具Tools如DuckDuckGoSearchRun进行网页搜索arxiv工具查询论文requests工具访问特定API获取数据。简单Worker Agent对于一些需要简单逻辑判断后再使用工具的操作可以设立Worker Agent。例如一个“链接提取与筛选Worker”它接收调研专家传来的一堆原始文本从中提取链接并过滤掉广告或不相关链接再将干净的链接列表返回。整个系统的数据流和控制流将通过LangGraph进行编排。下面我们就开始环境搭建和核心Agent的构建。3. 环境搭建与智能体核心构造3.1 初始化项目与安装依赖首先确保你有一个干净的Python环境3.10以上版本推荐。我们使用uv或pip进行包管理这里以pip为例。# 创建项目目录并进入 mkdir nested_agent_demo cd nested_agent_demo # 创建虚拟环境可选但推荐 python -m venv venv source venv/bin/activate # Linux/Mac # venv\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain langchain-openai langchain-community关键依赖说明langgraph: 本次的核心用于构建工作流图。langchain: 提供了构建Agent所需的大量基础组件如提示模板、记忆、输出解析器等。langchain-openai: 官方维护的OpenAI模型集成我们将使用GPT-4o或GPT-3.5-turbo作为Agent的“大脑”。langchain-community: 包含大量第三方工具和集成如我们即将用到的搜索工具。接下来你需要准备一个OpenAI的API Key。可以将其设置为环境变量export OPENAI_API_KEYyour-api-key-here # 或者在代码中直接设置不推荐用于生产环境3.2 构建基础智能体LangChain Agent as a Node在LangGraph中一切皆节点Node。一个Agent本质上就是一个接收状态、执行动作、并返回新状态的函数。LangChain的create_react_agent函数能快速创建一个基于ReAct范式的Agent我们可以把它包装成LangGraph的节点。我们先从构建最底层的工具调用AgentWorker层开始。假设我们创建一个专门用于“精确搜索”的Worker。import os from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain_core.tools import Tool from langchain_community.tools import DuckDuckGoSearchRun from langchain_core.prompts import PromptTemplate # 1. 初始化大语言模型 llm ChatOpenAI(modelgpt-4o, temperature0) # 使用gpt-4o以获得更好的推理能力temperature设为0保证稳定性 # 2. 定义工具 search_tool DuckDuckGoSearchRun(nameweb_search, description使用此工具在互联网上搜索最新信息。) # 3. 创建Worker Agent的提示词 worker_prompt PromptTemplate.from_template( 你是一个信息检索专家。你的唯一任务是理解用户的信息需求并使用提供的工具进行精准搜索。 用户的需求是{input} 你必须只使用工具来获取信息不要编造答案。 如果一次搜索没有找到满意答案可以分析结果并调整关键词再次搜索。 最终你需要整理并返回清晰、有引用来源的信息摘要。 开始思考 ) # 4. 创建React Agent及其执行器 worker_agent create_react_agent(llmllm, tools[search_tool], promptworker_prompt) worker_agent_executor AgentExecutor(agentworker_agent, tools[search_tool], verboseTrue, handle_parsing_errorsTrue) # 5. 将AgentExecutor包装成LangGraph Node所需的函数 def research_worker_node(state: dict): LangGraph节点函数。接收状态字典调用Worker Agent返回更新后的状态。 状态中应包含input键代表要搜索的查询。 user_query state.get(input, ) if not user_query: return {output: 错误未接收到查询输入。} # 调用Agent执行器 result worker_agent_executor.invoke({input: user_query}) # 将Agent的输出放入状态中假设我们用‘research_result’键来存储 return {research_result: result.get(output, 未找到相关信息。)}这个research_worker_node函数就是一个标准的LangGraph节点。它接收一个状态字典从中取出input调用真正的Agent去执行搜索然后将结果以research_result为键放回状态中。注意这里有一个关键设计点状态State的结构。我们提前规划状态字典里有哪些字段这就像团队协作的共享白板。在上面的Worker层我们定义了input和research_result。在后续的多层架构中状态会成为各Agent间通信的唯一桥梁因此设计一个清晰、一致的状态结构至关重要。我建议在项目开始时用注释或TypedDict明确状态模式State Schema。4. 设计核心三层图架构与状态流转现在我们来构建核心的三层图。我们将创建三个主要的图Graph每个图代表一层然后它们会嵌套调用。4.1 定义全局状态模式首先我们定义整个工作流共享的状态。这通常用一个TypedDict或简单的字典结构来约定。from typing import TypedDict, Annotated, Union from typing_extensions import TypedDict import operator # 定义整个系统的状态类型 class GraphState(TypedDict): 整个嵌套Agent工作流的共享状态。 # 原始输入与最终输出 original_input: str # 用户原始请求如“写一篇关于LangGraph的博客” final_output: Union[str, None] # 最终给用户的结果 # 第一层 Orchestrator 使用的字段 task_plan: Union[str, None] # Orchestrator分解后的任务计划如“1. 调研 2. 写作 3. 分发” current_step: Union[str, None] # 当前正在执行哪个步骤如“researching” # 第二层 Specialist 使用的字段 research_query: Union[str, None] # 传递给调研专家的具体查询 writing_brief: Union[str, None] # 传递给创作专家的内容简报 distribution_platforms: Union[list, None] # 需要分发的平台列表 # 第三层 Worker/Tools 产生的结果 research_result: Union[str, None] # 调研专家/Worker的搜索结果 draft_content: Union[str, None] # 创作专家生成的草稿 distribution_ready_posts: Union[dict, None] # 分发专家准备好的各平台帖子内容 # 错误与重试机制 error: Union[str, None] # 记录运行时错误 retry_count: int # 当前步骤重试次数这个GraphState定义了所有可能用到的字段。在实际运行中并非所有字段都会被同时填充但它提供了一个清晰的蓝图。4.2 构建第三层工具执行图Worker Graph第三层是最具体的执行层。我们构建一个简单的“调研工作流”图它只做一件事调用我们之前定义的research_worker_node。from langgraph.graph import StateGraph, END # 创建第三层图调研工作流 worker_builder StateGraph(GraphState) # 添加节点这个节点就是我们之前定义的函数 worker_builder.add_node(research_worker, research_worker_node) # 设置入口点 worker_builder.set_entry_point(research_worker) # 从research_worker节点执行完后直接结束这个子图。 worker_builder.add_edge(research_worker, END) # 编译第三层图 worker_graph worker_builder.compile()这个worker_graph现在是一个可执行的对象。你可以用它来测试worker_graph.invoke({original_input: 测试, research_query: LangGraph 2024 updates})。它会调用Worker Agent进行搜索并将结果存入research_result。4.3 构建第二层专家调度图Specialist Graph第二层的专家Agent其核心职责是“任务规划与调度”。以Research Specialist为例它不直接搜索而是分析Orchestrator给它的指令research_query判断是否需要搜索、如何搜索然后调用第三层的worker_graph最后对结果进行初步加工。# 1. 首先创建Research Specialist Agent一个更智能的规划者 research_specialist_prompt PromptTemplate.from_template( 你是资深调研分析师。你的目标是为内容创作收集充分、可靠的资料。 上级给你的调研主题是{topic} 请思考并完成以下步骤 1. **理解与拆解**这个主题涉及哪些关键子话题哪些是必须查证的核心概念 2. **制定搜索策略**针对每个子话题设计1-2个精准的搜索关键词。优先考虑官方文档、技术博客、权威社区。 3. **调度执行**你将拥有一个强大的搜索工具由下级Worker执行。请根据你的策略生成最终要执行的搜索查询。 4. **结果整合**拿到搜索结果后你需要过滤掉无关信息将不同来源的信息进行交叉验证和整合形成一份结构化的调研摘要。 请开始你的工作。首先输出你的搜索策略和决定执行的最终查询语句。 最终查询语句必须是清晰、可直接用于搜索的一句话。 ) research_specialist_agent create_react_agent(llmllm, tools[], promptresearch_specialist_prompt) research_specialist_executor AgentExecutor(agentresearch_specialist_agent, tools[], verboseTrue) # 2. 定义第二层图的节点函数 def research_specialist_node(state: GraphState): 调研专家节点。它规划搜索调用Worker图然后整合结果。 topic state.get(research_query) if not topic: return {error: Research specialist did not receive a topic.} # 步骤A专家自己进行规划不调用工具纯LLM推理 plan_result research_specialist_executor.invoke({input: f请为以下主题制定调研计划{topic}}) plan plan_result.get(output, ) # 这里一个简单的实现假设专家输出的最后一行就是“最终查询语句” # 在实际项目中你需要用更稳健的方式如Output Parser来提取这个查询 lines plan.strip().split(\n) final_query lines[-1] if lines else topic # 简化处理实际应解析出查询 print(f[Research Specialist] 分析完成。生成查询{final_query}) # 步骤B调用第三层Worker图来执行实际的搜索 # 注意我们通过更新状态然后调用worker_graph来实现嵌套 # 但更标准的LangGraph方式是使用Send和Subscribe这里为清晰起见采用直接调用子图的方式。 worker_result worker_graph.invoke({ **state, # 传入当前状态 input: final_query, # Worker节点需要input字段 }) raw_data worker_result.get(research_result, ) # 步骤C专家整合加工原始结果 integration_prompt f 以下是你要求的搜索查询结果 查询{final_query} 原始结果{raw_data[:2000]}...可能截断 请完成你的第4步整合信息形成一份给内容创作团队的摘要。 摘要需包含核心观点、数据来源、关键引用、以及尚存疑问或需要进一步核实的地方。 integration_result research_specialist_executor.invoke({input: integration_prompt}) final_research_summary integration_result.get(output, raw_data) # 如果整合失败返回原始数据 # 更新状态 return { research_result: final_research_summary, current_step: research_completed, # 通知上层本步骤完成 } # 3. 构建第二层图这里以调研专家为例写作和分发专家结构类似 specialist_builder StateGraph(GraphState) specialist_builder.add_node(conduct_research, research_specialist_node) specialist_builder.set_entry_point(conduct_research) specialist_builder.add_edge(conduct_research, END) research_specialist_graph specialist_builder.compile()这个research_specialist_graph就代表了一个具备规划能力的专家。它展示了第二层Agent的典型模式接收指令 - 自主规划 - 调度底层资源子图/工具- 加工结果 - 返回。4.4 构建第一层总指挥图Orchestrator Graph第一层Orchestrator是大脑中的大脑。它需要理解用户意图分解任务并决定调用哪个专家子图以及处理专家返回的结果决定下一步是继续还是结束。# 1. 创建Orchestrator Agent orchestrator_prompt PromptTemplate.from_template( 你是智能内容运营助手的总指挥。你的目标是将用户的复杂请求分解为有序的、可执行的任务流。 用户请求{request} 已知可用的专家团队有 1. **调研专家**负责信息搜集与验证。 2. **创作专家**负责撰写高质量文案草稿。 3. **分发专家**负责适配不同平台格式并安排发布。 请按以下步骤思考 1. **请求分析**用户的核心需求是什么最终要交付什么 2. **任务分解**要完成这个需求必须经历哪几个阶段例如调研 - 创作 - 分发。每个阶段的具体输入是什么 3. **流程编排**这些阶段的执行顺序是怎样的是否有并行可能例如调研和初步的框架创作可以同时进行。 4. **输出决策**根据当前状态和阶段结果决定下一步调用哪个专家还是结束流程并交付最终结果。 请输出你的任务分解计划和当前应执行的第一步。 输出格式必须严格遵循 计划[你的整体阶段计划如“1.调研2.创作3.分发”] 下一步[下一步要执行的专家名称如“调研专家”] ) orchestrator_agent create_react_agent(llmllm, tools[], promptorchestrator_prompt) orchestrator_executor AgentExecutor(agentorchestrator_agent, tools[], verboseTrue) # 2. 定义Orchestrator的节点函数 def orchestrator_node(state: GraphState): 总指挥节点。分析请求规划任务并设置下一步该谁执行。 request state.get(original_input, ) if not request: return {error: Orchestrator did not receive input.} # 调用Orchestrator Agent进行分析和规划 analysis orchestrator_executor.invoke({input: request}) analysis_output analysis.get(output, ) # 解析输出获取计划和下一步动作这里需要简单的解析逻辑 plan 未解析到计划 next_step 未知 for line in analysis_output.split(\n): if line.startswith(计划): plan line.replace(计划, ).strip() elif line.startswith(下一步): next_step line.replace(下一步, ).strip() # 根据计划更新状态并决定下一个节点 # 我们将“下一步”动作存入状态由条件边Conditional Edge来判断 return { task_plan: plan, current_step: next_step, # 例如 “调研专家” } # 3. 定义路由函数Conditional Edge def route_after_orchestrator(state: GraphState): 根据Orchestrator设定的current_step决定下一步去哪个节点。 next_step state.get(current_step) if next_step 调研专家: return call_research_specialist elif next_step 创作专家: return call_writing_specialist elif next_step 分发专家: return call_distribution_specialist elif next_step 结束: return compile_final_output else: # 如果无法识别返回错误节点或重新规划 return handle_error # 4. 定义调用各专家子图的节点函数 def call_research_specialist_node(state: GraphState): 调用调研专家子图 # 准备查询这里简化处理将原始输入作为查询。实际中Orchestrator应生成更精确的查询。 query state.get(original_input, ) # 调用第二层图 result_state research_specialist_graph.invoke({**state, research_query: query}) # 合并结果到当前状态 new_state {**state, **result_state} # 执行完后将控制权交回给Orchestrator让它决定下一步 new_state[current_step] orchestrator_review return new_state # 5. 构建完整的第一层图主图 main_builder StateGraph(GraphState) # 添加节点 main_builder.add_node(orchestrator, orchestrator_node) main_builder.add_node(call_research_specialist, call_research_specialist_node) # 这里省略了 call_writing_specialist_node 和 call_distribution_specialist_node 的定义和添加其结构与call_research_specialist_node类似。 main_builder.add_node(compile_final_output, lambda state: {final_output: 整合所有结果形成最终回复。此处为示意}) main_builder.add_node(handle_error, lambda state: {error: f路由失败未知步骤{state.get(current_step)}}) # 设置入口点 main_builder.set_entry_point(orchestrator) # 添加边包括条件边 main_builder.add_edge(handle_error, END) # 错误处理节点直接结束 main_builder.add_edge(compile_final_output, END) # 最终编译节点直接结束 # 从orchestrator节点出来后根据条件路由 main_builder.add_conditional_edges( orchestrator, route_after_orchestrator, { call_research_specialist: call_research_specialist, call_writing_specialist: call_writing_specialist, # 假设已定义 call_distribution_specialist: call_distribution_specialist, # 假设已定义 compile_final_output: compile_final_output, handle_error: handle_error, } ) # 专家节点执行完后都回到orchestrator进行下一轮决策 main_builder.add_edge(call_research_specialist, orchestrator) # ... 其他专家节点同理 # 编译主图 main_graph main_builder.compile()至此一个三层嵌套Agent架构的核心骨架就搭建完成了。main_graph是我们的总入口。它启动后由orchestrator节点分析任务然后通过条件边route_after_orchestrator动态决定调用哪个专家子图。专家子图在执行过程中又会调用更底层的Worker图或工具。执行完毕后流程返回orchestrator节点由其根据当前结果例如research_result是否已就绪和初始计划决定下一步是调用“创作专家”还是进入收尾阶段。5. 运行、调试与关键优化策略5.1 运行与可视化你可以像调用任何函数一样调用这个图# 定义初始状态 initial_state GraphState( original_input请帮我写一篇关于LangGraph三层Agent架构的博客并生成Twitter和LinkedIn的推广帖子。, final_outputNone, task_planNone, current_stepNone, # ... 其他字段初始化为None或默认值 ) # 运行工作流 final_state main_graph.invoke(initial_state) print(最终输出, final_state.get(final_output)) print(任务计划, final_state.get(task_plan)) print(调研结果, final_state.get(research_result)[:500]) # 查看部分调研结果LangGraph提供了一个强大的可视化工具让你能看清整个工作流的执行路径对于调试复杂流程不可或缺。from IPython.display import Image, display try: display(Image(main_graph.get_graph().draw_mermaid_png())) except: # 如果无法生成图片可以输出文本表示 print(main_graph.get_graph().draw_ascii())5.2 调试技巧与常见问题在开发这类嵌套架构时你一定会遇到各种问题。以下是我总结的几个关键调试点和解决方案状态管理混乱问题Agent A修改了状态字段XAgent B却读到了旧值或覆盖了它。解决严格定义状态模式Schema并使用Annotated类型提示来明确每个字段的合并策略是覆盖、追加还是其他。例如from typing import Sequence class GraphState(TypedDict): research_result: Annotated[Sequence[str], operator.add] # 表示追加到列表 current_step: str # 表示直接覆盖在StateGraph中正确配置state_schemaLangGraph会自动处理合并逻辑。条件路由Conditional Edge失灵问题route_after_orchestrator函数没有返回预期的节点名导致图停滞或报错。解决在路由函数中加入详细的日志打印确保state[“current_step”]的值完全符合你条件判断中的字符串。LLM的输出不稳定最好在Orchestrator的提示词中严格要求输出格式并在解析后做一次标准化如.strip().lower()。子图调用与状态隔离问题直接调用sub_graph.invoke(state)可能会无意间修改主图不希望被修改的状态字段。解决在调用子图时显式地传入一个新的、仅包含子图所需字段的状态字典而不是整个主状态。子图返回后再有选择地合并需要的结果。这增加了控制力避免了副作用。Agent陷入循环或“思考”过长问题某个Agent反复调用工具而不输出最终答案或生成过于冗长的中间链式思考。解决在创建AgentExecutor时设置max_iterations最大迭代次数和early_stopping_method提前停止方法。对于ReAct Agentmax_iterations5通常是个安全的起点。同时优化提示词明确要求“在最多X步内得出结论”。5.3 性能与成本优化嵌套调用意味着多次LLM调用成本和延迟会叠加。以下优化策略很有效缓存Caching对LLM调用和工具调用特别是搜索、API请求实施缓存。LangChain支持多种缓存后端内存、Redis、SQLite。对于不变的查询如“什么是LangGraph”缓存能极大节省成本和时间。异步执行Async如果子任务间没有严格的先后依赖例如调研和创作大纲可以并行使用LangGraph的异步接口ainvoke或利用asyncio.gather来并发执行多个节点能显著降低整体延迟。模型分级使用并非每个Agent都需要最强的GPT-4o。Orchestrator需要最强的规划和理解能力可以用GPT-4o。而Worker层执行简单、格式固定的任务如按模板提取信息完全可以使用更便宜、更快的模型如GPT-3.5-turbo甚至Claude Haiku。在ChatOpenAI初始化时通过model_name参数指定即可。精简提示词与少样本Few-Shot学习仔细打磨每个Agent的提示词移除冗余描述加入清晰的示例Few-Shot能提高一次成功率减少LLM的“胡思乱想”和反复纠正所需的交互轮次。6. 超越三层复杂工作流与高级模式三层架构是一个清晰的起点但真实世界的需求可能更复杂。LangGraph能让你轻松扩展动态子图调用上面的例子是“静态”嵌套即图结构在编译时就固定了。LangGraph支持动态节点允许Orchestrator在运行时根据情况决定创建并调用一个全新的、临时组装的子图实现真正的动态工作流。人工介入Human-in-the-Loop你可以在图中加入一个“人工审核”节点。例如在创作专家生成草稿后流程暂停将草稿通过接口发送给用户审批只有收到批准信号后才继续执行分发任务。这通过中断Interruption和外部事件机制实现。多专家协作与竞争一个任务可以同时分发给多个同类型的专家如三个不同的调研专家使用不同策略然后由一个“评审”节点对它们的结果进行综合比较或投票选出最佳答案。这类似于一个AI委员会。循环与迭代优化可以轻松设置循环边。例如如果Orchestrator对创作专家生成的初稿不满意它可以带着修改意见让流程跳回“创作专家”节点进行重写直到满足某个质量阈值通过条件边判断为止。搭建这样一个系统最深刻的体会是设计比编码更重要。在写第一行代码之前花时间在白板上画出Agent之间的数据流、状态字段的演变、以及异常分支的处理会节省你大量的调试时间。LangGraph提供的不仅是一个工具更是一种描述和实现复杂、有状态AI协作思维的范式。它迫使你以清晰、模块化的方式思考问题而这正是构建可靠、可维护AI应用的关键。