🦜🕸️LangGraph

⚡ 以图形构建语言代理 ⚡

Note

寻找 JS 版本？点击 这里 （JS 文档）。

概述

LangGraph 是一个用于构建有状态的多参与者应用程序的库，利用 LLM 创建代理和多代理工作流。与其他 LLM 框架相比，它提供了以下核心优势：循环性、可控性和持久性。LangGraph 允许您定义涉及循环的流程，这对于大多数代理架构至关重要，使其与基于 DAG 的解决方案区别开来。作为一个非常底层的框架，它提供了对应用程序流和状态的细粒度控制，这对于创建可靠的代理至关重要。此外，LangGraph 包含内置的持久性，支持先进的人机协作和记忆特性。

LangGraph 的灵感来源于 Pregel 和 Apache Beam。公共接口受 NetworkX 的启发。LangGraph 是由 LangChain Inc 开发的，它是 LangChain 的创建者，但可以在不使用 LangChain 的情况下使用。

LangGraph 平台 是用于部署 LangGraph 代理的基础设施。它是一个商业解决方案，用于将代理应用程序部署到生产环境，构建于开源的 LangGraph 框架之上。LangGraph 平台由多个组件组成，这些组件协同工作以支持 LangGraph 应用程序的开发、部署、调试和监控：LangGraph 服务器（API）、LangGraph SDK（API 客户端）、LangGraph CLI（构建服务器的命令行工具）、LangGraph Studio（用户界面/调试器）。

要了解更多关于 LangGraph 的信息，请查看我们的首个 LangChain 学院课程，LangGraph 入门，可以在 这里 免费获得。

主要特性

• 循环和分支：在您的应用程序中实现循环和条件。
• 持久性：在图中的每一步自动保存状态。在任何时刻暂停和恢复图的执行，以支持错误恢复、人机协作工作流、时间旅行等。
• 人机协作：中断图的执行以批准或编辑代理计划的下一个动作。
• 流式支持：在每个节点产生输出时进行流式传输（包括令牌流式传输）。
• 与 LangChain 的集成：LangGraph 与 LangChain 和 LangSmith 无缝集成（但不需要它们）。

LangGraph 平台

LangGraph 平台是一个商业解决方案，用于将代理应用程序部署到生产环境，构建于开源的 LangGraph 框架之上。 以下是一些在复杂部署中常见的问题，LangGraph 平台解决了这些问题：

• 流式支持：LangGraph 服务器提供 多种流式模式，优化以满足各种应用需求
• 后台运行：在后台异步运行代理
• 支持长时间运行的代理：能够处理长时间运行过程的基础设施
• 双重文本：处理用户在代理回复之前收到两条消息的情况
• 处理突发性：任务队列以确保请求在高负载下也能始终如一地处理，不会丢失

安装

pip install -U langgraph

示例

LangGraph 的一个核心概念是状态。每次图的执行都会创建一个状态，该状态在节点之间传递，节点执行后用其返回值更新此内部状态。图如何更新其内部状态由选择的图的类型或自定义函数定义。

让我们来看一个可以使用搜索工具的代理的简单示例。

pip install langchain-anthropic

export ANTHROPIC_API_KEY=sk-...

可选地，我们可以设置 LangSmith 以获得最佳的可观察性。

export LANGSMITH_TRACING=true
export LANGSMITH_API_KEY=lsv2_sk_...

from typing import Annotated, Literal, TypedDict

from langchain_core.messages import HumanMessage
from langchain_anthropic import ChatAnthropic
from langchain_core.tools import tool
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode


# 定义代理要使用的工具
@tool
def search(query: str):
    """调用以浏览网络。"""
    # 这是一个占位符，但不要告诉 LLM...
    if "sf" in query.lower() or "san francisco" in query.lower():
        return "温度是 60 华氏度，天气阴霾。"
    return "温度是 90 华氏度，天气晴朗。"


tools = [search]

tool_node = ToolNode(tools)

model = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0).bind_tools(tools)

# 定义决定是否继续的函数
def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state['messages']
    last_message = messages[-1]
    # 如果 LLM 发起了工具调用，我们将路由到 "tools" 节点
    if last_message.tool_calls:
        return "tools"
    # 否则，我们停止（回复用户）
    return END


# 定义调用模型的函数
def call_model(state: MessagesState):
    messages = state['messages']
    response = model.invoke(messages)
    # 我们返回一个列表，因为这将被添加到现有列表中
    return {"messages": [response]}


# 定义一个新图
workflow = StateGraph(MessagesState)

# 定义我们将循环的两个节点
workflow.add_node("agent", call_model)
workflow.add_node("tools", tool_node)

# 将入口点设置为 `agent`
# 这意味着这个节点是第一个被调用的
workflow.add_edge(START, "agent")

# 现在我们添加一个条件边
workflow.add_conditional_edges(
    # 首先，我们定义起始节点。我们使用 `agent`。
    # 这意味着这是在调用 `agent` 节点后走的边。
    "agent",
    # 接下来，我们传入将决定下一个调用哪个节点的函数。
    should_continue,
)

# 现在我们添加一条从 `tools` 到 `agent` 的普通边。
# 这意味着在调用 `tools` 后，接下来会调用 `agent` 节点。
workflow.add_edge("tools", 'agent')

# 初始化内存以在图运行之间保持状态
checkpointer = MemorySaver()

# 最后，我们编译它！
# 这将其编译成一个 LangChain 可运行对象，
# 意味着您可以像使用任何其他可运行对象一样使用它。
# 请注意，我们（可选地）在编译图时传递了内存
app = workflow.compile(checkpointer=checkpointer)

# 使用 Runnable
final_state = app.invoke(
    {"messages": [HumanMessage(content="旧金山的天气如何")]},
    config={"configurable": {"thread_id": 42}}
)
final_state["messages"][-1].content

"根据搜索结果，我可以告诉你当前旧金山的天气是：\n\n温度：60 华氏度\n天气状况：阴霾\n\n旧金山以其微气候和频繁的雾气而闻名，特别是在夏季。60 华氏度（约 15.5 摄氏度）是该市的典型气温，四季温和。雾气，常被当地人称为“卡尔雾”，是旧金山天气的一个特征，尤其是在早晨和晚上。\n\n还有什么其他您想知道的关于旧金山或其他地方的天气？"

现在当我们传递相同的 "thread_id" 时，对话上下文通过保存的状态（即存储的消息列表）得以保留

final_state = app.invoke(
    {"messages": [HumanMessage(content="纽约的天气如何")]},
    config={"configurable": {"thread_id": 42}}
)
final_state["messages"][-1].content

"根据搜索结果，我可以告诉你当前纽约市的天气是：\n\n温度：90 华氏度（约 32.2 摄氏度）\n天气状况：晴朗\n\n这个天气与我们刚看到的旧金山天气截然不同。纽约目前气温要高得多。这里有几个要点值得注意：\n\n1. 90 华氏度的温度相当炎热，典型的夏季天气。\n2. 晴朗的天气意味着天空晴朗，这对于户外活动非常好，但也意味着由于阳光直射，可能会感觉更热。\n3. 纽约这种天气通常伴随高湿度，这可能使得实际感温比实际温度更高。\n\n看到旧金山温和阴霾的天气与纽约炎热晴朗的天气之间的鲜明对比是很有趣的。这种差异说明了美国不同地区天气的多样性，即使在同一天。\n\n还有什么其他您想知道的关于纽约或其他地方的天气？"

逐步分解

1. 
   

   初始化模型和工具。

2. 我们使用 ChatAnthropic 作为我们的 LLM。**注意：**我们需要确保模型知道可以调用这些工具。我们可以通过使用 .bind_tools() 方法将 LangChain 工具转换为 OpenAI 工具调用的格式来实现。

3. 我们定义了要使用的工具——在我们的例子中是一个搜索工具。创建自己的工具非常简单——有关如何做到这一点的文档请查看 这里。

4. 
   

   初始化带状态的图。

5. 我们通过传递状态模式（在我们的例子中是 MessagesState）来初始化图（StateGraph）。

6. MessagesState 是一个预构建的状态模式，具有一个属性——LangChain Message 对象的列表，以及将来自每个节点的更新合并到状态中的逻辑。

7. 
   

   定义图节点。

我们需要两个主要节点：

 1. `agent` 节点：负责决定采取什么（如果有的话）行动。
 2. `tools` 节点，调用工具：如果代理决定采取行动，则该节点将执行该行动。
   </details>

1. 
   

   定义入口点和图边。

   首先，我们需要设置图执行的入口点——agent 节点。

   然后我们定义一条普通边和一条条件边。条件边意味着目的地取决于图的状态（MessageState）的内容。在我们的例子中，目的地在代理（LLM）决定之前是不确定的。

   1. 条件边：在调用代理后，我们应该：
   2. a. 如果代理说要采取行动，则运行工具，或者
   3. b. 如果代理没有要求运行工具，则结束（回复用户）
   4. 普通边：在调用工具后，图应该始终返回代理以决定接下来要做什么。

编译图。

当我们编译图时，我们将其转化为 LangChain 可运行对象，这自动启用使用 .invoke()、.stream() 和 .batch() 调用您的输入。 2.我们还可以选择传递检查点对象以在图运行之间持久化状态，并启用记忆、人机协作工作流、时间旅行等。在我们的例子中，我们使用 MemorySaver——一个简单的内存检查点。

执行图。

1. LangGraph 将输入消息添加到内部状态中，然后将状态传递给入口节点，"agent"。
2. "agent" 节点执行，调用聊天模型。
3. 聊天模型返回一个 AIMessage。LangGraph 将其添加到状态中。

4. 图循环执行以下步骤，直到没有更多的 tool_calls 在 AIMessage 上：

   • 如果 AIMessage 具有 tool_calls，则执行 "tools" 节点。
   • "agent" 节点再次执行并返回 AIMessage。

5. 执行进展到特殊的 END 值并输出最终状态。 结果，我们得到所有聊天消息的列表作为输出。

文档

• 教程: 通过指导示例学习使用 LangGraph。
• 操作指南: 在 LangGraph 中完成特定任务，从流式处理到添加记忆和持久性，再到常见设计模式（分支、子图等），这些都是您想要复制和运行特定代码片段的地方。
• 概念指南: 深入解释 LangGraph 背后的关键概念和原则，如节点、边、状态等。
• API 参考: 查看重要类和方法、如何使用图和检查点 API 的简单示例、较高级的预构建组件等。
• LangGraph 平台: LangGraph 平台是一个用于将代理应用程序部署到生产的商业解决方案，构建于开源的 LangGraph 框架。

贡献

有关如何贡献的更多信息，请参见 这里。