新手必看！LangGraph如何轻松搞定多语言模型协同工作

在当今数字化时代，利用大型语言模型（LLMs）构建应用程序变得越来越受欢迎。这些强大的工具能够帮助我们创建出智能且交互性强的系统，比如聊天机器人、智能助手等。然而，随着应用复杂度的提升，尤其是当多个LLMs协同工作时，问题也随之而来。如何管理这些模型之间的信息流动？如何确保它们能够顺畅地理解和执行任务？今天，就让我们一起走进LangGraph的世界，看看它是如何解决这些问题的。

一、初识LangGraph

LangGraph是LangChain工具集中的一个重要组成部分，它是一个强大的库，专门用于简化LLMs的集成过程。通过LangGraph，开发者可以轻松地构建和管理包含多个LLM代理的应用程序，确保这些代理能够无缝协作，高效地完成各种任务。

LangGraph的核心思想是将应用程序的工作流程表示为图结构，其中包含节点和边。节点代表工作流中的离散工作单元或计算任务，每个节点都是一个Python函数，负责处理当前状态并返回更新后的状态。节点可以执行诸如调用LLM、与工具或API交互以及操作数据等任务。而边则连接节点，定义了执行流程。边可以是简单的无条件转换，也可以是基于节点输出的条件分支逻辑，类似于if-else语句，从而实现工作流中的动态决策。

LangGraph特别适合使用有向无环图（DAG）来处理直线任务，但由于它支持循环结构，能够回溯，因此可以构建更复杂、更灵活的系统。比如，一个智能代理可以根据新信息重新思考并更新响应，或者改变决策路径。

二、LangGraph的关键概念

（一）图结构

如前文所述，LangGraph的工作流程基于图结构，由节点和边组成。节点是工作流的基本构建块，负责处理当前状态并返回更新后的状态。边则定义了节点之间的执行顺序，可以是简单的直接转换，也可以是基于条件的分支逻辑。

（二）状态管理

在多代理系统中，跟踪任务的当前状态至关重要。LangGraph通过自动管理状态来解决这一问题。它维护一个主状态对象，并在代理执行任务时实时更新该对象。状态对象中存储了重要的信息，比如聊天历史记录、用户偏好、过去的动作、外部数据等。这些信息可以帮助代理在执行任务时做出更明智的决策。

（三）多代理系统

多代理系统由多个独立的代理组成，它们可以合作或竞争以实现共同的目标。这些代理利用LLMs进行决策，并控制应用程序的流程。随着代理数量和任务复杂度的增加，系统可能会面临诸如决策质量下降、上下文管理困难等问题。多代理系统通过将系统分解为专注于特定任务的小型代理来解决这些问题，例如规划或研究。

多代理系统的主要优势在于模块化、专业化和控制能力。模块化使得开发、测试和维护变得更加容易；专业化确保了每个代理都能在其特定领域发挥专长，从而提高整体性能；控制能力则确保了开发者可以清晰地定义代理之间的通信方式。

（四）持久化

持久化是指保存流程的进度，以便在中断后能够从中断点继续执行。LangGraph通过检查点（checkpoints）来实现持久化。在执行过程中，LangGraph会在每个主要步骤后保存图的状态，每个保存的状态称为一个检查点。这些检查点被组织在一个线程（特定运行的对话历史）中。

检查点类似于图状态的快照，其中包含以下内容：

• 配置信息：在该步骤中使用的配置信息。
• 元数据：步骤的详细信息，例如正在运行的节点。
• 值：该点的实际状态值。
• 下一步：将要运行的下一个节点。
• 任务：即将执行的任务或错误信息。

在执行过程中，每个图都需要一个线程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"}
)

（五）人机协作（Human-in-the-Loop）

人机协作允许在自动化LangGraph工作流的关键步骤中加入人工反馈。这对于某些任务至关重要，因为LLMs可能会生成不确定或有风险的输出，例如在工具调用、内容生成或决策制定中。LangGraph的interrupt()函数使得这一过程成为可能，它可以通过暂停图的执行，将数据呈现给人类，并使用Command(resume=value)方法根据人类的输入恢复执行。这种方式支持多种模式，如批准/拒绝、编辑状态、提供输入或多轮对话。

要使用人机协作，需要定义一个检查点，并在节点中添加interrupt()。以下是一个示例代码：

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"}})

（六）流式输出（Streaming）

LangGraph支持流式输出，这意味着在生成输出的同时，用户可以实时看到结果。这种方式不仅提高了用户体验，还使得应用程序更具响应性。LangGraph支持三种主要的数据类型流式输出：工作流进度、LLM令牌和自定义更新。

可以使用.stream()（同步）或.astream()（异步）方法来流式输出结果，并通过设置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()方法，这对于迁移大型应用程序非常有用。

三、为什么选择LangGraph？

LangGraph非常适合开发智能且灵活的AI代理，原因如下：

（一）可靠且可控

LangGraph允许开发者添加内容审核检查和人工审批，确保在长时间任务中保持上下文的连贯性。这对于需要高可靠性和可控性的应用场景至关重要。

（二）定制化和可扩展性

LangGraph提供了低级工具，开发者可以根据自己的需求构建代理，设计具有特定角色的代理系统。这种灵活性使得开发者能够根据具体需求定制化解决方案。

（三）出色的流式输出

LangGraph支持实时流式输出，开发者可以实时查看每个令牌和步骤，跟踪代理的思考过程。这对于需要实时反馈的应用场景非常有帮助。

四、构建最简单的LangGraph

现在我们已经了解了LangGraph的关键组件，接下来让我们通过一个简单的例子来构建一个包含三个节点和一个条件边的基本图。这个例子将展示如何使用状态、节点和边等关键概念来调用一个图。

（一）定义图状态

状态定义了在节点之间共享的数据结构，它就像一个在图中流动的共享内存。我们可以使用Python的TypedDict来声明状态的结构。在这个例子中，我们定义了一个名为graph_state的键，它存储了一个字符串。

from typing_extensions import TypedDict
class State(TypedDict):
    graph_state: str

（二）创建节点

节点是简单的Python函数，每个节点接收当前状态，对其进行修改，并返回更新后的状态。在这个例子中，我们定义了三个节点，每个节点都会在graph_state中添加不同的内容。

def node_1(state):
    print("---Node 1---")
    return {"graph_state": state['graph_state'] + " I am"}

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!"}

（三）添加条件逻辑

有时候我们希望工作流的行为是动态的，即下一步骤取决于某些逻辑或随机性。条件边可以实现这一点。在这个例子中，我们定义了一个函数decide_mood，它随机选择node_2或node_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"

（四）构建图

接下来，我们使用LangGraph的StateGraph类将所有组件组合在一起。在这个类中，我们定义了完整的图结构。

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# 初始化图并指定状态结构
builder = StateGraph(State)

# 添加节点到图中
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

# 添加边以定义流程
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)

# 编译并可视化图
graph = builder.compile()
display(Image(graph.get_graph().draw_mermaid_png()))

（五）调用图

最后，我们可以通过invoke()方法运行这个图。我们将graph_state初始化为“Hi, this is Janvi.”，然后观察图的执行过程。

graph.invoke({"graph_state" : "Hi, this is Janvi."})

这个简单的例子展示了状态如何在图的每个步骤中流动和更新。

五、使用LangGraph构建支持聊天机器人

在上一部分中，我们已经学会了如何构建一个简单的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. 定义状态

状态定义了在图的节点之间传递的数据结构。在这个例子中，我们定义了一个名为messages的键，它将存储对话消息的列表。

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages

class State(TypedDict):
    messages: Annotated[list, add_messages]

1. 创建图构建器

StateGraph对象是定义图结构的入口点。我们使用刚刚定义的状态初始化它。

graph_builder = StateGraph(State)

1. 添加聊天机器人节点

我们定义了一个名为chatbot的Python函数，它接收当前状态，调用OpenAI的GPT模型，并将模型的响应作为更新返回到状态的messages键中。

import openai

# 初始化OpenAI GPT模型
openai.api_key = os.environ["OPENAI_API_KEY"]

def chatbot(state: State):
    response = openai.Completion.create(
        model="gpt-4",  # 也可以使用"gpt-3.5-turbo"或其他OpenAI模型
        prompt=state["messages"],
        max_tokens=150
    )
    return {"messages": [response.choices[0].text.strip()]}

graph_builder.add_node("chatbot", chatbot)

1. 设置入口和出口点

定义图的入口点（START）和出口点（END）。

graph_builder.add_edge(START, "chatbot")
graph_builder.add_edge("chatbot", END)

1. 编译图

定义完所有节点和边后，编译图结构。

graph = graph_builder.compile()

1. 可视化（可选）

LangGraph支持可视化编译后的图结构，这有助于理解执行流程。我们可以使用工具如pygraphviz或mermaid来可视化图。

from IPython.display import Image, display

try:
    display(Image(graph.get_graph().draw_mermaid_png()))
except Exception:
    pass  # 可选可视化

1. 运行聊天机器人

设置一个循环与聊天机器人进行交互。它接收用户输入，将其打包成预期的状态格式（{"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)

# 与机器人聊天的循环
whileTrue:
    try:
        user_input = input("User: ")
        if user_input.lower() in ["quit", "exit", "q"]:
            print("Goodbye!")
            break
        stream_graph_updates(user_input)
    except:  # 适用于没有输入功能的环境
        user_input = "What do you know about LangGraph?"
        print("User: " + user_input)
        stream_graph_updates(user_input)
        break

（三）增强聊天机器人功能

1. 添加工具集成

为了让聊天机器人能够获取最新的信息，我们可以为其添加网页搜索工具（例如Tavily）。这需要让LLM能够请求工具的使用，并在图中添加处理这些工具执行的组件。

（1）安装工具依赖

%pip install -U tavily-python langchain_community

（2）设置工具API密钥

_set_env("TAVILY_API_KEY")  # 使用前面定义的函数

（3）定义工具

实例化Tavily搜索工具，限制返回2个结果。这个工具将被LLM和图使用。

from langchain_community.tools.tavily_search import TavilySearchResults

# 创建Tavily搜索工具实例，限制返回2个结果
tool = TavilySearchResults(max_results=2)
tools = [tool]  # 机器人可以使用的工具列表

2. 添加记忆功能

为了让聊天机器人能够进行多轮对话并记住之前的对话内容，我们需要引入LangGraph的检查点功能。

（1）添加检查点

使用MemorySaver检查点将对话状态存储在内存中。在生产环境中，可以使用持久化后端，如SQLite或Postgres。

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()

3. 添加人机协作

有时，AI代理可能需要在继续之前获得人工输入。我们可以通过创建一个暂停图流程的工具来实现这一点。

（1）定义人工协助工具

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暂停图的执行并等待输入
    human_response = interrupt({"query": query})
    return human_response["data"]

这个工具会暂停图的执行，并等待人工输入后再继续。

（四）部署LangGraph应用

构建完LangGraph应用后，下一步就是将其部署到本地机器或云平台，以便进行进一步的开发和测试。LangGraph提供了多种部署选项，每种选项都有不同的工作流程和基础设施。

• 云SaaS模型：一切由LangChain托管，用户无需管理任何基础设施。
• 自托管数据平面：用户可以在自己的云环境中运行应用，同时使用LangChain的控制平面。
• 自托管控制平面：用户自行管理所有基础设施。
• 独立容器：使用Docker实现完全的灵活性。

六、LangGraph的用例

LangGraph可用于构建交互式和智能的AI代理，以下是一些常见的用例：

（一）改进客户服务

LangGraph能够开发高级的客户支持聊天机器人。这些聊天机器人能够回忆过去的购买记录和客户偏好，从而更快地解决客户问题。当需要人工干预时，它们还可以将对话转接给人工客服。

（二）AI研究助手

利用LangGraph，可以创建一个研究助手，它能够搜索学术文章并突出显示重要信息。研究人员和学生可以利用这些信息，从各个领域获取更多见解。

（三）个性化学习

LangGraph还可以用于构建个性化的学习系统，根据学习者的特点调整内容。它能够帮助学习者识别薄弱环节，并推荐相应的学习资源，从而提高学习效果。

（四）简化业务流程

LangGraph可以帮助自动化业务流程，例如文档审批和项目管理。它还可以用于数据分析，提高生产力，减少人为错误，让团队专注于更高层次的任务。

七、总结

在本篇LangGraph新手教程中，我们学习了如何构建交互式的AI系统。这些系统不仅仅是简单的问答机器人，而是可以通过LangGraph管理状态、集成多个代理，并允许人工输入。我们通过构建一个支持聊天机器人，展示了LangGraph如何处理网页搜索、记住过去的互动，甚至涉及人工干预。

LangGraph对于开发者来说是一个非常有价值的工具。它可以帮助我们创建强大的、由AI驱动的应用程序。通过LangGraph，我们可以构建灵活、适应性强的系统，能够处理复杂的任务。无论你是想构建聊天机器人、研究助手还是个性化学习工具，LangGraph都提供了所需的结构和工具，助力高效开发。