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基于以下几个关键考量：

1. 显式的控制流：LangGraph的核心是“图”，你需要显式地定义节点和边。这对于构建三层嵌套这种复杂、有条件分支和循环的流程来说，是天然契合的。你可以清晰地看到“失败重试”、“条件路由”这些逻辑在图中是如何体现的，调试起来非常直观。
2. 灵活的状态管理：它的状态（State）概念非常强大，可以自定义任何结构。我们的三层架构需要传递任务列表、中间结果等复杂数据，用LangGraph的状态机模型来管理比用简单的消息传递要稳健得多。
3. 对人类友好的调试：LangGraph内置了可视化工具，你可以把整个Agent的工作流画出来，看到每个节点的人（输入）和出（输出），对于理解多层嵌套的执行顺序至关重要。
4. 与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(model="gpt-4-turbo-preview", temperature=0.1) # 低随机性，保证规划稳定 # 执行层可根据需要选择 worker_llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0.7) # 创意任务可适当提高温度

设置好你的OpenAI API密钥在环境变量中。

3. 构建第一层：战略管理智能体

管理者的核心是生成一个可执行的任务列表。我们需要为其设计一个强大的提示词，并定义好输出的结构化格式。

3.1 设计管理者的系统提示词

提示词的质量直接决定了任务分解的合理性。一个好的管理者提示词需要包含：

1. 角色定义：明确告诉模型它现在是谁。
2. 终极目标：它要解决什么问题。
3. 约束条件：输出的格式、必须考虑的维度、禁止做的事情。
4. 示例：一两个清晰的例子，让模型学会输出格式。

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_name="messages"), # 这里会放入用户的输入 ])

### 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(content=user_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(content=f"我已将您的需求分解为{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 state

4.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 = f"A 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(llm=worker_llm, tools=creator_tools, prompt=worker_prompt) agent_executor = AgentExecutor(agent=agent, tools=creator_tools, verbose=True, handle_parsing_errors=True) # 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_list=None， current_task=None， 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拆解的任务可能逻辑混乱、顺序不对、或者不够原子化。解决方案：

1. 强化提示词：在管理者提示词中提供更详细、更优秀的示例。明确要求“后任务依赖前任务输出”。
2. 后置校验节点：在管理者节点后，增加一个“任务校验节点”。这个节点也是一个LLM调用，它检查生成的任务列表是否合理，并可以进行微调或重新排序。
3. 人工审核介入：对于关键流程，可以在生成任务列表后，将列表呈现给用户确认或修改，然后再进入执行流程。这可以通过在图中加入一个“人工审核”节点来实现，该节点暂停自动化，等待用户输入。

7.2 执行过程中的错误处理与重试

问题：某个执行者节点调用工具失败（如API超时、格式错误）。解决方案：

1. 节点内部的Try-Catch：就像我们在creator_worker_node里做的那样，每个执行节点都应该有完善的异常捕获，并将错误信息记录到结果中，而不是让整个图崩溃。
2. 全局错误处理节点：我们创建的error_handler_node可以做得更智能。它可以分析错误类型，决定是重试当前任务（例如，网络错误）、跳过该任务、还是升级到人工处理。
3. 设置重试逻辑：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_count+1}次重试…”) return state

7.3 状态管理的复杂性与性能

问题：随着任务增多，状态字典变得庞大，特别是messages历史可能很长，导致每次调用LLM的上下文窗口压力大、成本高。解决方案：

1. 精简状态：不是所有节点都需要完整的对话历史。可以为不同节点设计不同的提示词，只注入相关的历史片段。例如，执行者节点可能只需要它之前几个相关任务的结果，而不是所有消息。
2. 状态压缩：定期对messages进行总结。可以添加一个“总结器”节点，在历史消息过长时，调用LLM生成一个浓缩的摘要，然后用摘要替换掉旧消息。
3. 外部存储：对于非常大的中间结果（如图片二进制数据），不要放在内存状态里。可以将其存储到数据库或文件系统，在状态中只保存引用ID或路径。

7.4 图的监控与可观测性

问题：图运行起来像个黑盒，哪里慢了、哪里出错了不容易定位。解决方案：

1. 利用LangGraph的Streaming：如上例所示，使用app.stream()可以实时看到执行流经了哪些节点，并打印中间状态，这是最基本的调试手段。
2. 结构化日志：在每个节点的开始和结束处，记录结构化的日志，包括节点名、任务ID、耗时、输入输出摘要等。可以使用像structlog这样的库。
3. 持久化检查点：LangGraph支持将状态持久化。对于长时间运行的工作流，可以定期保存检查点。如果系统崩溃，可以从最近的检查点恢复，而不是从头开始。
4. 可视化：app.get_graph().draw_mermaid_png()生成的可视化图是理解和沟通架构的绝佳工具。

7.5 架构扩展：超越三层

三层架构是一个强大的范式，但并非一成不变。你可以根据需求扩展：

• 并行执行：如果任务间没有依赖，可以在协调者中实现并行路由，让多个执行者同时工作。这需要更复杂的状态管理来合并结果。
• 动态层级：某些复杂的子任务本身可能也需要被进一步分解。你可以让某个执行者（如analyzer_worker）在内部再调用一个子图，形成递归或动态的嵌套结构。
• 反馈循环：引入reviewer节点对执行结果进行审核，如果质量不达标，可以将其反馈给coordinator，由coordinator重新调度给creator进行修改。这就在图中形成了一个质量控制的循环。

搭建这样一个三层嵌套的Agent架构，初期投入的思考和时间会比较多，但一旦跑通，它带来的清晰度、可维护性和处理复杂任务的能力是简单链式调用无法比拟的。最关键的是理解“状态”如何在图中流动，以及每个节点如何读取和修改这个共享状态。从一个小而具体的场景开始，逐步增加复杂度和节点，是掌握LangGraph的最佳路径。