使用大型语言模型（LLM）构建应用程序是一件令人兴奋的事情，因为它能让我们创建智能的交互式系统。然而，让这些应用程序变得更加复杂也带来了挑战，尤其是当多个 LLM 一起工作时。那么，我们该如何管理它们之间的信息流？如何确保它们顺利工作并理解任务？LangGraph 就是所有这些问题的答案。本免费教程是初学者了解 LangGraph 如何解决这些问题的好方法。本指南将通过实际操作示例和完整代码，教你如何有效管理多个 LLM，让你的应用程序更强大、更高效。
了解LangGraph
LangGraph 是一个功能强大的库，是 LangChain 工具的一部分。它有助于简化 LLM 的集成，确保它们无缝协作，以理解和执行任务。它提供了一种简便的方法来构建和处理具有许多代理的 LLM 应用程序。
LangGraph 允许开发人员设置多个 LLM 代理如何相互对话。它将这些工作流程显示为循环图。这有助于保持通信顺畅并很好地执行复杂任务。在使用有向无环图（DAG）执行直线任务时，LangGraph 的效果最佳。但由于它是循环图，并增加了回环能力，因此可以实现更复杂、更灵活的系统。这就好比一个智能代理可能会重新思考问题，并利用新信息更新响应或改变选择。
LangGraph的关键概念
以下是您需要了解的 LangGraph 的一些关键概念：
1. 图形结构
LangGraph 的核心理念是为应用程序的工作流程使用图形。该图有两个主要部分–节点和边。
- 节点：节点是基本的构建模块，代表工作流中工作或计算的离散单元。每个节点都是一个 Python 函数，用于处理当前状态并返回更新后的状态。节点可以执行的任务包括调用 LLM 和与工具或 API 交互以操作数据。
- 边：边连接节点并定义执行流程。它们可以是
- 简单边：从一个节点到另一个节点的直接、无条件转换。
- 条件边：根据节点输出引导流程的分支逻辑，类似于 if-else 语句。这样就可以在工作流程中进行动态决策。
2. 状态管理
在有许多代理的情况下，跟踪正在发生的事情至关重要。所有代理都需要知道任务的当前状态。LangGraph 通过自动管理状态来解决这个问题。该库跟踪并更新一个主状态对象。当代理执行任务时，它就会这样做。状态对象包含重要信息。它可以在工作流的不同阶段使用。这可能包括聊天记录。
在聊天机器人中，状态可以保存对话。这有助于机器人使用之前的对话内容做出回应。它还可以存储上下文数据，如用户喜好、过去的操作等或外部数据。代理可以利用这些数据做出选择。内部变量也可以保存在这里。代理可以使用状态跟踪标志、计数或其他值。这些有助于指导它们的行动和决策。
3. 多代理系统
多代理系统由多个独立的代理组成，它们通过合作或竞争来实现共同的目标。这些代理使用 LLM 做出决策并控制应用程序的流程。随着代理和任务的增加，系统的复杂性也会随之增加。这可能会导致决策失误、上下文管理和专业化需求等挑战。多代理系统通过将系统分解成较小的代理来解决这些问题，每个代理都专注于特定的任务，如规划或研究。
使用多代理系统的主要好处是模块化、专业化和控制。模块化便于开发、测试和维护，而专业化则能确保专家代理提高整体性能。控制可确保您清楚地知道代理应如何交流。
多代理系统中的架构
以下是多代理系统中遵循的各类架构。
Source: LangChain
1. 网络架构：在这种架构中，每个代理都与其他代理进行通信，然后每个代理都可以决定下一步应该呼叫哪个代理。当没有明确的操作顺序时，这种架构非常有用。下面是一个使用状态图（StateGraph）的简单示例。
from langchain_openai import ChatOpenAI
from langgraph.types import Command
from langgraph.graph import StateGraph
model = ChatOpenAI()
def agent_1(state) -> Command:
response = model.invoke(...)
return Command(goto=response["next_agent"], update={"messages": [response["content"]]})
builder = StateGraph()
builder.add_node(agent_1)
builder.compile()
2. 主管架构：监管代理控制决策过程，并将任务分配给相应的代理。下面是一个示例：
def supervisor(state) -> Command: response = model.invoke(...) return Command(goto=response["next_agent"]) builder = StateGraph() builder.add_node(supervisor) builder.compile()
3. 带有工具调用功能的监督员：在这种架构中，监管代理使用工具调用代理来决定使用哪种工具（或代理）。工具执行任务并返回结果，为下一步控制流决策提供指导。一种常见的模式是使用工具包装功能：
def agent_1(state): response = model.invoke(...) return response.content
4. 分层架构：这种方法通过将代理组织成团队来解决多代理系统的复杂性问题，每个团队都有自己的主管。顶层主管负责指挥哪个团队。例如：
def top_level_supervisor(state): response = model.invoke(...) return Command(goto=response["next_team"]) builder = StateGraph() builder.add_node(top_level_supervisor) builder.compile()
5. 多代理系统中的切换：切换允许一个代理将控制权传递给另一个代理，从而促进从一个代理到下一个代理的流程。每个代理都会返回一个“命令”（Command）对象，指定下一个要调用和发送状态更新的代理。
def agent(state) -> Command:
goto = get_next_agent(...)
return Command(goto=goto, update={"my_state_key": "my_state_value"})
在复杂的系统中，代理可能嵌套在子图中，子图中的节点可以将控制权指向其图外的另一个代理：
def some_node_inside_alice(state): return Command(goto="bob", graph=Command.PARENT)
多代理系统采用模块化和专业化设计，代理可独立处理任务并进行交流，从而高效地解决问题。网络系统、监督系统和分层系统等架构都能满足特定需求，而交接则能确保代理之间的平稳过渡，保持灵活性和控制性。
4. 持久性
持久性指的是保存进程的进度，这样即使在中断之后，您也可以稍后再返回。每个步骤的状态都会被保存，这有助于错误恢复。它支持运行过程中的人为反馈。您还可以重放步骤来调试或尝试新的路径。
Source: LangChain
在 LangGraph 中，持久性是通过检查点来实现的。在这里，图的状态会在每个主要步骤后保存，每个保存的状态都称为一个检查点。所有检查点都集中在一个线程（特定运行的对话历史）中。
检查点是自动完成的，不需要总是手动配置。检查点就像是图表状态的快照，其中包括
- config：该步骤中使用的配置信息
- metadata：步骤详细信息（例如，哪个节点正在运行）
- values：此时的实际状态值
- next：将运行的下一个（多个）节点
- tasks：任务信息或错误信息
每个图形在执行时都需要一个线程 ID 来对其检查点进行分组。可以通过配置提供线程 ID： 下面是一个示例：
config = {"configurable": {"thread_id": "1"}}
要获取线程内的最新状态，请使用下面的代码：
graph.get_state({"configurable": {"thread_id": "1"}})
下面的代码显示了如何获取特定的检查点：
graph.get_state({
"configurable": {
"thread_id": "1",
"checkpoint_id": "your_checkpoint_id"
}
})
要获取状态历史记录或以前的所有状态，请使用此代码：
history = graph.get_state_history({"configurable": {"thread_id": "1"}})
您也可以随时手动更新或编辑状态：
graph.update_state(
config={"configurable": {"thread_id": "1"}},
values={"foo": "new_value"}
)
5. Human-in-the-Loop集成
在自动 LangGraph 工作流程的关键步骤中，您可以添加人的反馈。这在某些任务中至关重要，因为 LLM 可能会产生不确定或有风险的输出，例如在工具调用、内容生成或决策过程中。LangGraph 的 interrupt() 函数可以暂停图形，将数据显示给人类，然后使用 Command(resume=value) 方法根据人类的输入恢复图形。这样就可以进行审查、修正或数据输入。
Human-in-the-Loop 支持批准/拒绝、编辑状态、提供输入或多轮对话等模式。要使用它，请定义一个检查指针，并在节点内添加 interrupt()。在人工输入后，您可以使用 Command 恢复图形。
来源：LangChain
下面是如何在 LangGraph 中使用“Human-in-the-loop”的示例。
from langgraph.types import interrupt, Command
def human_node(state):
value = interrupt({"text_to_revise": state["some_text"]})
return {"some_text": value}
graph = graph_builder.compile(checkpointer=checkpointer)
graph.invoke(some_input, config={"configurable": {"thread_id": "some_id"}})
graph.invoke(Command(resume="Edited text"), config={"configurable": {"thread_id": "some_id"}})
这可以保持工作流程的互动性、可审计性和准确性，是高风险或协作式人工智能用例的完美选择。
6. 流
LangGraph 可在创建输出时将其串流起来，让用户更快地看到结果。这改善了用户使用 LLM 的体验。数据流可向您显示实时进度，从而帮助您构建响应式应用程序。有 3 种主要数据类型可用于流式传输：工作流进度、LLM 标记和自定义更新。
使用 .stream() (sync) 或 .astream() (async) 来实现流式输出。您可以设置 stream_mode 来控制输出内容：
- “values”：每个图形步骤后的完整状态
- “updates”：仅在每个节点后发生变化
- “custom”：在节点中记录的任何自定义数据
- “messages”：带有元数据的 LLM 标记流
- “debug”：整个运行过程中的所有信息
您可以像这样传递多种模式：
for stream_type, data in graph.stream(inputs, stream_mode=["updates", "messages"]): if stream_type == "messages": print(data[0].content) # AIMessageChunk elif stream_type == "updates": print(data) # State update
如果需要完整的事件流，请使用 .astream_events() 。这在迁移大型应用程序时非常适合。
专业建议：对于实时 UI 反馈，可使用“messages”进行令牌流，使用“updates”进行后台状态。
为什么使用LangGraph？
LangGraph 是开发人员构建智能、灵活的人工智能代理的理想选择。原因如下
- 可靠、可控：添加节制检查和人工审批。它能让长期任务的上下文保持活力。
- 自定义和可扩展：使用底层工具按自己的方式构建代理。设计系统时，每个代理都能发挥特定的作用。
- 流式传输：实时查看每个标记和步骤，跟踪代理的思考过程。
您还可以学习 Langchain 学院的这门课程 ，了解更多信息。
构建最简单的图表
既然我们已经看到了 LangGraph 的关键组件，那么让我们尝试用三个节点和一条条件边构建一个基本图。这个简单的示例展示了如何调用涉及状态、节点和边等关键概念的图形。
Step 1：定义图状态
状态定义了节点之间共享的数据结构。它就像一个在图中流动的共享内存。
from typing_extensions import TypedDict class State(TypedDict): graph_state: str
在这里，我们使用了 python 的 TypeDict 来声明我们的状态将有一个名为 graph_state 的单键，其中存储了一个字符串。
Step 2：创建节点
节点只是简单的 Python 函数。每个节点都接收当前状态，对其进行修改，然后返回更新后的状态。
def node_1(state):
print("---Node 1---")
return {"graph_state": state['graph_state'] + " I am"}
此函数将“I am”添加到 graph_state 中的任何字符串中。
def node_2(state):
print("---Node 2---")
return {"graph_state": state['graph_state'] + " extremely happy!"}
def node_3(state):
print("---Node 3---")
return {"graph_state": state['graph_state'] + " extremely sad!"}
在这里，这两个节点为句子添加了“happy!”或“sad!”的情感基调。
Step 3：添加条件逻辑
有时您需要动态行为，即下一步取决于逻辑或随机性。这就是条件边的作用。
import random from typing import Literal def decide_mood(state) -> Literal["node_2", "node_3"]: if random.random() < 0.5: return "node_2" return "node_3"
该函数以相等的概率在节点_2 和节点_3 之间随机选择，模拟一个简单的情绪选择器。
Step 4：构建图
让我们使用 LangGraph 的 StateGraph 类将这一切整合在一起。我们在这里定义完整的图结构。
from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END
# Initialize the graph with the state schema
builder = StateGraph(State)
# Add nodes to the graph
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)
我们从 START 节点开始，路由到节点_1。然后，我们使用 decide_mood 从节点_1 开始添加一条条件边。之后，图继续延伸至节点_2 或节点_3，并在终点节点处结束。
# Add edges to define flow
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)
# Compile and visualize the graph
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))
compile() 方法可执行基本验证，而 draw_mermaid_png() 则可将图形可视化为美人鱼图。
Step 5：调用图表
最后，我们可以使用 invoke() 方法运行图表。
graph.invoke({"graph_state" : "Hi, this is Janvi."})
这将在 START 节点处启动图形，并用句子 “嗨，我是 Janvi ”初始化 graph_state。
- node_1 添加 “I am”→“Hi, this is Janvi.”。
- decide_mood 随机选择路径
- node_2 或 node_3 应用“extremely happy!”或“extremely sad!”。
输出：
该输出显示了状态如何在图的每个步骤中流动和更新。
使用OpenAI通过LangGraph构建支持聊天机器人
在上一节中，我们已经构建了最简单的图，在本节中，我将向您展示如何使用 LangGraph 构建支持聊天机器人，从基本功能开始，逐步添加网络搜索、记忆和人在回路等功能。在此过程中，我们还将了解 LangGraph 的核心概念。
我们的目标是创建一个聊天机器人，它可以使用网络搜索回答问题、记住过去的对话、在需要时向人类寻求帮助、使用自定义状态行为以及倒带对话路径（通过检查点启用）。
设置
在构建聊天机器人之前，我们先安装必要的软件包。
!pip install -U langgraph langchain openai
此命令将安装：
- LangGraph：用于构建图形结构。
- LangChain：用于与 OpenAI 的语言模型交互。
- OpenAI：用于使用 OpenAI 的模型（如GPT-4）。
我们需要安全地提供 OpenAI API 密钥，以便应用程序可以验证和使用 GPT 模型。如果环境中尚未设置密钥，则此函数会提示输入密钥。
import getpass
import os
def _set_env(var: str):
if not os.environ.get(var):
os.environ[var] = getpass.getpass(f"{var}: ")
_set_env("OPENAI_API_KEY")
第 1 部分：构建基本聊天机器人
我们将从创建最简单的聊天机器人开始。
1. 定义状态
状态定义了在图中节点之间传递的数据结构。在这里，我们定义了一个带有单键 messages 的状态，它将保存对话信息列表。
from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END from langgraph.graph.messae import add_messages class State(TypedDict): # 'messages' holds the list of chat messages. # 'add_messages' ensures new messages are added, not replaced. messages: Annotated[list, add_messages]
2. 创建图形生成器
StateGraph 对象是定义图形结构的入口点。它使用我们刚刚创建的状态定义进行初始化。
graph_builder = StateGraph(State)
3. 添加聊天机器人节点
我们定义了一个 Python 函数 chatbot，它获取当前状态，用状态中的消息调用 OpenAI 的 GPT 模型，并返回 LLM 的响应，作为状态中消息键的更新。
import openai
# Initialize OpenAI GPT model
openai.api_key = os.environ["OPENAI_API_KEY"]
def chatbot(state: State):
response = openai.Completion.create(
model="gpt-4", # You can also use "gpt-3.5-turbo" or any other OpenAI model
prompt=state["messages"],
max_tokens=150
)
return {"messages": [response.choices[0].text.strip()]}
graph_builder.add_node("chatbot", chatbot)
4. 设置进入点和退出点
定义图形执行的入口点（START）和出口点（END）。
graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)
5. 编译图形
定义好所有节点和边后，编译图形结构。
graph = graph_builder.compile()
6. 可视化（可选）
LangGraph 允许将编译后的图形结构可视化。这有助于理解执行流程。我们可以使用 pygraphviz 或 mermaid 等工具将图可视化。
from IPython.display import Image, display try: display(Image(graph.get_graph().draw_mermaid_png())) except Exception: pass # Optional visualization
7. 运行聊天机器人
设置一个与聊天机器人交互的循环。它接收用户输入，将其打包成预期的 State 格式 ({"messages": […]})，然后使用 graph.stream 执行图表。随着图形的进行，stream 方法会返回事件，我们会打印助手的最终消息。
def stream_graph_updates(user_input: str):
for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
for value in event.values():
print("Assistant:", value["messages"][-1].content)
# Loop to chat with the bot
while True:
try:
user_input = input("User: ")
if user_input.lower() in ["quit", "exit", "q"]:
print("Goodbye!")
break
stream_graph_updates(user_input)
except: # Fallback for environments without input()
user_input = "What do you know about LangGraph?"
print("User: " + user_input)
stream_graph_updates(user_input)
break
第 2 部分：利用工具增强聊天机器人的功能
为了让聊天机器人掌握更多知识，尤其是最近的信息，我们将整合网络搜索工具（Tavily）。这需要让 LLM 能够请求使用工具，并添加图组件来处理这些工具的执行。
1. 安装工具要求
安装 Tavily 搜索工具所需的库。
%pip install -U tavily-python langchain_community
2. 设置工具API密钥
配置 Tavily 服务的 API 密钥。
_set_env("TAVILY_API_KEY") # Uses the function defined earlier
3. 定义工具
实例化 TavilySearchResults 工具，它将返回 2 个结果。LLM 和图表都将使用该工具。
from langchain_community.tools.tavily_search import TavilySearchResults # Create a Tavily search tool instance, limiting to 2 results tool = TavilySearchResults(max_results=2) tools = [tool] # List of tools the bot can use
第 2 部分：为聊天机器人添加记忆功能
为了让机器人在多轮对话中记住之前的信息，我们引入了 LangGraph 的检查点功能。
添加校验指针
使用 MemorySaver 校验指针将对话状态存储在内存中。在生产中，您可能会使用 SQLite 或 Postgres 等持久性后端。
from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver()
第 4 部分：Human-in-the-loop
有时，人工智能代理在继续工作之前可能需要人工输入。我们可以通过创建一个暂停图流的工具来实现这一点。
定义人工辅助工具
from langchain_core.tools import tool
from langgraph.types import interrupt
@tool
def human_assistance(query: str) -> str:
print(f"Pausing for human assistance regarding: {query}")
# interrupt pauses graph execution and waits for input
human_response = interrupt({"query": query})
return human_response["data"]
该工具会暂停图形，等待人工输入后再继续。
部署LangGraph应用程序
构建好 LangGraph 应用程序后，接下来需要做的就是在本地机器或云平台上运行应用程序，以便进一步开发和测试。LangGraph为我们提供了多种部署选项，这些选项可以有不同的工作流程和基础架构。
在部署方面，LangGraph支持多种选项。云 SaaS 模式可为您处理一切事务。自托管数据平面（Self-Hosted Data Plane）可让您在自己的云中运行应用程序，同时使用 LangChain 的控制平面。使用自托管控制平面，你可以自己管理一切。或者使用独立容器，使用 Docker 获得充分的灵活性。
LangGraph的使用案例
LangGraph 可用于构建交互式智能人工智能代理。让我们来探索和了解它的一些用例。
1. 改善客户服务：LangGraph 能够为客户支持开发高级聊天机器人。这些聊天机器人能够回忆起过去的购买记录和客户偏好。通过回忆过去的记录，它们可以回复有关订单的询问，并在必要时与人类建立联系。这样就能更快地解决客户的问题。
2. 人工智能研究助手：使用 LangGraph 也可以创建一个研究助手。它可以查找学术文章，然后突出显示重要信息。然后，助理可以提取信息，研究人员和学生可以利用这些信息从各个领域获得更多见解。
3. 个性化学习：利用 LangGraph，我们还可以建立个性化或定制化学习系统，根据学习者的情况调整学习内容。这将帮助学习者了解自己的薄弱环节，然后根据薄弱环节推荐资源。这将创造个性化的学习体验，提高参与度和学习效果。
4. 简化业务任务：LangGraph 还能帮助我们实现业务流程自动化。有了它，文件审批和项目管理可以实现自动化，代理还可以用来分析数据。自动化有助于提高生产力，减少人为错误，让团队专注于更高层次的任务。
小结
在这篇面向初学者的 LangGraph 教程中，您学会了如何构建交互式人工智能系统。这些系统不仅仅是简单的问答机器人。通过 LangGraph 示例，我们了解了 LangGraph 如何管理状态、集成多个代理并允许人类输入。该指南展示了如何构建一个支持聊天机器人，它可以处理网络搜索、记住过去的交互，甚至可以让人类介入。
对于开发人员来说，LangGraph 入门教程非常有用。它有助于创建强大的人工智能驱动应用程序。通过使用 LangGraph，我们可以构建灵活、自适应的系统，处理复杂的任务。无论您是要创建聊天机器人、研究助手还是个性化学习工具，LangGraph 都能为您提供高效开发所需的结构和工具。
评论留言