🦜🕸️LangGraph

⚡ 将语言代理构建为图形 ⚡

[!注意] 寻找JS版本？点击这里（JS文档）。

概述

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

LangGraph的灵感来自Pregel和Apache Beam。其公共接口借鉴了NetworkX。LangGraph由LangChain Inc.（LangChain的创建者）开发，但可以独立于LangChain使用。

主要特性

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

安装

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, 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.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 Runnable，
# 意味着你可以像使用任何其他runnable一样使用它。
# 注意，我们在编译图时（可选）传递了内存
app = workflow.compile(checkpointer=checkpointer)

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

"根据搜索结果，我可以告诉你旧金山当前的天气情况是：

温度：60华氏度（约15.5摄氏度）
天气状况：有雾

旧金山以其微气候和频繁的雾气而闻名，尤其是在夏季月份。60°F的温度对这个城市来说很典型，全年温度都比较温和。雾气，当地人常称之为"卡尔雾"，是旧金山天气的一个特征，尤其是在早晨和傍晚。

你还想了解关于旧金山或其他地方的天气情况吗？"

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

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

"根据搜索结果，我可以告诉你纽约市当前的天气情况是：

温度：90华氏度（约32.2摄氏度）
天气状况：阳光明媚

这个天气与我们刚刚看到的旧金山的天气有很大不同。纽约现在经历着更温暖的温度。以下是几个要点：

1. 90°F的温度相当热，这是纽约市夏季天气的典型特征。
2. 阳光明媚的条件表明天空晴朗，这很适合户外活动，但也意味着由于阳光直射，可能会感觉更热。
3. 纽约这种天气通常伴随着高湿度，这可能会使体感温度比实际温度更高。

有趣的是看到旧金山温和多雾的天气与纽约炎热晴朗的条件形成鲜明对比。这种差异说明了美国不同地区的天气在同一天可以有多么大的变化。

你还想了解关于纽约或其他地方的天气情况吗？"

步骤分解

<details> <summary>初始化模型和工具。</summary>
- 我们使用ChatAnthropic作为我们的LLM。**注意：**我们需要确保模型知道它可以调用这些工具。我们可以通过使用.bind_tools()方法将LangChain工具转换为OpenAI工具调用格式来实现这一点。
- 我们定义我们想要使用的工具 - 在我们的例子中是一个搜索工具。创建自己的工具非常简单 - 请参阅这里的文档了解如何操作。
</details>
<details> <summary>用状态初始化图。</summary>
- 我们通过传递状态模式（在我们的例子中是MessagesState）来初始化图（StateGraph）
- MessagesState是一个预构建的状态模式，它有一个属性 -- 一个LangChain Message对象列表，以及将每个节点的更新合并到状态中的逻辑
</details>
<details> <summary>定义图节点。</summary>
我们需要两个主要节点：
- agent节点：负责决定采取什么（如果有）行动。
- 调用工具的tools节点：如果代理决定采取行动，这个节点将执行该行动。
</details>
<details> <summary>定义入口点和图边。</summary>
首先，我们需要为图执行设置入口点 - agent节点。

然后我们定义一个普通边和一个条件边。条件边意味着目标取决于图的状态（MessageState）的内容。在我们的例子中，直到代理（LLM）做出决定，目标才能确定。
- 条件边：在调用代理后，我们应该：
  - a. 如果代理说要采取行动，则运行工具，或者
  - b. 如果代理没有要求运行工具，则完成（回应用户）
- 普通边：在调用工具后，图应该总是返回到代理以决定下一步要做什么
</details>
<details> <summary>编译图。</summary>
- 当我们编译图时，我们将其转换为LangChain Runnable，这自动启用了用你的输入调用.invoke()、.stream()和.batch()
- 我们还可以选择传递检查点对象，用于在图运行之间保持状态，并启用记忆、人机交互工作流、时间旅行等功能。在我们的例子中，我们使用MemorySaver - 一个简单的内存检查点器
</details>
<details> <summary>执行图。</summary>
1. LangGraph将输入消息添加到内部状态，然后将状态传递给入口点节点"agent"。
2. "agent"节点执行，调用聊天模型。
3. 聊天模型返回一个AIMessage。LangGraph将其添加到状态中。
4. 图循环以下步骤，直到AIMessage上没有更多的tool_calls：
  - 如果AIMessage有tool_calls，"tools"节点执行
  - "agent"节点再次执行并返回AIMessage
5. 执行进展到特殊的END值并输出最终状态。结果，我们得到一个包含所有聊天消息的列表作为输出。
</details>

文档

教程：通过指导示例学习使用LangGraph构建。
操作指南：在LangGraph中完成特定任务，从流式处理，到添加内存和持久性，再到常见设计模式（分支、子图等），如果你想复制和运行特定的代码片段，这里是最佳去处。
概念指南：深入解释LangGraph背后的关键概念和原理，如节点、边、状态等。
API参考：查看重要的类和方法，使用图和检查点API的简单示例，更高级别的预构建组件等。
云服务（测试版）：一键将LangGraph应用部署到LangGraph云。