1. 项目概述当AI Agent需要“记忆”与“协作”最近在设计和实现一个复杂的AI Agent系统时我遇到了一个核心架构选择难题是使用StateGraph还是MessageGraph这不仅仅是LangGraph框架里的两个类名更是两种截然不同的Agent状态管理哲学。简单来说StateGraph像是给Agent配备了一个结构化的“工作笔记本”所有信息分门别类地记录而MessageGraph则更像是一个线性的“聊天记录”一切流转皆由消息驱动。这个选择直接决定了你的Agent系统是更擅长处理有复杂内部状态的长期任务还是更专注于轻量、灵活的多轮对话。对于任何想要深入AI Agent开发特别是涉及多Agent协作、复杂工作流的开发者而言理解这两者的权衡是绕不开的一课。本文将结合我踩过的坑和实战经验为你彻底拆解StateGraph与MessageGraph的选型逻辑。2. 核心理念拆解状态容器 vs 消息总线要做出正确的选型首先得抛开代码从设计理念上理解它们各自代表了什么。2.1 StateGraph基于结构化状态的有限状态机StateGraph的核心思想来源于有限状态机。它将Agent的执行过程抽象为在不同“状态”之间的转移。这里的“状态”是一个强类型的、结构化的数据容器。什么是“状态”在StateGraph中状态State通常是一个Pydantic模型或字典明确定义了Agent在某一时刻所知晓的所有信息。例如一个研究Agent的状态可能包含topic研究主题、collected_papers已收集的论文列表、analysis_report分析报告草稿等字段。每个字段都有明确的类型和含义。如何工作Agent的执行被分解为多个节点Nodes每个节点是一个函数。这个函数接收当前完整的状态作为输入执行某些操作如调用LLM、查询网络然后返回一个更新后的状态。通过边Edges定义状态如何从一个节点流转到下一个节点。编译器compile会将这个图结构转化为一个可执行的、维护着单一状态对象的链式调用。注意StateGraph中的状态是累积且共享的。一个节点对状态的修改例如向collected_papers列表添加一篇新论文会直接反映在全局状态中并被后续所有节点看到。这既是其强大之处也带来了数据耦合的风险。2.2 MessageGraph基于消息传递的参与者模型MessageGraph的核心思想则更接近Actor模型或事件驱动架构。它不维护一个中心化的、结构化的状态对象而是依靠消息在节点之间传递。什么是“消息”在MessageGraph中信息传递的基本单元是消息Message通常对应LangChain的AIMessage,HumanMessage,SystemMessage等。节点之间的通信完全通过向一个共享的“消息列表”追加消息来完成。如何工作每个节点也是一个函数但它接收和返回的不是一个状态字典而是一个消息列表或类似的可追加结构。节点读取之前消息列表中的内容生成新的消息并追加到列表末尾从而将信息传递给下一个节点。图的流转条件也往往基于最新消息的内容来决定。提示MessageGraph的思维模式更贴近我们熟悉的聊天机器人。你不需要一个“用户偏好”状态字段只需要看历史对话消息里用户说过什么。它的状态是隐式的蕴含在完整的消息序列中。2.3 核心差异对比表为了更直观地对比我将两者的核心差异总结如下特性维度StateGraphMessageGraph数据模型中心化的、结构化的状态对象如Pydantic Model。线性的、序列化的消息列表List[BaseMessage]。思维模式有限状态机FSM。关心“当前处于什么情况拥有什么数据”。参与者模型Actor。关心“收到了什么信息需要回复什么”。数据流状态对象在节点间传递和变异。节点直接修改共享状态。消息在节点间追加和广播。节点通过读写消息列表通信。优势状态管理清晰适合复杂、多步骤且需要维护丰富中间数据的业务流程。易于实现条件分支和循环。轻量、灵活天然适合对话场景。节点间耦合度低易于测试和复用单个节点。劣势状态结构需预先定义变更成本较高。节点间因共享状态而耦合需要谨慎设计以避免副作用。复杂状态难以维护例如需要从历史消息中频繁提取、汇总信息。不适合数据密集型操作。适用场景研究助手、数据分析流水线、复杂决策工作流、需要“长期记忆”的Agent。聊天机器人、简单的多轮工具调用、基于对话历史的路由、消息转换管道。3. 选型决策的实战权衡点理解了理念差异后在实际项目中该如何抉择我通常会从以下几个维度进行考量。3.1 场景复杂度与数据形态这是最根本的决策点。问自己一个问题我的Agent流程需要操作的数据是结构化的业务对象还是非结构化的自然语言对话选择StateGraph如果你的流程有明确的阶段每个阶段需要产生和消费结构化的数据。例如一个“市场报告生成Agent”步骤可能是1. 获取关键词字符串2. 爬取新闻列表[文章]3. 情感分析字典{文章: 情感}4. 生成报告字符串。这里的“文章”、“情感字典”都是结构化数据非常适合用State来承载。你需要对中间结果进行复杂的条件判断。例如根据“已收集资料是否充足”这个状态字段的值决定是进入“深入分析”节点还是“补充收集”节点。在StateGraph中这可以通过条件边conditional_edge优雅实现。选择MessageGraph如果你的流程本质上是多轮对话。例如一个“客服Agent”根据用户的最新问题结合对话历史决定调用知识库查询工具还是转人工。整个上下文就是对话历史用Message列表表示再自然不过。你的节点功能非常独立只是对输入消息进行某种转换或响应不需要关心一个庞大的全局状态。例如一个消息过滤节点、一个格式美化节点。3.2 开发体验与维护成本不同的模型对开发者的心智负担和项目后期的维护成本影响巨大。StateGraph的开发体验前期设计成本高你需要像设计数据库Schema一样仔细设计State模型。字段名、类型、默认值都需要考虑周全。一旦定义后期修改如增加字段可能涉及多个节点的修改。类型安全与IDE支持如果使用Pydantic模型定义State你将获得完美的类型提示和自动补全能极大减少运行时错误。这是StateGraph在大型项目中的一大杀手锏。调试直观因为所有数据都在一个状态对象里你可以在任何节点打印或记录完整状态对理解流程执行逻辑非常有帮助。MessageGraph的开发体验上手快速对于熟悉聊天机器人开发的开发者来说MessageGraph的模式非常直观。你只需要关心当前节点输入了什么消息要返回什么消息。灵活度高节点功能单一耦合度低很容易单独测试和复用。你可以像搭积木一样组合不同的消息处理节点。状态追踪困难当流程复杂后你需要从冗长的消息历史中手动解析出当前所需的“状态”比如用户的偏好代码会变得难以维护。虽然可以通过设计特定的消息类型如包含结构化数据的ToolMessage来缓解但这本质上是在MessageGraph中模拟一个轻量级State。3.3 与LangChain生态的集成你的技术栈选择也会影响决策。StateGraph与LangChain的Runnable协议集成得非常好。因为每个节点本质上是一个接收状态、返回状态的函数它可以很容易地封装RunnableLambda或与其他Runnable组件链式调用。这对于构建复杂、可组合的工作流非常有利。MessageGraph则与LangChain的LCEL和Runnable中的消息处理部分天然契合。许多现成的Runnable组件如ChatPromptTemplate,LLM,Tools都直接处理消息列表。在MessageGraph中集成这些组件几乎不需要任何适配。实操心得我个人的经验是对于中型以上、业务逻辑复杂的Agent项目从StateGraph开始往往是更稳妥的选择。虽然前期设计费点劲但清晰的状态结构就像项目的“骨架”能迫使你更好地抽象业务后期扩展和调试收益巨大。而MessageGraph更适合快速原型验证一个对话式创意或者作为大系统中一个处理纯对话的子模块。4. 从零实现两个Graph的代码级对比光说不练假把式。我们通过一个具体的场景来对比实现一个“智能任务分解与执行Agent”。它接收一个复杂任务如“策划一场线上技术分享会”先将其分解为子任务然后为每个子任务选择并执行合适的工具。4.1 使用StateGraph的实现首先我们需要定义状态模型。这是最关键的一步。from typing import List, Optional, Annotated from typing_extensions import TypedDict from pydantic import BaseModel import operator # 1. 定义强类型状态模型 class AgentState(TypedDict): Agent的全局状态容器 original_task: str # 原始任务 subtasks: List[str] # 分解后的子任务列表 current_subtask_index: int # 当前正在执行的子任务索引 current_subtask: Optional[str] # 当前子任务内容 tool_for_subtask: Optional[str] # 为当前子任务选择的工具名 tool_result: Optional[str] # 工具执行结果 final_results: List[str] # 所有子任务的结果汇总 is_complete: bool # 所有任务是否完成 # 2. 定义各个节点函数 def task_decomposition_node(state: AgentState) - AgentState: 节点任务分解 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4) prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深的项目规划助手。请将以下复杂任务分解为3-5个清晰的、可顺序执行的子任务。只输出子任务列表每行一个。), (human, 任务{task}) ]) chain prompt | llm response chain.invoke({task: state[original_task]}) # 解析LLM返回更新状态 subtasks [line.strip(- ).strip() for line in response.content.split(\n) if line.strip()] state[subtasks] subtasks state[current_subtask_index] 0 state[final_results] [] return state def select_tool_node(state: AgentState) - AgentState: 节点为当前子任务选择工具 current_task state[subtasks][state[current_subtask_index]] state[current_subtask] current_task # 简单的规则匹配实际中可用LLM路由 if 邮件 in current_task or 邀请 in current_task: state[tool_for_subtask] send_email_tool elif 调研 in current_task or 搜索 in current_task: state[tool_for_subtask] web_search_tool elif 撰写 in current_task or 文档 in current_task: state[tool_for_subtask] draft_document_tool else: state[tool_for_subtask] general_qa_tool return state def execute_tool_node(state: AgentState) - AgentState: 节点执行工具 tool_name state[tool_for_subtask] task state[current_subtask] # 模拟工具执行 mock_tool_results { send_email_tool: f已发送关于{task}的邮件。, web_search_tool: f已完成对{task}的搜索找到10条相关信息。, draft_document_tool: f已起草关于{task}的文档初稿。, general_qa_tool: f已处理通用任务{task}。 } state[tool_result] mock_tool_results.get(tool_name, 工具执行失败) state[final_results].append(f子任务[{state[current_subtask_index]1}]: {task} - 结果: {state[tool_result]}) return state def check_completion_node(state: AgentState) - AgentState: 节点检查是否所有子任务完成 next_index state[current_subtask_index] 1 if next_index len(state[subtasks]): state[is_complete] True else: state[current_subtask_index] next_index state[current_subtask] None state[tool_for_subtask] None state[tool_result] None return state # 3. 构建并编译StateGraph from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 添加节点 workflow.add_node(decompose, task_decomposition_node) workflow.add_node(select_tool, select_tool_node) workflow.add_node(execute_tool, execute_tool_node) workflow.add_node(check_completion, check_completion_node) # 设置边 workflow.set_entry_point(decompose) workflow.add_edge(decompose, select_tool) workflow.add_edge(select_tool, execute_tool) workflow.add_edge(execute_tool, check_completion) # 条件边如果未完成循环回select_tool如果完成结束。 workflow.add_conditional_edges( check_completion, # 判断函数根据状态中的 is_complete 字段决定下一步 lambda state: END if state[is_complete] else select_tool, {END: END, select_tool: select_tool} ) # 编译图 app workflow.compile() # 4. 执行 initial_state {original_task: 策划一场面向开发者的线上AI技术分享会, is_complete: False} final_state app.invoke(initial_state) print(final_state[final_results])关键解读状态驱动整个流程的推进完全依赖于AgentState这个中心对象的演变。current_subtask_index和is_complete字段是控制循环的关键。清晰的数据流每个节点接收旧状态返回新状态。你可以清晰地看到数据如何被一步步加工。条件逻辑add_conditional_edges使得实现“循环执行直到完成”这种逻辑变得非常简单直观。4.2 使用MessageGraph的实现现在我们用MessageGraph来实现类似功能。思维模式需要转换不再有中心状态一切靠消息传递。from langgraph.graph import MessageGraph from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage from typing import List # 1. 定义节点函数现在它们接收和返回的是消息列表 def task_decomposition_node(messages: List) - List: 节点从最新消息中提取任务并分解 last_message messages[-1] if isinstance(last_message, HumanMessage): user_task last_message.content # 模拟LLM分解任务同上略 subtasks [f1. 确定分享主题与大纲, f2. 邀请讲师与嘉宾, f3. 宣传推广, f4. 进行线上直播, f5. 收集反馈] # 关键不返回状态而是返回新的AIMessage # 我们将子任务列表作为内容返回。在实际中可能需要更结构化的消息如使用ToolMessage。 response_content f任务已分解为以下子任务\n \n.join(subtasks) # 为了后续节点能识别我们可以把结构化数据放在一个特殊的键里这是一种变通 # 这里为了简单我们只是返回文本。更复杂的做法需要自定义消息类型。 return messages [AIMessage(contentresponse_content, additional_kwargs{subtasks: subtasks})] return messages [AIMessage(content请提供一个任务让我来分解。)] def select_tool_node(messages: List) - List: 节点选择工具。需要从历史消息中‘解析’出当前子任务。 # 这里就体现出MessageGraph的麻烦了我们需要手动回溯消息找到子任务信息。 # 假设上一步的AIMessage的additional_kwargs里存了subtasks并且我们用一个索引记录进度。 # 我们需要维护一个“隐式”的索引。这通常需要借助图的“内存”功能或者将索引放在某条消息的元数据中。 # 为了示例我们简化假设总是处理第一个子任务。 subtasks None for msg in reversed(messages): if isinstance(msg, AIMessage) and subtasks in msg.additional_kwargs: subtasks msg.additional_kwargs[subtasks] break if not subtasks: return messages [AIMessage(content未找到可处理的子任务。)] current_task subtasks[0] # 简化处理 # ... 工具选择逻辑同上 tool_name web_search_tool if 调研 in current_task else general_qa_tool # 返回一条包含工具调用信息的消息 return messages [AIMessage(contentf将为子任务『{current_task}』选择工具{tool_name}, additional_kwargs{selected_tool: tool_name, current_task: current_task})] def execute_tool_node(messages: List) - List: 节点执行工具 # 同样需要从历史消息中找到 selected_tool 和 current_task selected_tool None current_task None for msg in reversed(messages): if isinstance(msg, AIMessage) and selected_tool in msg.additional_kwargs: selected_tool msg.additional_kwargs[selected_tool] current_task msg.additional_kwargs.get(current_task) break # ... 模拟执行工具同上 result f模拟执行工具 {selected_tool} 于任务『{current_task}』的结果。 # 返回工具执行结果消息 return messages [ToolMessage(contentresult, tool_call_idmock_id)] # 2. 构建MessageGraph workflow MessageGraph() workflow.add_node(decompose, task_decomposition_node) workflow.add_node(select_tool, select_tool_node) workflow.add_node(execute_tool, execute_tool_node) workflow.set_entry_point(decompose) workflow.add_edge(decompose, select_tool) workflow.add_edge(select_tool, execute_tool) workflow.add_edge(execute_tool, END) # 简单结束没有循环 app workflow.compile() # 3. 执行 initial_messages [HumanMessage(content策划一场线上AI技术分享会)] result app.invoke(initial_messages) for msg in result: print(f{type(msg).__name__}: {msg.content[:50]}...)关键解读与困境消息传递每个节点都在消息列表末尾追加新消息信息通过消息流传递。状态解析困境在select_tool_node和execute_tool_node中我们需要从过往的所有消息中“翻找”所需的信息如子任务列表、当前任务索引。这通过reversed(messages)循环和检查additional_kwargs实现代码变得冗长且低效。循环逻辑难以实现我们这个简化的例子无法实现“循环处理所有子任务”。因为MessageGraph本身没有内置的循环计数器。要实现它你必须要么在消息中携带一个显式的“当前索引”污染了消息内容的主要目的。要么使用MessageGraph提供的add_messages功能来维护一个简单的、非结构化的“内存”但这本质上是在向StateGraph靠拢。要么将循环逻辑外置通过多次调用图来实现这破坏了图的完整性和封装性。踩坑实录在MessageGraph中实现一个需要维护多个数据字段和循环状态的复杂流程你会发现自己花大量精力在“消息考古学”上——不断解析历史消息来重建当前状态。这不仅代码难看性能也差。此时StateGraph的结构化状态优势就无比明显。5. 高级模式与混合使用策略在实际的大型项目中非此即彼的选择并不多见。LangGraph的巧妙之处在于它允许甚至鼓励混合使用。5.1 在StateGraph中嵌入消息处理这是最常见也最强大的模式。StateGraph的状态模型中可以包含一个messages: List[BaseMessage]字段。这样你既拥有了结构化的业务状态如current_step,collected_data又保留了完整的对话历史供LLM节点消费。class HybridState(TypedDict): structured_data: dict # 你的业务数据 messages: List[BaseMessage] # 对话历史 current_step: str def llm_node_with_memory(state: HybridState): 一个既能看到业务数据又能看到对话历史的节点 # 从structured_data中提取业务上下文 business_context state[structured_data] # 将业务上下文和对话历史一起构建Prompt prompt build_prompt(business_context, state[messages]) # 调用LLM... new_ai_message llm.invoke(prompt) # 更新状态既更新业务数据也追加消息 state[messages].append(new_ai_message) state[structured_data][last_llm_output] extract_structured_info(new_ai_message) state[current_step] decide_next_step(new_ai_message) return state这种模式完美结合了两者的优点用结构化状态控制复杂流程逻辑用消息列表保持与LLM交互的灵活性和上下文完整性。5.2 将MessageGraph作为StateGraph的一个节点另一种模式是将一个完整的MessageGraph例如一个负责处理用户单轮问答的子对话系统编译成一个Runnable然后将其作为StateGraph的一个节点。这样StateGraph负责宏观流程和状态管理而具体的、对话密集的子系统则交给更擅长的MessageGraph去处理。# 假设我们已经定义好一个客服对话的MessageGraph并编译为 customer_service_graph from langchain_core.runnables import RunnableLambda def customer_service_node(state: MainState): StateGraph的节点内部调用一个MessageGraph子流程 user_query state[latest_query] # 将主状态中的某些信息作为初始消息传入子图 initial_messages [HumanMessage(contentuser_query), SystemMessage(contentf用户信息{state[user_profile]})] # 运行子MessageGraph conversation_result customer_service_graph.invoke(initial_messages) # 从子图的结果消息中提取关键信息更新主状态 resolution extract_resolution(conversation_result[-1].content) state[query_resolution] resolution state[conversation_history].extend(conversation_result) # 可选保存子对话历史 return state # 在主StateGraph中添加这个节点 main_workflow.add_node(handle_customer_service, RunnableLambda(customer_service_node))5.3 选型决策流程图为了帮助你在项目初期快速决策我总结了一个简单的决策流程你的核心流程是否严重依赖多轮、自由的对话历史是- 优先考虑MessageGraph或HybridState带messages字段。否- 进入下一步。你的流程是否需要维护多个明确的、结构化的中间数据字段并基于它们做复杂路由或循环是- 毫不犹豫选择StateGraph。否- 进入下一步。你的流程是否简单、线性主要是对输入进行一系列转换是-MessageGraph可能更轻量、合适。否- 回到第一步重新评估或直接采用StateGraph以获得更好的可扩展性。个人经验法则对于大多数涉及“任务”、“流程”、“工作流”这些字眼的AI Agent应用StateGraph通常是更优的起点。它的结构化约束在项目增长时带来的益处远大于初期的设计成本。MessageGraph则是你工具箱里一把锋利的“手术刀”专门用于处理纯对话切片或作为大系统的组件。6. 常见陷阱与性能调优指南即使选对了模型在实际开发中依然会遇到不少坑。以下是一些高频问题和解决方案。6.1 StateGraph的常见陷阱状态模型设计过载或不足问题一开始把所有能想到的字段都塞进State导致状态臃肿每个节点都需要处理大量无关字段。或者相反字段设计不足后期频繁修改模型破坏兼容性。解决遵循“最小化”和“前瞻性”原则。只定义当前节点确实需要读写字段。但对于核心业务实体如User,Order可以将其作为整体字段放入即使初期只用其中一两个属性。节点副作用污染状态问题多个节点意外修改了同一个状态字段导致难以调试的竞态条件或逻辑错误。解决为节点函数编写清晰的文档说明其读写哪些字段。在复杂流程中可以考虑使用不可变数据模式如返回全新的状态字典但这会牺牲一些便利性。更实用的方法是在关键节点前后打印或记录状态快照。条件边conditional_edge逻辑过于复杂问题判断下一个节点的函数router写得非常复杂难以理解和测试。解决将复杂的路由逻辑封装成独立的、可测试的函数。甚至可以为不同的路由条件创建专门的“路由节点”该节点只负责检查状态并返回下一个节点名称使主图结构更清晰。6.2 MessageGraph的常见陷阱消息列表无限膨胀问题长时间运行的对话导致消息列表巨大每次调用LLM的token成本剧增且影响节点查找历史信息的性能。解决实现消息摘要或窗口化。可以定期用一个节点将冗长的历史消息总结成一条系统消息然后清空或截断旧消息。LangGraph的Checkpointer机制也能帮助管理长上下文。从消息中提取状态信息效率低下问题每个节点都需要for msg in reversed(messages):来查找信息O(n)复杂度。解决约定俗成地将关键结构化数据放在最新一条消息的additional_kwargs中。或者直接使用Hybrid模式在MessageGraph的“内存”或父StateGraph的状态中维护一个轻量级的状态索引。难以实现复杂循环和状态判断问题如前文所述这是MessageGraph的天然短板。解决不要强求。如果你的流程逻辑变得复杂这正是考虑切换到StateGraph或采用混合架构的信号。6.3 性能调优要点图的编译与预热app workflow.compile()会产生一些开销。在生产环境中考虑在服务启动时预先编译好常用的图并将其缓存。异步支持LangGraph节点函数天然支持async def。如果你的节点涉及网络IO如调用LLM API、查询数据库务必使用异步函数并用ainvoke来并发执行可以极大提升吞吐量。状态序列化如果你使用LangGraph的持久化或检查点功能StateGraph的状态对象需要能被序列化如Pickle或JSON。确保你的状态模型中只包含可序列化的Python类型。对于复杂对象考虑将其转换为字典或字符串存储。节点粒度不要把所有逻辑塞进一个巨型节点。将节点拆分为功能单一、职责明确的单元有利于复用、测试和并行化。但节点过多也会增加图的管理开销和调用延迟需要在清晰度和性能间取得平衡。最后无论选择哪种Graph充分的测试都至关重要。为每个节点函数编写单元测试模拟输入状态或消息验证输出。对于整个图编写集成测试验证从初始输入到最终输出的完整流程是否符合预期。LangGraph的确定性执行在给定相同输入和随机种子下特性使得测试变得相对可行。记住一个设计良好的AI Agent系统其核心逻辑的可靠性和可维护性很大程度上就取决于你对StateGraph和MessageGraph这些基础构建块的理解与运用。 AI Agent开发：StateGraph与MessageGraph的选型指南
1. 项目概述当AI Agent需要“记忆”与“协作”最近在设计和实现一个复杂的AI Agent系统时我遇到了一个核心架构选择难题是使用StateGraph还是MessageGraph这不仅仅是LangGraph框架里的两个类名更是两种截然不同的Agent状态管理哲学。简单来说StateGraph像是给Agent配备了一个结构化的“工作笔记本”所有信息分门别类地记录而MessageGraph则更像是一个线性的“聊天记录”一切流转皆由消息驱动。这个选择直接决定了你的Agent系统是更擅长处理有复杂内部状态的长期任务还是更专注于轻量、灵活的多轮对话。对于任何想要深入AI Agent开发特别是涉及多Agent协作、复杂工作流的开发者而言理解这两者的权衡是绕不开的一课。本文将结合我踩过的坑和实战经验为你彻底拆解StateGraph与MessageGraph的选型逻辑。2. 核心理念拆解状态容器 vs 消息总线要做出正确的选型首先得抛开代码从设计理念上理解它们各自代表了什么。2.1 StateGraph基于结构化状态的有限状态机StateGraph的核心思想来源于有限状态机。它将Agent的执行过程抽象为在不同“状态”之间的转移。这里的“状态”是一个强类型的、结构化的数据容器。什么是“状态”在StateGraph中状态State通常是一个Pydantic模型或字典明确定义了Agent在某一时刻所知晓的所有信息。例如一个研究Agent的状态可能包含topic研究主题、collected_papers已收集的论文列表、analysis_report分析报告草稿等字段。每个字段都有明确的类型和含义。如何工作Agent的执行被分解为多个节点Nodes每个节点是一个函数。这个函数接收当前完整的状态作为输入执行某些操作如调用LLM、查询网络然后返回一个更新后的状态。通过边Edges定义状态如何从一个节点流转到下一个节点。编译器compile会将这个图结构转化为一个可执行的、维护着单一状态对象的链式调用。注意StateGraph中的状态是累积且共享的。一个节点对状态的修改例如向collected_papers列表添加一篇新论文会直接反映在全局状态中并被后续所有节点看到。这既是其强大之处也带来了数据耦合的风险。2.2 MessageGraph基于消息传递的参与者模型MessageGraph的核心思想则更接近Actor模型或事件驱动架构。它不维护一个中心化的、结构化的状态对象而是依靠消息在节点之间传递。什么是“消息”在MessageGraph中信息传递的基本单元是消息Message通常对应LangChain的AIMessage,HumanMessage,SystemMessage等。节点之间的通信完全通过向一个共享的“消息列表”追加消息来完成。如何工作每个节点也是一个函数但它接收和返回的不是一个状态字典而是一个消息列表或类似的可追加结构。节点读取之前消息列表中的内容生成新的消息并追加到列表末尾从而将信息传递给下一个节点。图的流转条件也往往基于最新消息的内容来决定。提示MessageGraph的思维模式更贴近我们熟悉的聊天机器人。你不需要一个“用户偏好”状态字段只需要看历史对话消息里用户说过什么。它的状态是隐式的蕴含在完整的消息序列中。2.3 核心差异对比表为了更直观地对比我将两者的核心差异总结如下特性维度StateGraphMessageGraph数据模型中心化的、结构化的状态对象如Pydantic Model。线性的、序列化的消息列表List[BaseMessage]。思维模式有限状态机FSM。关心“当前处于什么情况拥有什么数据”。参与者模型Actor。关心“收到了什么信息需要回复什么”。数据流状态对象在节点间传递和变异。节点直接修改共享状态。消息在节点间追加和广播。节点通过读写消息列表通信。优势状态管理清晰适合复杂、多步骤且需要维护丰富中间数据的业务流程。易于实现条件分支和循环。轻量、灵活天然适合对话场景。节点间耦合度低易于测试和复用单个节点。劣势状态结构需预先定义变更成本较高。节点间因共享状态而耦合需要谨慎设计以避免副作用。复杂状态难以维护例如需要从历史消息中频繁提取、汇总信息。不适合数据密集型操作。适用场景研究助手、数据分析流水线、复杂决策工作流、需要“长期记忆”的Agent。聊天机器人、简单的多轮工具调用、基于对话历史的路由、消息转换管道。3. 选型决策的实战权衡点理解了理念差异后在实际项目中该如何抉择我通常会从以下几个维度进行考量。3.1 场景复杂度与数据形态这是最根本的决策点。问自己一个问题我的Agent流程需要操作的数据是结构化的业务对象还是非结构化的自然语言对话选择StateGraph如果你的流程有明确的阶段每个阶段需要产生和消费结构化的数据。例如一个“市场报告生成Agent”步骤可能是1. 获取关键词字符串2. 爬取新闻列表[文章]3. 情感分析字典{文章: 情感}4. 生成报告字符串。这里的“文章”、“情感字典”都是结构化数据非常适合用State来承载。你需要对中间结果进行复杂的条件判断。例如根据“已收集资料是否充足”这个状态字段的值决定是进入“深入分析”节点还是“补充收集”节点。在StateGraph中这可以通过条件边conditional_edge优雅实现。选择MessageGraph如果你的流程本质上是多轮对话。例如一个“客服Agent”根据用户的最新问题结合对话历史决定调用知识库查询工具还是转人工。整个上下文就是对话历史用Message列表表示再自然不过。你的节点功能非常独立只是对输入消息进行某种转换或响应不需要关心一个庞大的全局状态。例如一个消息过滤节点、一个格式美化节点。3.2 开发体验与维护成本不同的模型对开发者的心智负担和项目后期的维护成本影响巨大。StateGraph的开发体验前期设计成本高你需要像设计数据库Schema一样仔细设计State模型。字段名、类型、默认值都需要考虑周全。一旦定义后期修改如增加字段可能涉及多个节点的修改。类型安全与IDE支持如果使用Pydantic模型定义State你将获得完美的类型提示和自动补全能极大减少运行时错误。这是StateGraph在大型项目中的一大杀手锏。调试直观因为所有数据都在一个状态对象里你可以在任何节点打印或记录完整状态对理解流程执行逻辑非常有帮助。MessageGraph的开发体验上手快速对于熟悉聊天机器人开发的开发者来说MessageGraph的模式非常直观。你只需要关心当前节点输入了什么消息要返回什么消息。灵活度高节点功能单一耦合度低很容易单独测试和复用。你可以像搭积木一样组合不同的消息处理节点。状态追踪困难当流程复杂后你需要从冗长的消息历史中手动解析出当前所需的“状态”比如用户的偏好代码会变得难以维护。虽然可以通过设计特定的消息类型如包含结构化数据的ToolMessage来缓解但这本质上是在MessageGraph中模拟一个轻量级State。3.3 与LangChain生态的集成你的技术栈选择也会影响决策。StateGraph与LangChain的Runnable协议集成得非常好。因为每个节点本质上是一个接收状态、返回状态的函数它可以很容易地封装RunnableLambda或与其他Runnable组件链式调用。这对于构建复杂、可组合的工作流非常有利。MessageGraph则与LangChain的LCEL和Runnable中的消息处理部分天然契合。许多现成的Runnable组件如ChatPromptTemplate,LLM,Tools都直接处理消息列表。在MessageGraph中集成这些组件几乎不需要任何适配。实操心得我个人的经验是对于中型以上、业务逻辑复杂的Agent项目从StateGraph开始往往是更稳妥的选择。虽然前期设计费点劲但清晰的状态结构就像项目的“骨架”能迫使你更好地抽象业务后期扩展和调试收益巨大。而MessageGraph更适合快速原型验证一个对话式创意或者作为大系统中一个处理纯对话的子模块。4. 从零实现两个Graph的代码级对比光说不练假把式。我们通过一个具体的场景来对比实现一个“智能任务分解与执行Agent”。它接收一个复杂任务如“策划一场线上技术分享会”先将其分解为子任务然后为每个子任务选择并执行合适的工具。4.1 使用StateGraph的实现首先我们需要定义状态模型。这是最关键的一步。from typing import List, Optional, Annotated from typing_extensions import TypedDict from pydantic import BaseModel import operator # 1. 定义强类型状态模型 class AgentState(TypedDict): Agent的全局状态容器 original_task: str # 原始任务 subtasks: List[str] # 分解后的子任务列表 current_subtask_index: int # 当前正在执行的子任务索引 current_subtask: Optional[str] # 当前子任务内容 tool_for_subtask: Optional[str] # 为当前子任务选择的工具名 tool_result: Optional[str] # 工具执行结果 final_results: List[str] # 所有子任务的结果汇总 is_complete: bool # 所有任务是否完成 # 2. 定义各个节点函数 def task_decomposition_node(state: AgentState) - AgentState: 节点任务分解 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4) prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深的项目规划助手。请将以下复杂任务分解为3-5个清晰的、可顺序执行的子任务。只输出子任务列表每行一个。), (human, 任务{task}) ]) chain prompt | llm response chain.invoke({task: state[original_task]}) # 解析LLM返回更新状态 subtasks [line.strip(- ).strip() for line in response.content.split(\n) if line.strip()] state[subtasks] subtasks state[current_subtask_index] 0 state[final_results] [] return state def select_tool_node(state: AgentState) - AgentState: 节点为当前子任务选择工具 current_task state[subtasks][state[current_subtask_index]] state[current_subtask] current_task # 简单的规则匹配实际中可用LLM路由 if 邮件 in current_task or 邀请 in current_task: state[tool_for_subtask] send_email_tool elif 调研 in current_task or 搜索 in current_task: state[tool_for_subtask] web_search_tool elif 撰写 in current_task or 文档 in current_task: state[tool_for_subtask] draft_document_tool else: state[tool_for_subtask] general_qa_tool return state def execute_tool_node(state: AgentState) - AgentState: 节点执行工具 tool_name state[tool_for_subtask] task state[current_subtask] # 模拟工具执行 mock_tool_results { send_email_tool: f已发送关于{task}的邮件。, web_search_tool: f已完成对{task}的搜索找到10条相关信息。, draft_document_tool: f已起草关于{task}的文档初稿。, general_qa_tool: f已处理通用任务{task}。 } state[tool_result] mock_tool_results.get(tool_name, 工具执行失败) state[final_results].append(f子任务[{state[current_subtask_index]1}]: {task} - 结果: {state[tool_result]}) return state def check_completion_node(state: AgentState) - AgentState: 节点检查是否所有子任务完成 next_index state[current_subtask_index] 1 if next_index len(state[subtasks]): state[is_complete] True else: state[current_subtask_index] next_index state[current_subtask] None state[tool_for_subtask] None state[tool_result] None return state # 3. 构建并编译StateGraph from langgraph.graph import StateGraph, END workflow StateGraph(AgentState) # 添加节点 workflow.add_node(decompose, task_decomposition_node) workflow.add_node(select_tool, select_tool_node) workflow.add_node(execute_tool, execute_tool_node) workflow.add_node(check_completion, check_completion_node) # 设置边 workflow.set_entry_point(decompose) workflow.add_edge(decompose, select_tool) workflow.add_edge(select_tool, execute_tool) workflow.add_edge(execute_tool, check_completion) # 条件边如果未完成循环回select_tool如果完成结束。 workflow.add_conditional_edges( check_completion, # 判断函数根据状态中的 is_complete 字段决定下一步 lambda state: END if state[is_complete] else select_tool, {END: END, select_tool: select_tool} ) # 编译图 app workflow.compile() # 4. 执行 initial_state {original_task: 策划一场面向开发者的线上AI技术分享会, is_complete: False} final_state app.invoke(initial_state) print(final_state[final_results])关键解读状态驱动整个流程的推进完全依赖于AgentState这个中心对象的演变。current_subtask_index和is_complete字段是控制循环的关键。清晰的数据流每个节点接收旧状态返回新状态。你可以清晰地看到数据如何被一步步加工。条件逻辑add_conditional_edges使得实现“循环执行直到完成”这种逻辑变得非常简单直观。4.2 使用MessageGraph的实现现在我们用MessageGraph来实现类似功能。思维模式需要转换不再有中心状态一切靠消息传递。from langgraph.graph import MessageGraph from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage from typing import List # 1. 定义节点函数现在它们接收和返回的是消息列表 def task_decomposition_node(messages: List) - List: 节点从最新消息中提取任务并分解 last_message messages[-1] if isinstance(last_message, HumanMessage): user_task last_message.content # 模拟LLM分解任务同上略 subtasks [f1. 确定分享主题与大纲, f2. 邀请讲师与嘉宾, f3. 宣传推广, f4. 进行线上直播, f5. 收集反馈] # 关键不返回状态而是返回新的AIMessage # 我们将子任务列表作为内容返回。在实际中可能需要更结构化的消息如使用ToolMessage。 response_content f任务已分解为以下子任务\n \n.join(subtasks) # 为了后续节点能识别我们可以把结构化数据放在一个特殊的键里这是一种变通 # 这里为了简单我们只是返回文本。更复杂的做法需要自定义消息类型。 return messages [AIMessage(contentresponse_content, additional_kwargs{subtasks: subtasks})] return messages [AIMessage(content请提供一个任务让我来分解。)] def select_tool_node(messages: List) - List: 节点选择工具。需要从历史消息中‘解析’出当前子任务。 # 这里就体现出MessageGraph的麻烦了我们需要手动回溯消息找到子任务信息。 # 假设上一步的AIMessage的additional_kwargs里存了subtasks并且我们用一个索引记录进度。 # 我们需要维护一个“隐式”的索引。这通常需要借助图的“内存”功能或者将索引放在某条消息的元数据中。 # 为了示例我们简化假设总是处理第一个子任务。 subtasks None for msg in reversed(messages): if isinstance(msg, AIMessage) and subtasks in msg.additional_kwargs: subtasks msg.additional_kwargs[subtasks] break if not subtasks: return messages [AIMessage(content未找到可处理的子任务。)] current_task subtasks[0] # 简化处理 # ... 工具选择逻辑同上 tool_name web_search_tool if 调研 in current_task else general_qa_tool # 返回一条包含工具调用信息的消息 return messages [AIMessage(contentf将为子任务『{current_task}』选择工具{tool_name}, additional_kwargs{selected_tool: tool_name, current_task: current_task})] def execute_tool_node(messages: List) - List: 节点执行工具 # 同样需要从历史消息中找到 selected_tool 和 current_task selected_tool None current_task None for msg in reversed(messages): if isinstance(msg, AIMessage) and selected_tool in msg.additional_kwargs: selected_tool msg.additional_kwargs[selected_tool] current_task msg.additional_kwargs.get(current_task) break # ... 模拟执行工具同上 result f模拟执行工具 {selected_tool} 于任务『{current_task}』的结果。 # 返回工具执行结果消息 return messages [ToolMessage(contentresult, tool_call_idmock_id)] # 2. 构建MessageGraph workflow MessageGraph() workflow.add_node(decompose, task_decomposition_node) workflow.add_node(select_tool, select_tool_node) workflow.add_node(execute_tool, execute_tool_node) workflow.set_entry_point(decompose) workflow.add_edge(decompose, select_tool) workflow.add_edge(select_tool, execute_tool) workflow.add_edge(execute_tool, END) # 简单结束没有循环 app workflow.compile() # 3. 执行 initial_messages [HumanMessage(content策划一场线上AI技术分享会)] result app.invoke(initial_messages) for msg in result: print(f{type(msg).__name__}: {msg.content[:50]}...)关键解读与困境消息传递每个节点都在消息列表末尾追加新消息信息通过消息流传递。状态解析困境在select_tool_node和execute_tool_node中我们需要从过往的所有消息中“翻找”所需的信息如子任务列表、当前任务索引。这通过reversed(messages)循环和检查additional_kwargs实现代码变得冗长且低效。循环逻辑难以实现我们这个简化的例子无法实现“循环处理所有子任务”。因为MessageGraph本身没有内置的循环计数器。要实现它你必须要么在消息中携带一个显式的“当前索引”污染了消息内容的主要目的。要么使用MessageGraph提供的add_messages功能来维护一个简单的、非结构化的“内存”但这本质上是在向StateGraph靠拢。要么将循环逻辑外置通过多次调用图来实现这破坏了图的完整性和封装性。踩坑实录在MessageGraph中实现一个需要维护多个数据字段和循环状态的复杂流程你会发现自己花大量精力在“消息考古学”上——不断解析历史消息来重建当前状态。这不仅代码难看性能也差。此时StateGraph的结构化状态优势就无比明显。5. 高级模式与混合使用策略在实际的大型项目中非此即彼的选择并不多见。LangGraph的巧妙之处在于它允许甚至鼓励混合使用。5.1 在StateGraph中嵌入消息处理这是最常见也最强大的模式。StateGraph的状态模型中可以包含一个messages: List[BaseMessage]字段。这样你既拥有了结构化的业务状态如current_step,collected_data又保留了完整的对话历史供LLM节点消费。class HybridState(TypedDict): structured_data: dict # 你的业务数据 messages: List[BaseMessage] # 对话历史 current_step: str def llm_node_with_memory(state: HybridState): 一个既能看到业务数据又能看到对话历史的节点 # 从structured_data中提取业务上下文 business_context state[structured_data] # 将业务上下文和对话历史一起构建Prompt prompt build_prompt(business_context, state[messages]) # 调用LLM... new_ai_message llm.invoke(prompt) # 更新状态既更新业务数据也追加消息 state[messages].append(new_ai_message) state[structured_data][last_llm_output] extract_structured_info(new_ai_message) state[current_step] decide_next_step(new_ai_message) return state这种模式完美结合了两者的优点用结构化状态控制复杂流程逻辑用消息列表保持与LLM交互的灵活性和上下文完整性。5.2 将MessageGraph作为StateGraph的一个节点另一种模式是将一个完整的MessageGraph例如一个负责处理用户单轮问答的子对话系统编译成一个Runnable然后将其作为StateGraph的一个节点。这样StateGraph负责宏观流程和状态管理而具体的、对话密集的子系统则交给更擅长的MessageGraph去处理。# 假设我们已经定义好一个客服对话的MessageGraph并编译为 customer_service_graph from langchain_core.runnables import RunnableLambda def customer_service_node(state: MainState): StateGraph的节点内部调用一个MessageGraph子流程 user_query state[latest_query] # 将主状态中的某些信息作为初始消息传入子图 initial_messages [HumanMessage(contentuser_query), SystemMessage(contentf用户信息{state[user_profile]})] # 运行子MessageGraph conversation_result customer_service_graph.invoke(initial_messages) # 从子图的结果消息中提取关键信息更新主状态 resolution extract_resolution(conversation_result[-1].content) state[query_resolution] resolution state[conversation_history].extend(conversation_result) # 可选保存子对话历史 return state # 在主StateGraph中添加这个节点 main_workflow.add_node(handle_customer_service, RunnableLambda(customer_service_node))5.3 选型决策流程图为了帮助你在项目初期快速决策我总结了一个简单的决策流程你的核心流程是否严重依赖多轮、自由的对话历史是- 优先考虑MessageGraph或HybridState带messages字段。否- 进入下一步。你的流程是否需要维护多个明确的、结构化的中间数据字段并基于它们做复杂路由或循环是- 毫不犹豫选择StateGraph。否- 进入下一步。你的流程是否简单、线性主要是对输入进行一系列转换是-MessageGraph可能更轻量、合适。否- 回到第一步重新评估或直接采用StateGraph以获得更好的可扩展性。个人经验法则对于大多数涉及“任务”、“流程”、“工作流”这些字眼的AI Agent应用StateGraph通常是更优的起点。它的结构化约束在项目增长时带来的益处远大于初期的设计成本。MessageGraph则是你工具箱里一把锋利的“手术刀”专门用于处理纯对话切片或作为大系统的组件。6. 常见陷阱与性能调优指南即使选对了模型在实际开发中依然会遇到不少坑。以下是一些高频问题和解决方案。6.1 StateGraph的常见陷阱状态模型设计过载或不足问题一开始把所有能想到的字段都塞进State导致状态臃肿每个节点都需要处理大量无关字段。或者相反字段设计不足后期频繁修改模型破坏兼容性。解决遵循“最小化”和“前瞻性”原则。只定义当前节点确实需要读写字段。但对于核心业务实体如User,Order可以将其作为整体字段放入即使初期只用其中一两个属性。节点副作用污染状态问题多个节点意外修改了同一个状态字段导致难以调试的竞态条件或逻辑错误。解决为节点函数编写清晰的文档说明其读写哪些字段。在复杂流程中可以考虑使用不可变数据模式如返回全新的状态字典但这会牺牲一些便利性。更实用的方法是在关键节点前后打印或记录状态快照。条件边conditional_edge逻辑过于复杂问题判断下一个节点的函数router写得非常复杂难以理解和测试。解决将复杂的路由逻辑封装成独立的、可测试的函数。甚至可以为不同的路由条件创建专门的“路由节点”该节点只负责检查状态并返回下一个节点名称使主图结构更清晰。6.2 MessageGraph的常见陷阱消息列表无限膨胀问题长时间运行的对话导致消息列表巨大每次调用LLM的token成本剧增且影响节点查找历史信息的性能。解决实现消息摘要或窗口化。可以定期用一个节点将冗长的历史消息总结成一条系统消息然后清空或截断旧消息。LangGraph的Checkpointer机制也能帮助管理长上下文。从消息中提取状态信息效率低下问题每个节点都需要for msg in reversed(messages):来查找信息O(n)复杂度。解决约定俗成地将关键结构化数据放在最新一条消息的additional_kwargs中。或者直接使用Hybrid模式在MessageGraph的“内存”或父StateGraph的状态中维护一个轻量级的状态索引。难以实现复杂循环和状态判断问题如前文所述这是MessageGraph的天然短板。解决不要强求。如果你的流程逻辑变得复杂这正是考虑切换到StateGraph或采用混合架构的信号。6.3 性能调优要点图的编译与预热app workflow.compile()会产生一些开销。在生产环境中考虑在服务启动时预先编译好常用的图并将其缓存。异步支持LangGraph节点函数天然支持async def。如果你的节点涉及网络IO如调用LLM API、查询数据库务必使用异步函数并用ainvoke来并发执行可以极大提升吞吐量。状态序列化如果你使用LangGraph的持久化或检查点功能StateGraph的状态对象需要能被序列化如Pickle或JSON。确保你的状态模型中只包含可序列化的Python类型。对于复杂对象考虑将其转换为字典或字符串存储。节点粒度不要把所有逻辑塞进一个巨型节点。将节点拆分为功能单一、职责明确的单元有利于复用、测试和并行化。但节点过多也会增加图的管理开销和调用延迟需要在清晰度和性能间取得平衡。最后无论选择哪种Graph充分的测试都至关重要。为每个节点函数编写单元测试模拟输入状态或消息验证输出。对于整个图编写集成测试验证从初始输入到最终输出的完整流程是否符合预期。LangGraph的确定性执行在给定相同输入和随机种子下特性使得测试变得相对可行。记住一个设计良好的AI Agent系统其核心逻辑的可靠性和可维护性很大程度上就取决于你对StateGraph和MessageGraph这些基础构建块的理解与运用。