LangGraph 是一款用于构建、管理和部署长期运行的有状态代理的底层编排框架和运行时，深受塑造代理未来的企业信赖，包括 Klarna、Uber、摩根大通等。LangGraph 属于非常底层的框架，完全专注于代理编排。在使用 LangGraph 之前，我们建议您先熟悉用于构建代理的一些组件，从模型和工具开始。LangGraph 专注于代理编排中至关重要的底层能力：持久化执行、流式处理、人机协同等。
-> 前往星空笔记
LangGraph 是一款用于构建、管理和部署长期运行的有状态代理的底层编排框架和运行时，深受塑造代理未来的企业信赖，包括 Klarna、Uber、摩根大通等。LangGraph 属于非常底层的框架，完全专注于代理编排。在使用 LangGraph 之前，我们建议您先熟悉用于构建代理的一些组件，从模型和工具开始。LangGraph 专注于代理编排中至关重要的底层能力：持久化执行、流式处理、人机协同等。
-> 前往星空笔记
官网 |
LangGraph文档 |
LangGraph Github |
langgraph 1.0手册 |
LangGraph解决线性序列的局限性问题，而解决的方法就是循环图。在LangGraph框架中，用图来管理代理的生命周期并在其状态内将暂存器作为消息进行跟踪，增加了以循环方式跨各种计算步骤协调多个链或参与者的功能。LangGraph通过组合Nodes和Edges去创建复杂的循环工作流程，通过消息传递的方式串联所有的节点形成一个通路。那么维持消息能够及时的更新并向该去的地方传递，则依赖langGraph构建的State概念。LangGraph的优势则是：virtualenv lang_env # 创建虚拟环境 source lang_env/Scripts/activate //windows pip install langgraph -i https://pypi.tuna.tsinghua.edu.cn/simple （清华镜像） 0.4.8 / 1.1.6 pip install langchain-openai python-dotenv -i https://pypi.tuna.tsinghua.edu.cn/simple # 可以不安装langchain参考
在
LangGraph中，节点(Nodes)是一个python函数（sync或async），接收当前State作为输入，执行自定义的计算，并返回更新的State。所以其中第一个位置参数是state。def agent_node(state:InputState): print("我是一个AI Agent。") return定义好了节点以后，我们需要使用
add_node方法将这些节点添加到图中。在将节点添加到图中的时候，可以自定义节点的名称。而如果不指定名称，则会为自动指定一个与函数名称等效的默认名称。代码如下：builder.add_node("agent_node", agent_node) builder.add_node("action_node", action_node)参考
Edges（边）用来定义逻辑如何路由以及图何时开始与停止。这是智能体工作以及不同节点如何相互通信的重要组成部分。有几种关键的边类型：
同样，我们先看普通边。如果直接想从节点
A到节点B，可以直接使用add_edge方法。注意：LangGraph有两个特殊的节点：START和END。START表示将用户输入发送到图的节点。使用该节点的主要目的是确定应该首先调用哪些节点。END节点是代表终端节点的特殊节点。当想要指示哪些边完成后没有任何操作时，将使用该节点。因此，一个完整的图就可以使用如下代码进行定义：from langgraph.graph import START, END builder.add_edge(START, "agent_node") builder.add_edge("agent_node", "action_node") builder.add_edge("action_node", END) # 最后，通过`compile`编译图。在编译过程中，会对图结构执行一些基本检查（如有没有孤立节点等）。 graph = builder.compile() graph.invoke({"question":"hello，你好"})在代理架构模式下，大模型能通过多种方式进行操作控制。最基本的功能是在两个潜在路径之间做出选择。进而在每条路径上，如果存在多个可用工具，大模型能够自主决定调用哪一个。更复杂的情况下，它还能评估生成的答案是否满足问题的需求。如有必要进行额外工作，它将自行执行，直到得到一个充分满足条件的答案为止。
LangGraph框架则是从这个角度出发，接入了路由代理，工具代理，自主循环代理以及多代理这四大类代理架构，以支持不同的场景需求。在实际应用中，流式输出尤其适用于需要快速反馈的业务场景，如聊天机器人，因为大语言模型可能需要几秒钟才能生成对查询的完整响应，这远远慢于应用程序对最终用户的响应速度约为 200-300 毫秒的阈值，如果是涉及多个大模型调用的复杂应用程序，这种延时会变得更加明显。让应用程序感觉响应更快的关键策略是显示中间进度；即，通过
token流式传输大模型Token的输出，以此来显著提升用户体验。而在开发阶段，利用流式输出功能可以准确追踪到事件的具体执行阶段，并捕获相关数据，从而接入不同逻辑的数据处理和决策流程。是我们在应用开发中必须理解和掌握的技术点。流式输出功能在
LangGraph框架中的实现方式比较简单，因为LangGraph底层是基于LangChain构建的，所有就直接把LangChain中的回调系统拿过来使用了。在LangChain中的流式输出是：以块的形式传输最终输出，即一旦监测到有可用的块，就直接生成它。最常见和最关键的流数据是大模型本身生成的输出。 大模型通常需要时间才能生成完整的响应，通过实时流式传输输出，用户可以在生成时看到部分结果，这可以提供即时反馈并有助于减少用户的等待时间。LangGraph框架中的工作流中由各个步骤的节点和边组成。这里的流式传输涉及在各个节点请求更新时跟踪图状态的变化。这样可以更精细地监控工作流中当前处于活动状态的节点，并在工作流经过不同阶段时提供有关工作流状态的实时更新。其实现方式也是和LangChain一样通过.stream和.astream方法执行流式输出，只不过适配到了图结构中。调用.stream和.astream方法时可以指定几种不同的模式，即：messages中的增量token。LangGraph 的 StreamWriter方法async def main(): async for event in graph.astream_events({"messages": ["what is labubu"]}): kind = event["event"] if kind == "on_chat_model_stream": print(event["data"]["chunk"].content, flush=True) asyncio.run(main())多代理系统其实就可以非常简单的理解为：将原本的应用程序拆分成多个较小的独立代理，从而组合而成的系统。这些小的独立代理可以是简单的大模型交互代理，也可以是复杂的
ReAct代理。举个比较热门的案例，假设我们需要建立一个用于数据分析的Agent，则可以设计代理配置：Agent 1作为用户意图识别代理，集成大模型用来解析用户的查询和指令，理解其意图和需求，并将用户输入转化为具体的任务。Agent 2作为数据分析代理，集成大模型并绑定若干个处理不同数据和需求的工具，提供统计分析、趋势预测和数据可视化服务。当任务涉及到代码生成时，Agent 3，即代码执行代理，会接收用户输入的代码，在安全的Python环境中执行这些代码，并返回运行结果，用于代码测试、执行特定算法或自动化任务。LangGraph利用基于图的结构来定义代理并在它们之间建立连接。在此框架中，每个代理都表示为图中的一个节点，并通过边链接到其它代理。每个代理通过接收来自其他代理的输入并将控制权传递给下一个代理来执行其指定的操作。在LangGraph框架的设计中，主要通过如下几种方法来建立各个子代理之间的通信连接：Supervisor代理通信。由Supervisor代理决定接下来应调用哪个代理。Supervisor架构的一个特例。每个代理都是一个工具。由Supervisor代理通过工具调用的方式来决定调用哪些子代理执行任务，以及要传递给这些代理程序的参数supervisor嵌套supervisor多代理系统。这是Supervisor架构的一种泛化，允许更复杂的控制流。-- Subgraphs
Subgraphs（子图）指的是能够用作另一个图中的节点的图。简单理解就是：把一个已经编译好的图，嵌入到另一个已经编译好的图中，并且两个独立图的中的状态可以信息共享。一个典型的应用就是构建多代理系统架构。它所做的事情是：当把每个独立的Agent图结构定义为一个子图时，只要遵守子图的接口（输入和输出模式）规范，那么子图中定义的共享状态就可以在父图中进行使用添加子图主要解决的问题就是解决各
Single-Agent之间的通信问题，即它们如何在图执行期间在彼此之间传递状态。这主要有两种情况：