🦜🕸️LangGraph

⚡ 构建语言智能体的图形 ⚡

[!NOTE] 在找 JS 版本？点击这里（JS 文档）。

概述

LangGraph 是一个用于构建有状态的多角色应用程序的库，使用大语言模型（LLMs）来创建智能体和多智能体工作流程。与其他 LLM 框架相比，它提供了以下核心优势：循环、可控性和持久性。LangGraph 允许您定义包含循环的流程，这对于大多数智能体架构至关重要，从而区别于基于有向无环图（DAG）的解决方案。作为一个非常底层的框架，它提供了对应用程序流程和状态的精细控制，这对于创建可靠的智能体至关重要。此外，LangGraph 包含内置的持久性功能，支持高级的人机协作和记忆功能。

LangGraph 的灵感来源于 Pregel 和 Apache Beam。公共接口借鉴了 NetworkX 的设计。LangGraph 由 LangChain Inc 开发，虽然与 LangChain 紧密集成，但可以独立使用。

主要特性

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

安装

pip install -U langgraph

示例

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

让我们来看一个使用搜索工具的简单智能体示例。

pip install lang

chain-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 import MemorySaver
from langgraph.graph import END, StateGraph, MessagesState
from langgraph.prebuilt import ToolNode


# 定义智能体使用的工具
@tool
def search(query: str):
    """调用上网搜索。"""
    # 这是一个占位符，但不要告诉大语言模型……
    if "sf" in查询.lower()或"san francisco"在查询.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]
    # 如果大语言模型发出工具调用，则路由到 "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.set_entry_point("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)

# 使用可运行对象
final_state = app.invoke(
    {"messages": [HumanMessage(content="sf 的天气如何")]},
    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="ny 的天气如何")]},
    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. 初始化模型和工具。

   • 我们使用 ChatAnthropic 作为我们的 LLM。注意： 我们需要确保模型知道它有这些可调用的工具。我们可以通过使用 .bind_tools() 方法将 LangChain 工具转换为 OpenAI 工具调用格式来实现这一点。
   • 我们定义了要使用的工具 - 在我们的例子中是一个搜索工具。创建自己的工具非常简单 - 有关如何执行此操作的文档请参见这里。
2. 初始化带状态的图形。

   • 我们通过传递状态架构（在我们的例子中为 MessagesState）来初始化图形（StateGraph）
   • MessagesState 是一个预构建的状态架构，具有一个属性——LangChain Message 对象列表，以及用于合并每个节点的更新到状态中的逻辑
3. 定义图形节点。

   我们需要的两个主要节点是：

   • agent 节点：负责决定是否采取任何行动。
   • tools 节点：调用工具：如果智能体决定采取行动，则该节点将执行该行动。
4. 定义入口点和图形边。

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

   然后我们定义一个普通边和一个条件边。条件边意味着目标取决于图形状态的内容（MessageState）。在我们的例子中，目标直到智能体（LLM）决定后才确定。

   • 条件边：在调用智能体后，我们应该：
     • a. 如果智能体说要采取行动，则运行工具，或者
     • b. 如果智能体没有要求运行工具，则结束（回复用户）
   • 普通边：在调用工具后，图形应始终返回到智能体，以决定下一步该做什么
5. 编译图形。

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

   1. LangGraph 将输入消息添加到内部状态，然后将状态传递到入口点节点，"agent"。

   2. "agent" 节点执行，调用聊天模型。

   3. 聊天模型返回一个 AIMessage。LangGraph 将其添加到状态中。

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

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

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

文档

• 教程：通过指导示例学习使用 LangGraph 进行

构建。

• 操作指南：在 LangGraph 中完成特定任务，从流式传输、添加内存和持久性，到常见的设计模式（分支、子图等），这些指南是您想要复制和运行特定代码片段时的首选。
• 概念指南：深入解释 LangGraph 的关键概念和原理，如节点、边、状态等。
• API 参考：查看重要的类和方法，如何使用图形和检查点 API 的简单示例，更高级别的预构建组件等。
• 云（测试版）：一键将 LangGraph 应用程序部署到 LangGraph Cloud。