本文主要介绍多智能体集成框架LangGraph的相关概念及使用，关于大模型应用开发的基本的流程可以阅读：《GitHub 12w Star神器！一文详解大模型集成框架LangChain》，详细介绍了构建大模型应用的步骤及概念（提示词模版、输出解析器、记忆、工具、Agent、RAG等）。01 LangGraph概述1.1什么是LangGraphLangGraph 是LangChain 生态的一部分，专门用于构建基于大模型（LLM）的复杂、有状态、多智能体应用的框架，核心思想是将应用的工作流程抽象为一个有向图结构，通过节点和边来定义任务的执行步骤和逻辑流，从而提供了远超传统线性链式调用的灵活性和控制力。相比传统的线性执行模式，LangGraph 支持条件分支、循环、并行等复杂控制流，能够实现状态持久化、断点续跑、时间旅行、人机协作等高级功能，并提供了多智能体协作、层级架构等多种架构模式。在实际应用中，LangGraph已成功应用于智能客服、自动化运维、研究 Agent 等场景，展现出卓越的适应性和扩展性。LangGraph在Github上的热度变化1.2为什么使用LangGraphLangGraph 相比传统的线性执行模式具有显著的技术优势。LangGraph 提供了强大的状态管理机制，允许 Agent 在不同节点之间传递和维护信息，从而实现长期的记忆和多轮对话能力。这种集中式的状态管理避免了传统方法中状态分散在多个变量中的问题，提高了系统的可维护性和可观测性。通过定义节点和边，可以精确控制 Agent 的执行逻辑，包括条件分支、循环和并行执行等LangGraph 能够无缝集成各种外部工具（如搜索引擎、数据库、API 等），让 Agent 能够获取实时信息、执行特定操作，极大地扩展了 LLM 的能力边界。图结构使得 Agent 的运行路径清晰可见，便于理解 Agent 的决策过程，并在出现问题时进行快速定位和调试。模块化与可复用性。每个节点都可以是一个独立的、可复用的组件，维护性高且易于扩展。通过子图机制，复杂的工作流可以被分解为多个可独立开发和测试的模块，提高了开发和测试效率。1.3安装使用安装LangGraphpipinstall -U langgraph使用LangGraph创建一个简单的Agentdefget_weather(city:str) -str: """获取指定城市的天气信息。 Args: city: 城市名称 Returns: 返回该城市的天气描述 """ returnf"今天{city}是晴天"# 创建模型model = ChatOpenAI( model_name=model_name, base_url=base_url, api_key=api_key)# 使用LangGraph提供的API创建Agentagent = create_react_agent( model=model, # 添加模型 tools=[get_weather], # 添加工具 prompt="你是一个天气助手")human_message = HumanMessage(content="今天深圳天气怎么样？")response = agent.invoke( {"messages": [human_message]})print(response)运行模式：Agent可以通过两种主要模式执行同步：使用 .invoke() 或 .stream()异步：使用 await .ainvoke() 或 async for 配合 .astream()最大迭代次数：为了避免Agent无限循环执行，可以设置一个递归限制response = agent.invoke( {"messages": [{"role":"user","content":"预定一个深圳到北京的机票"}]}, {"recursion_limit": 10} # 指定最大迭代次数)02 LangGraph核心2.1Graph（图）图是一种由节点和边组成的用于描述节点之间关系的数据结构，分为无向图和有向图，有向图是带有方向的图。LangGraph通过有向图定义AI工作流中的执行步骤和执行顺序，从而实现复杂、有状态、可循环的应用程序逻辑。2.2LangGraph核心要素（State、Edge、Node）1、State（状态）在LangGraph中，State是一个贯穿整个工作流执行过程中的共享数据的结构，代表当前快照，它存储了从工作流开始到结束的所有必要的信息（历史对话、检索到的文档、工具执行结果等），在各个节点中共享，且每个节点都可以修改。State可以是TypedDict类型，也可以是pydantic中的BaseModel类型。# 定义状态classGraphState(TypedDict): process_data:dict# 默认更新策略为替换（后续会讲更新策略）# 创建一个状态图，并指定状态graph = StateGraph(GraphState)2、Node（节点）Node是LangGraph中的一个基本处理单元，代表工作流中的一个操作步骤，可以是一个Agent、调用大模型、工具或一个函数（说白了就是绑定一个函数，具体逻辑可以干任何事情）。Node的设计原则：单一职责原则：每个节点应该只负责一项职责，避免功能过于复杂无状态设计：节点本身不应该保存状态，所有数据都通过输入状态传递幂等性：相同的输入应该产生相同的输出，确保可重试性可测试性：节点逻辑应该易于单元测试如下是添加一个节点的例子：# 定义一个节点，入参为statedefinput_node(state: GraphState) - GraphState: print(state) return{"process_data": {"input":"input_value"}}# 定义带参数的node节点defprocess_node(state:dict, param1:int, param2:str) -dict: print(state, param1, param2) return{"process_data": {"process":"process_value"}}graph = StateGraph(GraphState)# 添加inpu节点graph.add_node("input", input_node)# 给process_node节点绑定参数process_with_params = partial(process_node, param1=100, param2="test")# 添加带参数的node节点graph.add_node("process", process_with_params)特殊节点：在LangGraph中有两个特殊的节点 __START__ （开始节点）和 __END__（结束节点）__START__节点：开始节点，确定应该首先调用哪些节点。fromlanggraph.graphimportSTART# 第一个执行的节点是 node_startgraph.add_edge(START,"node_start")也可以通过graph.set_entry_point("node_start") 函数设置起始节点，等价于graph.add_edge(START, "node_start")__END__节点：终止节点，表示后续没有其他节点可以继续执行了（非必须）。fromlanggraph.graphimportEND# node_end 节点执行后，没有后续节点了graph.add_edge("node_a", END)也可以通过graph.set_finish_point("node_end") 函数设置结束节点，等价于graph.add_edge("node_start"， END)错误处理和重试机制：LangGraph还提供了错误处理和重试机制来指定重试次数、重试间隔、重试异常等，用于保证系统的可靠性。# 重试策略retry_policy = RetryPolicy( max_attempts=3, # 最大重试次数 initial_interval=1, # 初始间隔 jitter=True, # 抖动（添加随机性避免重试风暴） backoff_factor=2, # 退避乘数（每次重试间隔时间的增长倍数） retry_on=[RequestException, Timeout] # 只重试这些异常)graph.add_node("process", process_node, retry=retry_policy)节点缓存：LangGraph 支持根据节点输入对节点进行缓存，用于加快节点的响应速度缓存键与命中：当一个节点开始执行时，系统会使用其配置的 key_func 根据当前节点的输入数据生成一个唯一的键。LangGraph 会检查缓存中是否存在这个键。如果存在（缓存命中），则直接返回之前存储的结果，跳过该节点的实际执行。如果不存在（缓存未命中），则正常执行节点函数，并将结果与缓存键关联后存入缓存。缓存有效期：ttl 参数能控制缓存的有效期。例如，对于依赖实时数据的天气查询节点，可以设置较短的 ttl（如60秒）。而对于处理静态信息或变化不频繁数据的节点，则可以设置较长的 ttl甚至不设置（None），让缓存永久有效，直到手动清除classState(TypedDict): x:int result:intbuilder = StateGraph(State)defexpensive_node(state: State) -dict[str,int]: # 模拟耗时 time.sleep(2) return{"result": state["x"] *2}#添加节点，并指定缓存策略builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))builder.set_entry_point("expensive_node")builder.set_finish_point("expensive_node")graph = builder.compile(cache=InMemoryCache())3、Edge（边）Edge定义了节点之间的连接和执行顺序，以及不同节点之间是如何通讯的，一个节点可以有多个出边（指向多个节点），多个节点也可以指向同一个节点（Map-Reduce），如下是添加边的代码：# 添加固定边，执行顺序：start - input - process - output - endgraph.add_edge(START,"input")graph.add_edge("input","process")graph.add_edge("process","output")graph.add_edge("output",END)# 编译图，保证生成的图是正确的，如果添加了边，没添加节点，会报错app = graph.compile()app.invoke({})4、构建一个完整的图图的构建流程：1、初始化一个StateGraph实例。2、添加节点。3、定义边，将所有的节点连接起来。4、设置特殊节点，入口和出口（可选）。5、编译图。6、执行工作流。classGraphState(TypedDict): process_data:dictdefinput_node(state: GraphState) - GraphState: print(state) return{"process_data": {"input":"input_value"}}defoutput_node(state: GraphState) - GraphState: print(state) return{"process_data": {"output":"output_value"}}defprocess_node(state:dict) -dict: print(state) return{"process_data": {"process":"process_value"}}# 创建一个状态图，并指定状态graph = StateGraph(GraphState)# 添加input、process、output节点graph.add_node("input", input_node)graph.add_node("process", process_node)graph.add_node("output", output_node)# 添加固定边，执行顺序：start - input - process - output - endgraph.add_edge(START,"input")graph.add_edge("input","process")graph.add_edge("process","output")graph.add_edge("output", END)# 编译图，保证生成的图是正确的，如果添加了边，没添加节点，会报错app = graph.compile()# 执行app.invoke({})2.3状态合并策略（Reducers）LangGraph工作流中，State作为贯穿整个节点之间共享数据的结构，每一个节点都可以读取当前State的数据，并且可以更新。Reducer是定义多个节点之间State如何更新的（覆盖、合并、添加等）。1、直接覆盖：如果没有为状态字段指定 Reducer，默认会覆盖更新。也就是说，后执行的节点返回的值会直接覆盖先执行节点的值，即下一个节点的State数据是上一个节点的返回。classOverrideState(TypedDict): process_data :dict# 未指定合并策略，默认覆盖，上一个节点的返回是下一个节点的值2、Annotated：使用类型注解指定合并策略classAddState(TypedDict): data_int: Annotated[int, add] # 数字相加 data_list: Annotated[list, add]# 合并两个列表 data_str: Annotated[str, add] # 字符串拼接defadd_node1(state: AddState) - AddState: print(state) return{"data_int":1,"data_list": [1],"data_str":"hello "}defadd_node2(state: AddState) - AddState: print(state) return{"data_int":2,"data_list": [2],"data_str":"world"}3、内置Reducer：add_messages（消息列表合并）LangGraph提供的专用Reducer函数，能智能的合并消息列表，不只是简单的追加，add_messages能够保证消息列表正确被累计，常用在多轮对话系统中，主要逻辑包括：追加新消息：如果新消息的 ID 不在现有列表中，则将其追加到列表末尾。覆盖旧消息：如果新消息的 ID 与列表中某条现有消息的 ID 相同，则用新消息替换掉旧消息。用于处理工具调用中间结果或更新流式生成的临时消息。自动类型转换：如果传入一个字符串（如 "Hello World"），add_messages会自动将其转换为HumanMessage（用户消息）classMessageState(TypedDict): # 消息列表,使用add_messages合并消息列表 messages: Annotated[list, add_messages]defsystem_node(state: MessageState) -dict: return{"messages": [SystemMessage(content="你是一个精通LangGraph的专家工程师.")]}defuser_input_node(state: MessageState) -dict: return{"messages": [HumanMessage(content="什么是LangGraph?")]}defai_response_node(state: MessageState) -dict: return{"messages": [AIMessage(content="LangGraph是一个...")]}deftool_node(state: MessageState) -dict: return{"messages": [ToolMessage(content="工具调用参数params1", tool_call_id="tool_call_id")]}4、自定义Reducer：实现自定义合并逻辑defmerge_dict_reducer(source:dict, new:dict) -dict: # 自定义合并逻辑 result = source.copy() result.update(new) returnresultdefmax_reducer(source:int, new:int) -int: # 自定义合并逻辑 returnmax(source, new)classCustomReducerState(TypedDict): # 使用自定义Reducer的状态 max_score: Annotated[int, max_reducer] # 保留最大值 metadata: Annotated[dict, merge_dict_reducer] # 字典合并2.4条件边（Conditional Edge）实际应用中，工作流的下一个节点可能并不是固定的，需要根据当前的执行状态去确定需要路由到哪一个节点。条件边可以动态控制执行流程，LangGraph中可以指定路由函数，来选择具体要执行的节点（可以是多个节点）defroute_by_sentiment(state: GraphState) -str: # 路由逻辑...返回最终的条件 return"condition_1"graph = StateGraph(GraphState)graph.add_node("node1", node1)graph.add_node("node2", node2)graph.add_node("node3", node3)# 添加路由函数，参数：当前节点，路由函数，路由函数返回的条件与node的映射graph.add_conditional_edges( START, route_by_sentiment, { "condition_1":"node1", "condition_2":"node2", "condition_3":"node3" })# 所有处理节点都连接到ENDgraph.add_edge("node1", END)graph.add_edge("node2", END)graph.add_edge("node3", END)app = graph.compile()LangGraph 提供了图的可视化，可以通过调用函数保存图，用于查看工作流是否与预期定义的规则一致。png_data = app.get_graph().draw_mermaid_png()withopen("graph.png","wb")asf: f.write(png_data)2.5Send 和 CommandSend和Command是两种用于实现高级工作流控制的核心机制，用于支持动态地决定下一步执行哪个节点1、Send：动态创建多个执行分支，实现并行处理，每个Send对象都指定了一个执行目标节点和传递给该节点的参数，LangGraph会并行执行所有的这些任务。比如可以用在Map-Reduce的场景，并行执行多个子节点并最终汇总到一个总节点。def route_tasks(state: MapReduce