人工智能（AI）已经取得了显著进步，语言模型（LLM）现在正在执行复杂任务并做出动态决策。然而，支持这些高级功能的基础设施往往难以跟上步伐。检索增强生成（RAG）系统虽然对于简单查询很有用，但在管理复杂和多步骤流程时却力不从心。

LangGraph是LangChain生态系统中的一个改变游戏规则的库，旨在简化复杂AI应用程序的创建过程。LangGraph提供了一个框架，用于以结构化、循环的方式高效地定义、协调和执行多个LLM代理（或链）。

在本文中，我们将探讨LangGraph如何帮助构建健壮、可扩展且灵活的多代理系统，使智能AI工作流的开发更加顺畅和高效。

什么是LangGraph

在LangGraph之前，LangChain中的代理执行器类是构建AI代理的主要工具。它通过在一个循环中使用代理来做出决策、执行动作和记录观察结果来运行。然而，这个类的灵活性有限，将开发人员限制在固定的工具调用和错误处理模式中，使得构建动态和可适应的代理运行时变得困难。

LangGraph是LangChain生态系统中的一个高级库，通过引入循环计算能力解决了这些限制。虽然LangChain支持用于线性工作流的有向无环图（DAG），但LangGraph允许创建循环，使LLM代理能够动态地循环通过过程并根据不断变化的条件做出决策。这个框架使开发人员能够构建更复杂、更灵活和更自适应的代理系统。

概述

LangGraph是建立在LangChain之上的库，旨在为基于LLM的AI代理创建循环图。

它为工作流启用循环图拓扑，允许比线性模型更灵活和细微的代理行为。

利用关键元素：

• 节点：表示函数或LangChain可运行项。
• 边：定义执行和数据流。
• 有状态图：在管理执行周期中管理持久数据。

支持多代理协调，允许每个代理在单个图中拥有自己的提示、LLM、工具和自定义代码。

引入了聊天代理执行器，将代理状态表示为消息列表，非常适合基于聊天的模型。

关键组件

• 状态：随着进程的推进维护和更新上下文或内存，使每个步骤都能访问先前步骤中的相关信息以进行动态决策。
• 节点：表示单个计算步骤或任务，执行特定功能，如数据处理、决策制定或系统交互。
• 边：连接节点，规定计算流，并允许条件逻辑根据当前状态指导工作流，支持复杂和多步骤操作。
• 有状态图是其架构的核心，其中每个节点表示计算中的一个步骤。图随着进程的推进维护和更新共享状态。

LangGraph的工作原理

LangGraph支持带有状态持久性的循环LLM调用执行，这是代理行为所必需的。LangGraph受Pregel和Apache Beam的启发，并基于用户友好的基于图的编程库NetworkX进行建模，为代理运行时提供了更先进的方法。与其依赖基本循环的前辈不同，LangGraph支持复杂的节点和边系统，允许开发人员创建更复杂的决策过程和动作序列。

图示展示了LangGraph如何为AI代理提供一个动态、循环的工作流程。流程从一个初始状态开始，该状态包含输入数据或上下文（在此例中，是客户询问太阳能板优势的消息）。这个状态被传递给一个代理节点，该节点与客户进行交互以收集信息，例如获取他们的电费账单。交互完成后，状态会根据收集到的信息进行更新，并被传递给一个边，这个边代表了一个决策点。

在决策点，系统会评估当前状态，并根据更新后的信息决定下一步的行动方向。这可能导向与工具的交互，或者直接通向结束状态。

如果决定与工具交互，则会触发另一个节点，该节点上的工具（在此例中为能源成本计算器）会处理数据。这一步涉及执行特定操作的函数，再次更新状态，随着工作流程的继续。最后，系统要么达到结束状态，完成流程，要么因为需要进一步的行动而循环回之前的步骤，使用更新后的状态。这种反馈循环允许根据不断变化的情况进行持续的动态决策，展示了LangGraph的循环图如何使代理行为更加灵活和迭代。

使用LangGraph进行实验

为了更深入地理解LangGraph，我们将通过构建一个代理来进行实验。我们将从实现工具调用开始，然后使用预构建的代理，最后在LangGraph中创建我们自己的自定义代理。

前提条件

本教程使用了OpenAI、Weather API、Together和Tavily。如果你有OpenAI的API密钥，可以在整个教程中使用它，或者你也可以选择其他强大的专有模型，如GPT-4和Gemini Pro。此外，还可以选择开放访问的模型，如Mixtral和Llama-3.2。对于大型语言模型（LLM）的推理，有多个平台可用，如Abacus、Anyscale和Together。

在本教程中，我们将使用Together AI来推理Llama 3.1，因此请确保在开始之前从Together获取API密钥。此外，你还需要从WeatherAPI获取API密钥以访问天气数据，以及从Tavily获取API密钥以优化AI代理的搜索。

• WeatherAPI仪表盘：WeatherAPI.com
• Together AI接口：api.together.ai
• Tavily AI

安装依赖

通过执行以下命令来创建并激活一个虚拟环境。

python -m venv venv
source venv/bin/activate #for ubuntu
venv/Scripts/activate #for windows

使用pip安装langgraph、langchain、langchain-community、langchain-openai和python-dotenv库。

pip install langgraph langchain langchain-community langchain-openai python-dotenv

设置环境变量

• 首先，为你的项目创建一个新文件夹，并选择一个能够反映项目目的的名称。
• 在你的新项目文件夹内，创建一个名为.env的文件。这个文件将用于存储你的环境变量，包括Weather、Together和Tavily的API密钥。
• 打开.env文件，并添加以下代码来指定你的Gemini API密钥：

WEATHER_API_KEY=b37fb08d051440ef......
TAVILY_API_KEY=tvly-iSIylB0XSj.......
TOGETHER_API_KEY=05e1a66f6........

1. 在LangGraph中调用工具

导入密钥

让我们将API密钥导入到代码中。

# Import the keys
import os
from dotenv import load_dotenv
load_dotenv('.env')
WEATHER_API_KEY = os.environ['WEATHER_API_KEY']
TAVILY_API_KEY = os.environ['TAVILY_API_KEY']
TOGETHER_API_KEY = os.environ['TOGETHER_API_KEY']

导入必要的库

让我们导入项目所需的必要库和模块。

# Import the required libraries and methods
import requests
from typing import List, Literal
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

定义工具

我们将创建两个工具：

• 用于获取与天气相关查询的天气数据。
• 当大型语言模型（LLM）无法提供答案时，用于互联网搜索。

@tool
def get_weather(query: str) -> list:
    """Search weatherapi to get the current weather"""
    endpoint = f"http://api.weatherapi.com/v1/current.json?key={WEATHER_API_KEY}&q={query}"
    response = requests.get(endpoint)
    data = response.json()
    if data.get("location"):
        return data
    else:
        return "Weather Data Not Found"
@tool
def search_web(query: str) -> list:
    """Search the web for a query"""
    tavily_search = TavilySearchResults(api_key=TAVILY_API_KEY, max_results=2, search_depth='advanced', max_tokens=1000)
    results = tavily_search.invoke(query)
    return results

为了使这些工具能够被大型语言模型（LLM）访问，我们可以将它们绑定到模型上，如下所示。在这个例子中，我们使用Together API来利用Llama 3模型。如果你有OpenAI的密钥，你也可以通过指定你的密钥来继续使用它。

llm = ChatOpenAI(base_url="https://api.together.xyz/v1","https://api.together.xyz/v1",
    api_key=TOGETHER_API_KEY,
    model="meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo"
# If you have OpenAI key
# llm = ChatOpenAI(model="gpt-4o-mini", api_key="sk-U7tijaa4jwHvhVWGr....", temperature=0)
tools = [search_web, get_weather]
llm_with_tools = llm.bind_tools(tools)

现在，让我们用一个提示来调用大型语言模型（LLM），以查看结果：

prompt = """"""
    Given only the tools at your disposal, mention tool calls for the following tasks:
    Do not change the query given for any search tasks
        1. What is the current weather in Trivandrum today
        2. Can you tell me about Kerala
        3. Why is the sky blue?
    """
results = llm_with_tools.invoke(prompt)
print(results.tool_calls)
query = "What is the current weather in Trivandrum today"
response = llm.invoke(query)
print(response.content)

Llama 3 / GPT模型没有实时访问当前天气信息的能力，因此它无法提供最新的天气详情。

使用预构建的代理

LangGraph提供了一个预构建的React（推理与行动）代理，旨在简化决策和行动执行的过程。

让我们来探索一下它是如何工作的。

from langgraph.prebuilt import create_react_agent
# system prompt is used to inform the tools available to when to use each
system_prompt = """Act as a helpful assistant.
    Use the tools at your disposal to perform tasks as needed.
        - get_weather: whenever user asks get the weather of a place.
        - search_web: whenever user asks for information on current events or if you don't know the answer.
    Use the tools only if you don't know the answer.
    """
# we can initialize the agent using the llama3 model, tools, and system prompt.
agent = create_react_agent(model=llm, tools=tools, state_modifier=system_prompt)
# Let’s query the agent to see the result.
def print_stream(stream):
    for s in stream:
        message = s["messages"][-1]
        if isinstance(message, tuple):
            print(message)
        else:
            message.pretty_print()
inputs = {"messages": [("user", "What is the current weather in Trivandrum today")]}
print_stream(agent.stream(inputs, stream_mode="values"))

现在，让我们运行脚本来查看结果：

输出展示了大型语言模型（LLM）如何利用外部工具来回答用户查询。LLM调用了search_web工具和get_weather API来获取相关信息。search_web工具成功获取了一个URL并提取了内容，然后LLM利用这些内容生成了对查询的响应。同样地，LLM利用get_weather API获取了当前的天气信息，并将其融入到输出中，以提供准确的答案。

构建自定义代理

现在，让我们使用LangGraph来创建一个代理。

导入所需的库。

from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END

定义一个包含可用工具的tool_node。ToolNode是LangChain中的一个可运行对象，它接收包括消息列表在内的图形状态作为输入，并产生更新后的状态作为工具调用的结果。

tools = [search_web, get_weather]
tool_node = ToolNode(tools)

定义调用模型或工具的函数。

def call_model(state: MessagesState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}
def call_tools(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END

call_model函数接收来自状态的消息作为输入，这些消息可以包括查询、提示或工具内容，并返回一个响应。call_tools函数也接受状态消息作为输入，并且如果最后一条消息包含工具调用，则返回工具节点，否则它将终止。

现在，让我们创建节点和边。

# initialize the workflow from StateGraph
workflow = StateGraph(MessagesState)
# add a node named LLM, with call_model function. This node uses an LLM to make decisions based on the input given
workflow.add_node("LLM", call_model)
# Our workflow starts with the LLM node
workflow.add_edge(START, "LLM")
# Add a tools node
workflow.add_node("tools", tool_node)
# Add a conditional edge from LLM to call_tools function. It can go tools node or end depending on the output of the LLM. 
workflow.add_conditional_edges("LLM", call_tools)
# tools node sends the information back to the LLM
workflow.add_edge("tools", "LLM")
agent = workflow.compile()

上述代码使用带有MessagesState的StateGraph类初始化了一个工作流。它首先添加了一个名为LLM的节点，该节点利用call_model函数根据提供的输入做出决策。工作流从LLM节点开始，并建立到工具节点的连接，该工具节点是使用tool_node创建的。然后添加了条件边，允许工作流根据LLM节点的输出选择继续到工具节点还是终止，如果选择继续到工具节点，处理完成后会再返回到LLM进行进一步处理。

现在，让我们来查询这个代理。

for chunk in agent.stream(
    {"messages": [("user", "Will it rain in Trivandrum today?")]},
    stream_mode="values",):
    chunk["messages"][-1].pretty_print()

以下是提供的完整代码。

# import the required methods
from langgraph.prebuilt import ToolNode
from langgraph.graph import StateGraph, MessagesState, START, END
# define a tool_node with the available tools
tools = [search_web, get_weather]
tool_node = ToolNode(tools)
# define functions to call the LLM or the tools
def call_model(state: MessagesState):
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}
def call_tools(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    if last_message.tool_calls:
        return "tools"
    return END
# initialize the workflow from StateGraph
workflow = StateGraph(MessagesState)
# add a node named LLM, with call_model function. This node uses an LLM to make decisions based on the input given
workflow.add_node("LLM", call_model)
# Our workflow starts with the LLM node
workflow.add_edge(START, "LLM")
# Add a tools node
workflow.add_node("tools", tool_node)
# Add a conditional edge from LLM to call_tools function. It can go tools node or end depending on the output of the LLM. 
workflow.add_conditional_edges("LLM", call_tools)
# tools node sends the information back to the LLM
workflow.add_edge("tools", "LLM")
agent = workflow.compile()
# display(Image(agent.get_graph().draw_mermaid_png()))
for chunk in agent.stream(
    {"messages": [("user", "Will it rain in Trivandrum today?")]},
    stream_mode="values",):
    chunk["messages"][-1].pretty_print()

现在，让我们运行代码来查看结果。

工作流被编译成一个代理，该代理针对用户关于特里凡得琅天气的查询，以可读格式实时显示每条接收到的消息作为响应。

在这个例子中，调用了get_weather工具来获取各种与天气相关的数值，使得大型语言模型（LLM）能够得出结论，很可能会下雨。这展示了如何将不同的工具与LLM集成，从而增强其处理自己可能无法回答的查询的能力。