尧图网站开发 
1. 项目概述为什么需要三层嵌套的智能体架构最近在折腾一些复杂的自动化流程比如一个需要先分析需求、再规划步骤、最后调用不同工具执行的场景。用单个大模型LLM或者一个简单的链式调用Chain去处理总感觉力不从心。要么是上下文太长导致模型“失忆”忘了最初的目标要么是任务类型太杂一个智能体Agent无法精通所有领域结果就是规划混乱、执行出错。这时候一个清晰的层级架构就变得至关重要。这就像管理一个项目你不能让CEO直接去写每一行代码而是需要设立部门经理、技术主管和一线工程师。LangGraph这个库正是为了构建这种有状态、可循环、多参与者的智能体系统而生的。它把整个流程抽象成一个“图”Graph节点是处理单元可以是LLM调用、工具执行或条件判断边定义了控制流。而“三层嵌套Agent架构”就是在这种图上玩出的一个高级模式。它不是一个简单的线性链条而是构建了一个“管理者-协调者-执行者”的层级体系。管理者负责顶层目标拆解和战略制定协调者接收子任务并调度资源执行者则专注于调用具体工具完成动作。这种架构的核心优势在于职责分离、上下文隔离和错误收敛。每个层级的智能体只需要关注自己层面的问题避免了单一智能体因任务过载而产生的“思维混乱”也让整个系统的可维护性和可扩展性大大提升。接下来我就带你从零开始手把手搭建这样一个三层架构。我们会用一个相对复杂但贴近实际的场景来贯穿始终“智能内容运营助手”。它的目标是当你输入一个模糊的营销想法例如“为我们的新款智能水杯做一次小红书种草”时系统能自动完成从策略分析、内容规划到文案生成和排期发布的全流程。2. 核心架构设计与组件选型2.1 三层架构的职责定义与数据流在动手写代码之前我们必须把每一层的职责和数据交互方式定义清楚。一个混乱的架构设计后期调试会是噩梦。第一层战略管理智能体Manager Agent这是系统的大脑。它的输入是用户的原始、模糊的指令。它的核心职责是进行任务分解与战略规划。它不关心具体怎么写文案、用什么工具它只做宏观决策。输入用户原始指令String。处理调用LLM分析指令拆解成一个有序的、原子化的子任务列表。每个子任务必须包含明确的目标、所需的资源类型如图文生成、数据分析和成功标准。输出一个结构化的任务清单List[Dict]。例如针对“小红书种草”它可能输出[{id: 1, goal: 分析目标用户画像与热门话题, agent_type: analyzer}, {id: 2, goal: 生成3篇不同角度的图文草稿, agent_type: creator}, {id: 3, goal: 制定未来一周的发布排期, agent_type: planner}]。工具它通常只有思考能力不直接调用外部API。它的“工具”就是LLM本身加上我们定义的提示词Prompt来约束其输出格式。第二层任务协调智能体Coordinator Agent这是系统的中枢神经。它接收来自管理者的具体子任务并负责调度与编排合适的执行者。它理解不同执行者的能力并准备执行任务所需的上下文。输入单个子任务对象Dict以及从全局状态中获取的相关历史信息。处理根据子任务的agent_type或goal关键词决定路由到哪个第三层执行者。它可能需要为执行者收集或预处理一些信息例如从之前的任务结果中提取关键数据。输出一个格式化的执行指令包含目标、上下文和约束条件直接发送给特定的第三层智能体。工具它可能需要一些轻量级的工具比如从状态中查询数据的“工具”但其主要功能是逻辑判断和路由。第三层技能执行智能体Worker Agent这是系统的手和脚。每个执行者都是某个垂直领域的专家拥有执行特定任务所需的专用工具集。输入来自协调者的格式化执行指令。处理调用其专属的工具来完成工作。例如“文案生成执行者”会调用DALL-E API和文案LLM“数据分析执行者”可能会调用网络搜索工具和数据分析库。输出任务执行的具体结果String, Image, Data等并附带执行状态成功/失败。工具拥有丰富的外部工具调用权限如search_web,generate_image,write_copy,query_database等。数据流与状态管理 整个LangGraph的运行依赖于一个共享的“状态”State。这个状态是一个字典随着图的执行而更新。典型的状态字段包括messages: 整个对话历史是所有智能体沟通的媒介。task_list: 管理者生成的总任务列表。current_task: 当前正在处理的任务。results: 一个字典收集每个已完成任务的结果键为任务ID。next: 指示下一个应该执行的节点。数据流是单向且清晰的用户输入 - 更新状态 - 管理者节点运行 - 更新状态加入task_list- 协调者节点被触发 - 根据任务列表和状态调用对应的执行者节点 - 执行者运行并更新结果 - 循环直至所有任务完成 - 返回最终状态。2.2 为什么选择LangGraph而非其他框架市面上构建Agent的框架不少比如LangChain的原生Agent、AutoGen、CrewAI等。选择LangGraph基于以下几个关键考量显式的控制流LangGraph的核心是“图”你需要显式地定义节点和边。这对于构建三层嵌套这种复杂、有条件分支和循环的流程来说是天然契合的。你可以清晰地看到“失败重试”、“条件路由”这些逻辑在图中是如何体现的调试起来非常直观。灵活的状态管理它的状态State概念非常强大可以自定义任何结构。我们的三层架构需要传递任务列表、中间结果等复杂数据用LangGraph的状态机模型来管理比用简单的消息传递要稳健得多。对人类友好的调试LangGraph内置了可视化工具你可以把整个Agent的工作流画出来看到每个节点的人输入和出输出对于理解多层嵌套的执行顺序至关重要。与LangChain生态无缝集成我们的执行者智能体Worker Agent很可能需要用到LangChain的Tool和AgentExecutor。LangGraph本身就是LangChain家族的一员集成起来毫无障碍可以直接在节点函数里调用现有的LangChain组件。注意不要试图用单一的、超级复杂的提示词Prompt去让一个Agent完成所有三层的工作。那样做的结果通常是提示词极其臃肿成本高昂上下文长且稳定性极差。架构的意义就在于“分而治之”。2.3 基础环境搭建与模型选择首先准备好你的Python环境。我强烈建议使用虚拟环境。# 创建并激活虚拟环境 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # 或 langgraph-env\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain langchain-openai模型的选择取决于你的需求和预算。对于管理者和协调者它们需要进行复杂的逻辑分析和规划建议使用能力较强的模型如GPT-4系列。对于执行者可以根据任务性质选择创意生成类如文案可用GPT-4简单工具调用类可用GPT-3.5 Turbo以节约成本。# 在代码中初始化LLM from langchain_openai import ChatOpenAI # 战略层使用更强的模型 manager_llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.1) # 低随机性保证规划稳定 # 执行层可根据需要选择 worker_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 创意任务可适当提高温度设置好你的OpenAI API密钥在环境变量中。3. 构建第一层战略管理智能体管理者的核心是生成一个可执行的任务列表。我们需要为其设计一个强大的提示词并定义好输出的结构化格式。3.1 设计管理者的系统提示词提示词的质量直接决定了任务分解的合理性。一个好的管理者提示词需要包含角色定义明确告诉模型它现在是谁。终极目标它要解决什么问题。约束条件输出的格式、必须考虑的维度、禁止做的事情。示例一两个清晰的例子让模型学会输出格式。from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder manager_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深的项目主管和策略分析师。你的任务是将用户模糊的需求分解为具体、可执行、有序的子任务。 请遵循以下规则 1. **深度分析需求**仔细理解用户目标的深层意图、目标受众和期望效果。 2. **任务拆解**将大目标拆解为顺序执行的原子任务。后一个任务可以依赖前一个任务的输出。 3. **任务描述**每个任务必须包含清晰的目标Goal、负责此任务的智能体类型Agent Type、以及简要的上下文说明Context。 4. **智能体类型**只能使用以下几种类型analyzer分析者负责调研、分析、creator创造者负责内容生成、planner规划者负责排期、规划、reviewer审查者负责质量检查。 5. **输出格式**你必须以严格的JSON列表格式输出列表中的每个元素是一个任务对象。 示例 用户输入“我想推广我的新咖啡店” 你的输出 json [ { “id”: 1, “goal”: “分析本地咖啡市场趋势、竞争对手及目标顾客喜好” “agent_type”: “analyzer” “context”: “为后续内容创作提供数据支持和方向指导” }, { “id”: 2, “goal”: “为咖啡店设计3个营销主题并生成对应的社交媒体文案和图片创意” “agent_type”: “creator” “context”: “基于分析结果创作吸引人的内容” }, { “id”: 3, “goal”: “制定为期两周的社交媒体发布日历” “agent_type”: “planner” “context”: “合理安排内容发布节奏和渠道” } ]现在开始处理用户的需求。), MessagesPlaceholder(variable_namemessages), # 这里会放入用户的输入 ])### 3.2 创建管理者节点函数 在LangGraph中节点就是一个普通的Python函数它接收并返回状态。 python from langchain_core.messages import HumanMessage import json def manager_node(state): 战略管理节点分析需求生成任务列表。 # 1. 从状态中获取最新的用户消息 user_input state[messages][-1].content if state[messages] else # 2. 构建提示词并调用LLM prompt_value manager_prompt.invoke({messages: [HumanMessage(contentuser_input)]}) response manager_llm.invoke(prompt_value) # 3. 解析LLM的返回提取JSON任务列表 try: # 模型返回的是字符串我们需要提取JSON部分。有时模型会在文本中包裹JSON。 response_content response.content # 简单的提取找到第一个[和最后一个] start_idx response_content.find([) end_idx response_content.rfind(]) 1 if start_idx ! -1 and end_idx ! 0: task_list_json response_content[start_idx:end_idx] task_list json.loads(task_list_json) else: # 如果没找到尝试直接解析整个内容模型可能只返回了JSON task_list json.loads(response_content) except json.JSONDecodeError as e: # 如果解析失败说明模型没有按要求输出这是一个关键错误点 print(f管理者节点解析JSON失败: {e}) print(f模型原始返回: {response_content}) # 可以在这里设计一个降级策略例如返回一个默认的简单任务列表 task_list [{id: 1, goal: 分析需求失败请检查输入或模型响应。, agent_type: analyzer, context: 错误处理任务}] # 4. 更新状态 # 将任务列表存入状态并初始化结果收集器 state[task_list] task_list state[results] {} # 用于存放每个任务的执行结果 # 指示下一个节点应该是协调者 state[next] coordinator # 也可以将管理者的思考过程存入消息历史便于追溯可选 # state[messages].append(AIMessage(contentf我已将您的需求分解为{len(task_list)}个任务{task_list})) return state实操心得JSON解析是这类工作流中最脆弱的环节之一。模型并不总是乖乖输出纯净的JSON。除了用字符串查找更稳健的做法是使用LangChain的OutputParser如JsonOutputParser或者让模型使用支持结构化输出的功能如OpenAI的response_format。这里为了演示清晰使用了手动解析。在生产环境中务必加入更完善的错误处理和重试机制。4. 构建第二层任务协调智能体协调者是一个路由中心。它查看当前需要处理哪个任务然后根据任务类型决定下一步该调用哪个执行者节点。4.1 设计协调者的决策逻辑协调者不需要复杂的LLM调用它的逻辑通常是基于规则的。我们可以用一个简单的字典来映射agent_type到对应的执行者节点名。def coordinator_node(state): 任务协调节点决定下一个执行哪个任务并路由到对应的执行者。 # 1. 获取任务列表和当前结果 task_list state.get(task_list, []) results state.get(results, {}) # 2. 找出第一个未完成的任务即不在results中的任务 current_task None for task in task_list: if str(task[id]) not in results: current_task task break if not current_task: # 所有任务都已完成 state[next] __end__ return state # 3. 根据任务类型路由到不同的执行者节点 agent_type current_task.get(agent_type, ).lower() # 更新当前任务到状态中供执行者读取 state[current_task] current_task # 4. 路由决策 if agent_type analyzer: state[next] analyzer_worker elif agent_type creator: state[next] creator_worker elif agent_type planner: state[next] planner_worker elif agent_type reviewer: state[next] reviewer_worker else: # 未知类型可以路由到一个默认的错误处理节点或者直接标记失败 state[next] error_handler # 记录错误结果 task_id str(current_task[id]) state[results][task_id] { status: failed, error: f未知的智能体类型: {agent_type} } return state4.2 将协调者定义为条件边在LangGraph中协调者的这种“根据状态决定下一步”的功能通常通过定义一个条件边Conditional Edge的函数来实现而不是一个独立的节点。上面的coordinator_node函数返回的state[“next”]就指明了下一个节点。我们会在构建图的时候创建一个以coordinator_node为路由函数的条件边。5. 构建第三层技能执行智能体执行者是真正干活的。我们以creator_worker内容创造者为例展示如何构建一个具备工具调用能力的智能体。5.1 为执行者配备工具假设我们的创造者需要两个工具一个用于生成图片创意描述另一个用于撰写文案。from langchain.tools import tool from typing import Optional tool def generate_image_prompt(subject: str, style: str) - str: 根据主题和风格生成一个详细的、可供AI绘画模型使用的提示词。 # 这里可以是一个复杂的提示词模板也可以调用另一个LLM # 为了简化我们直接返回一个组合字符串 prompt fA high-quality, {style} style image of {subject}, suitable for social media marketing, clear focus, vibrant colors. return prompt tool def write_marketing_copy(topic: str, tone: str, length: str medium) - str: 根据主题、语气和长度撰写一篇营销文案。 # 在实际应用中这里会调用LLM。我们模拟一个复杂调用。 # 注意我们在这个函数内部使用LLM但这个函数本身对LangGraph的智能体来说是一个“工具”。 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser copy_prompt ChatPromptTemplate.from_template( 你是一位专业的社交媒体文案写手。 请围绕以下主题以{tone}的语气撰写一篇{length}长度的营销文案。 主题{topic} 文案 ) chain copy_prompt | worker_llm | StrOutputParser() result chain.invoke({topic: topic, tone: tone, length: length}) return result # 将工具打包成列表 creator_tools [generate_image_prompt, write_marketing_copy]5.2 创建执行者节点函数执行者节点需要做几件事从状态中获取任务指令调用合适的工具可能通过一个LangChain Agent并将结果保存回状态。from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate def creator_worker_node(state): 内容创造执行者节点生成营销内容和图片创意。 current_task state[current_task] task_id str(current_task[id]) goal current_task[goal] context current_task.get(context, ) # 1. 为这个执行者创建专属的提示词 worker_prompt PromptTemplate.from_template( 你是一个专业的社交媒体内容创造者。 你的当前任务是{goal} 上下文信息{context} 你已经完成了之前的所有任务相关结果已汇总在下方。 请运用你拥有的工具高质量地完成上述任务。 请逐步思考并最终输出你的创作成果。 如果使用工具请清晰说明。 之前任务汇总 {previous_results} 开始你的工作 ) # 2. 准备之前任务的结果作为上下文 previous_results_str for tid, result in state.get(results, {}).items(): if isinstance(result, dict): previous_results_str f任务{tid}: {result.get(output, str(result))}\n else: previous_results_str f任务{tid}: {result}\n # 3. 创建智能体执行器 # 使用ReAct范式让智能体能够思考并调用工具 agent create_react_agent(llmworker_llm, toolscreator_tools, promptworker_prompt) agent_executor AgentExecutor(agentagent, toolscreator_tools, verboseTrue, handle_parsing_errorsTrue) # 4. 执行任务 try: task_input { goal: goal, context: context, previous_results: previous_results_str } # 调用执行器 full_response agent_executor.invoke(task_input) final_output full_response.get(output, 任务执行完成但无输出。) except Exception as e: final_output f任务执行过程中出现错误: {str(e)} # 5. 将结果保存到状态中 state[results][task_id] { status: completed, output: final_output, agent: creator_worker } # 6. 指示下一个节点回到协调者以处理下一个任务 state[next] coordinator # 清理当前任务避免干扰下一次执行可选 # state.pop(current_task, None) return state按照类似的模式我们可以创建analyzer_worker配备网络搜索和数据分析工具、planner_worker配备日历API工具等。6. 组装与调试构建完整的LangGraph工作流现在我们将所有节点和边组装起来形成一个完整的图。6.1 定义状态结构首先我们需要定义一个状态结构告诉LangGraph我们的状态字典里都有哪些字段以及它们的类型。from typing import TypedDict, List, Dict, Any, Optional from langgraph.graph import StateGraph, END class AgentState(TypedDict): 定义工作流的状态结构。 messages: List[Any] # 消息历史 task_list: Optional[List[Dict]] # 管理者生成的任务列表 current_task: Optional[Dict] # 当前正在处理的任务 results: Dict[str, Any] # 任务ID到结果的映射 next: str # 指示下一个节点6.2 创建图并添加节点# 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(manager, manager_node) workflow.add_node(coordinator, coordinator_node) # 协调者也是一个节点但主要功能是路由 workflow.add_node(creator_worker, creator_worker_node) # 假设我们已经定义了其他worker节点 # workflow.add_node(analyzer_worker, analyzer_worker_node) # workflow.add_node(planner_worker, planner_worker_node) workflow.add_node(error_handler, error_handler_node) # 一个简单的错误处理节点6.3 设置边与条件路由这是最关键的一步定义了工作流的逻辑。# 设置入口点从manager开始 workflow.set_entry_point(manager) # 从manager出来后进入coordinator workflow.add_edge(manager, coordinator) # 从coordinator出来根据其设置的state[“next”]值动态路由到下一个节点 # 这需要一个条件边函数但我们已经把逻辑放在coordinator_node里了。 # 我们可以让coordinator_node不直接返回state而是返回next的值。 # 但更LangGraph的方式是使用add_conditional_edges。 # 首先定义一个路由函数它读取state[“next”]来决定去向 def route_next(state): next_node state.get(next) if next_node __end__: return END return next_node # 为coordinator节点添加条件边 workflow.add_conditional_edges( “coordinator” route_next # 这个函数返回下一个节点的名字 # 下面列出所有可能的目的地包括END { “creator_worker”: “creator_worker” “analyzer_worker”: “analyzer_worker” “planner_worker”: “planner_worker” “error_handler”: “error_handler” “__end__”: END # 理论上也可以路由回“coordinator”形成循环直到任务完成。 # 但我们的设计是worker完成后回到coordinator所以需要在worker节点设置state[“next”] “coordinator” } ) # 为各个worker节点添加边执行完后回到coordinator检查下一个任务 workflow.add_edge(“creator_worker” “coordinator”) # workflow.add_edge(“analyzer_worker” “coordinator”) # workflow.add_edge(“planner_worker” “coordinator”) workflow.add_edge(“error_handler” “coordinator”) # 错误处理后也继续流程6.4 编译并运行图# 编译图得到一个可执行的应用 app workflow.compile() # 为了可视化我们可以将图画出来需要安装graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(“可视化需要graphviz和IPython环境。”) # 运行工作流 initial_state AgentState( messages[HumanMessage(content“为我们的新款智能水杯做一次小红书种草”)], task_listNone current_taskNone results{} next“” ) # 流式输出执行过程 final_state None for event in app.stream(initial_state, stream_mode“values”): node_name list(event.keys())[0] print(f”\n{‘’*20} 进入节点[{node_name}] {‘’*20}”) state event[node_name] if “current_task” in state and state[“current_task”]: print(f”当前处理任务: {state[‘current_task’].get(‘goal’)}”) if “results” in state: print(f”累计结果: {list(state[‘results’].keys())}”) final_state state print(f”\n{‘’*20} 所有任务执行完毕 {‘’*20}”) print(“最终任务结果汇总”) for task_id result in final_state[“results”].items(): print(f”任务{task_id}: {result.get(‘status’ ‘N/A’)} - 输出摘要: {str(result.get(‘output’ ‘N/A’))[:200]}…”)7. 常见问题、优化策略与避坑指南在实际搭建和运行过程中你肯定会遇到各种问题。下面是我踩过坑后总结的一些核心要点。7.1 任务分解不合理的应对策略问题管理者LLM拆解的任务可能逻辑混乱、顺序不对、或者不够原子化。解决方案强化提示词在管理者提示词中提供更详细、更优秀的示例。明确要求“后任务依赖前任务输出”。后置校验节点在管理者节点后增加一个“任务校验节点”。这个节点也是一个LLM调用它检查生成的任务列表是否合理并可以进行微调或重新排序。人工审核介入对于关键流程可以在生成任务列表后将列表呈现给用户确认或修改然后再进入执行流程。这可以通过在图中加入一个“人工审核”节点来实现该节点暂停自动化等待用户输入。7.2 执行过程中的错误处理与重试问题某个执行者节点调用工具失败如API超时、格式错误。解决方案节点内部的Try-Catch就像我们在creator_worker_node里做的那样每个执行节点都应该有完善的异常捕获并将错误信息记录到结果中而不是让整个图崩溃。全局错误处理节点我们创建的error_handler_node可以做得更智能。它可以分析错误类型决定是重试当前任务例如网络错误、跳过该任务、还是升级到人工处理。设置重试逻辑LangGraph本身不直接提供重试但可以在节点函数内实现。例如当捕获到可重试错误时将state[“next”]重新指向自己并设置一个重试计数器在状态中避免无限循环。def robust_worker_node(state): task_id str(state[“current_task”][“id”]) retry_count state.get(“retries” {}).get(task_id 0) if retry_count 2: state[“results”][task_id] {“status”: “failed” “error”: “超过最大重试次数”} state[“next”] “coordinator” return state try: # … 执行任务 … state[“results”][task_id] {“status”: “completed” “output”: result} state[“next”] “coordinator” except TransientError as e: # 假设定义了一个临时错误类型 # 更新重试计数 state.setdefault(“retries” {})[task_id] retry_count 1 # 下次再进入本节点 state[“next”] “robust_worker_node” print(f”任务{task_id}第{retry_count1}次重试…”) return state7.3 状态管理的复杂性与性能问题随着任务增多状态字典变得庞大特别是messages历史可能很长导致每次调用LLM的上下文窗口压力大、成本高。解决方案精简状态不是所有节点都需要完整的对话历史。可以为不同节点设计不同的提示词只注入相关的历史片段。例如执行者节点可能只需要它之前几个相关任务的结果而不是所有消息。状态压缩定期对messages进行总结。可以添加一个“总结器”节点在历史消息过长时调用LLM生成一个浓缩的摘要然后用摘要替换掉旧消息。外部存储对于非常大的中间结果如图片二进制数据不要放在内存状态里。可以将其存储到数据库或文件系统在状态中只保存引用ID或路径。7.4 图的监控与可观测性问题图运行起来像个黑盒哪里慢了、哪里出错了不容易定位。解决方案利用LangGraph的Streaming如上例所示使用app.stream()可以实时看到执行流经了哪些节点并打印中间状态这是最基本的调试手段。结构化日志在每个节点的开始和结束处记录结构化的日志包括节点名、任务ID、耗时、输入输出摘要等。可以使用像structlog这样的库。持久化检查点LangGraph支持将状态持久化。对于长时间运行的工作流可以定期保存检查点。如果系统崩溃可以从最近的检查点恢复而不是从头开始。可视化app.get_graph().draw_mermaid_png()生成的可视化图是理解和沟通架构的绝佳工具。7.5 架构扩展超越三层三层架构是一个强大的范式但并非一成不变。你可以根据需求扩展并行执行如果任务间没有依赖可以在协调者中实现并行路由让多个执行者同时工作。这需要更复杂的状态管理来合并结果。动态层级某些复杂的子任务本身可能也需要被进一步分解。你可以让某个执行者如analyzer_worker在内部再调用一个子图形成递归或动态的嵌套结构。反馈循环引入reviewer节点对执行结果进行审核如果质量不达标可以将其反馈给coordinator由coordinator重新调度给creator进行修改。这就在图中形成了一个质量控制的循环。搭建这样一个三层嵌套的Agent架构初期投入的思考和时间会比较多但一旦跑通它带来的清晰度、可维护性和处理复杂任务的能力是简单链式调用无法比拟的。最关键的是理解“状态”如何在图中流动以及每个节点如何读取和修改这个共享状态。从一个小而具体的场景开始逐步增加复杂度和节点是掌握LangGraph的最佳路径。 
1. 项目概述为什么需要三层嵌套的智能体架构最近在折腾一些复杂的自动化流程比如一个需要先分析需求、再规划步骤、最后调用不同工具执行的场景。用单个大模型LLM或者一个简单的链式调用Chain去处理总感觉力不从心。要么是上下文太长导致模型“失忆”忘了最初的目标要么是任务类型太杂一个智能体Agent无法精通所有领域结果就是规划混乱、执行出错。这时候一个清晰的层级架构就变得至关重要。这就像管理一个项目你不能让CEO直接去写每一行代码而是需要设立部门经理、技术主管和一线工程师。LangGraph这个库正是为了构建这种有状态、可循环、多参与者的智能体系统而生的。它把整个流程抽象成一个“图”Graph节点是处理单元可以是LLM调用、工具执行或条件判断边定义了控制流。而“三层嵌套Agent架构”就是在这种图上玩出的一个高级模式。它不是一个简单的线性链条而是构建了一个“管理者-协调者-执行者”的层级体系。管理者负责顶层目标拆解和战略制定协调者接收子任务并调度资源执行者则专注于调用具体工具完成动作。这种架构的核心优势在于职责分离、上下文隔离和错误收敛。每个层级的智能体只需要关注自己层面的问题避免了单一智能体因任务过载而产生的“思维混乱”也让整个系统的可维护性和可扩展性大大提升。接下来我就带你从零开始手把手搭建这样一个三层架构。我们会用一个相对复杂但贴近实际的场景来贯穿始终“智能内容运营助手”。它的目标是当你输入一个模糊的营销想法例如“为我们的新款智能水杯做一次小红书种草”时系统能自动完成从策略分析、内容规划到文案生成和排期发布的全流程。2. 核心架构设计与组件选型2.1 三层架构的职责定义与数据流在动手写代码之前我们必须把每一层的职责和数据交互方式定义清楚。一个混乱的架构设计后期调试会是噩梦。第一层战略管理智能体Manager Agent这是系统的大脑。它的输入是用户的原始、模糊的指令。它的核心职责是进行任务分解与战略规划。它不关心具体怎么写文案、用什么工具它只做宏观决策。输入用户原始指令String。处理调用LLM分析指令拆解成一个有序的、原子化的子任务列表。每个子任务必须包含明确的目标、所需的资源类型如图文生成、数据分析和成功标准。输出一个结构化的任务清单List[Dict]。例如针对“小红书种草”它可能输出[{id: 1, goal: 分析目标用户画像与热门话题, agent_type: analyzer}, {id: 2, goal: 生成3篇不同角度的图文草稿, agent_type: creator}, {id: 3, goal: 制定未来一周的发布排期, agent_type: planner}]。工具它通常只有思考能力不直接调用外部API。它的“工具”就是LLM本身加上我们定义的提示词Prompt来约束其输出格式。第二层任务协调智能体Coordinator Agent这是系统的中枢神经。它接收来自管理者的具体子任务并负责调度与编排合适的执行者。它理解不同执行者的能力并准备执行任务所需的上下文。输入单个子任务对象Dict以及从全局状态中获取的相关历史信息。处理根据子任务的agent_type或goal关键词决定路由到哪个第三层执行者。它可能需要为执行者收集或预处理一些信息例如从之前的任务结果中提取关键数据。输出一个格式化的执行指令包含目标、上下文和约束条件直接发送给特定的第三层智能体。工具它可能需要一些轻量级的工具比如从状态中查询数据的“工具”但其主要功能是逻辑判断和路由。第三层技能执行智能体Worker Agent这是系统的手和脚。每个执行者都是某个垂直领域的专家拥有执行特定任务所需的专用工具集。输入来自协调者的格式化执行指令。处理调用其专属的工具来完成工作。例如“文案生成执行者”会调用DALL-E API和文案LLM“数据分析执行者”可能会调用网络搜索工具和数据分析库。输出任务执行的具体结果String, Image, Data等并附带执行状态成功/失败。工具拥有丰富的外部工具调用权限如search_web,generate_image,write_copy,query_database等。数据流与状态管理 整个LangGraph的运行依赖于一个共享的“状态”State。这个状态是一个字典随着图的执行而更新。典型的状态字段包括messages: 整个对话历史是所有智能体沟通的媒介。task_list: 管理者生成的总任务列表。current_task: 当前正在处理的任务。results: 一个字典收集每个已完成任务的结果键为任务ID。next: 指示下一个应该执行的节点。数据流是单向且清晰的用户输入 - 更新状态 - 管理者节点运行 - 更新状态加入task_list- 协调者节点被触发 - 根据任务列表和状态调用对应的执行者节点 - 执行者运行并更新结果 - 循环直至所有任务完成 - 返回最终状态。2.2 为什么选择LangGraph而非其他框架市面上构建Agent的框架不少比如LangChain的原生Agent、AutoGen、CrewAI等。选择LangGraph基于以下几个关键考量显式的控制流LangGraph的核心是“图”你需要显式地定义节点和边。这对于构建三层嵌套这种复杂、有条件分支和循环的流程来说是天然契合的。你可以清晰地看到“失败重试”、“条件路由”这些逻辑在图中是如何体现的调试起来非常直观。灵活的状态管理它的状态State概念非常强大可以自定义任何结构。我们的三层架构需要传递任务列表、中间结果等复杂数据用LangGraph的状态机模型来管理比用简单的消息传递要稳健得多。对人类友好的调试LangGraph内置了可视化工具你可以把整个Agent的工作流画出来看到每个节点的人输入和出输出对于理解多层嵌套的执行顺序至关重要。与LangChain生态无缝集成我们的执行者智能体Worker Agent很可能需要用到LangChain的Tool和AgentExecutor。LangGraph本身就是LangChain家族的一员集成起来毫无障碍可以直接在节点函数里调用现有的LangChain组件。注意不要试图用单一的、超级复杂的提示词Prompt去让一个Agent完成所有三层的工作。那样做的结果通常是提示词极其臃肿成本高昂上下文长且稳定性极差。架构的意义就在于“分而治之”。2.3 基础环境搭建与模型选择首先准备好你的Python环境。我强烈建议使用虚拟环境。# 创建并激活虚拟环境 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # 或 langgraph-env\Scripts\activate # Windows # 安装核心依赖 pip install langgraph langchain langchain-openai模型的选择取决于你的需求和预算。对于管理者和协调者它们需要进行复杂的逻辑分析和规划建议使用能力较强的模型如GPT-4系列。对于执行者可以根据任务性质选择创意生成类如文案可用GPT-4简单工具调用类可用GPT-3.5 Turbo以节约成本。# 在代码中初始化LLM from langchain_openai import ChatOpenAI # 战略层使用更强的模型 manager_llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0.1) # 低随机性保证规划稳定 # 执行层可根据需要选择 worker_llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.7) # 创意任务可适当提高温度设置好你的OpenAI API密钥在环境变量中。3. 构建第一层战略管理智能体管理者的核心是生成一个可执行的任务列表。我们需要为其设计一个强大的提示词并定义好输出的结构化格式。3.1 设计管理者的系统提示词提示词的质量直接决定了任务分解的合理性。一个好的管理者提示词需要包含角色定义明确告诉模型它现在是谁。终极目标它要解决什么问题。约束条件输出的格式、必须考虑的维度、禁止做的事情。示例一两个清晰的例子让模型学会输出格式。from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder manager_prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深的项目主管和策略分析师。你的任务是将用户模糊的需求分解为具体、可执行、有序的子任务。 请遵循以下规则 1. **深度分析需求**仔细理解用户目标的深层意图、目标受众和期望效果。 2. **任务拆解**将大目标拆解为顺序执行的原子任务。后一个任务可以依赖前一个任务的输出。 3. **任务描述**每个任务必须包含清晰的目标Goal、负责此任务的智能体类型Agent Type、以及简要的上下文说明Context。 4. **智能体类型**只能使用以下几种类型analyzer分析者负责调研、分析、creator创造者负责内容生成、planner规划者负责排期、规划、reviewer审查者负责质量检查。 5. **输出格式**你必须以严格的JSON列表格式输出列表中的每个元素是一个任务对象。 示例 用户输入“我想推广我的新咖啡店” 你的输出 json [ { “id”: 1, “goal”: “分析本地咖啡市场趋势、竞争对手及目标顾客喜好” “agent_type”: “analyzer” “context”: “为后续内容创作提供数据支持和方向指导” }, { “id”: 2, “goal”: “为咖啡店设计3个营销主题并生成对应的社交媒体文案和图片创意” “agent_type”: “creator” “context”: “基于分析结果创作吸引人的内容” }, { “id”: 3, “goal”: “制定为期两周的社交媒体发布日历” “agent_type”: “planner” “context”: “合理安排内容发布节奏和渠道” } ]现在开始处理用户的需求。), MessagesPlaceholder(variable_namemessages), # 这里会放入用户的输入 ])### 3.2 创建管理者节点函数 在LangGraph中节点就是一个普通的Python函数它接收并返回状态。 python from langchain_core.messages import HumanMessage import json def manager_node(state): 战略管理节点分析需求生成任务列表。 # 1. 从状态中获取最新的用户消息 user_input state[messages][-1].content if state[messages] else # 2. 构建提示词并调用LLM prompt_value manager_prompt.invoke({messages: [HumanMessage(contentuser_input)]}) response manager_llm.invoke(prompt_value) # 3. 解析LLM的返回提取JSON任务列表 try: # 模型返回的是字符串我们需要提取JSON部分。有时模型会在文本中包裹JSON。 response_content response.content # 简单的提取找到第一个[和最后一个] start_idx response_content.find([) end_idx response_content.rfind(]) 1 if start_idx ! -1 and end_idx ! 0: task_list_json response_content[start_idx:end_idx] task_list json.loads(task_list_json) else: # 如果没找到尝试直接解析整个内容模型可能只返回了JSON task_list json.loads(response_content) except json.JSONDecodeError as e: # 如果解析失败说明模型没有按要求输出这是一个关键错误点 print(f管理者节点解析JSON失败: {e}) print(f模型原始返回: {response_content}) # 可以在这里设计一个降级策略例如返回一个默认的简单任务列表 task_list [{id: 1, goal: 分析需求失败请检查输入或模型响应。, agent_type: analyzer, context: 错误处理任务}] # 4. 更新状态 # 将任务列表存入状态并初始化结果收集器 state[task_list] task_list state[results] {} # 用于存放每个任务的执行结果 # 指示下一个节点应该是协调者 state[next] coordinator # 也可以将管理者的思考过程存入消息历史便于追溯可选 # state[messages].append(AIMessage(contentf我已将您的需求分解为{len(task_list)}个任务{task_list})) return state实操心得JSON解析是这类工作流中最脆弱的环节之一。模型并不总是乖乖输出纯净的JSON。除了用字符串查找更稳健的做法是使用LangChain的OutputParser如JsonOutputParser或者让模型使用支持结构化输出的功能如OpenAI的response_format。这里为了演示清晰使用了手动解析。在生产环境中务必加入更完善的错误处理和重试机制。4. 构建第二层任务协调智能体协调者是一个路由中心。它查看当前需要处理哪个任务然后根据任务类型决定下一步该调用哪个执行者节点。4.1 设计协调者的决策逻辑协调者不需要复杂的LLM调用它的逻辑通常是基于规则的。我们可以用一个简单的字典来映射agent_type到对应的执行者节点名。def coordinator_node(state): 任务协调节点决定下一个执行哪个任务并路由到对应的执行者。 # 1. 获取任务列表和当前结果 task_list state.get(task_list, []) results state.get(results, {}) # 2. 找出第一个未完成的任务即不在results中的任务 current_task None for task in task_list: if str(task[id]) not in results: current_task task break if not current_task: # 所有任务都已完成 state[next] __end__ return state # 3. 根据任务类型路由到不同的执行者节点 agent_type current_task.get(agent_type, ).lower() # 更新当前任务到状态中供执行者读取 state[current_task] current_task # 4. 路由决策 if agent_type analyzer: state[next] analyzer_worker elif agent_type creator: state[next] creator_worker elif agent_type planner: state[next] planner_worker elif agent_type reviewer: state[next] reviewer_worker else: # 未知类型可以路由到一个默认的错误处理节点或者直接标记失败 state[next] error_handler # 记录错误结果 task_id str(current_task[id]) state[results][task_id] { status: failed, error: f未知的智能体类型: {agent_type} } return state4.2 将协调者定义为条件边在LangGraph中协调者的这种“根据状态决定下一步”的功能通常通过定义一个条件边Conditional Edge的函数来实现而不是一个独立的节点。上面的coordinator_node函数返回的state[“next”]就指明了下一个节点。我们会在构建图的时候创建一个以coordinator_node为路由函数的条件边。5. 构建第三层技能执行智能体执行者是真正干活的。我们以creator_worker内容创造者为例展示如何构建一个具备工具调用能力的智能体。5.1 为执行者配备工具假设我们的创造者需要两个工具一个用于生成图片创意描述另一个用于撰写文案。from langchain.tools import tool from typing import Optional tool def generate_image_prompt(subject: str, style: str) - str: 根据主题和风格生成一个详细的、可供AI绘画模型使用的提示词。 # 这里可以是一个复杂的提示词模板也可以调用另一个LLM # 为了简化我们直接返回一个组合字符串 prompt fA high-quality, {style} style image of {subject}, suitable for social media marketing, clear focus, vibrant colors. return prompt tool def write_marketing_copy(topic: str, tone: str, length: str medium) - str: 根据主题、语气和长度撰写一篇营销文案。 # 在实际应用中这里会调用LLM。我们模拟一个复杂调用。 # 注意我们在这个函数内部使用LLM但这个函数本身对LangGraph的智能体来说是一个“工具”。 from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser copy_prompt ChatPromptTemplate.from_template( 你是一位专业的社交媒体文案写手。 请围绕以下主题以{tone}的语气撰写一篇{length}长度的营销文案。 主题{topic} 文案 ) chain copy_prompt | worker_llm | StrOutputParser() result chain.invoke({topic: topic, tone: tone, length: length}) return result # 将工具打包成列表 creator_tools [generate_image_prompt, write_marketing_copy]5.2 创建执行者节点函数执行者节点需要做几件事从状态中获取任务指令调用合适的工具可能通过一个LangChain Agent并将结果保存回状态。from langchain.agents import create_react_agent, AgentExecutor from langchain_core.prompts import PromptTemplate def creator_worker_node(state): 内容创造执行者节点生成营销内容和图片创意。 current_task state[current_task] task_id str(current_task[id]) goal current_task[goal] context current_task.get(context, ) # 1. 为这个执行者创建专属的提示词 worker_prompt PromptTemplate.from_template( 你是一个专业的社交媒体内容创造者。 你的当前任务是{goal} 上下文信息{context} 你已经完成了之前的所有任务相关结果已汇总在下方。 请运用你拥有的工具高质量地完成上述任务。 请逐步思考并最终输出你的创作成果。 如果使用工具请清晰说明。 之前任务汇总 {previous_results} 开始你的工作 ) # 2. 准备之前任务的结果作为上下文 previous_results_str for tid, result in state.get(results, {}).items(): if isinstance(result, dict): previous_results_str f任务{tid}: {result.get(output, str(result))}\n else: previous_results_str f任务{tid}: {result}\n # 3. 创建智能体执行器 # 使用ReAct范式让智能体能够思考并调用工具 agent create_react_agent(llmworker_llm, toolscreator_tools, promptworker_prompt) agent_executor AgentExecutor(agentagent, toolscreator_tools, verboseTrue, handle_parsing_errorsTrue) # 4. 执行任务 try: task_input { goal: goal, context: context, previous_results: previous_results_str } # 调用执行器 full_response agent_executor.invoke(task_input) final_output full_response.get(output, 任务执行完成但无输出。) except Exception as e: final_output f任务执行过程中出现错误: {str(e)} # 5. 将结果保存到状态中 state[results][task_id] { status: completed, output: final_output, agent: creator_worker } # 6. 指示下一个节点回到协调者以处理下一个任务 state[next] coordinator # 清理当前任务避免干扰下一次执行可选 # state.pop(current_task, None) return state按照类似的模式我们可以创建analyzer_worker配备网络搜索和数据分析工具、planner_worker配备日历API工具等。6. 组装与调试构建完整的LangGraph工作流现在我们将所有节点和边组装起来形成一个完整的图。6.1 定义状态结构首先我们需要定义一个状态结构告诉LangGraph我们的状态字典里都有哪些字段以及它们的类型。from typing import TypedDict, List, Dict, Any, Optional from langgraph.graph import StateGraph, END class AgentState(TypedDict): 定义工作流的状态结构。 messages: List[Any] # 消息历史 task_list: Optional[List[Dict]] # 管理者生成的任务列表 current_task: Optional[Dict] # 当前正在处理的任务 results: Dict[str, Any] # 任务ID到结果的映射 next: str # 指示下一个节点6.2 创建图并添加节点# 初始化图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(manager, manager_node) workflow.add_node(coordinator, coordinator_node) # 协调者也是一个节点但主要功能是路由 workflow.add_node(creator_worker, creator_worker_node) # 假设我们已经定义了其他worker节点 # workflow.add_node(analyzer_worker, analyzer_worker_node) # workflow.add_node(planner_worker, planner_worker_node) workflow.add_node(error_handler, error_handler_node) # 一个简单的错误处理节点6.3 设置边与条件路由这是最关键的一步定义了工作流的逻辑。# 设置入口点从manager开始 workflow.set_entry_point(manager) # 从manager出来后进入coordinator workflow.add_edge(manager, coordinator) # 从coordinator出来根据其设置的state[“next”]值动态路由到下一个节点 # 这需要一个条件边函数但我们已经把逻辑放在coordinator_node里了。 # 我们可以让coordinator_node不直接返回state而是返回next的值。 # 但更LangGraph的方式是使用add_conditional_edges。 # 首先定义一个路由函数它读取state[“next”]来决定去向 def route_next(state): next_node state.get(next) if next_node __end__: return END return next_node # 为coordinator节点添加条件边 workflow.add_conditional_edges( “coordinator” route_next # 这个函数返回下一个节点的名字 # 下面列出所有可能的目的地包括END { “creator_worker”: “creator_worker” “analyzer_worker”: “analyzer_worker” “planner_worker”: “planner_worker” “error_handler”: “error_handler” “__end__”: END # 理论上也可以路由回“coordinator”形成循环直到任务完成。 # 但我们的设计是worker完成后回到coordinator所以需要在worker节点设置state[“next”] “coordinator” } ) # 为各个worker节点添加边执行完后回到coordinator检查下一个任务 workflow.add_edge(“creator_worker” “coordinator”) # workflow.add_edge(“analyzer_worker” “coordinator”) # workflow.add_edge(“planner_worker” “coordinator”) workflow.add_edge(“error_handler” “coordinator”) # 错误处理后也继续流程6.4 编译并运行图# 编译图得到一个可执行的应用 app workflow.compile() # 为了可视化我们可以将图画出来需要安装graphviz try: from IPython.display import Image, display display(Image(app.get_graph().draw_mermaid_png())) except: print(“可视化需要graphviz和IPython环境。”) # 运行工作流 initial_state AgentState( messages[HumanMessage(content“为我们的新款智能水杯做一次小红书种草”)], task_listNone current_taskNone results{} next“” ) # 流式输出执行过程 final_state None for event in app.stream(initial_state, stream_mode“values”): node_name list(event.keys())[0] print(f”\n{‘’*20} 进入节点[{node_name}] {‘’*20}”) state event[node_name] if “current_task” in state and state[“current_task”]: print(f”当前处理任务: {state[‘current_task’].get(‘goal’)}”) if “results” in state: print(f”累计结果: {list(state[‘results’].keys())}”) final_state state print(f”\n{‘’*20} 所有任务执行完毕 {‘’*20}”) print(“最终任务结果汇总”) for task_id result in final_state[“results”].items(): print(f”任务{task_id}: {result.get(‘status’ ‘N/A’)} - 输出摘要: {str(result.get(‘output’ ‘N/A’))[:200]}…”)7. 常见问题、优化策略与避坑指南在实际搭建和运行过程中你肯定会遇到各种问题。下面是我踩过坑后总结的一些核心要点。7.1 任务分解不合理的应对策略问题管理者LLM拆解的任务可能逻辑混乱、顺序不对、或者不够原子化。解决方案强化提示词在管理者提示词中提供更详细、更优秀的示例。明确要求“后任务依赖前任务输出”。后置校验节点在管理者节点后增加一个“任务校验节点”。这个节点也是一个LLM调用它检查生成的任务列表是否合理并可以进行微调或重新排序。人工审核介入对于关键流程可以在生成任务列表后将列表呈现给用户确认或修改然后再进入执行流程。这可以通过在图中加入一个“人工审核”节点来实现该节点暂停自动化等待用户输入。7.2 执行过程中的错误处理与重试问题某个执行者节点调用工具失败如API超时、格式错误。解决方案节点内部的Try-Catch就像我们在creator_worker_node里做的那样每个执行节点都应该有完善的异常捕获并将错误信息记录到结果中而不是让整个图崩溃。全局错误处理节点我们创建的error_handler_node可以做得更智能。它可以分析错误类型决定是重试当前任务例如网络错误、跳过该任务、还是升级到人工处理。设置重试逻辑LangGraph本身不直接提供重试但可以在节点函数内实现。例如当捕获到可重试错误时将state[“next”]重新指向自己并设置一个重试计数器在状态中避免无限循环。def robust_worker_node(state): task_id str(state[“current_task”][“id”]) retry_count state.get(“retries” {}).get(task_id 0) if retry_count 2: state[“results”][task_id] {“status”: “failed” “error”: “超过最大重试次数”} state[“next”] “coordinator” return state try: # … 执行任务 … state[“results”][task_id] {“status”: “completed” “output”: result} state[“next”] “coordinator” except TransientError as e: # 假设定义了一个临时错误类型 # 更新重试计数 state.setdefault(“retries” {})[task_id] retry_count 1 # 下次再进入本节点 state[“next”] “robust_worker_node” print(f”任务{task_id}第{retry_count1}次重试…”) return state7.3 状态管理的复杂性与性能问题随着任务增多状态字典变得庞大特别是messages历史可能很长导致每次调用LLM的上下文窗口压力大、成本高。解决方案精简状态不是所有节点都需要完整的对话历史。可以为不同节点设计不同的提示词只注入相关的历史片段。例如执行者节点可能只需要它之前几个相关任务的结果而不是所有消息。状态压缩定期对messages进行总结。可以添加一个“总结器”节点在历史消息过长时调用LLM生成一个浓缩的摘要然后用摘要替换掉旧消息。外部存储对于非常大的中间结果如图片二进制数据不要放在内存状态里。可以将其存储到数据库或文件系统在状态中只保存引用ID或路径。7.4 图的监控与可观测性问题图运行起来像个黑盒哪里慢了、哪里出错了不容易定位。解决方案利用LangGraph的Streaming如上例所示使用app.stream()可以实时看到执行流经了哪些节点并打印中间状态这是最基本的调试手段。结构化日志在每个节点的开始和结束处记录结构化的日志包括节点名、任务ID、耗时、输入输出摘要等。可以使用像structlog这样的库。持久化检查点LangGraph支持将状态持久化。对于长时间运行的工作流可以定期保存检查点。如果系统崩溃可以从最近的检查点恢复而不是从头开始。可视化app.get_graph().draw_mermaid_png()生成的可视化图是理解和沟通架构的绝佳工具。7.5 架构扩展超越三层三层架构是一个强大的范式但并非一成不变。你可以根据需求扩展并行执行如果任务间没有依赖可以在协调者中实现并行路由让多个执行者同时工作。这需要更复杂的状态管理来合并结果。动态层级某些复杂的子任务本身可能也需要被进一步分解。你可以让某个执行者如analyzer_worker在内部再调用一个子图形成递归或动态的嵌套结构。反馈循环引入reviewer节点对执行结果进行审核如果质量不达标可以将其反馈给coordinator由coordinator重新调度给creator进行修改。这就在图中形成了一个质量控制的循环。搭建这样一个三层嵌套的Agent架构初期投入的思考和时间会比较多但一旦跑通它带来的清晰度、可维护性和处理复杂任务的能力是简单链式调用无法比拟的。最关键的是理解“状态”如何在图中流动以及每个节点如何读取和修改这个共享状态。从一个小而具体的场景开始逐步增加复杂度和节点是掌握LangGraph的最佳路径。